mock测试基础:unittest.mock入门与实践

Python 测试与调试专题 · 用Mock对象隔离外部依赖进行测试

专题:Python 测试与调试系统学习

关键词:Python, 测试, 调试, mock, unittest.mock, patch, MagicMock, 测试隔离, 模拟, Python测试

一、Mock概述

在编写单元测试时,一个核心挑战是如何隔离被测试代码与外部依赖。实际项目中的函数和方法往往依赖于数据库、网络API、文件系统、外部服务等组件,这些依赖在测试环境中可能不可用、不稳定或成本过高。Mock对象(模拟对象)正是为解决这一问题而生的技术——它通过创建"替身"对象来替换真实依赖,使测试能够专注于被测试代码本身的逻辑,而不受外部因素干扰。

Mock对象的核心原理是基于Python的动态特性,在运行时创建一个可编程的虚拟对象,该对象能模拟真实对象的接口和行为。当测试代码调用Mock对象的方法或访问其属性时,Mock会自动记录这些操作(包括调用的参数、次数、顺序等),同时允许开发者预设返回值或副作用。这种"记录-回放"机制使得测试可以在完全可控的环境中验证代码行为。Python标准库中的unittest.mock模块(自Python 3.3起加入标准库)提供了完整的Mock框架支持,无需安装任何第三方包。

unittest.mock模块是Python官方推荐的Mock方案,其设计简洁而功能强大。它包含几个核心类:Mock(基础模拟类)、MagicMock(预置魔术方法的Mock子类)、PropertyMock(属性模拟专用)、AsyncMock(异步Mock,Python 3.8+),以及核心辅助函数patch(用于在测试期间替换对象)、sentinel(创建唯一标识对象)等。这些组件协同工作,能够覆盖几乎所有测试隔离场景。

值得一提的是,unittest.mock与第三方库pytest-mock之间是互补而非替代关系。pytest-mock是一个pytest插件,它将unittest.mock的功能封装为pytest fixture(mocker),提供了更简洁的语法和自动清理功能。使用pytest-mock时,底层仍然是unittest.mock的Mock和patch机制,只是包装了更友好的API。选择哪一个取决于项目风格:如果项目使用pytest作为测试框架,pytest-mock能提升代码可读性;如果希望减少依赖,直接用unittest.mock同样完整可靠。

from unittest.mock import Mock, MagicMock, patch # 创建一个基本的Mock对象 mock_obj = Mock() mock_obj.some_method(1, 2, 3) mock_obj.some_method.assert_called_once_with(1, 2, 3) print(mock_obj.some_method.return_value) # 输出: <Mock name='mock.some_method()' id='...'>
# pytest-mock 使用示例(需安装pytest-mock) # test_example.py def test_with_mocker(mocker): # mocker 自动整合了 patch 功能 mock_requests_get = mocker.patch('requests.get') mock_requests_get.return_value.json.return_value = {'key': 'value'} # 被测试函数内部调用 requests.get result = my_function_that_calls_api() assert result == {'key': 'value'} mock_requests_get.assert_called_once()
# 对比:不使用mock的脆弱测试 import requests def test_real_api(): # 依赖外部API,不稳定 resp = requests.get('https://api.example.com/data') assert resp.status_code == 200 # 使用mock的隔离测试 @patch('requests.get') def test_mocked_api(mock_get): mock_get.return_value.status_code = 200 mock_get.return_value.json.return_value = {'ok': True} resp = requests.get('https://api.example.com/data') assert resp.status_code == 200 assert resp.json() == {'ok': True}

二、Mock对象

Mock对象的创建和配置是mock测试的基础。通过Mock()构造函数可以创建最基础的模拟对象,该对象的所有属性和方法都是动态的——访问一个不存在的属性会自动创建一个子Mock对象,调用一个不存在的方法也会自动创建并返回一个子Mock。这种"惰性创建"机制使得Mock对象极其灵活:你不需要预先定义所有接口,直接调用即可。然而,这种灵活性也可能隐藏代码中的拼写错误,因此需要结合后面的spec机制一起使用。

