Python后端工程化与测试11：让数据库真的锁住，再验证失败能够恢复
一个Mock仓库可以随时抛出StorageBusy，让我们确认HTTP层把它转成503。但事故现场的路径更长：SQLite先等待写锁，超时后产生具体数据库异常，仓库识别异常并回滚关闭连接，服务传播领域化的错误，API输出受控响应，日志保留关联ID。只验证最后一小段，前面任何一段都可能断掉。这一课亲手占住SQLite写锁，再观察整个失败链和恢复链。
我们不随机断网、不随便结束系统进程，也不把“等一秒看看”当作实验设计。失败注入应当有明确控制点、确定的释放方式和可观察结果。SQLite文件与连接都由测试创建，写锁只影响当前临时数据库。测试结束后的电脑状态应该与测试之前一致，这既是可靠测试的基础，也是并行运行其他任务的前提。
E10答案：先确认两个仓库承诺的是同一件事
上一课的仓库合同在两个实现上执行完全相同的断言。固定clock使时间可精确比较，固定id让重复键场景可控。由于repository是function scope，每个测试都有自己的仓库或文件，固定id不会在测试之间冲突。答案中terminal状态的第二次转换必须带当前version，否则会先命中过期版本判断，无法证明终态规则。
&.\.venv\Scripts\python.exe-m pytest.\lessons\11\answer_previous.py-qlessons/11/answer_previous.py
"""The same observable repository contract must hold for both implementations."""fromcollections.abcimportIteratorfromdatetimeimportUTC,datetimefrompathlibimportPathimportpytestfromquoteledger.domainimport(CreateQuote,DuplicateQuote,InvalidState,NotFound,VersionConflict,)fromquoteledger.memoryimportMemoryRepositoryfromquoteledger.migrationsimportmigratefromquoteledger.portsimportRepositoryfromquoteledger.serviceimportQuoteServicefromquoteledger.sqlite_repoimportSQLiteRepository@pytest.fixture(params=["memory","sqlite"])defrepository(request:pytest.FixtureRequest,tmp_path:Path)->Iterator[Repository]:"""Function scope gives each example an independent store and database file."""ifrequest.param=="memory":yieldMemoryRepository()else:database=tmp_path/"contract.db"migrate(database)# The repository owns short-lived connections, not a suite-wide connection.yieldSQLiteRepository(database)@pytest.fixturedefservice(repository:Repository)->QuoteService:returnQuoteService(repository,clock=lambda:datetime(2026,10,9,2,30,tzinfo=UTC),id_factory=lambda:"contract-quote",)deftest_missing_is_a_domain_error(repository:Repository)->None:withpytest.raises(NotFound):repository.get("absent")withpytest.raises(NotFound):repository.transition("absent","confirmed",1)deftest_create_and_reload_preserve_every_field(repository:Repository,service:QuoteService)->None:original=service.create(CreateQuote("SKU-1","supplier-A",2,5,3))loaded=repository.get(original.id)assertloaded==originalassertloaded.total_cents==13assertloaded.created_at==datetime(2026,10,9,2,30,tzinfo=UTC)assertloaded.version==1assertloaded.status=="draft"deftest_duplicate_id_does_not_overwrite(repository:Repository,service:QuoteService)->None:original=service.create(CreateQuote("original","supplier-A",2,5,3))withpytest.raises(DuplicateQuote):service.create(CreateQuote("replacement","supplier-B",1,999))assertrepository.get(original.id)==originaldeftest_stale_version_does_not_change_the_winner(repository:Repository,service:QuoteService)->None:original=service.create(CreateQuote("SKU-1","supplier-A",2,5,3))winner=service.confirm(original.id,1)withpytest.raises(VersionConflict):service.cancel(original.id,1)assertrepository.get(original.id)==winnerassertwinner.version==2assertoriginal.status=="draft"@pytest.mark.parametrize("first",["confirmed","cancelled"])@pytest.mark.parametrize("second",["confirmed","cancelled"])deftest_terminal_state_never_changes(repository:Repository,service:QuoteService,first:str,second:str)->None:original=service.create(CreateQuote("SKU-1","supplier-A",1,50))# Literal values are narrowed at the public boundary, without a type-ignore.initial="confirmed"iffirst=="confirmed"else"cancelled"target="confirmed"ifsecond=="confirmed"else"cancelled"terminal=repository.transition(original.id,initial,1)withpytest.raises(InvalidState):repository.transition(original.id,target,terminal.version)assertrepository.get(original.id)==terminal这份合同让Fake可用于服务测试，但它不把Fake升级成数据库。MemoryRepository没有磁盘同步、SQL语法、数据库锁和文件权限。两份实现都通过合同，只说明合同列出的业务行为一致。对文件持久化和原子更新的结论，还需要下面的独立证据。
先把“重启之后还在”拆开检查
集成测试的第一条路径从HTTP进入，经过Pydantic校验、服务、仓库，最后落到临时SQLite文件。创建响应必须是201，Location指向新资源，返回金额必须13。关闭第一个TestClient后，构造第二个应用，它拥有新的仓库对象和新的连接；读取结果必须与之前返回的全部JSON一致。
然后第二个应用确认报价单，关闭以后再创建一个全新SQLiteRepository读取，验证status和version。这样的安排同时检查读路径和写路径。如果只用同一个客户端立即GET，错误的内存缓存可能让测试看起来像持久化成功；如果只断言文件存在，里面可能没有正确记录；如果只断言id相同，金额和时间可能已经在序列化时损坏。
这里的应用重建仍然发生在同一个Python进程内，所以我们准确称为“新应用、新连接读取”。它是快速且稳定的存储集成测试，不能假称已经完成操作系统进程重启测试。第15课会用安装好的wheel、真实HTTP端口及服务进程重启补上后一个证据。分清证据强度比给测试取一个很大的名字更重要。
SQLite的:memory:很适合一些局部实验，却不适合替代这个文件场景。普通内存数据库通常与连接绑定；本项目每次操作新建连接，如果随手把文件路径换成:memory:，建表的连接关闭以后，下一次连接可能看到另一个空库。即使通过共享缓存或保活连接绕过去，也已经换了生命周期，无法证明文件关闭、重新打开与落盘行为。
占住真实写锁，验证503而不是碰运气等超时
我们用独立的sqlite3.connect(database)连接执行BEGIN IMMEDIATE。这条命令在事务开始时申请写事务所需的锁，不必插入一条伪业务数据，也不用等待被测代码碰巧撞上竞态窗口。随后同一测试通过HTTP创建报价单，仓库里的另一连接尝试写入。timeout设置为0.01秒，足够让锁冲突迅速变成可控失败。
这个数字是故障注入配置，不是性能指标。不要断言请求必须在10毫秒内结束：线程池调度、文件系统和CI机器负载都会增加延迟。我们验证的是有限等待后的503、错误code为storage_busy、响应头和body包含指定request_id、不泄露数据库本地路径。超时是为了测试不长期卡住，业务合同是受控失败。
锁要放在try/finally里释放。无论HTTP响应错误还是断言失败，finally都rollback和close。释放之后，同一个应用与同一个客户端再次创建必须201。若只验证第一次503，仓库可能泄漏连接、遗留事务或进入不可恢复状态；第二次成功把恢复能力也纳入合同。最后直接查询表的记录数为1，证明失败的写入没有留下半条报价或重复数据。
这里采用rollback不是因为阻塞连接已经写了业务数据，而是明确结束我们主动打开的事务。close也仍然需要调用；事务结束和连接释放不是同一件事。未来若测试还创建了线程、临时服务或文件句柄，应逐一明确所有者和释放位置，不要把pytest会清理临时目录误当作所有资源都自动安全关闭。
两个客户拿到版本1，数据库怎样只让一个人赢
竞态测试创建一份draft，然后启动两个工作线程。每个线程内部构造新的SQLiteRepository，文件路径相同，仓库对象不同；一个确认，一个取消，都提交expected_version=1。Barrier(2)让两个参与者都完成准备再发起请求。我们设置屏障与future的有限超时，让错误实现能够失败并给出位置，而不是永久挂住测试进程。
结果不预言哪一个线程赢。操作系统调度没有义务优先确认或取消；正确合同是一个Quote、一个VersionConflict，最终状态属于两个合法终态之一、version恰好2。把“确认必须赢”写成断言会让测试随机失败，把两个失败也算“没有重复成功”又会漏掉可用性错误。
为什么SQL条件更新重要？如果实现先SELECT检查version，再在另一个不受保护步骤无条件UPDATE，两个线程都可能看到1，都自认有权写入。最终数据库可能只保存最后一个结果，但两个调用方都收到成功，业务承诺已经破坏。因此必须检查双方返回结果和最终存储，不能只看最终version等于2。CAS把版本和状态条件放入更新动作，rowcount决定是否赢得修改资格。
08开始约定过期版本先于终态判断。败者依然提交1，胜者已经写成2，所以它应该得到VersionConflict，而不是泛泛的InvalidState。反之，针对已确认记录持有当前版本2再确认，才是终态规则的InvalidState。两种409有不同语义，客户端可以据此决定重新读取还是停止操作。
一次竞态测试不是数学证明，Barrier也不会精确控制数据库内部每条指令的时间。它把两个真实参与者放进竞争条件，配合检查SQL条件、事务范围和返回结果，形成可信证据。如果长期观察到偶发失败，应保留失败现场并分析锁范围，不能无依据增加sleep直到“看上去稳定”。本课程刻意不用sleep来协调正确性。
失败应该穿过哪些边界
| 注入点 | 直接观察 | 后续观察 | 能排除的缺陷 |
|---|---|---|---|
| SQLite独立写锁 | HTTP503与storage_busy | 释放后201，表内仅1条 | 异常没映射、事务泄漏、失败写入残留 |
| 两个独立仓库提交同版本 | 一成功一VersionConflict | 新仓库读到版本2 | 先查后写、双成功、只改返回值 |
| 关闭应用后另建应用 | GET与创建响应一致 | 确认后再次重开读取 | 只写缓存、字段映射丢失 |
这些测试与纯领域测试互补。失败后排查先看最内层：领域金额性质是否通过，仓库合同是否通过，随后看真实SQLite集成，最后看HTTP映射。每一层的职责清楚，测试失败才更接近定位线索，而不是一个只能重跑的红灯。
下面是本课新增集成测试全文。测试没有替换sqlite3，没有伪造rowcount，也没有Mock服务；用到的依赖注入只选择本测试独有的真实文件和短超时。同步数据库操作仍然由def路由在线程池中执行，这也是为何不能拿async函数包装sqlite3就声称已经非阻塞。
project/tests/test_course_integration.py
"""Exercise real files, real SQLite locks, and competing independent repositories."""importsqlite3fromconcurrent.futuresimportThreadPoolExecutorfromcontextlibimportclosingfrompathlibimportPathfromthreadingimportBarrierfromfastapi.testclientimportTestClientfromquoteledger.apiimportcreate_appfromquoteledger.domainimportCreateQuote,Quote,VersionConflictfromquoteledger.migrationsimportmigratefromquoteledger.serviceimportQuoteServicefromquoteledger.settingsimportSettingsfromquoteledger.sqlite_repoimportSQLiteRepository PAYLOAD={"sku":"integration-sku","supplier":"supplier-A","quantity":2,"unit_cents":5,"shipping_cents":3,}deftest_http_to_file_survives_new_application(tmp_path:Path)->None:database=tmp_path/"restart.db"migrate(database)settings=Settings(database=database,environment="test")withTestClient(create_app(settings=settings))asfirst:created=first.post("/quotes",json=PAYLOAD)assertcreated.status_code==201original=created.json()assertoriginal["total_cents"]==13location=created.headers["location"]# Creating a second app builds another repository and another connection.withTestClient(create_app(settings=settings))assecond:loaded=second.get(location)assertloaded.status_code==200assertloaded.json()==original confirmed=second.post(f"{location}/confirm",json={"expected_version":original["version"]})assertconfirmed.status_code==200assertconfirmed.json()["version"]==2persisted=SQLiteRepository(database).get(original["id"])assertpersisted.status=="confirmed"assertpersisted.version==2assertpersisted.total_cents==13deftest_real_write_lock_returns_503_then_recovers(tmp_path:Path)->None:database=tmp_path/"busy.db"migrate(database)app=create_app(settings=Settings(database=database,environment="test"),repository=SQLiteRepository(database,timeout=0.01),)withTestClient(app)asclient:blocker=sqlite3.connect(database,timeout=0.01)try:blocker.execute("BEGIN IMMEDIATE")failed=client.post("/quotes",json=PAYLOAD,headers={"X-Request-ID":"busy-case"})assertfailed.status_code==503assertfailed.json()["error"]["code"]=="storage_busy"assertfailed.json()["request_id"]=="busy-case"assertfailed.headers["X-Request-ID"]=="busy-case"assertstr(database)notinfailed.textfinally:# Even a failed assertion must release the lock before the next test.blocker.rollback()blocker.close()recovered=client.post("/quotes",json=PAYLOAD)assertrecovered.status_code==201assertrecovered.json()["version"]==1withclosing(sqlite3.connect(database))asconnection:count=connection.execute("SELECT COUNT(*) FROM quotes").fetchone()[0]assertcount==1# The rejected transaction must not leave a partial quote.deftest_two_connections_compete_for_one_version(tmp_path:Path)->None:database=tmp_path/"race.db"migrate(database)original=QuoteService(SQLiteRepository(database)).create(CreateQuote("race-sku","supplier-A",2,5,3))barrier=Barrier(2)defcompete(confirm:bool)->Quote|VersionConflict:service=QuoteService(SQLiteRepository(database,timeout=2.0))barrier.wait(timeout=5)try:ifconfirm:returnservice.confirm(original.id,1)returnservice.cancel(original.id,1)exceptVersionConflictaserror:returnerrorwithThreadPoolExecutor(max_workers=2)asexecutor:futures=[executor.submit(compete,True),executor.submit(compete,False)]outcomes=[future.result(timeout=10)forfutureinfutures]successes=[resultforresultinoutcomesifisinstance(result,Quote)]conflicts=[resultforresultinoutcomesifisinstance(result,VersionConflict)]assertlen(successes)==1assertlen(conflicts)==1persisted=SQLiteRepository(database).get(original.id)assertpersisted==successes[0]assertpersisted.statusin{"confirmed","cancelled"}assertpersisted.version==2assertpersisted.id==original.idassertpersisted.total_cents==original.total_cents&.\.venv\Scripts\python.exe-m pytest.\project\tests\test_course_integration.py-q若遇到“database is locked”直接冒出测试，不要首先延长timeout。检查被测仓库是否转换了锁忙异常、是否在finally关闭连接，再检查测试自己的阻塞连接是否按时释放。如果后台还有人手工打开同一文件，改用测试的tmp_path，不要让测试碰课程示例数据。若提示no such table，检查是否显式执行了migrate，及迁移和仓库是否指向同一绝对路径。
E11：失败之后，原来的应用还能继续写吗
创建并迁移真实临时SQLite文件，注入timeout=0.01的SQLiteRepository到应用。用另一个sqlite3连接执行BEGIN IMMEDIATE并保持事务；通过HTTP发起创建，断言503、storage_busy、合法且一致的request_id、不含本地路径。在finally中rollback并close阻塞连接，随后使用同一个TestClient再创建，断言201且version=1；直接查询数据库，确认总共只有一条报价。不得Mock sqlite3，不得用sleep猜测锁是否获得，不得重新创建应用来掩盖故障后的状态污染。
下一课给出完整答案，再讨论测试全绿以后怎样继续追问：执行过一行代码，究竟证明了多少？当输入空间比手工案例大得多时，怎样把“金额公式和状态转换不变量”交给工具生成数据验证？
