如何在测试中检查无返回值函数内部的sql及execute变量内容？

要验证create_tables函数内部的sql变量内容以及cur.execute的执行内容，你需要完善Mock配置，同时修正模板变量的匹配问题，具体步骤如下：

1. 修正模板变量匹配问题 #

原函数中使用template.render(TABLE_NAME=table)传递变量，但你的测试里mock的模板内容用的是{{table}}，变量名不匹配会导致渲染后的SQL不符合预期。需要把mock的模板内容改为{{TABLE_NAME}}。

2. 追踪游标（cursor）的execute调用 #

你需要让pg_mock.cursor()返回的Mock对象能记录execute方法的调用参数，这样就能验证实际执行的SQL内容。

3. 验证模板渲染的SQL内容 #

可以通过两种方式验证：

• 间接方式：通过检查cursor.execute的调用参数，确认渲染后的SQL是否正确；
• 直接方式：Mockjinja2.Template类，捕获render方法的调用参数和返回值，直接验证模板渲染结果。

完整测试代码示例 #

方式一：通过游标调用验证（简洁直接） #

from unittest.mock import MagicMock, mock_open, call
from your_module import create_tables  # 替换为实际的模块名
from airflow.hooks.postgres_hook import PostgresHook

def test_create_tables(mocker):
    # 1. Mock PostgresHook的get_conn方法，返回自定义的conn mock
    pg_mock = MagicMock(name="pg_conn")
    mocker.patch.object(PostgresHook, "get_conn", return_value=pg_mock)
    
    # 2. 配置cursor mock，适配上下文管理器语法
    cursor_mock = MagicMock(name="pg_cursor")
    pg_mock.cursor.return_value.__enter__.return_value = cursor_mock
    
    # 3. Mock模板文件内容，变量名与函数内保持一致
    mock_file_content = "CREATE TABLE IF NOT EXISTS {{TABLE_NAME}}"
    mocker.patch("builtins.open", mock_open(read_data=mock_file_content))
    
    # 4. 执行测试函数
    create_tables(tables=["mock_table_1", "mock_table_2"])
    
    # 5. 验证游标execute的调用内容与顺序
    expected_calls = [
        call("CREATE TABLE IF NOT EXISTS mock_table_1"),
        call("CREATE TABLE IF NOT EXISTS mock_table_2")
    ]
    cursor_mock.execute.assert_has_calls(expected_calls, any_order=False)
    
    # 6. 验证conn.commit()是否被正确调用
    pg_mock.commit.assert_called_once()

方式二：直接Mock模板类验证渲染结果 #

如果需要单独验证模板渲染逻辑，不依赖游标调用，可以这样做：

from unittest.mock import MagicMock, mock_open, call
from your_module import create_tables
from airflow.hooks.postgres_hook import PostgresHook
import jinja2

def test_create_tables_with_template_mock(mocker):
    # 1. Mock PostgresHook相关逻辑
    pg_mock = MagicMock(name="pg_conn")
    mocker.patch.object(PostgresHook, "get_conn", return_value=pg_mock)
    cursor_mock = MagicMock(name="pg_cursor")
    pg_mock.cursor.return_value.__enter__.return_value = cursor_mock
    
    # 2. Mock jinja2.Template，捕获render调用
    template_mock = MagicMock(name="jinja_template")
    template_mock.render.side_effect = lambda TABLE_NAME: f"CREATE TABLE IF NOT EXISTS {TABLE_NAME}"
    mocker.patch("jinja2.Template", return_value=template_mock)
    
    # 3. 执行测试函数
    create_tables(tables=["mock_table_1", "mock_table_2"])
    
    # 验证模板render的调用参数
    template_mock.render.assert_has_calls(
        [call(TABLE_NAME="mock_table_1"), call(TABLE_NAME="mock_table_2")],
        any_order=False
    )
    
    # 同时验证游标execute的调用
    cursor_mock.execute.assert_has_calls(
        [call("CREATE TABLE IF NOT EXISTS mock_table_1"), call("CREATE TABLE IF NOT EXISTS mock_table_2")],
        any_order=False
    )
    
    pg_mock.commit.assert_called_once()

关键说明 #

• 针对with conn.cursor() as cur这种上下文管理器语法，必须设置pg_mock.cursor.return_value.__enter__.return_value = cursor_mock，才能让as后的cur指向我们的Mock对象；
• 使用assert_has_calls可以按顺序验证多次调用的参数，确保每个表对应的SQL都被正确执行；
• 两种方式可以结合使用，同时验证模板渲染逻辑和数据库执行逻辑。

内容的提问来源于stack exchange，提问作者Claudiu