配置返回值(return_value)是最常见的Mock配置操作。当测试代码调用被模拟的方法时,Mock会返回预设的return_value。如果同时配置了side_effect(副作用),则side_effect的优先级更高。配置side_effect可以发挥更大的作用:它可以是一个函数(每次调用时执行该函数并返回其结果)、一个异常(调用时抛出异常用于测试异常处理)、或一个可迭代对象(每次调用依次返回迭代中的下一个值)。这种机制非常适合模拟不同的调用场景。

Mock对象的属性设置也非常直观,直接赋值即可。例如mock_obj.some_attr = 42为Mock设置一个固定值属性,之后访问mock_obj.some_attr就会返回42。对于需要按方法调用的属性(如属性是一个可调用对象),可以通过mock_obj.method.return_value = value链式配置。此外,name参数虽然看起来不起眼,但在调试时非常有用——它会在Mock的repr输出中显示,帮助区分不同Mock对象在测试失败时的角色。

from unittest.mock import Mock # 创建并配置return_value mock = Mock() mock.get_data.return_value = {'id': 1, 'name': 'Alice'} result = mock.get_data() print(result) # {'id': 1, 'name': 'Alice'} # 链式配置 mock.db.query.return_value = [('row1',), ('row2',)] rows = mock.db.query('SELECT * FROM table') print(rows) # [('row1',), ('row2',)]
# side_effect 的各种用法 from unittest.mock import Mock # 1. side_effect 为异常 mock = Mock() mock.fetch.side_effect = ConnectionError('API不可达') try: mock.fetch() except ConnectionError as e: print(e) # API不可达 # 2. side_effect 为可迭代对象(每次调用返回下一个值) mock = Mock() mock.next_value.side_effect = [10, 20, 30, StopIteration] print(mock.next_value()) # 10 print(mock.next_value()) # 20 print(mock.next_value()) # 30 # 3. side_effect 为函数(动态计算返回值) mock = Mock() mock.calc.side_effect = lambda x: x * 2 print(mock.calc(5)) # 10 print(mock.calc(100)) # 200
# name 参数在调试中的作用 from unittest.mock import Mock mock_a = Mock(name='db_connection') mock_b = Mock(name='cache_client') mock_a.query.return_value = 'db result' mock_b.get.return_value = 'cache result' # 如果测试失败,name帮助快速定位是哪个Mock print(repr(mock_a)) # <Mock name='db_connection' id='...'> print(repr(mock_b)) # <Mock name='cache_client' id='...'> # 设置属性 mock_a.host = 'localhost' mock_a.port = 5432 print(mock_a.host, mock_a.port) # localhost 5432

三、断言方法

Mock对象的断言方法是验证测试行为正确性的关键工具。当一个Mock对象被调用后,它会记录每一次调用的详细信息,包括调用的参数、调用顺序、调用次数等。断言方法就是对记录进行验证的接口。最基本的断言包括assert_called()(验证至少被调用过一次)、assert_called_once()(验证恰好被调用一次)、assert_called_with(*args, **kwargs)(验证最近一次调用使用了指定参数)和assert_called_once_with(*args, **kwargs)(验证恰好调用一次且参数匹配)。这些断言方法在参数不匹配时会抛出AssertionError,并提供详细的失败信息,包括实际调用记录与期望调用的对比。

除了上述基本断言,assert_any_call(*args, **kwargs)用于验证Mock曾在任何时候以指定参数被调用过(不关心调用次数和顺序),这在测试一个方法被多次调用且只关心某次特定调用的场景下非常实用。assert_not_called()则用于验证Mock从未被调用过,常用于验证某些条件分支下的代码路径未被触发。此外,call对象是一个辅助工具,它可以与mock_callsmethod_calls等属性结合使用,对多次调用进行精确的顺序验证。

理解断言方法的失败信息对调试非常有帮助。当断言失败时,unittest.mock会打印出详细的调用记录对比,显示"预期调用"和"实际调用"之间的差异。例如,assert_called_with失败时会列出所有实际调用及其参数,方便快速定位问题。在实际项目中,推荐在关键的交互边界上使用较为严格的断言(如assert_called_once_with),在辅助路径上使用较宽松的断言(如assert_called),以达到测试严格性与可维护性之间的平衡。

