英文:
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?
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论