为什么需要 pytest？
Pytest 的设计哲学是：让测试代码回归普通 Python 逻辑，把精力放在业务断言上，而不是框架的条条框框。
相比 unittest 的显式继承（class TestXxx(unittest.TestCase)），Pytest 采用约定大于配置：只要文件名是 test*.py 或 *_test.py，函数名是 test*，就能自动发现并执行。函数就是最天然的测试单元。
断言也更简洁 —— 只需一个原生 assert：
1
2
3
4
5
|
# unittest 写法
self.assertEqual(result["status"], 200, f"期望200，实际得到{result['status']}")
# Pytest 写法（失败时自动拆解表达式，给出详细上下文）
assert result["status"] == 200
|
替身演员：Mock 与 Patch
Mock：制造空壳对象
Mock 是替身演员，用来替代真实对象。只有 Mock 对象才有 .return_value、.side_effect 等属性。
1
2
3
4
5
6
7
8
9
10
11
12
|
from unittest.mock import Mock
# ❌ 真实对象没有 return_value
real_list = [1, 2, 3]
real_list.append.return_value = "ok" # 报错
# ✅ Mock 对象可以随意设定
mock_list = Mock()
mock_list.append.return_value = "ok"
result = mock_list.append(1) # 返回 "ok"，不会真的追加
mock_list.append.assert_called_once_with(1) # 验证调用
|
patch：临时替换真实对象
Mock 只是造了替身，但代码里写死的是真实类怎么办？
patch 的作用：在指定作用域内，把目标模块里的某个名字临时替换成你的 Mock，作用域结束立即恢复。
1
2
3
4
5
6
7
8
9
10
11
12
|
from unittest.mock import patch
# 装饰器用法
@patch('module.ClassName')
def test_something(MockClass):
instance = MockClass.return_value
instance.method.return_value = "mocked"
# 上下文管理器用法
def test_something():
with patch('module.ClassName') as MockClass:
instance = MockClass.return_value
|
pytest Fixture（夹具）
如果说 Mock 是解决“依赖隔离”的利器，那么 Fixture 就是解决“测试数据准备与清理”的终极武器。它是 Pytest 灵魂所在
替代传统的 Setup/Teardown
当你需要为每个测试准备相同的数据库连接时
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
import pytest
# 定义一个 Fixture（本质就是一个带有 @pytest.fixture 装饰器的函数）
@pytest.fixture
def db_connection():
print("\n[Setup] 连接数据库...")
conn = "模拟的数据库连接对象"
yield conn # yield 之前的代码是 Setup，yield 出去的值就是注入给测试函数的数据
print("\n[Teardown] 关闭数据库连接...")
# 测试函数：参数名与 Fixture 名字一致，Pytest 会自动注入
def test_query_user(db_connection):
assert db_connection == "模拟的数据库连接对象"
# 执行完后，自动执行 yield 后面的 Teardown 逻辑
|
理解 Fixture 的作用域
如果你的测试有 1000 个，每个都去连一次数据库，测试会慢到无法忍受。作用域决定了 Fixture 的生命周期。
1
2
3
4
5
6
7
8
9
10
11
12
13
|
# scope="session"：整个测试运行期间只执行一次 Setup 和 Teardown
# scope="module"：当前 .py 文件只执行一次
# scope="class"：当前测试类只执行一次
# scope="function"：（默认值）每个测试函数都执行一次
@pytest.fixture(scope="session")
def global_redis():
r = "启动沉重的 Redis 模拟"
yield r
r = "销毁 Redis"
def test_redis_set(global_redis): ...
def test_redis_get(global_redis): ...
# 上面两个函数共享同一个 global_redis 实例，极大地提升了速度
|
Fixture 的组合与依赖注入
1
2
3
4
5
6
7
8
9
10
11
12
13
|
@pytest.fixture
def mock_user():
return {"id": 1, "name": "张三"}
@pytest.fixture
def mock_db(mock_user):
# 依赖注入：直接把 mock_user 作为参数拿过来用
db = {"users": [mock_user]}
return db
def test_user_in_db(mock_db):
# 测试函数只管拿最终结果
assert len(mock_db["users"]) == 1
|
隐藏的管家：conftest.py
随着项目变大，你的 Fixture 可能会被多个测试文件复用。如果每个文件都 import，会导致循环引用等问题。
Pytest 提供了一个特殊的文件：conftest.py。
你可以在每个测试文件夹下放一个 conftest.py。
不需要 import，同目录及子目录下的所有 test_*.py 都能直接使用里面定义的 Fixture。
它可以覆盖：比如根目录有一个全局的 db_connection，某个子模块的 conftest.py 里也定义了一个同名的 db_connection，子模块会优先使用自己的。
1
2
3
4
5
6
|
tests/
├── conftest.py (定义全局的 mock_engine)
├── test_login.py (直接使用全局 mock_engine)
└── order/
├── conftest.py (定义局部的 order_mock_engine)
└── test_order.py (使用局部的 order_mock_engine)
|
消灭重复代码：参数化（parametrize）
在实际业务中，一个校验规则往往需要测试十几组正反例。如果写十几个 test_xxx 函数，代码会非常臃肿。@pytest.mark.parametrize 就是用来消灭这种重复代码的。
场景：测试一个计算订单折扣的函数。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
# 被测代码
def calculate_discount(total_amount, vip_level):
if total_amount < 0:
raise ValueError("金额不能为负")
if vip_level == 3:
return total_amount * 0.7
if vip_level == 2 and total_amount >= 100:
return total_amount * 0.9
return total_amount
# 测试代码：使用参数化
import pytest
# 参数化装饰器：第一个字符串定义变量名，后面是测试数据列表（元组形式）
@pytest.mark.parametrize("amount, vip, expected", [
(100, 3, 70.0), # 正常：SVIP 打 7 折
(200, 2, 180.0), # 正常：VIP 满 100 打 9 折
(50, 2, 50.0), # 边界：VIP 不足 100 不打折
(0, 1, 0.0), # 边界：0 元
])
def test_calculate_discount_normal(amount, vip, expected):
assert calculate_discount(amount, vip) == expected
# 测试异常情况：配合 pytest.raises
@pytest.mark.parametrize("amount, vip, error_msg", [
(-1, 1, "金额不能为负"),
])
def test_calculate_discount_exception(amount, vip, error_msg):
with pytest.raises(ValueError, match=error_msg):
calculate_discount(amount, vip)
|
运行上述代码时，Pytest 会在底层自动把这一组数据拆分成 5 个独立的测试用例执行，任何一个失败都不影响其他用例的断言。
入门实践
Step 1: 先写最简单的测试
1
2
3
4
5
6
7
8
9
10
11
12
|
# tests/unit/test_simple.py
import pytest
def test_request_validator_simple():
from your_module import RequestValidator
validator = RequestValidator()
assert validator.validate_get_order_type_msg("CGD20250101") == "purchase"
assert validator.validate_get_order_type_msg("WW20250101") == "outsourcing"
with pytest.raises(Exception, match="未知单据类别"):
validator.validate_get_order_type_msg("INVALID")
|
Step 2: 使用 fixture 简化测试数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
# tests/conftest.py
import pytest
@pytest.fixture
def simple_purchase_request():
return {
"wmsOrderCode": "CGD20250101",
"u8Code": "PO20250101",
"u8Id": "100001",
"operationType": "1",
}
# tests/unit/test_validator.py
def test_validate_initiation_data_simple(simple_purchase_request):
from your_module import RequestValidator
validator = RequestValidator()
validator.validate_initiation_data(simple_purchase_request) # 不抛异常即可
|
Step 3: 测试有外部依赖的类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
# tests/unit/test_order_checker.py
from unittest.mock import Mock, patch
import pandas as pd
def test_order_status_checker_with_mock():
"""测试OrderStatusChecker - 使用Mock"""
from your_module import OrderStatusChecker
# 创建Mock的数据库引擎
mock_engine = Mock()
# 创建检查器实例
with patch('your_module.sql_engine', mock_engine), \
patch('your_module.flwms_engine', mock_engine):
checker = OrderStatusChecker()
# Mock查询方法
with patch.object(checker, 'query_u8_stockin_qty') as mock_query:
# 设置返回值
mock_query.return_value = pd.DataFrame({
'POID': ['100001'],
'iReceivedQTY': [10.0]
})
# 执行测试
result = checker.check_u8_stockin_status("100001")
# 验证
assert result is True
mock_query.assert_called_once_with(u8_id="100001")
|
最佳实践
将复杂的测试方法拆分成更小的方法单独测试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
# 假设我们把复杂方法拆分成小方法
class PurchaseOmModificationApplicationService:
def _handle_initiate_request(self, req):
"""这个大方法可以拆分成多个小方法"""
# 1. 验证请求
self._validate_initiate_request(req)
# 2. 检查状态
self._check_order_status(req)
# 3. 准备审批数据
approval_data = self._prepare_approval_data(req)
# 4. 发起审批
result = self._initiate_approval(approval_data)
# 5. 处理后续操作
self._post_approval_actions(req, result)
return result
# 这些小方法可以单独测试
def _validate_initiate_request(self, req):
pass
def _check_order_status(self, req):
pass
def _initiate_approval(self, approval_data):
pass
def _post_approval_actions(self, req, result):
pass
|
然后分别测试每个小方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
class TestPurchaseServiceSmallMethods:
"""测试拆分后的小方法"""
def test_validate_initiate_request(self):
"""测试验证请求"""
with patch.object(PurchaseOmModificationApplicationService, '__init__', return_value=None):
service = PurchaseOmModificationApplicationService()
service.request_validator = Mock()
# 测试数据
valid_request = {"wmsOrderCode": "CGD20250101", "operationType": "1"}
# 执行
service._validate_initiate_request(valid_request)
# 验证
service.request_validator.validate_initiation_data.assert_called_once_with(valid_request)
def test_check_order_status_no_stockin(self):
"""测试检查订单状态（无入库）"""
with patch.object(PurchaseOmModificationApplicationService, '__init__', return_value=None):
service = PurchaseOmModificationApplicationService()
service.status_checker = Mock()
service.request_validator = Mock()
# 设置Mock
service.status_checker.check_u8_stockin_status.return_value = False
service.status_checker.check_saleorder_audit_status.return_value = False
service.request_validator.validate_get_order_type_msg.return_value = "purchase"
# 测试数据
test_request = {
"wmsOrderCode": "CGD20250101",
"u8Id": "100001",
"u8Code": "PO20250101"
}
# 执行 - 应该不抛异常
service._check_order_status(test_request)
# 验证
service.status_checker.check_u8_stockin_status.assert_called_once_with("100001")
|
实战演练
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
|
# tests/unit/test_service_refactored.py
import pytest
from unittest.mock import Mock, patch
# ------------------------------------------------------------
# 测试1: 纯逻辑方法（无依赖）
# ------------------------------------------------------------
class TestStaticMethods:
"""测试静态方法"""
def test_process_dingtalk_form_data(self):
"""测试处理钉钉表单数据"""
from architecture.purchaseOmModificationPurchaseOmModification import PurchaseOmModificationApplicationService
# 测试数据
form_data = [
{"name": "类型", "value": "单据修改"},
{"name": "产品数据中心单据号", "value": "CGD20250101"},
{"name": "无名称字段", "value": "应该被忽略"}, # 没有name, 应该被忽略
]
# 执行静态方法
result = PurchaseOmModificationApplicationService._process_dingtalk_form_data(form_data)
# 验证
assert result["类型"] == "1" # "单据修改"应该转为"1"
assert result["产品数据中心单据号"] == "CGD20250101"
assert "无名称字段" not in result # 应该被过滤
# ------------------------------------------------------------
# 测试2: 简单类（依赖少）
# ------------------------------------------------------------
class TestRequestValidatorSimple:
"""简单测试RequestValidator"""
def test_order_type_validation(self):
"""测试单据类型验证"""
from architecture.purchaseOmModificationPurchaseOmModification import RequestValidator
validator = RequestValidator()
test_cases = [
("CGD20250101", "purchase"),
("cgd20250101", "purchase"), # 小写
("CGD", "purchase"), # 短格式
("WW20250101", "outsourcing"),
("ww20250101", "outsourcing"),
]
for input_code, expected in test_cases:
result = validator.validate_get_order_type_msg(input_code)
assert result == expected, f"输入: {input_code}, 期望: {expected}, 实际: {result}"
# 测试异常情况
with pytest.raises(Exception, match="未知单据类别"):
validator.validate_get_order_type_msg("INVALID")
# ------------------------------------------------------------
# 测试3: 有依赖的类（使用Mock）
# ------------------------------------------------------------
class TestOrderStatusCheckerMocked:
"""使用Mock测试OrderStatusChecker"""
@pytest.fixture
def mocked_checker(self):
"""创建带有Mock数据库的检查器"""
with patch('architecture.purchaseOmModificationPurchaseOmModification.sql_engine'), \
patch('architecture.purchaseOmModificationPurchaseOmModification.flwms_engine'):
from architecture.purchaseOmModificationPurchaseOmModification import OrderStatusChecker
checker = OrderStatusChecker()
# 替换查询方法为Mock
checker.query_u8_stockin_qty = Mock()
checker.check_wms_stockin_status = Mock()
checker.check_saleorder_audit_status = Mock()
return checker
def test_check_u8_stockin_with_data(self, mocked_checker):
"""测试有U8入库数据的情况"""
import pandas as pd
# 设置Mock返回值
mocked_checker.query_u8_stockin_qty.return_value = pd.DataFrame({
'POID': ['100001'],
'iReceivedQTY': [10.0] # 有入库数量
})
result = mocked_checker.check_u8_stockin_status("100001")
assert result is True
mocked_checker.query_u8_stockin_qty.assert_called_once_with(u8_id="100001")
def test_check_u8_stockin_no_data(self, mocked_checker):
"""测试无U8入库数据的情况"""
import pandas as pd
# 设置Mock返回值
mocked_checker.query_u8_stockin_qty.return_value = pd.DataFrame({
'POID': ['100001'],
'iReceivedQTY': [0.0] # 入库数量为0
})
result = mocked_checker.check_u8_stockin_status("100001")
assert result is False
# ------------------------------------------------------------
# 测试4: 主服务类（复杂, 需要大量Mock）
# ------------------------------------------------------------
class TestMainServiceIntegration:
"""集成测试主服务"""
def create_mocked_service(self):
"""创建完全Mock的服务"""
from unittest.mock import create_autospec
from architecture.purchaseOmModificationPurchaseOmModification import PurchaseOmModificationApplicationService
# 创建自动spec的Mock
service = create_autospec(PurchaseOmModificationApplicationService)
# 设置属性
service.request_validator = Mock()
service.status_checker = Mock()
service.dingtalk_service = Mock()
service.wms_action = Mock()
return service
def test_service_method_calls(self):
"""测试服务方法调用链"""
service = self.create_mocked_service()
# 测试数据
test_request = {
"wmsOrderCode": "CGD20250101",
"u8Id": "100001"
}
# 设置返回值
service.request_validator.validate_get_order_type_msg.return_value = "purchase"
service._handle_initiate_request.return_value = {
"code": 200,
"status": True
}
# 执行
result = service._handle_initiate_request(test_request)
# 验证
assert result["code"] == 200
service._handle_initiate_request.assert_called_once_with(test_request)
# ------------------------------------------------------------
# 测试5: 真实场景测试（使用patch管理依赖）
# ------------------------------------------------------------
def test_real_scenario_with_patches():
"""使用patch的真实场景测试"""
from architecture.purchaseOmModificationPurchaseOmModification import PurchaseOmModificationApplicationService
# 测试数据
test_request = {
"callType": "initiate",
"wmsOrderCode": "CGD20250101",
"u8Code": "PO20250101",
"u8Id": "100001",
"operationType": "1",
"isChangedPrice": False,
"isInternalSaleOrder": False,
"initiateName": "张三",
"initiateWorkNo": "ZS001",
"operationReason": "原因",
"operationDesc": "描述"
}
# 使用patch替换所有外部依赖
with patch('architecture.purchaseOmModificationPurchaseOmModification.generic_database_interact.DingdingGeneralService') as mock_dgs, \
patch('architecture.purchaseOmModificationPurchaseOmModification.common') as mock_common, \
patch('architecture.purchaseOmModificationPurchaseOmModification.WmsAction') as mock_wms, \
patch.object(PurchaseOmModificationApplicationService, '_handle_initiate_request') as mock_handle:
# 创建服务实例
service = PurchaseOmModificationApplicationService()
# 设置Mock返回值
mock_handle.return_value = {
"code": 200,
"status": True,
"message": "成功",
"data": {"instanceId": "123"}
}
# 执行主入口方法
result = service.purchase_out_modify_ding_approval(test_request)
# 验证
assert result["code"] == 200
assert result["status"] is True
mock_handle.assert_called_once_with(test_request)
if __name__ == "__main__":
pytest.main([__file__, "-v"])
|
运行测试的命令
1
2
3
4
5
6
7
8
9
10
11
|
# 只运行简单测试
pytest tests/unit/test_simple.py -v
# 运行所有测试
pytest tests/unit/ -v
# 运行并打印详细信息
pytest tests/unit/test_service_refactored.py::TestStaticMethods -v -s
# 遇到错误时进入pdb调试
pytest tests/unit/ -v --pdb
|
更进一步
可以去了解 Pytest 的 Plugin System（插件系统） 和 Hook 函数（如 pytest_configure、pytest_collection_modifyitems）
当我们有一天需要定制“测试失败时自动重试 2 次”或者“按特定顺序执行用例”这种底层逻辑时，Hook 函数是唯一的出路
目前的笔记主要集中在“单元测试”。接下来可以学习 pytest 的高级 Fixture 机制（fixture 的 scope 参数管理、yield 进行测试后清理），掌握如何编写优雅的集成测试
思考一个问题：如果把业务代码全部 Mock 掉，我们到底在测什么？
今后需要学习**“测试覆盖率”**的工具（如 coverage），明白“100% 覆盖率不等于没有 Bug”。学会在“Mock 掉外部世界”和“保留内部真实逻辑”之间寻找平衡点
记录生活