Python后端工程化与测试10：测试越写越脆，先查替身和fixture放在哪里

同事把“确认报价”的实现拆成两个私有函数，业务结果完全相同，二十个测试却变红了。另一组测试始终全绿，直到上线才发现SQL语句拼错。前者知道实现细节太多，后者接触真实依赖太少。本课从这两个症状出发，决定哪些东西应当替换、哪些证据必须来自真实实现，然后把资源生命周期写进 fixture。

先接上E09。答案使用一个很小的捕获 handler，把实际 JSON formatter 输出保存为字符串。它替换的是日志的最终出口——不让测试依赖终端——但没有替换字段选择和格式化规则。随后json.loads把输出转回对象，分别比较状态、路由模板、ID和耗时。这使断言覆盖“记录最终长什么样”，比logger.info.assert_called_once()有更直接的用户价值。

捕获器安装在命名logger上，因此它是进程级资源。fixture在yield之前安装，yield之后的finally中移除和关闭。测试中途失败，pytest仍会继续执行已经建立的fixture清理。两个应用的handler幂等性检查独立存在；不能让一个测试同时吞下十种失败而只留下一个模糊断言。

答案还直接调用ASGI中间件验证ContextVar恢复。仅检查TestClient调用结束后主线程的值不够，因为客户端请求可能在另一执行上下文运行；即使中间件漏掉reset，主线程也可能原样不动。我们在同一个异步任务中设置外层值，分别让下游成功和抛异常，再检查中间件返回后的值已恢复。这个测试贴近资源生命周期的边界，补足HTTP端到端断言看不到的细节。

本课命令在课程根目录运行，使用最终project，业务合同与09一致。若此前为旧阶段设置过环境变量，先恢复它，否则你可能明明修改project却仍在测试checkpoints。

&.\.venv\Scripts\python.exe-m pytest.\lessons\10\answer_previous.py-q &.\.venv\Scripts\python.exe-m pytest.\project\tests\test_course_logging.py-q

lessons/10/answer_previous.py

"""Inspect the actual configured JSON formatter without patching HTTP internals."""importasyncioimportjsonimportloggingimportrefromcollections.abcimportIteratorfrompathlibimportPathimportpytestfromfastapi.testclientimportTestClientfromstarlette.typesimportMessage,Receive,Scope,Sendfromquoteledger.apiimportcreate_appfromquoteledger.logging_setupimport(RequestContextMiddleware,configure_logging,request_id_var,)fromquoteledger.memoryimportMemoryRepositoryfromquoteledger.settingsimportSettingsclassCapturedEvents(logging.Handler):def__init__(self,formatter:logging.Formatter)->None:super().__init__()self.setFormatter(formatter)self.lines:list[str]=[]defemit(self,record:logging.LogRecord)->None:self.lines.append(self.format(record))@pytest.fixturedefaccess_events()->Iterator[CapturedEvents]:configure_logging("INFO")logger=logging.getLogger("quoteledger.access")formatter=next(handler.formatterforhandlerinlogger.handlersifhandler.formatterisnotNone)capture=CapturedEvents(formatter)logger.addHandler(capture)try:yieldcapturefinally:logger.removeHandler(capture)capture.close()deftest_request_ids_and_logs_exclude_secrets(access_events:CapturedEvents,tmp_path:Path)->None:app=create_app(settings=Settings(database=tmp_path/"unused.db",environment="test"),repository=MemoryRepository(),)token=request_id_var.set("outside-request")try:withTestClient(app)asclient:created=client.post("/quotes?token=QUERY_SECRET_987",json={"sku":"BODY_SECRET_654","supplier":"supplier","quantity":2,"unit_cents":5,"shipping_cents":3,},headers={"X-Request-ID":"trusted-format_123","Authorization":"Bearer AUTH_SECRET_321",},)assertcreated.status_code==201assertcreated.headers["X-Request-ID"]=="trusted-format_123"loaded=client.get(created.headers["Location"],headers={"X-Request-ID":"has spaces"})replacement=loaded.headers["X-Request-ID"]assertre.fullmatch(r"[A-Za-z0-9_-]{1,64}",replacement)assertreplacement!="has spaces"missing=client.get("/quotes/PRIVATE_UNKNOWN_ID",headers={"X-Request-ID":"error-case"})assertmissing.status_code==404assertmissing.json()["request_id"]=="error-case"assertmissing.headers["X-Request-ID"]=="error-case"assertrequest_id_var.get()=="outside-request"finally:request_id_var.reset(token)assertlen(access_events.lines)==3events=[json.loads(line)forlineinaccess_events.lines]assert[event["status"]foreventinevents]==[201,200,404]assert[event["request_id"]foreventinevents]==["trusted-format_123",replacement,"error-case",]assert[event["path"]foreventinevents]==["/quotes","/quotes/{quote_id}","/quotes/{quote_id}",]assertall(event["duration_ms"]>=0foreventinevents)text="\n".join(access_events.lines)forsecretin("QUERY_SECRET_987","BODY_SECRET_654","AUTH_SECRET_321","Authorization","PRIVATE_UNKNOWN_ID",):assertsecretnotintextdeftest_two_apps_do_not_accumulate_log_handlers(tmp_path:Path)->None:configure_logging("INFO")logger=logging.getLogger("quoteledger.access")original=tuple(logger.handlers)for_inrange(2):app=create_app(settings=Settings(database=tmp_path/"unused.db",environment="test"),repository=MemoryRepository(),)withTestClient(app)asclient:assertclient.get("/health").status_code==200asserttuple(logger.handlers)==originaldeftest_invalid_environment_fails_at_configuration_boundary(monkeypatch:pytest.MonkeyPatch,)->None:monkeypatch.setenv("QUOTEL_ENV","misspelled-production")withpytest.raises(ValueError):Settings.from_env()@pytest.mark.parametrize("fail",[False,True])deftest_middleware_restores_same_task_context_on_success_and_failure(access_events:CapturedEvents,fail:bool)->None:"""TestClient runs another task, so inspect reset directly in the calling task."""messages:list[Message]=[]asyncdefdownstream(scope:Scope,receive:Receive,send:Send)->None:assertrequest_id_var.get()=="inner-request"iffail:raiseRuntimeError("PRIVATE_EXCEPTION_DETAIL")awaitsend({"type":"http.response.start","status":204,"headers":[]})awaitsend({"type":"http.response.body","body":b""})asyncdefreceive()->Message:return{"type":"http.request","body":b"","more_body":False}asyncdefsend(message:Message)->None:messages.append(message)asyncdefexercise()->None:token=request_id_var.set("outer-context")try:middleware=RequestContextMiddleware(downstream)awaitmiddleware({"type":"http","method":"GET","path":"/context-check","headers":[(b"x-request-id",b"inner-request")],},receive,send,)assertrequest_id_var.get()=="outer-context"finally:request_id_var.reset(token)asyncio.run(exercise())start=next(messageformessageinmessagesifmessage["type"]=="http.response.start")assertstart["status"]==(500iffailelse204)assert(b"x-request-id",b"inner-request")instart["headers"]assert"PRIVATE_EXCEPTION_DETAIL"notinrepr(messages)event=json.loads(access_events.lines[-1])assertevent["status"]==(500iffailelse204)assertevent["request_id"]=="inner-request"

