pytest/unittest：模块中的 mock.patch 函数？

Question

给定这样的文件夹结构：

dags/
  **/
    code.py
tests/
  dags/
    **/
      test_code.py
  conftest.py

其中 dags 作为 src 文件的根目录，将 'dags/a/b/c.py' 导入为 'a.b.c'。

我想在code.py中测试以下函数：

from dag_common.connections import get_conn
from utils.database import dbtypes

def select_records(
    conn_id: str,
    sql: str,
    bindings,
):
    conn: dbtypes.Connection = get_conn(conn_id)
    with conn.cursor() as cursor:
        cursor.execute(
            sql, bindings
        )
        records = cursor.fetchall()
    return records

但我面临的问题是我无法找到从dag_common.connections 修补get_conn 的方法。我尝试了以下方法：

(1) 全局在conftest.py中

import os
import sys

# adds dags to sys.path for tests/*.py files to be able to import them
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "dags"))

{{fixtures}}

我测试了{{fixtures}} 的以下替换：

(1.a) - 默认

@pytest.fixture(autouse=True, scope="function")
def mock_get_conn():
    with mock.patch("dag_common.connections.get_conn") as mock_getter:
        yield mock_getter

(1.b) - 以 dags 为前缀的路径

@pytest.fixture(autouse=True, scope="function")
def mock_get_conn():
    with mock.patch("dags.dag_common.connections.get_conn") as mock_getter:
        yield mock_getter

(1.c) - 1.a，范围="session"

(1.d) - 1.b，范围="session"

(1.e) - 修补模块本身的对象

@pytest.fixture(autouse=True, scope="function")
def mock_get_conn():
    import dags.dag_common.connections
    mock_getter = mock.MagicMock()
    with mock.patch.object(dags.dag_common.connections, 'get_conn', mock_getter):
        yield mock_getter

(1.f) - 1.a，但使用 pytest-mock 夹具

@pytest.fixture(autouse=True, scope="function")
def mock_get_conn(mocker):
    with mocker.patch("dag_common.connections.get_conn") as mock_getter:
        yield mock_getter

(1.g) - 1.b，但使用 pytest-mock 夹具

(1.h) - 1.a，但使用 pytest 的猴子补丁

@pytest.fixture(autouse=True, scope="function")
def mock_get_conn(mocker, monkeypatch):
    import dags.dag_common.connections
    mock_getter = mocker.MagicMock()
    monkeypatch.setattr(dags.dag_common.connections, 'get_conn', mock_getter)
    yield mock_getter

(2) 在测试中本地应用mock.patch/作为装饰器

(2.a) - 装饰器@mock.patch("dag_common.connections.get_conn")

    @mock.patch("dag_common.connections.get_conn")
    def test_executes_sql_with_default_bindings(mock_getter, mock_context):
        # arrange
        sql = "SELECT * FROM table"
        records = [RealDictRow(col1=1), RealDictRow(col1=2)]
        mock_conn = mock_getter.return_value
        mock_cursor = mock_conn.cursor.return_value
        mock_cursor.execute.return_value = records
        # act
        select_records(conn_id="orca", sql=sql, ) # ...
        # assert
        mock_cursor.execute.assert_called_once_with(
            sql, # ...
        )

(2.b) - (2.a) 但带有“dags”。前缀

(2.c) - 上下文管理器

    def test_executes_sql_with_default_bindings(mock_context):
        # arrange
        sql = "SELECT * FROM table"
        records = [RealDictRow(col1=1), RealDictRow(col1=2)]
        with mock.patch("dag_common.connections.get_conn") as mock_getter:
            mock_conn = mock_getter.return_value
            mock_cursor = mock_conn.cursor.return_value
            mock_cursor.execute.return_value = records
            # act
            select_records(conn_id="orca", sql=sql, ) # ...
            # assert
            mock_cursor.execute.assert_called_once_with(
                sql, # ...
            )

(2.d) - (2.c) 但带有“dags”。前缀

---

结论

但是，不管我选择什么解决方案，要模拟的函数仍然会被调用。 我确保分别尝试每个解决方案，并在两次尝试之间终止/清除/重新启动我的 pytest-watch 进程。

我觉得这可能与我在 conftest.py 中干预 sys.path 有关，因为除此之外我觉得我已经用尽了所有可能性。

知道如何解决这个问题吗？

Answer 1

是的。最初，当我学习修补和模拟时，我也曾与此作斗争，并且知道这是多么令人沮丧，因为您似乎做的一切都正确，但它不起作用。我很同情你！

这实际上是模拟导入的东西的工作原理，一旦你意识到这一点，它实际上是有道理的。

问题在于导入的工作方式是使导入的模块在导入所在的上下文中可用。

让我们假设您的code.py 模块位于“my_package”文件夹中。您的代码随后以my_package.code 的形式提供。一旦您在 code 模块中使用 from dag_common.connections import get_conn - 导入的 get_conn 将变为 .... my_package.code.get_conn

在这种情况下，您需要修补 my_package.code.get_conn 而不是您从中导入 get_conn 的原始包。

一旦你意识到这一点，修补就会变得容易得多。