无法在使用 pytest_postgresql 进行测试时在测试数据库中创建表格。

huangapple go评论60阅读模式
英文:

Can't create tables in test database while testing with pytest_postgresql

问题

我尝试编写一个使用 fixtures 和 pytest_postgresql 在 Postgres 中进行模型和数据库的 pytest 测试。

运行测试时出现了以下错误:

FAILED tests/test_model_with_test_db.py::test_authors - sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedTable) relation "authors" does not exist

为什么它没有使用 model.Base.metadata.create_all(con) 创建所有表格?

我的测试代码如下:

import pytest
from pytest_postgresql import factories
from pytest_postgresql.janitor import DatabaseJanitor
from sqlalchemy import create_engine, select
from sqlalchemy.orm.session import sessionmaker

import model

test_db = factories.postgresql_proc(port=None, dbname="test_db")


@pytest.fixture(scope="session")
def db_session(test_db):
    pg_host = test_db.host
    pg_port = test_db.port
    pg_user = test_db.user
    pg_password = test_db.password
    pg_db = test_db.dbname

    with DatabaseJanitor(pg_user, pg_host, pg_port, pg_db, test_db.version, pg_password):
        connection_str = f"postgresql+psycopg2://{pg_user}@{pg_host}:{pg_port}/{pg_db}"
        engine = create_engine(connection_str)
        with engine.connect() as con:
            model.Base.metadata.create_all(con)
            yield sessionmaker(bind=engine, expire_on_commit=False)


@pytest.fixture(scope="module")
def create_test_data():
    authors = [
        ["John", "Smith", "john@gmail.com"],
        ["Bill", "Miles", "bill@gmail.com"],
        ["Frank", "James", "frank@gmail.com"]
    ]
    return [model.Author(firstname=firstname, lastname=lastname, email=email)
            for firstname, lastname, email in authors]


def test_persons(db_session, create_test_data):
    s = db_session()
    for obj in create_test_data:
        s.add(obj)
    s.commit()

    query_result = s.execute(select(model.Author)).all()
    s.close()

    assert len(query_result) == len(create_test_data)

model.py:

from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, ForeignKey
from sqlalchemy.engine import URL
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime

Base = declarative_base()

class Author(Base):
    __tablename__ = 'authors'

    id = Column(Integer(), primary_key=True)
    firstname = Column(String(100))
    lastname = Column(String(100))
    email = Column(String(255), nullable=False)
    joined = Column(DateTime(), default=datetime.now)

    articles = relationship('Article', backref='author')


class Article(Base):
    __tablename__ = 'articles'

    id = Column(Integer(), primary_key=True)
    slug = Column(String(100), nullable=False)
    title = Column(String(100), nullable=False)
    created_on = Column(DateTime(), default=datetime.now)
    updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)
    content = Column(Text)
    author_id = Column(Integer(), ForeignKey('authors.id'))


url = URL.create(
    drivername="postgresql",
    username="postgres",
    host="localhost",
    port=5433,
    database="andy"
)

engine = create_engine(url)
Session = sessionmaker(bind=engine)

(请注意,我已经修复了代码中的 HTML 转义字符,以便正确显示代码部分。)

英文:

I'm trying to write a pytest for models and database in Postgres using fixtures and pytest_postgresql.

Running test gives:

FAILED tests/test_model_with_test_db.py::test_authors - sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedTable) relation "authors" does not exist

Why it doesn't created all tables with model.Base.metadata.create_all(con)?

My test code is following:

import pytest
from pytest_postgresql import factories
from pytest_postgresql.janitor import DatabaseJanitor
from sqlalchemy import create_engine, select
from sqlalchemy.orm.session import sessionmaker

import model

test_db = factories.postgresql_proc(port=None, dbname="test_db")


@pytest.fixture(scope="session")
def db_session(test_db):
    pg_host = test_db.host
    pg_port = test_db.port
    pg_user = test_db.user
    pg_password = test_db.password
    pg_db = test_db.dbname

    with DatabaseJanitor(pg_user, pg_host, pg_port, pg_db, test_db.version, pg_password):
        connection_str = f"postgresql+psycopg2://{pg_user}:@{pg_host}:{pg_port}/{pg_db}"
        engine = create_engine(connection_str)
        with engine.connect() as con:
            model.Base.metadata.create_all(con)
            yield sessionmaker(bind=engine, expire_on_commit=False)


@pytest.fixture(scope="module")
def create_test_data():
    authors = [
        ["John", "Smith", "john@gmail.com"],
        ["Bill", "Miles", "bill@gmail.com"],
        ["Frank", "James", "frank@gmail.com"]
    ]
    return [model.Author(firstname=firstname, lastname=lastname, email=email)
            for firstname, lastname, email in authors]


def test_persons(db_session, create_test_data):
    s = db_session()
    for obj in create_test_data:
        s.add(obj)
    s.commit()

    query_result = s.execute(select(model.Author)).all()
    s.close()

    assert len(query_result) == len(create_test_data)

model.py:

from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, ForeignKey
from sqlalchemy.engine import URL
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime

Base = declarative_base()


class Author(Base):
    __tablename__ = 'authors'

    id = Column(Integer(), primary_key=True)
    firstname = Column(String(100))
    lastname = Column(String(100))
    email = Column(String(255), nullable=False)
    joined = Column(DateTime(), default=datetime.now)

    articles = relationship('Article', backref='author')


class Article(Base):
    __tablename__ = 'articles'

    id = Column(Integer(), primary_key=True)
    slug = Column(String(100), nullable=False)
    title = Column(String(100), nullable=False)
    created_on = Column(DateTime(), default=datetime.now)
    updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)
    content = Column(Text)
    author_id = Column(Integer(), ForeignKey('authors.id'))


url = URL.create(
    drivername="postgresql",
    username="postgres",
    host="localhost",
    port=5433,
    database="andy"
)

engine = create_engine(url)
Session = sessionmaker(bind=engine)

答案1

得分: 2

Figured out myself.
Modified db_session fixture like following:

@pytest.fixture(scope="session")
def db_session(test_db):
    pg_host = test_db.host
    pg_port = test_db.port
    pg_user = test_db.user
    pg_password = test_db.password
    pg_db = test_db.dbname

    with DatabaseJanitor(pg_user, pg_host, pg_port, pg_db, test_db.version, pg_password):
        connection_str = f"postgresql+psycopg2://{pg_user}:{pg_password}@{pg_host}:{pg_port}/{pg_db}"
        engine = create_engine(connection_str)
        model.Base.metadata.create_all(engine)
        yield sessionmaker(bind=engine, expire_on_commit=False)

And test passes.

But why it doesn't work with connection instead?

英文:

Figured out myself.
Modified db_session fixture like following:

@pytest.fixture(scope="session")
def db_session(test_db):
    pg_host = test_db.host
    pg_port = test_db.port
    pg_user = test_db.user
    pg_password = test_db.password
    pg_db = test_db.dbname

    with DatabaseJanitor(pg_user, pg_host, pg_port, pg_db, test_db.version, pg_password):
        connection_str = f"postgresql+psycopg2://{pg_user}:@{pg_host}:{pg_port}/{pg_db}"
        engine = create_engine(connection_str)
        model.Base.metadata.create_all(engine)
        yield sessionmaker(bind=engine, expire_on_commit=False)

And test passes.

But why it doesn't work with connection instead?

huangapple
  • 本文由 发表于 2023年6月13日 02:46:12
  • 转载请务必保留本文链接:https://go.coder-hub.com/76459471.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定