How would you unit test this SQLAlchemy Core query/function?

Question

I'm working on learning how to unit test properly. Given this function...

def get_user_details(req_user_id):
    users = sa.Table('users', db.metadata, autoload=True)

    s = sa.select([users.c.username,
                   users.c.favorite_color,
                   users.c.favorite_book
                   ]).select_from(users)

    s = s.where(users.c.user_id == req_user_id)

    result = db.connection.execute(s).fetchone()
    return dict(result)

...what is the proper way to unit test it?

Here's where I am now...

1. From what I've read, testing the "construct" of the query is unnecessary as that's part of the already well-tested SQLAlchemy library. So I don't need to test the raw SQL generated, right? But should I test the parameters passed and if so, how?
2. 我已经读过关于嘲笑来自 的响应，db.connection.execute但这如何真正测试任何东西？最终，我想确保该函数生成正确的 SQL 并获得正确的数据库结果？

非常感谢任何建议/指导。谢谢！

Answer 1

\n

以下是此评论：

\n

\n

您需要测试的是代码中的语句是否产生\n预期的结果。\xe2\x80\x93 肖德

\n

\n

和你的代码，这就是我要回答的内容：\n如何测试动态生成 SQLAlchemy 查询的方法？

\n

这就是我遇到的问题，因为我想确保生成的查询确实正确 - 不是因为 SQL Alchemy，而是因为将它们组合在一起的逻辑。

\n

我们将测试的代码

\n

def add_filters(query, target_table, filter_type: str, filter_value: str):\n    if filter_type == "favourite_book":\n        query = query.filter(target_table.c.favourite_book == filter_value)\n    elif filter_type == "favourite_color":\n        query = query.filter(target_table.c.favourite_color == filter_value)\n\n    return query\n

\n

所以我想测试 favourite_book 过滤器是否确实正确添加到查询中。

\n

为此，我们将创建一个临时 sqlite3 数据库，其中包含一个包含数据的表，并对其运行查询。最后我们测试查询的结果。注意：您需要数据中的好数据和坏数据进行全面测试。

\n

设置测试数据库

\n

import pytest\nfrom sqlalchemy.orm import Session\nfrom sqlalchemy import create_engine, MetaData, Table, Column, Integer, String\nimport pandas as pd\n\n@pytest.fixture\ndef engine():\n    my_engine = create_engine("sqlite:///file:test_db?mode=memory&cache=shared&uri=true", echo=True)\n    return my_engine\n\n\n@pytest.fixture\ndef target_table(engine):\n    meta = MetaData()\n    table_name = "users"\n\n    users = Table(\n        table_name,\n        meta,\n        Column("id", Integer, primary_key=True),\n        Column("username", String),\n        Column("favourite_color", String),\n        Column("favourite_book", String),\n    )\n\n    meta.create_all(engine)\n\n    # you can choose to skip the whole table declaration as df.to_sql will create\n    # the table for you if it doesn\'t exist\n\n    records = [\n        {"username": "john", "favourite_color": "blue", "favourite_book": "Harry Potter"},\n        {"username": "jane", "favourite_color": "red", "favourite_book": "The Power of Now"},\n        {"username": "bob", "favourite_color": "green", "favourite_book": "Extreme Ownership"},\n    ]\n\n    df = pd.DataFrame(records)\n    df.to_sql(table_name, engine, if_exists="append", index=False)\n    return users\n

\n

最后是实际测试

\n

def test_query(engine, target_table):\n    with Session(engine) as session:\n        query = session.query(target_table)\n        query = add_filters(query, target_table, "favourite_book", "Extreme Ownership")\n        df = pd.read_sql_query(query.statement, session.bind)\n        assert df["favourite_book"].unique().tolist() == ["Extreme Ownership"]\n

\n

你可以看到，测试不是很全面，因为它只测试了一种情况。\n但是，我们可以使用 pytest.mark.parametrize 来扩展它。（见最后一个参考）

\n

@pytest.mark.parametrize(\n    "filter_type,filter_value,expected_result",\n    [\n        ("favourite_book", "Extreme Ownership", True),\n        ("favourite_book", "Extreme", False),\n        ("favourite_color", "blue", True),\n        ("favourite_color", "purple", False),\n    ],\n)\ndef test_query(engine, target_table, filter_type, filter_value, expected_result):\n    with Session(engine) as session:\n        query = session.query(target_table)\n        query = add_filters(query, target_table, filter_type, filter_value)\n        df = pd.read_sql_query(query.statement, session.bind)\n        assert (df[filter_type].unique().tolist() == [filter_value]) == expected_result\n

\n

一些参考资料：

\n

\n
• https://www.tutorialspoint.com/pytest/pytest_fixtures.htm
\n
• https://smirnov-am.github.io/pytest-testing_database/
\n
• Python - 如何将 SQLAlchemy 连接到内存中的现有数据库
\n
• 使用 pytest-mock 在 python 中模拟数据库调用
\n

\n