Pytest mocker 补丁 - 如何排除故障？

Question

我在使用模拟修补时遇到了一个我认为是常见问题，因为我无法确定要修补的正确的事情。

我有两个问题希望得到帮助。

1. 关于如何解决以下示例中特定问题的想法
2. 可能最重要的专业提示/指针/想法/建议是关于如何最好地解决“我要修补哪些内容”问题的。我遇到的问题是，在不完全了解修补程序的工作原理的情况下，我什至不知道我应该寻找什么并且发现自己在玩猜谜游戏。

使用pyarrow 的示例目前让我很痛苦：

mymodule.py

import pyarrow

class HdfsSearch:
    def __init__(self):
        self.fs = self._connect()

    def _connect(self) -> object:
        return pyarrow.hdfs.connect(driver="libhdfs")

    def search(self, path: str):
        return self.fs.ls(path=path)

test_module.py

import pyarrow
import pytest

from mymodule import HdfsSearch

@pytest.fixture()
def hdfs_connection_fixture(mocker):
    mocker.patch("pyarrow.hdfs.connect")
    yield HdfsSearch()

def test_hdfs_connection(hdfs_connection_fixture):
    pyarrow.hdfs.connect.assert_called_once() # <-- succeeds

def test_hdfs_search(hdfs_connection_fixture):
    hdfs_connection_fixture.search(".")
    pyarrow.hdfs.HadoopFileSystem.ls.assert_called_once() # <-- fails

pytest 输出：

$ python -m pytest --verbose test_module.py
=========================================================================================================== test session starts ============================================================================================================
platform linux -- Python 3.7.4, pytest-5.0.1, py-1.8.0, pluggy-0.12.0 -- /home/bbaur/miniconda3/envs/dev/bin/python
cachedir: .pytest_cache
rootdir: /home/user1/work/app
plugins: cov-2.7.1, mock-1.10.4
collected 2 items

test_module.py::test_hdfs_connection PASSED                                                                                                                                                                                          [ 50%]
test_module.py::test_hdfs_search FAILED                                                                                                                                                                                              [100%]

================================================================================================================= FAILURES =================================================================================================================
_____________________________________________________________________________________________________________ test_hdfs_search _____________________________________________________________________________________________________________

hdfs_connection_fixture = <mymodule.HdfsSearch object at 0x7fdb4ec2a610>

    def test_hdfs_search(hdfs_connection_fixture):
        hdfs_connection_fixture.search(".")
>       pyarrow.hdfs.HadoopFileSystem.ls.assert_called_once()
E       AttributeError: 'function' object has no attribute 'assert_called_once'

test_module.py:16: AttributeError

Answer 1

要了解修补在 python 中的工作原理，我们首先要了解 import 语句。

当我们在模块中使用import pyarrow 时（在本例中为 mymodule.py），它会执行两个操作：

1. 它在sys.modules 中搜索pyarrow 模块
2. 它将搜索结果绑定到本地范围内的名称 (pyarrow)。 通过执行以下操作：pyarrow = sys.modules['pyarrow']

注意：python 中的import 语句不执行代码。 import 语句将名称带入本地范围。只有当 python 在 sys.modules 中找不到模块时，代码的执行才会作为副作用发生

因此，要修补 mymodule.py 中导入的 pyarrow 我们需要修补 mymodule.py 的本地范围中存在的 pyarrow 名称

patch('mymodule.pyarrow', autospec=True)

test_module.py

import pytest
from mock import Mock, sentinel
from pyarrow import hdfs

from mymodule import HdfsSearch


class TestHdfsSearch(object):
    @pytest.fixture(autouse=True, scope='function')
    def setup(self, mocker):
        self.hdfs_mock = Mock(name='HadoopFileSystem', spec=hdfs.HadoopFileSystem)
        self.connect_mock = mocker.patch("mymodule.pyarrow.hdfs.connect", return_value=self.hdfs_mock)

    def test_initialize_HdfsSearch_should_connect_pyarrow_hdfs_file_system(self):
        HdfsSearch()

        self.connect_mock.assert_called_once_with(driver="libhdfs")

    def test_initialize_HdfsSearch_should_set_pyarrow_hdfs_as_file_system(self):
        hdfs_search = HdfsSearch()

        assert self.hdfs_mock == hdfs_search.fs

    def test_search_should_retrieve_directory_contents(self):
        hdfs_search = HdfsSearch()
        self.hdfs_mock.ls.return_value = sentinel.contents

        result = hdfs_search.search(".")

        self.hdfs_mock.ls.assert_called_once_with(path=".")
        assert sentinel.contents == result

使用上下文管理器修补内置插件

def test_patch_built_ins():
    with patch('os.curdir') as curdir_mock:  # curdir_mock lives only inside with block. Doesn't lives outside
        assert curdir_mock == os.curdir
    assert os.curdir == '.'

Answer 2

你没有在 Mock 对象上调用断言，这是正确的断言：

hdfs_connection_fixture.fs.ls.assert_called_once()

说明：

当您访问 Mock 对象中的任何属性时，它将返回另一个 Mock 对象。

由于您修补了 "pyarrow.hdfs.connect"，因此您已将其替换为 Mock，我们将其称为 Mock A。您的 _connect 方法将返回该 Mock A，您将其分配给 self.fs。

现在让我们分解当您调用self.fs.ls 时search 方法中发生的情况。

self.fs 返回您的 Mock A 对象，然后 .ls 将返回一个不同的 Mock 对象，我们称之为 Mock B。在此 Mock B 对象中，您正在执行传递 (path=path) 的调用。

在您的断言中，您尝试访问pyarrow.hdfs.HadoopFileSystem，但从未对其进行修补。您需要对 Mock B 对象进行断言，该对象位于 hdfs_connection_fixture.fs.ls

修补什么

如果您将mymodule.py 中的导入更改为此from pyarrow.hdfs import connect，您的补丁将停止工作。

这是为什么呢？

当您修补某些内容时，您正在更改 name 指向的内容，而不是实际的对象。

您当前的补丁正在修补名称 pyarrow.hdfs.connect，而在 mymodule 中您使用的是相同的名称 pyarrow.hdfs.connect，所以一切都很好。

但是，如果您使用from pyarrow.hdfs import connect，mymodule 将导入真正的pyarrow.hdfs.connect 并为其创建一个名为mymodule.connect 的引用。

因此，当您在 mymodule 中调用 connect 时，您访问的是未修补的名称 mymodule.connect。

这就是为什么在使用 from import 时需要修补 mymodule.connect。

我建议在进行此类修补时使用from x import y。它使您尝试模拟的内容更加明确，并且补丁将仅限于该模块，这可以防止不可预见的副作用。

来源，Python 文档中的这一部分：Where to patch