不同替身解决的是不同问题

这些名称描述用途，不是判断工具高级程度的阶梯。同一个Python对象也可能在不同测试中扮演不同角色。为了避免术语遮住问题，我们用QuoteLedger中的依赖逐一解释。

Stub给预定答案，Fake实现可工作的简化行为，Mock经常关注交互是否符合预期。概念边界有重叠，但测试目的必须明确。我们的内存仓库有真实状态、重复id检查、版本规则和锁，它不是一个随便返回Quote的桩。正因为它复杂，必须定期与SQLite实现跑同一合同，防止两个实现逐渐对错误规则产生不同理解。

例如测试服务计算金额时，传入固定clock和id_factory就足够了。patchdatetime.now、patch UUID库、patch服务内部导入名，会把测试绑在实现路径上。明明已有QuoteService(repository, clock=..., id_factory=...)，就应沿公开构造函数控制依赖。这不是测试专用后门：系统本来就需要把时间、身份与业务规则区分开，测试只让这个边界变得可见。

反过来，如果要证明收到版本冲突时返回409，可以选择稳定抛出VersionConflict的Stub来单测HTTP映射；但这份测试不能拿去证明SQLite真的发现了冲突。后一个结论需要两个真实连接争抢同一版本。一个测试只负责一类证据，组合起来覆盖需求，比一个“端到端”标题却内部处处Mock的测试更诚实。

fixture先回答“谁拥有它”，再选择scope

fixture是带依赖关系的准备与清理机制，不是避免重复代码的万能袋子。tmp_path由pytest为每次测试提供独立临时目录，适合保存这个测试拥有的SQLite文件。repository依赖tmp_path，service依赖repository，pytest按依赖顺序构造它们，再按反向顺序清理。测试函数只声明需要什么，创建顺序不用藏在模块导入和全局变量中。

默认function scope意味着每个测试调用得到新资源；参数化会为每个参数组合重新执行。我们让仓库保持function scope，因为“已经确认过的报价单”是可变状态。改成session scope表面上更快，实际会让前一个测试的插入、状态变更或固定ID污染后一个测试。单独运行全部通过、一起运行随机失败，通常先检查这里。

