我正在用pytest测试一个FastAPI应用程序。我创建了一个客户端设备,其中包括一个从CSV创建的sqlite DB:
import pytest
from os import path, listdir, remove
from pandas import read_csv
from fastapi.testclient import TestClient
from api.main import app
from api.db import engine, db_url
@pytest.fixture(scope="session")
def client():
db_path = db_url.split("///")[-1]
if path.exists(db_path):
remove(db_path)
file_path = path.dirname(path.realpath(__file__))
table_path = path.join(file_path, "mockdb")
for table in listdir(table_path):
df = read_csv(path.join(table_path, table))
df.to_sql(table.split('.')[0], engine, if_exists="append", index=False)
client = TestClient(app)
yield client
我在FastAPI应用程序中的数据库设置:
import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
dirname = os.path.dirname(__file__)
if "pytest" in modules:
mock_db_path = os.path.join(dirname, '../test/mockdb/test.db')
db_url = f"sqlite:///{mock_db_path}"
else:
db_url = os.environ.get("DATABASE_URL", None)
if "sqlite" in db_url:
engine = create_engine(db_url, connect_args={"check_same_thread": False})
else:
engine = create_engine(db_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
这很有效:我可以为查询DB的应用程序端点设置测试,并且我放入CSV中的数据会被返回,例如在向mockdb/person.csv
:添加一行之后
from api.db import SessionLocal
db = SessionLocal()
all = db.query(Person).all()
print(all)
[<tables.Person object at 0x7fc829f81430>]
我现在正在尝试测试向数据库中的表添加新行的代码。
只有当我指定ID时,这才有效(假设这发生在pytest运行期间(:
db.add(Person(id=2, name="Alice"))
db.commit()
all = db.query(Person).all()
print(all)
[<tables.Person object at 0x7fc829f81430>, <tables.Person object at 0x7fc829f3bdc0>]
上面的结果是我所期望的程序行为。但是,如果我没有指定ID,那么结果是None
:
db.add(Person(name="Alice"))
db.commit()
all = db.query(Person).all()
print(all)
[<tables.Person object at 0x7fc829f81430>, None]
这个结果不是我所期望的程序的行为。
我想要测试的代码没有指定ID,它使用自动增量,这是一种很好的做法。因此,我无法测试此代码。它只是创建这些None
s。
起初,我认为罪魁祸首不是用Base.metadata.create_all()
创建表。然而,我已经尝试将其放在我的客户端设备中,并按照我的DB设置(即上面的前两个代码块(进行设置,但结果是一样的:None
s。
通过调试器,当添加Person行时,会出现以下错误:
sqlalchemy.orm.exc.ObjectDeletedError: Instance '<Person at 0x7fc829f3bdc0>' has been deleted, or its row is otherwise not present.
为什么生成的行是None
?如何解决此错误?
错误的原因是我的数据库中有一个与SQLite不兼容的列类型,即PostgresSQL的ARRAY
类型。不幸的是,没有任何错误消息暗示这一点。最简单的解决方案是删除或更改此列的类型。
还可以通过如下更改client()
来保留柱和SQLite固定装置:
from mytableschema import MyOffendingTable
@pytest.fixture(scope="session")
def client():
table_meta = SBEvent.metadata.tables[MyOffendingTable.__tablename__]
table_meta._columns.remove(table_meta._columns["my_offending_column"])
Base.metadata.create_all(bind=engine)
db_path = db_url.split("///")[-1]
if path.exists(db_path):
remove(db_path)
file_path = path.dirname(path.realpath(__file__))
table_path = path.join(file_path, "mockdb")
for table in listdir(table_path):
df = read_csv(path.join(table_path, table))
df.to_sql(table.split('.')[0], engine, if_exists="append", index=False)
client = TestClient(app)
yield client
如果从MyOffendingTable
CSV中删除my_offending_column
,现在可以正常进行。不再有None
s!
遗憾的是,在测试运行期间查询有问题的表仍然会遇到问题,因为SELECT语句将查找不存在的my_offending_column
。对于那些需要查询上述表的人,我建议使用方言特定的编译规则。