pytest本身不能直接模拟竞争条件,需配合threading/multiprocessing手动构造并发场景,如用Barrier同步线程触发高概率竞态,并通过多轮运行验证原子性修复效果。

pytest 能不能直接模拟竞争条件
不能。pytest 本身是单元测试框架,不提供线程/进程调度控制能力,无法精确触发竞态窗口。想用它测原子性,必须配合 threading 或 multiprocessing 手动构造并发场景,否则只是在跑单线程逻辑,完全测不出问题。
常见错误现象:pytest 里写一堆 Thread(target=inc).start() 就以为在测并发——其实线程启动顺序、执行时机全由 OS 决定,失败概率极低,且不可复现。
- 必须显式控制线程数量、启动节奏(比如用
threading.Barrier同步到同一时刻) - 避免依赖
time.sleep()做“等待”,它不保证精度,也不跨平台 - 测试断言要放在所有线程结束后,且检查最终状态是否符合原子操作预期(如:100 个线程各加 1,结果必须是 100)
用 threading + Barrier 触发高概率竞态
Barrier 是最可控的方式:让 N 个线程同时抵达临界区入口,极大增加对共享变量的读-改-写重叠概率。
示例场景:测试一个非线程安全的计数器类
import threading
<p>class Counter:
def <strong>init</strong>(self):
self.value = 0</p><pre class="brush:php;toolbar:false;">def increment(self):
self.value += 1 # 非原子:读取、+1、写回三步
快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构,支持CLI、API、爬虫、Bot、Django等多种项目类型,包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
对应测试:
import pytest
import threading
<p>def test_counter_racing():
counter = Counter()
barrier = threading.Barrier(10) # 10 个线程同步点</p><pre class="brush:php;toolbar:false;">def worker():
barrier.wait() # 全部卡在这里,直到第 10 个到达
counter.increment()
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
assert counter.value == 10 # 大概率失败:实际常为 7~9
- Barrier 数量必须等于线程数,否则
wait()会永久阻塞 - 不要在
worker中捕获异常——竞态导致的逻辑错误不会抛异常,只会让结果错 - 可重复运行多次(用
pytest -xvs --count=50),提高暴露概率
如何让测试真正验证“原子性修复”
修完 bug 后,不能只靠一次 assert 通过就认为 OK。原子操作的正确性需要排除偶然性。
- 用
threading.Lock或threading.RLock包裹临界区后,必须把测试改成多轮压力运行(例如 100 次循环,每次起 50 线程) - 避免在测试中引入新锁(比如给
increment加锁后,又在测试里用Lock同步线程——这掩盖了真实并发路径) - 对比未加锁 vs 加锁版本的失败率:未加锁版应频繁失败(>80%),加锁后应稳定 100% 通过
- 注意 GIL 影响:纯 Python 的
+=在 CPython 下仍可能被中断,不能假设“有 GIL 就安全”
multiprocessing 适合测什么,不适合测什么
用 multiprocessing 测试原子性,本质是测进程间共享内存(如 multiprocessing.Value)或文件/数据库等外部资源的并发访问,不是测 Python 对象本身的线程安全。
典型适用场景:
- 测试
multiprocessing.Value('i', 0)配合.get_lock()的使用是否正确 - 模拟多个 Flask/Gunicorn 进程并发写同一个 SQLite 文件(需开启 WAL 模式并处理
OperationalError: database is locked) - 不适合用来测
list.append()或dict[key] = val的线程安全性——这些是线程内对象,进程间不共享
关键区别:threading 测的是同一内存空间内的指令交错;multiprocessing 测的是 IPC 或持久化层的并发控制,两者失败表现和修复方式完全不同。
最容易被忽略的一点:很多开发者用 multiprocessing 跑 pytest,却忘了每个进程都有独立的全局解释器状态,mock、单例、模块级变量都不会跨进程生效——这不是竞态,是测试设计错误。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