from unittest.mock import Mock, call mock = Mock() mock.process(1) mock.process(2, key='value') # 基本断言 mock.process.assert_called() # 通过:至少调用过一次 mock.process.assert_called_once() # 失败:调用了两次 mock.process.assert_called_with(2, key='value') # 通过:最近一次参数匹配 # 查看实际调用记录 print(mock.mock_calls) # [call.process(1), call.process(2, key='value')] # assert_any_call mock.process(1) mock.process.assert_any_call(1) # 通过:参数(1)的调用确实存在
# assert_not_called 和 assert_called_once_with from unittest.mock import Mock def test_conditional_branch(): mock = Mock() def process(is_admin): if is_admin: mock.admin_action() mock.common_action() process(is_admin=False) mock.admin_action.assert_not_called() # 验证未进入管理员分支 mock.common_action.assert_called_once() # 验证公共动作执行了一次 test_conditional_branch()
# call 对象和多次调用顺序验证 from unittest.mock import Mock, call mock = Mock() mock.start() mock.update('config') mock.stop() # 验证整体调用顺序 expected_calls = [ call.start(), call.update('config'), call.stop(), ] assert mock.mock_calls == expected_calls # 严格的顺序验证 # 过滤特定方法的调用记录 mock2 = Mock() mock2.open('file1') mock2.write('data1') mock2.open('file2') mock2.write('data2') open_calls = [c for c in mock2.mock_calls if c[0] == 'open'] print(open_calls) # [call.open('file1'), call.open('file2')]

四、patch方法

patch是unittest.mock模块中最核心也最强大的功能,它允许在测试期间临时替换目标对象(通常是模块或类中的某个属性),测试结束后自动恢复。patch支持多种使用风格:作为装饰器、作为上下文管理器、或通过start/stop方法手动控制生命周期。patch的目标路径使用点号分隔的字符串,例如'module.ClassName.method_name'。需要特别注意:patch的路径应该指向对象被导入和使用的位置(即测试代码中被测试模块的命名空间),而不是对象定义的位置,这是一个常见的容易出错的点。

patch.object用于直接替换指定对象上的属性,适合已知对象引用的场景。patch.multiple允许同时替换一个对象上的多个属性,一次patch操作即可完成多个模拟资源的设置。patch.dict专门用于临时修改字典内容,如环境变量os.environ或配置字典,它会在测试结束后自动恢复原始内容。这些变体共同构成了完整的patch工具集,覆盖了几乎所有需要临时替换的场景。

理解patched对象(即patch注入的参数)的引用方式很重要。当patch作为装饰器使用时,patched对象作为额外参数传递给被装饰函数,参数顺序与patch装饰器的顺序相反(从下往上)。如果使用上下文管理器方式,patcher.start()返回patched对象。测试完成后,务必确保patch被正确停止——上下文管理器和装饰器自动处理停止,而手工start/stop方式需要在teardown中显式调用stop()或在测试类中使用stopall()

from unittest.mock import patch import os # 方式1:装饰器风格 @patch('os.getenv') def test_with_decorator(mock_getenv): mock_getenv.return_value = 'mocked_value' result = os.getenv('DATABASE_URL') assert result == 'mocked_value' # 方式2:上下文管理器风格 def test_with_context_manager(): with patch('os.getenv') as mock_getenv: mock_getenv.return_value = 'mocked_value' result = os.getenv('DATABASE_URL') assert result == 'mocked_value' # 方式3:手动start/stop(适用于setUp/tearDown) def test_manual(): patcher = patch('os.getenv') mock_getenv = patcher.start() mock_getenv.return_value = 'mocked_value' result = os.getenv('DATABASE_URL') assert result == 'mocked_value' patcher.stop()
# patch.object 和 patch.multiple from unittest.mock import patch, Mock import time # patch.object:替换指定对象上的方法 @patch.object(time, 'sleep') def test_sleep(mock_sleep): time.sleep(5) mock_sleep.assert_called_once_with(5) # patch.multiple:同时替换多个属性 @patch.multiple('os.path', exists=Mock(return_value=True), isfile=Mock(return_value=True)) def test_path_checks(exists, isfile): import os.path assert os.path.exists('/any/path') is True assert os.path.isfile('/any/file') is True exists.assert_called_once_with('/any/path') isfile.assert_called_once_with('/any/file')
# patch.dict 临时修改字典/环境变量 from unittest.mock import patch import os @patch.dict('os.environ', {'DEBUG': 'true', 'DB_URL': 'sqlite:///test.db'}) def test_with_env(): assert os.environ['DEBUG'] == 'true' assert os.environ['DB_URL'] == 'sqlite:///test.db' # 也可以只添加/更新部分键 def test_partial_env(): with patch.dict('os.environ', {'MY_FLAG': '1'}, clear=False): original_path = os.environ.get('PATH') # 仍然存在 assert os.environ['MY_FLAG'] == '1' # 测试结束后 MY_FLAG 被移除 # 多个patch装饰器的顺序(从下往上) @patch('module_a.func') @patch('module_b.func') def test_multi_patch(mock_b_func, mock_a_func): # 注意:参数顺序与装饰器顺序相反 # 最靠近函数的装饰器先注入,所以mock_b_func对应module_b.func pass

