pytest本身不能直接模拟竞争条件，需配合threading/multiprocessing手动构造并发场景，如用Barrier同步线程触发高概率竞态，并通过多轮运行验证原子性修复效果。
pytest 能不能直接模拟竞争条件
不能。pytest 本身是单元测试框架，不提供线程/进程调度控制能力，无法精确触发竞态窗口。想用它测原子性，必须配合
或
手动构造并发场景，否则只是在跑单线程逻辑，完全测不出问题。
常见错误现象：
里写一堆
就以为在测并发——其实线程启动顺序、执行时机全由 OS 决定，失败概率极低，且不可复现。
必须显式控制线程数量、启动节奏（比如用
同步到同一时刻）
避免依赖
做“等待”，它不保证精度，也不跨平台
测试断言要放在所有线程结束后，且检查最终状态是否符合原子操作预期（如：100 个线程各加 1，结果必须是 100）
用 threading + Barrier 触发高概率竞态
Barrier 是最可控的方式：让 N 个线程同时抵达临界区入口，极大增加对共享变量的读-改-写重叠概率。
示例场景：测试一个非线程安全的计数器类
立即学习
“
Python免费学习笔记（深入）
”；
对应测试：
Python 3.14.3
微软官方的 Python 扩展，是 VS Code 安装量最高的扩展（209M+）。集成 IntelliSense（通过 Pylance）、调试（通过 Python Debugger）、代码检查、格式化、重构和单元测试等功能。支持 Jupyter Notebook、虚拟环境管理和多 Python 版本切换。
下载
Barrier 数量必须等于线程数，否则
会永久阻塞
不要在
中捕获异常——竞态导致的逻辑错误不会抛异常，只会让结果错
可重复运行多次（用
），提高暴露概率
如何让测试真正验证“原子性修复”
修完 bug 后，不能只靠一次
通过就认为 OK。原子操作的正确性需要排除偶然性。
用
或
包裹临界区后，必须把测试改成多轮压力运行（例如 100 次循环，每次起 50 线程）
避免在测试中引入新锁（比如给
加锁后，又在测试里用
同步线程——这掩盖了真实并发路径）
对比未加锁 vs 加锁版本的失败率：未加锁版应频繁失败（>80%），加锁后应稳定 100% 通过
注意 GIL 影响：纯 Python 的
在 CPython 下仍可能被中断，不能假设“有 GIL 就安全”
multiprocessing 适合测什么，不适合测什么
用
测试原子性，本质是测进程间共享内存（如
）或文件/数据库等外部资源的并发访问，不是测 Python 对象本身的线程安全。
典型适用场景：
测试
配合
的使用是否正确
模拟多个 Flask/Gunicorn 进程并发写同一个 SQLite 文件（需开启 WAL 模式并处理
）
不适合用来测
或
的线程安全性——这些是线程内对象，进程间不共享
关键区别：
测的是同一内存空间内的指令交错；
测的是 IPC 或持久化层的并发控制，两者失败表现和修复方式完全不同。
最容易被忽略的一点：很多开发者用 multiprocessing 跑 pytest，却忘了每个进程都有独立的全局解释器状态，mock、单例、模块级变量都不会跨进程生效——这不是竞态，是测试设计错误。
threadingmultiprocessingpytestThread(target=inc).start()threading.Barriertime.sleep()import threading
class Counter:
def init(self):
self.value = 0
def increment(self):
self.value += 1 # 非原子：读取、+1、写回三步import pytest
import threading
def test_counter_racing():
counter = Counter()
barrier = threading.Barrier(10) # 10 个线程同步点
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~9wait()workerpytest -xvs --count=50assertthreading.Lockthreading.RLockincrementLock+=multiprocessingmultiprocessing.Valuemultiprocessing.Value('i', 0).get_lock()OperationalError: database is lockedlist.append()dict[key] = valthreadingmultiprocessing