摘自此评论：
您需要测试的是代码中的语句是否产生
预期成绩。 – 肖德
和你的代码，这就是我要回答的内容：如何测试动态生成 SQLAlchemy 查询的方法？
这就是我遇到的问题，因为我想确保生成的查询确实正确 - 不是因为 SQL Alchemy，而是因为将它们组合在一起的逻辑。
我们将测试的代码
def add_filters(query, target_table, filter_type: str, filter_value: str):
if filter_type == "favourite_book":
query = query.filter(target_table.c.favourite_book == filter_value)
elif filter_type == "favourite_color":
query = query.filter(target_table.c.favourite_color == filter_value)
return query
所以我想测试 favourite_book 过滤器是否确实正确添加到查询中。
为此，我们将创建一个临时 sqlite3 数据库，其中包含一个包含数据的表，并对其运行查询。最后我们测试查询的结果。注意：您需要数据中的好数据和坏数据进行全面测试。
设置测试数据库
import pytest
from sqlalchemy.orm import Session
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
import pandas as pd
@pytest.fixture
def engine():
my_engine = create_engine("sqlite:///file:test_db?mode=memory&cache=shared&uri=true", echo=True)
return my_engine
@pytest.fixture
def target_table(engine):
meta = MetaData()
table_name = "users"
users = Table(
table_name,
meta,
Column("id", Integer, primary_key=True),
Column("username", String),
Column("favourite_color", String),
Column("favourite_book", String),
)
meta.create_all(engine)
# you can choose to skip the whole table declaration as df.to_sql will create
# the table for you if it doesn't exist
records = [
{"username": "john", "favourite_color": "blue", "favourite_book": "Harry Potter"},
{"username": "jane", "favourite_color": "red", "favourite_book": "The Power of Now"},
{"username": "bob", "favourite_color": "green", "favourite_book": "Extreme Ownership"},
]
df = pd.DataFrame(records)
df.to_sql(table_name, engine, if_exists="append", index=False)
return users
最后是实际测试
def test_query(engine, target_table):
with Session(engine) as session:
query = session.query(target_table)
query = add_filters(query, target_table, "favourite_book", "Extreme Ownership")
df = pd.read_sql_query(query.statement, session.bind)
assert df["favourite_book"].unique().tolist() == ["Extreme Ownership"]
可以看到，测试不是很全面，因为它只测试了一种情况。
但是，我们可以使用 pytest.mark.parametrize 来扩展它。 （参见最后的参考资料）
@pytest.mark.parametrize(
"filter_type,filter_value,expected_result",
[
("favourite_book", "Extreme Ownership", True),
("favourite_book", "Extreme", False),
("favourite_color", "blue", True),
("favourite_color", "purple", False),
],
)
def test_query(engine, target_table, filter_type, filter_value, expected_result):
with Session(engine) as session:
query = session.query(target_table)
query = add_filters(query, target_table, filter_type, filter_value)
df = pd.read_sql_query(query.statement, session.bind)
assert (df[filter_type].unique().tolist() == [filter_value]) == expected_result
一些参考资料：
- https://www.tutorialspoint.com/pytest/pytest_fixtures.htm https://www.tutorialspoint.com/pytest/pytest_fixtures.htm
- https://smirnov-am.github.io/pytest-testing_database/ https://smirnov-am.github.io/pytest-testing_database/
- Python - 如何将 SQLAlchemy 连接到内存中的现有数据库 https://stackoverflow.com/questions/62502827/python-how-to-connect-sqlalchemy-to-existing-database-in-memory
- 使用 pytest-mock 在 python 中模拟数据库调用 https://stackoverflow.com/questions/66223290/mocking-database-calls-in-python-using-pytest-mock
