如何在模拟类中模拟/修补多个方法？

Question

我对 Python 自动化测试没有经验，而且我很难尝试模拟（或修补？）一个类及其一些方法。

这是我们的文件夹结构：

src
|--- main
     |--- run_data_process.py
tests
|--- main_tests
     |--- test_run_data_process.py

以下是目前这些文件的一些代码：

run_data_process.py:

import logging
import pandas as pd
import sys

from company.connections import DBConnection # internal library from my company
from company.data_tools import DataStorageManager # another internal library

logger = logging.getLogger(__name__)


class DataProcessException(Exception):
    pass


class DataProcess:
    def run_task(self):
        try:
            df_raw = self.gather_data()
            self.validate_gathered_data(df_raw)
        except DataProcessException as e:
            logger.error(f"Error on the Data Process task: {e}")
            raise
        else:
            self.save_output_data(df_raw)
            logger.info("Data Process task executed successfully.")

    def gather_data(self) -> pd.DataFrame:
        """This calls the database and returns a DataFrame. I don't have access to the implementation of DBConnection()."""
        conn = DBConnection()
        return conn.run()

    def validate_gathered_data(self, df_raw: pd.DataFrame) -> None:
        if len(df_raw) == 0:
            raise DataProcessException("Raw dataset is empty.")

    def save_output_data(self, df_raw: pd.DataFrame) -> None:
        """Another internal library that serialises the DataFrame and sends it to another server. Don't have the implementation either."""
        DataStorageManager().save_output_object(df_raw)

test_run_data_process.py:

from unittest.mock import Mock, MagicMock, patch
import pandas as pd
import pytest

from src.main.run_data_process import DataProcess, RawDataException


class TestDataProcess:

    def __get_mocked_dataset():
        return pd.DataFrame(
            [[111, 222, 333], ['text_1', 'text_2', 'text_3'], [True, False, False]],
            columns=['col_1', 'col_2', 'col_3'])

    def test_gathered_data_is_validated_successfully(self):
        # arrange / act
        with patch('src.main.run_data_process.DataProcess', return_value=MagicMock()) as p:
            p.return_value.gather_data = Mock(return_value=self.__get_mocked_dataset)
            p.return_value.save_output_data = Mock(return_value=None)
            p.return_value.run_task()

        # assert
        p.return_value.validate_gathered_data.assert_called_with(p.return_value.gather_data.return_value)

我要做的是测试validate_gathered_data，同时模拟gather_data（避免数据库查询）和save_output_data（避免调用我们的服务器），但我无法让它工作，我'我用这个把我的头发拉出来。

上面的代码给了我这个错误：

./tests/main_tests/test_run_data_process.py::TestDataProcess::test_gathered_data_is_validated_successfully Failed: [undefined]AssertionError: expected call not found.
Expected: validate_gathered_data(<bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>)
Actual: not called.
__wrapped_mock_method__ = <function NonCallableMock.assert_called_with at 0x108964550>
args = (<MagicMock name='mock.validate_gathered_data' id='4894604448'>, <bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>)
kwargs = {}, __tracebackhide__ = True
msg = 'expected call not found.\nExpected: validate_gathered_data(<bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>)\nActual: not called.'
__mock_self = <MagicMock name='mock.validate_gathered_data' id='4894604448'>

    def assert_wrapper(
        __wrapped_mock_method__: Callable[..., Any], *args: Any, **kwargs: Any
    ) -> None:
        __tracebackhide__ = True
        try:
>           __wrapped_mock_method__(*args, **kwargs)

../../../../Library/Python/3.8/lib/python/site-packages/pytest_mock/plugin.py:414: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <MagicMock name='mock.validate_gathered_data' id='4894604448'>
args = (<bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>,)
kwargs = {}
expected = 'validate_gathered_data(<bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>)'
actual = 'not called.'
error_message = 'expected call not found.\nExpected: validate_gathered_data(<bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>)\nActual: not called.'

    def assert_called_with(self, /, *args, **kwargs):
        """assert that the last call was made with the specified arguments.
    
        Raises an AssertionError if the args and keyword args passed in are
        different to the last call to the mock."""
        if self.call_args is None:
            expected = self._format_mock_call_signature(args, kwargs)
            actual = 'not called.'
            error_message = ('expected call not found.\nExpected: %s\nActual: %s'
                    % (expected, actual))
>           raise AssertionError(error_message)
E           AssertionError: expected call not found.
E           Expected: validate_gathered_data(<bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>)
E           Actual: not called.

/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/unittest/mock.py:904: AssertionError

During handling of the above exception, another exception occurred:

self = <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>

    def test_gathered_data_is_validated_successfully(self):
        # arrange / act
        with patch('src.main.run_data_process.DataProcess', return_value=MagicMock()) as p:
            p.return_value.gather_data = Mock(return_value=self.__get_mocked_dataset)
            p.return_value.save_output_artifact = Mock(return_value=None)
            p.return_value.run_task()
    
        # assert
>       p.return_value.validate_gathered_data.assert_called_with(p.return_value.gather_data.return_value)
E       AssertionError: expected call not found.
E       Expected: validate_gathered_data(<bound method TestDataProcess.__get_mocked_dataset of <tests.main_tests.test_run_data_process.TestDataProcess object at 0x123b9fe50>>)
E       Actual: not called.

tests/main_tests/test_run_data_process.py:23: AssertionError

在我之前的镜头中，我已经在实例化 DataProcess 时尝试了 @mock.patch.multiple（我在此处看到过的一个解决方案，这里有无数的问题），但卡在了 assert_called_with 部分。

任何帮助将不胜感激。

Answer 1

通常最好避免修补被测类。在这种情况下，由于DataProcess.gather() 返回DBConnection().run()，并且由于您要避免的具体事情是制作真正的DBConnection，因此直接的解决方案是修补DBConnection：

    def test_gathered_data_is_validated_successfully(self):
        with patch('src.main.run_data_process.DBConnection') as db:
            db.return_value.run.return_value = self.__get_mocked_dataset()
            DataProcess().run_task()