
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-qlessons/10/answer_previous.pyInspect 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)loggerlogging.getLogger(quoteledger.access)formatternext(handler.formatterforhandlerinlogger.handlersifhandler.formatterisnotNone)captureCapturedEvents(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:appcreate_app(settingsSettings(databasetmp_path/unused.db,environmenttest),repositoryMemoryRepository(),)tokenrequest_id_var.set(outside-request)try:withTestClient(app)asclient:createdclient.post(/quotes?tokenQUERY_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_code201assertcreated.headers[X-Request-ID]trusted-format_123loadedclient.get(created.headers[Location],headers{X-Request-ID:has spaces})replacementloaded.headers[X-Request-ID]assertre.fullmatch(r[A-Za-z0-9_-]{1,64},replacement)assertreplacement!has spacesmissingclient.get(/quotes/PRIVATE_UNKNOWN_ID,headers{X-Request-ID:error-case})assertmissing.status_code404assertmissing.json()[request_id]error-caseassertmissing.headers[X-Request-ID]error-caseassertrequest_id_var.get()outside-requestfinally: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)loggerlogging.getLogger(quoteledger.access)originaltuple(logger.handlers)for_inrange(2):appcreate_app(settingsSettings(databasetmp_path/unused.db,environmenttest),repositoryMemoryRepository(),)withTestClient(app)asclient:assertclient.get(/health).status_code200asserttuple(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-requestiffail: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:tokenrequest_id_var.set(outer-context)try:middlewareRequestContextMiddleware(downstream)awaitmiddleware({type:http,method:GET,path:/context-check,headers:[(bx-request-id,binner-request)],},receive,send,)assertrequest_id_var.get()outer-contextfinally:request_id_var.reset(token)asyncio.run(exercise())startnext(messageformessageinmessagesifmessage[type]http.response.start)assertstart[status](500iffailelse204)assert(bx-request-id,binner-request)instart[headers]assertPRIVATE_EXCEPTION_DETAILnotinrepr(messages)eventjson.loads(access_events.lines[-1])assertevent[status](500iffailelse204)assertevent[request_id]inner-request不同替身解决的是不同问题这些名称描述用途不是判断工具高级程度的阶梯。同一个Python对象也可能在不同测试中扮演不同角色。为了避免术语遮住问题我们用QuoteLedger中的依赖逐一解释。依赖与目的适合的方式这份证据不能说明什么创建时间必须固定注入返回固定时间的clock属于Stub不能证明真实系统时钟准确生成可预期ID注入简单id_factory不能证明UUID永不冲突服务规则不依赖磁盘使用实现完整仓库行为的MemoryRepository属于Fake不能证明SQL、事务或文件权限正确确认外部通知是否发送可用Mock核对调用参数与次数不能证明网络请求真正到达远端验证持久化和锁竞争临时文件上的真实SQLiteRepository不自动覆盖另一数据库引擎的语义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先回答“谁拥有它”再选择scopefixture是带依赖关系的准备与清理机制不是避免重复代码的万能袋子。tmp_path由pytest为每次测试提供独立临时目录适合保存这个测试拥有的SQLite文件。repository依赖tmp_pathservice依赖repositorypytest按依赖顺序构造它们再按反向顺序清理。测试函数只声明需要什么创建顺序不用藏在模块导入和全局变量中。默认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随后创建并完整读取报价单。金额选quantity2、unit5、shipping3总价13特意让运费有非零值、数量大于1如果所有样例都quantity1错误实现shipping * quantity也会通过。重复id测试第二次使用明显不同的sku和金额然后断言第一份对象仍完整存在堵住“抛异常前已覆盖”的漏洞。终态测试用两个参数的笛卡尔积confirmed/cancelled各自再次请求confirmed/cancelled共四种情形两种仓库共八次。这里传的是当前版本2用来隔离状态规则如果一直传旧版本1程序可能先抛VersionConflict测试并没有走到InvalidState分支。过期版本另有专门测试失败之后重新读取证明胜者没有被修改。这套合同不要求两个实现内部一致。内存可以用锁和字典SQLite可以用条件SQL和rowcount测试只依赖公开方法与结果。这样的测试允许你重构实现也能揭露Fake与真实依赖的行为偏差。它仍不检查SQLite特有的文件寿命、锁等待、事务回滚这些留给集成测试。project/tests/test_course_contract.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.\.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映射错误还是存储边界出了问题。