session scope并非错误。例如只读的昂贵参考数据、编译好的不可变资源可以共享。关键是读者能说出谁会修改它、如何恢复、并发执行时是否隔离。fixture scope也不能随意跨层依赖：长寿命fixture不能要求短寿命fixture为它长期保存状态，这正是pytest会报告ScopeMismatch的原因。需要会话级目录时要用对应的tmp_path_factory，而不是强迫function级tmp_path延寿。

数据库连接的寿命与fixture寿命又是两回事。本项目仓库对象可以在一次测试中存活，但每次操作自行获取和关闭连接；fixture不长期霸占连接。用with sqlite3.connect(...)也不能想当然认为退出必然close，标准库连接上下文管理主要负责事务提交和回滚。仓库显式关闭连接，测试自己创建的阻塞连接也必须显式close。Windows删除文件时的“文件被占用”，常常是这层责任没有划清。

yield不是装饰，它是失败之后仍然可靠的清理边界

把资源创建写在测试前十行、清理写在测试最后一行，测试第十一行断言失败后，最后一行不会执行。try/finally能保证清理；yield fixture把这份责任集中给拥有资源的一方。访问日志捕获器就是具体例子。如果忘记移除它，下一个测试的日志会写进上一份list，事件数量断言随测试顺序变化。

清理也有边界：如果fixture在yield之前就失败，yield后代码尚未登记执行。对分步获取多个外部资源的fixture，应在每次成功获取之后立即建立对应的try/finally或上下文管理，不能指望最后一个yield包治所有异常。本课的捕获器在安装后立刻进入try块；真实数据库锁的例子在下一课展示同样原则。

monkeypatch.setenv适合配置输入边界。它记录原值并在测试结束恢复，比手写os.environ['QUOTEL_ENV']=...安全。它也可以替换一个外部调用，但应patch被使用的位置：模块A导入了B中的函数后，patch B原始属性不一定改变A已经绑定的名字。若为了测试每个函数都得解释复杂导入路径，通常说明依赖边界需要重新设计，而不是继续堆patch装饰器。

一份仓库合同，两种实现

项目里的test_course_contract.py用@pytest.fixture(params=['memory', 'sqlite'])创建两种实现，所有业务断言原封不动运行两遍。接口的Protocol只说明方法形状，不能证明“重复id不能覆盖原数据”“过期版本不能偷偷修改终态”等行为；这些语义必须通过测试表达。

测试先检查缺失id，随后创建并完整读取报价单。金额选quantity=2、unit=5、shipping=3，总价13，特意让运费有非零值、数量大于1；如果所有样例都quantity=1，错误实现shipping * quantity也会通过。重复id测试第二次使用明显不同的sku和金额，然后断言第一份对象仍完整存在，堵住“抛异常前已覆盖”的漏洞。

终态测试用两个参数的笛卡尔积：confirmed/cancelled各自再次请求confirmed/cancelled，共四种情形，两种仓库共八次。这里传的是当前版本2，用来隔离状态规则；如果一直传旧版本1，程序可能先抛VersionConflict，测试并没有走到InvalidState分支。过期版本另有专门测试，失败之后重新读取，证明胜者没有被修改。

这套合同不要求两个实现内部一致。内存可以用锁和字典，SQLite可以用条件SQL和rowcount；测试只依赖公开方法与结果。这样的测试允许你重构实现，也能揭露Fake与真实依赖的行为偏差。它仍不检查SQLite特有的文件寿命、锁等待、事务回滚；这些留给集成测试。

project/tests/test_course_contract.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

&.\.venv\Scripts\python.exe-m pytest.\project\tests\test_course_contract.py-q

测试名称若带参数后缀，如[sqlite-confirmed-cancelled]，它不是多余噪声，而是失败定位：哪个实现、哪组状态转换出问题。不要为了输出短而把所有分支塞进一个for循环；小型不变量可以循环，但独立业务情形更适合参数化，让pytest为每种情况单独报告。

E10：补一份有行为约束的仓库合同

使用同一个参数化repository fixture，分别返回新的MemoryRepository与已迁移的真实临时SQLiteRepository。断言缺失get和transition抛NotFound；创建后所有字段保持；重复id抛DuplicateQuote且原记录不被覆盖；用当前版本对confirmed/cancelled再次执行两个终态目标均抛InvalidState；旧版本操作抛VersionConflict且已有胜者不变。不得Mock SQLiteRepository，不得为两种仓库复制两套略有不同的断言。说明function scope为何能隔离固定id，及这份合同为何还不能证明真正的并发原子性。

本课已经展示完整合同实现，你可以先关掉源码按要求重写，再把遗漏的断言逐个对照。下一课保留完整可运行答案，并往测试里放入真实文件、真实写锁和真实线程竞争。做完之后，你应能判断一项测试失败是领域语义错误、HTTP映射错误，还是存储边界出了问题。