五、MagicMock与特殊方法

MagicMockMock的子类,最大的区别是它预置了所有Python魔术方法(magic methods)的默认实现。魔术方法是以双下划线开头和结尾的特殊方法,如__len____iter____enter____exit____getitem__等。使用普通的Mock对象时,这些魔术方法不会被自动模拟,导致代码在调用len(mock_obj)for x in mock_obj等操作时会失败。而MagicMock预先为这些魔术方法配置了合理的默认返回值,使得模拟对象可以像真实对象一样参与Python的各种语法糖操作。

模拟上下文管理器(支持with语句)是MagicMock的典型应用场景。当需要模拟一个文件打开操作或数据库连接时,被测试代码通常会使用with语句。MagicMock自动支持__enter____exit__方法,其中__enter__返回自身(这使得可以链式配置返回值的属性)。同样,模拟可迭代对象(__iter__)和序列访问(__getitem____setitem__)也变得非常自然。在需要模拟这些特殊行为的测试中,使用MagicMock可以显著减少手动配置量。

属性访问模拟是另一个重要功能。Python的属性访问机制包括__getattr____setattr____getattribute__。Mock对象默认的__getattr__行为是动态创建子Mock,这意味着访问任何不存在的属性都不会报错,而是返回一个新的Mock。这在某些场合方便(访问深层嵌套属性),但在另一些场合可能隐藏错误。通过spec参数或PropertyMock可以精确控制属性访问行为。使用PropertyMock结合property()函数,可以让Mock属性在访问时触发特定逻辑。

from unittest.mock import MagicMock # MagicMock 自动支持魔术方法 mock = MagicMock() # __len__ mock.__len__.return_value = 42 print(len(mock)) # 42 # __getitem__ / __setitem__ mock.__getitem__.return_value = 'mocked_item' print(mock[0]) # 'mocked_item' mock['key'] = 'value' # 自动记录调用 # __iter__ mock.__iter__.return_value = iter([1, 2, 3]) result = list(mock) print(result) # [1, 2, 3]
# 模拟上下文管理器(with语句) from unittest.mock import MagicMock, patch # MagicMock自动支持__enter__/__exit__ mock_file = MagicMock(name='file') mock_file.read.return_value = 'file content' # open 返回 MagicMock,其 __enter__ 返回自身 with patch('builtins.open', return_value=mock_file): with open('test.txt') as f: content = f.read() assert content == 'file content' # 手动验证上下文管理器的调用 mock_file.__enter__.assert_called_once() mock_file.__exit__.assert_called_once() mock_file.read.assert_called_once()
# PropertyMock 模拟属性 from unittest.mock import MagicMock, PropertyMock, patch class User: @property def name(self): return 'real_name' # 方式1:使用PropertyMock替换类属性 with patch('__main__.User.name', new_callable=PropertyMock) as mock_name: mock_name.return_value = 'mocked_name' user = User() print(user.name) # 'mocked_name' # 验证属性被访问 mock_name.assert_called_once() # 方式2:在MagicMock中使用 obj = MagicMock() type(obj).computed_value = PropertyMock(return_value=42) print(obj.computed_value) # 42

六、spec与配置

