pytest需在单个测试函数内用threading.Thread启动多线程并发调用被测函数，通过共享可变对象、控制同步时机、捕获子线程异常并join等待，以复现确定性竞态问题。
pytest如何触发多线程并发执行测试用例
pytest 本身默认是单线程顺序执行测试，想测竞态条件，必须主动启动多个线程去并发调用被测函数——不能靠 pytest-xdist（它分的是进程，且隔离了状态），也不能靠 @pytest.mark.parametrize（只是重复跑，非并发）。关键动作是：在单个测试函数内部用 threading.Thread 启动 N 个线程，统一操作共享资源（如全局变量、类属性或传入的可变对象）。
实操建议：
- 用
threading.Event或time.sleep()控制线程“同时”开始，避免因启动时序差异漏掉竞态 - 每个线程应捕获异常并存入列表（如
errors.append(sys.exc_info())），否则主线程无法感知子线程内抛出的AssertionError - 务必调用
thread.join()等待全部完成，否则测试可能提前结束，漏判失败 - 别用
concurrent.futures.ThreadPoolExecutor简化——它默认复用线程、隐藏异常传播路径，反而掩盖问题
如何构造可复现的竞态场景（以计数器为例）
竞态不是“偶尔出错”，而是代码本身缺乏同步机制导致的确定性错误。最典型的是非原子的“读-改-写”：比如 counter += 1 在字节码层面拆成 LOAD、INPLACE_ADD、STORE 三步，多线程交错执行就会丢更新。
实操建议：
- 用一个全局
int或list（如shared = [0]）作为共享状态，避免引用不可变对象导致误判 - 被测函数不要加锁——加了就不是竞态了；但可以加
time.sleep(0.001)在读写之间，人为放大交错窗口 - 运行 100 次线程，预期结果应为 100；若最终值
- 示例片段：
def increment(): global counter time.sleep(0.001) # 增大竞态概率 counter += 1
pytest中如何断言竞态真实发生而非偶然失败
单纯看最终值不等于预期，只能说明“可能”有竞态；要确认是并发导致的，得验证行为模式符合竞态特征：比如多次运行中结果不稳定、或能观测到中间态脏值。
实操建议：
- 不要只断言
assert counter == 100；改为assert counter ，因为竞态只会让结果变小，不会变大 - 在被测逻辑中插入日志或快照（如把每次读到的旧值存进
reads.append(shared[0])），检查是否出现“跳变”或重复读取 - 用
pytest-repeat插件跑 50 轮测试（pytest test_race.py --count=50），观察失败率——稳定在 20%~80% 才可信；100% 失败说明没并发，0% 失败说明竞态窗口太小 - 避免在 CI 中依赖“运气型”断言；真正可靠的竞态测试，应配合
threading.Lock修复后对比验证：修复前失败率高，修复后稳定通过
为什么 multiprocessing.Pool 会干扰竞态复现
用 multiprocessing 替代 threading 测试，看似更“重”，实则完全无效：Python 的进程间内存不共享，每个子进程都有一份独立的 counter 副本，根本不会产生数据竞争——你测的只是进程启动开销，不是并发 bug。
实操建议：
- 确认所有线程操作的是同一对象 ID：
print(id(shared))在各线程中应一致；若用multiprocessing.Manager().dict()，那是进程安全代理，已自带锁，也测不出原始竞态 - 调试时加
threading.current_thread().name日志，确认确实是多个线程在跑，而不是单线程循环 - Windows 下注意 pytest 默认使用
spawn方式启动子进程，会重新导入模块，导致全局变量被重置——这会让多进程测试看起来“总是成功”，纯属假象
竞态测试最难的不是写多线程，而是让失败稳定出现又不误报；重点始终放在共享状态的可见性、操作的非原子性、以及观测手段是否真的反映并发行为。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
