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.pyThe same observable repository contract must hold for both implementations.fromcollections.abcimportIteratorfromdatetimeimportUTC,datetimefrompathlibimportPathimportpytestfromquoteledger.domainimport(CreateQuote,DuplicateQuote,InvalidState,NotFound,VersionConflict,)fromquoteledger.memoryimportMemoryRepositoryfromquoteledger.migrationsimportmigratefromquoteledger.portsimportRepositoryfromquoteledger.serviceimportQuoteServicefromquoteledger.sqlite_repoimportSQLiteRepositorypytest.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.parammemory:yieldMemoryRepository()else:databasetmp_path/contract.dbmigrate(database)# The repository owns short-lived connections, not a suite-wide connection.yieldSQLiteRepository(database)pytest.fixturedefservice(repository:Repository)-QuoteService:returnQuoteService(repository,clocklambda:datetime(2026,10,9,2,30,tzinfoUTC),id_factorylambda: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:originalservice.create(CreateQuote(SKU-1,supplier-A,2,5,3))loadedrepository.get(original.id)assertloadedoriginalassertloaded.total_cents13assertloaded.created_atdatetime(2026,10,9,2,30,tzinfoUTC)assertloaded.version1assertloaded.statusdraftdeftest_duplicate_id_does_not_overwrite(repository:Repository,service:QuoteService)-None:originalservice.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:originalservice.create(CreateQuote(SKU-1,supplier-A,2,5,3))winnerservice.confirm(original.id,1)withpytest.raises(VersionConflict):service.cancel(original.id,1)assertrepository.get(original.id)winnerassertwinner.version2assertoriginal.statusdraftpytest.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:originalservice.create(CreateQuote(SKU-1,supplier-A,1,50))# Literal values are narrowed at the public boundary, without a type-ignore.initialconfirmediffirstconfirmedelsecancelledtargetconfirmedifsecondconfirmedelsecancelledterminalrepository.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文件。创建响应必须是201Location指向新资源返回金额必须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_version1。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.pyExercise 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:databasetmp_path/restart.dbmigrate(database)settingsSettings(databasedatabase,environmenttest)withTestClient(create_app(settingssettings))asfirst:createdfirst.post(/quotes,jsonPAYLOAD)assertcreated.status_code201originalcreated.json()assertoriginal[total_cents]13locationcreated.headers[location]# Creating a second app builds another repository and another connection.withTestClient(create_app(settingssettings))assecond:loadedsecond.get(location)assertloaded.status_code200assertloaded.json()original confirmedsecond.post(f{location}/confirm,json{expected_version:original[version]})assertconfirmed.status_code200assertconfirmed.json()[version]2persistedSQLiteRepository(database).get(original[id])assertpersisted.statusconfirmedassertpersisted.version2assertpersisted.total_cents13deftest_real_write_lock_returns_503_then_recovers(tmp_path:Path)-None:databasetmp_path/busy.dbmigrate(database)appcreate_app(settingsSettings(databasedatabase,environmenttest),repositorySQLiteRepository(database,timeout0.01),)withTestClient(app)asclient:blockersqlite3.connect(database,timeout0.01)try:blocker.execute(BEGIN IMMEDIATE)failedclient.post(/quotes,jsonPAYLOAD,headers{X-Request-ID:busy-case})assertfailed.status_code503assertfailed.json()[error][code]storage_busyassertfailed.json()[request_id]busy-caseassertfailed.headers[X-Request-ID]busy-caseassertstr(database)notinfailed.textfinally:# Even a failed assertion must release the lock before the next test.blocker.rollback()blocker.close()recoveredclient.post(/quotes,jsonPAYLOAD)assertrecovered.status_code201assertrecovered.json()[version]1withclosing(sqlite3.connect(database))asconnection:countconnection.execute(SELECT COUNT(*) FROM quotes).fetchone()[0]assertcount1# The rejected transaction must not leave a partial quote.deftest_two_connections_compete_for_one_version(tmp_path:Path)-None:databasetmp_path/race.dbmigrate(database)originalQuoteService(SQLiteRepository(database)).create(CreateQuote(race-sku,supplier-A,2,5,3))barrierBarrier(2)defcompete(confirm:bool)-Quote|VersionConflict:serviceQuoteService(SQLiteRepository(database,timeout2.0))barrier.wait(timeout5)try:ifconfirm:returnservice.confirm(original.id,1)returnservice.cancel(original.id,1)exceptVersionConflictaserror:returnerrorwithThreadPoolExecutor(max_workers2)asexecutor:futures[executor.submit(compete,True),executor.submit(compete,False)]outcomes[future.result(timeout10)forfutureinfutures]successes[resultforresultinoutcomesifisinstance(result,Quote)]conflicts[resultforresultinoutcomesifisinstance(result,VersionConflict)]assertlen(successes)1assertlen(conflicts)1persistedSQLiteRepository(database).get(original.id)assertpersistedsuccesses[0]assertpersisted.statusin{confirmed,cancelled}assertpersisted.version2assertpersisted.idoriginal.idassertpersisted.total_centsoriginal.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文件注入timeout0.01的SQLiteRepository到应用。用另一个sqlite3连接执行BEGIN IMMEDIATE并保持事务通过HTTP发起创建断言503、storage_busy、合法且一致的request_id、不含本地路径。在finally中rollback并close阻塞连接随后使用同一个TestClient再创建断言201且version1直接查询数据库确认总共只有一条报价。不得Mock sqlite3不得用sleep猜测锁是否获得不得重新创建应用来掩盖故障后的状态污染。下一课给出完整答案再讨论测试全绿以后怎样继续追问执行过一行代码究竟证明了多少当输入空间比手工案例大得多时怎样把“金额公式和状态转换不变量”交给工具生成数据验证