Mock对象默认的"接受一切"特性虽然灵活,但也带来了风险:测试中访问了不存在的属性或调用了不存在的方法不会被及时发现,导致测试通过而生产代码运行时才报错。spec参数正是为了解决这个问题而设计的。当创建Mock时传入spec参数(可以是一个类、一个对象或一个字符串列表),Mock会限制只能访问spec中定义的属性和方法。通过这种方式,Mock既提供了模拟功能,又保留了对接口的契约检查,确保模拟对象的行为与真实对象一致。

spec_setspec的严格版本:它不仅限制属性的读取,还限制属性的写入——尝试设置spec_set中不存在的属性会抛出AttributeError。这在需要确保测试不会意外地给模拟对象附加新属性时非常有用。而autospec则更进一步:它自动从被patch的真实对象推断出完整的接口签名,包括参数名称和默认值。使用autospec后,即使不显式指定spec,Mock也会自动匹配真实函数的签名,并以正确的参数调用验证。autospec是生产级测试的首选配置,因为它最大程度地降低了测试与真实代码之间的偏差。

wraps参数提供了另一种配置思路:它包装一个真实对象,未模拟的调用会透传到真实对象执行,而已模拟的调用则返回模拟结果。这在需要保留部分真实行为、只模拟特定方法的场景下非常有用。例如,可以创建一个包装了真实数据库连接对象的Mock,只模拟close()方法但不模拟execute()方法,这样测试既能验证关闭操作,又能真实地执行查询。三种配置方式(spec、autospec、wraps)各有适用场景,合理组合使用可以构建出既灵活又严格的测试体系。

from unittest.mock import Mock # spec 限制接口访问 class Database: def connect(self): pass def query(self, sql): pass def close(self): pass mock_db = Mock(spec=Database) mock_db.connect() # 允许 mock_db.query('SQL') # 允许 mock_db.unknown() # 抛出 AttributeError # spec_set 严格模式(禁止写入未定义属性) mock_db2 = Mock(spec_set=Database) mock_db2.connect = lambda: None # 允许:connect 在 spec 中 mock_db2.new_attr = 42 # 抛出 AttributeError
# autospec 自动适配真实函数签名 from unittest.mock import create_autospec, patch def real_function(a, b, c=None): return a + b # create_autospec 创建自动适配的Mock mock_fn = create_autospec(real_function, return_value=10) print(mock_fn(1, 2)) # 10 print(mock_fn(1, 2, c=3)) # 10 try: mock_fn(1, 2, 3, 4) # 抛出 TypeError:参数过多 except TypeError as e: print(e) # patch 配合 autospec 参数 @patch('module.real_function', autospec=True) def test_with_autospec(mock_func): mock_func.return_value = 42 result = mock_func(1, 2) assert result == 42
# wraps 包装真实对象 from unittest.mock import Mock class Calculator: def add(self, a, b): return a + b def multiply(self, a, b): return a * b real_calc = Calculator() # 只模拟 multiply,保留 add 的真实行为 mock_calc = Mock(wraps=real_calc) mock_calc.multiply.return_value = 999 print(mock_calc.add(2, 3)) # 5(透传到真实方法) print(mock_calc.multiply(2, 3)) # 999(返回模拟值) # 验证真实方法也被调用了 mock_calc.add.assert_called_once_with(2, 3) mock_calc.multiply.assert_called_once_with(2, 3)

七、高级Mock技术

在实际项目中,测试场景往往比简单的函数调用复杂得多。链式调用是常见模式之一,例如response.json()['data'][0]['id']这样的深层访问。Mock通过其惰性创建机制天然支持链式调用:访问mock.a.b.c会自动创建中间Mock对象,每个中间步骤都返回一个新的Mock。要配置链式调用的返回值,可以使用mock.a.b.c.return_value = value,或者使用mock.configure_mock()批量配置。更简洁的方式是使用return_value嵌套赋值,例如mock.json.return_value = {'data': [{'id': 1}]}

属性访问控制和__getattr__模拟是高级Mock技术的另一个重要方面。Mock对象默认的__getattr__行为是为任何未定义的属性创建一个新的子Mock并返回。如果想改变这个行为,可以自定义__getattr__的side_effect,或者在创建Mock时使用spec限制。例如,可以实现一个"只读"Mock,访问未定义属性时抛出异常而不是静默返回子Mock。此外,通过side_effect动态计算返回值是实现复杂行为的关键技术——你可以传入一个函数,根据调用参数的不同返回不同的结果,实现状态依赖的模拟行为。

异常模拟和回调模拟也是测试中常见的需求。模拟异常抛出使用side_effect = SomeException('message'),可以测试代码的错误处理路径。回调模拟则适用于被测试代码注册了回调函数然后触发它的场景——通过side_effect函数,可以在Mock被调用时执行自定义逻辑,包括收集参数、修改状态、记录日志等。结合wrapsside_effect,可以实现非常精细的模拟控制,几乎能模拟任何真实场景中的交互模式。

from unittest.mock import Mock # 链式调用模拟 mock_response = Mock() mock_response.json.return_value = { 'data': [ {'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}, ] } # 被测试代码中的链式调用 data = mock_response.json()['data'] user_id = data[0]['id'] print(user_id) # 1 # 深层嵌套链式调用 mock = Mock() mock.a.b.c.d.return_value = 'deep_value' print(mock.a.b.c.d()) # 'deep_value' # configure_mock 批量配置 mock2 = Mock() mock2.configure_mock( host='localhost', port=8080, timeout=30, ) print(mock2.host, mock2.port) # localhost 8080
# side_effect 动态计算和状态管理 from unittest.mock import Mock # 根据参数动态返回不同结果 mock = Mock() def dynamic_response(url): if 'users' in url: return {'status': 200, 'data': [{'id': 1}]} elif 'posts' in url: return {'status': 200, 'data': [{'title': 'Hello'}]} else: return {'status': 404} mock.fetch.side_effect = dynamic_response print(mock.fetch('/api/users')) # {'status': 200, 'data': [{'id': 1}]} print(mock.fetch('/api/unknown')) # {'status': 404} # 回调模拟(模拟事件监听器) handler = Mock() def trigger_event(data): handler(data) trigger_event({'type': 'click', 'x': 10, 'y': 20}) handler.assert_called_once_with({'type': 'click', 'x': 10, 'y': 20})
# 异常模拟和__getattr__模拟 from unittest.mock import Mock # 模拟异常 mock_api = Mock() mock_api.get_user.side_effect = TimeoutError('请求超时') try: mock_api.get_user(123) except TimeoutError as e: print(f'捕获异常: {e}') # 自定义 __getattr__ 控制属性访问 class StrictMock(Mock): def __getattr__(self, name): if name.startswith('_'): raise AttributeError(f'不允许访问 {name}') return super().__getattr__(name) strict = StrictMock() strict.allowed_attr = 42 print(strict.allowed_attr) # 42 try: _ = strict._private # 抛出 AttributeError except AttributeError as e: print(f'拒绝访问: {e}')

八、清理与重置

在测试生命周期管理中,Mock的清理和重置是一个容易被忽视但非常重要的环节。如果一个Mock对象在多个测试用例中被复用,前一个测试中的调用记录会污染后一个测试的判断——调用计数、参数记录等都是累积的,导致断言在不经意间失效或通过。使用reset_mock()方法可以清除Mock对象的所有调用记录(包括mock_callsmethod_callscall_args等),同时也可以选择性地重置return_valueside_effect配置。推荐在每个测试用例的setup阶段创建新的Mock,或者在teardown阶段调用reset_mock,确保测试之间的隔离性。

attach_mock是一种高级Mock管理技术,它允许将一个子Mock"附加"到父Mock的某个属性上,同时让子Mock的调用记录合并到父Mock的method_calls中。这在需要统一跟踪多个相关Mock的调用顺序时非常有用。当被测试代码通过不同的路径调用不同的Mock方法时,通过父Mock可以观察全局调用顺序。此外,stopall()函数是unittest.mock提供的批量停止工具——它可以一次性停止所有通过patcher.start()启动的patch,在复杂的setUp/tearDown模式中特别有用,避免忘记手动停止某个patch而导致的测试间泄漏。

清理注意事项中最关键的一点是:永远不要在生产代码路径中创建Mock对象。Mock是测试工具,应该只在测试代码中使用。另一个常见问题是patch的作用域——如果两个测试用例使用相同的patch路径但期望不同的行为,务必确保每个测试用例独立配置Mock的返回值和副作用,而不是在模块级别共享Mock配置。最后,使用pytest-mockmocker fixture时,清理是自动的——每个测试结束后,mocker会自动停止所有通过它启动的patch,这是推荐使用它的原因之一。

from unittest.mock import Mock # reset_mock 的基本使用 mock = Mock() mock.get_data() mock.get_data() print(mock.get_data.call_count) # 2 mock.reset_mock() print(mock.get_data.call_count) # 0(调用记录已清除) # reset_mock 同时可以重置 return_value 和 side_effect mock.return_value = 42 mock.side_effect = ValueError('test') mock.reset_mock(return_value=True, side_effect=True) print(mock.return_value) # 新的 Mock 对象(已重置) print(mock.side_effect) # None(已重置)
# attach_mock 合并调用记录 from unittest.mock import Mock, call parent = Mock() child_db = Mock(name='db') child_cache = Mock(name='cache') parent.attach_mock(child_db, 'db') parent.attach_mock(child_cache, 'cache') # 被测试代码通过不同路径调用 child_db.query('SELECT 1') child_cache.get('key') child_db.close() # 通过 parent 可以观察到全局调用顺序 print(parent.mock_calls) # [call.db.query('SELECT 1'), call.cache.get('key'), call.db.close()] # stopall 批量停止所有patch from unittest.mock import patch def test_with_stopall(): patcher1 = patch('os.getenv') patcher2 = patch('os.path.exists') patcher1.start() patcher2.start() try: # 测试逻辑... pass finally: # 确保无论如何都会停止 patch.stopall()
# 测试类中的清理最佳实践 from unittest.mock import Mock, patch import unittest class TestService(unittest.TestCase): def setUp(self): self.mock_db = Mock(name='db') self.mock_cache = Mock(name='cache') self.patcher_db = patch('myapp.database.Connection', return_value=self.mock_db) self.patcher_cache = patch('myapp.cache.Client', return_value=self.mock_cache) self.patcher_db.start() self.patcher_cache.start() def tearDown(self): # 清理所有patch patch.stopall() # 或者逐个停止:self.patcher_db.stop() # 重置所有Mock self.mock_db.reset_mock() self.mock_cache.reset_mock() def test_query_user(self): self.mock_db.query.return_value = {'id': 1, 'name': 'Alice'} # 测试逻辑... self.mock_db.query.assert_called_once()

九、实战案例

前面的章节涵盖了Mock的各项技术细节,本节通过三个完整的实战案例展示如何将这些技术综合应用于真实的测试场景中。每个案例都模拟了开发中常见的依赖类型:外部HTTP API、数据库查询、以及文件系统操作。通过这些案例,可以直观地理解Mock在实际项目中带来的价值——隔离测试、提高速度、增强稳定性。

案例一:模拟外部API调用。在微服务架构中,一个服务经常需要调用另一个服务的HTTP接口。在单元测试中,我们不应该真正发起网络请求——外部服务可能不可用、测试数据可能污染生产环境、网络延迟会拖慢测试速度。通过Mock替换requests.get或使用responses库,可以在毫秒级别完成测试,并且完全控制返回的数据。下面的示例展示了如何模拟一个天气查询服务的API调用,涵盖成功响应、错误响应和超时三种场景。

案例二:模拟数据库查询。数据库操作是另一个常见的需要模拟的场景。在实际项目中,数据库可能包含大量数据,建立测试数据库需要额外的维护工作。通过Mock替换数据库连接或ORM的查询方法,可以验证业务逻辑正确处理了查询结果。示例中模拟了SQLAlchemy的session查询,验证用户服务层的查询和创建逻辑是否正确。关键点在于只模拟数据库交互层,而非整个应用——这样既达到了隔离目的,又保持了较高的测试覆盖率。

案例三:模拟文件系统操作。文件读取和写入是许多应用的基础功能,但在测试中直接读写磁盘文件会引入环境依赖和状态污染。通过Mock替换open内置函数或使用pyfakefs/tempfile,可以在内存中模拟文件操作,测试完成后不需要清理临时文件。示例展示了如何测试一个日志解析器函数,它需要读取日志文件并解析其中的关键信息。通过Mock控制文件内容,可以轻松测试各种日志格式和边界情况。

# 案例一:模拟外部API调用 from unittest.mock import Mock, patch import json # 被测试代码:天气查询服务 def get_weather(city): import requests url = f'https://api.weather.com/v1/{city}' resp = requests.get(url, timeout=5) data = resp.json() return { 'city': data['location'], 'temperature': data['current']['temp_c'], 'humidity': data['current']['humidity'], } # 测试代码 @patch('requests.get') def test_get_weather_success(mock_get): # 模拟成功的API响应 mock_response = Mock() mock_response.json.return_value = { 'location': 'Beijing', 'current': {'temp_c': 22, 'humidity': 65}, } mock_get.return_value = mock_response result = get_weather('Beijing') assert result == {'city': 'Beijing', 'temperature': 22, 'humidity': 65} mock_get.assert_called_once_with( 'https://api.weather.com/v1/Beijing', timeout=5 ) @patch('requests.get') def test_get_weather_timeout(mock_get): # 模拟超时异常 mock_get.side_effect = TimeoutError('连接超时') try: get_weather('Unknown') except TimeoutError: pass # 期望捕获异常
# 案例二:模拟数据库查询 from unittest.mock import Mock, patch, MagicMock from dataclasses import dataclass # 被测试代码:用户服务 @dataclass class User: id: int name: str email: str class UserService: def __init__(self, db_session): self.db = db_session def get_user(self, user_id): user = self.db.query(User).filter_by(id=user_id).first() if not user: raise ValueError(f'用户 {user_id} 不存在') return {'id': user.id, 'name': user.name, 'email': user.email} def create_user(self, name, email): user = User(id=None, name=name, email=email) self.db.add(user) self.db.commit() return user # 测试代码 def test_user_service_get_user(): mock_session = MagicMock() mock_user = User(id=1, name='Alice', email='alice@test.com') mock_session.query.return_value.filter_by.return_value.first.return_value = mock_user service = UserService(mock_session) result = service.get_user(1) assert result == {'id': 1, 'name': 'Alice', 'email': 'alice@test.com'} def test_user_service_user_not_found(): mock_session = MagicMock() mock_session.query.return_value.filter_by.return_value.first.return_value = None service = UserService(mock_session) try: service.get_user(999) assert False, '应该抛出异常' except ValueError as e: assert str(e) == '用户 999 不存在' def test_user_service_create_user(): mock_session = MagicMock() service = UserService(mock_session) user = service.create_user('Bob', 'bob@test.com') mock_session.add.assert_called_once() mock_session.commit.assert_called_once()
# 案例三:模拟文件系统操作 from unittest.mock import patch, mock_open # 被测试代码:日志解析器 def parse_log_file(filepath): with open(filepath, 'r') as f: lines = f.readlines() errors = [] warnings = [] for line in lines: if '[ERROR]' in line: errors.append(line.strip()) elif '[WARNING]' in line: warnings.append(line.strip()) return { 'total_lines': len(lines), 'errors': errors, 'warnings': warnings, 'error_count': len(errors), 'warning_count': len(warnings), } # 测试代码 def test_parse_log_file(): mock_content = """[INFO] 服务启动 [ERROR] 数据库连接失败 [INFO] 重试连接 [WARNING] 连接超时 [ERROR] 磁盘空间不足 """ m = mock_open(read_data=mock_content) with patch('builtins.open', m): result = parse_log_file('/var/log/app.log') assert result['total_lines'] == 5 assert result['error_count'] == 2 assert result['warning_count'] == 1 assert '[ERROR] 数据库连接失败' in result['errors'] assert '[WARNING] 连接超时' in result['warnings'] # 验证文件是用正确的路径和模式打开的 m.assert_called_once_with('/var/log/app.log', 'r') def test_parse_log_file_empty(): m = mock_open(read_data='') with patch('builtins.open', m): result = parse_log_file('/var/log/empty.log') assert result['total_lines'] == 0 assert result['error_count'] == 0 def test_parse_log_file_no_errors(): m = mock_open(read_data='[INFO] 正常运行\n[INFO] 一切正常\n') with patch('builtins.open', m): result = parse_log_file('/var/log/clean.log') assert result['total_lines'] == 2 assert result['error_count'] == 0 assert result['warning_count'] == 0