第 3 周周测答案¶

输出已在 Python 3.14.4 / httpx 0.28 上实际运行核对（耗时类数字取近似）。

一、预测输出¶

1. 10 0.1 0.1（并发那一段实测约 0.06，四舍五入到一位是 0.1；关键是它 ≈ 一次 sleep 的时间，而顺序段是两次之和 0.1）。await 一个接一个是串行；gather 让两个协程同时挂起等待。
2. coroutine True → False → True 2。协程对象只是"待执行的配方"；create_task 立刻调度但还没跑完，所以 done() 是 False。追问：coro 从未被 await，退出时打印 RuntimeWarning: coroutine 'work' was never awaited。
3. ['two'] [False, False, True] —— boom(2) 在 0.02s 抛错，TaskGroup 立刻取消还没完成的 boom(3)（cancelled() == True）；boom(1) 已在 0.01s 正常结束。异常被聚合成 ExceptionGroup，用 except* 拆开。
4. timeout → [1, 'ValueError'] —— asyncio.timeout 超时抛 TimeoutError（3.11+ 与内置 TimeoutError 是同一个）；gather(return_exceptions=True) 把异常当结果返回而不是抛出。
5. [0, 1, 4, 9]；总耗时约 0.05–0.1 秒（4 个任务 4 个线程并行；Windows 上线程启动有些开销），远小于顺序的 0.2 秒。map 内部按提交顺序逐个 result()，所以顺序与输入一致（"哪个先完成"要用 as_completed）。
6. 5 1 ('a',)，然后 m.nope 抛 AttributeError —— spec=Svc 让 mock 只接受 Svc 有的属性，拼错方法名立刻暴露。
7. 42 和 42 —— 第一个 patch 直接替换 shop.now；第二个替换 time 模块的 time，而 shop.now 里是 time.time() 每次动态取属性，所以也生效。追问：from time import time 把函数对象复制进了 shop 命名空间，patch("time.time") 改的是 time 模块，shop.time 仍是旧函数——无效，必须 patch("shop.time")（打到使用处）。
8. ['slow', 'fast'] 0.2 —— time.sleep(0.2) 阻塞整个事件循环，fast() 只能等它结束后才开始，总耗时 ≈ 0.2 而不是 0.01。协程里应 await asyncio.sleep 或 await asyncio.to_thread(time.sleep, 0.2)。
9. 3 out err —— 非零退出码不抛异常，stdout/stderr 分开捕获。追问：check=True 时抛 subprocess.CalledProcessError（属性里仍有 returncode/stdout/stderr）。
10. 503 busy，随后 raise_for_status() 抛 httpx.HTTPStatusError（消息含 503 Service Unavailable）。不 raise_for_status 就 .json()，拿到的是错误页的数据。

二、找 bug¶

1. 只创建 Task 却不保存引用、不 await：函数返回后任务可能被垃圾回收或在 asyncio.run 结束时被取消，结果也拿不到。用 TaskGroup（或收集 task 列表后 gather），让"所有子任务结束"成为函数返回的前提。
2. 三个问题：每次新建 AsyncClient 丢掉连接复用（慢且耗端口）；没有显式超时（默认 5s 也应写明，生产里常需更短/更长）；不 raise_for_status 会把 4xx/5xx 的错误体当正常 JSON。客户端在外面建一次传进来；httpx.Timeout(...)；先 raise_for_status()。
3. Windows 用 spawn 启动子进程：子进程会重新 import 主模块，模块顶层的 ProcessPoolExecutor 又会启动子进程……无限递归（实际会报 RuntimeError: An attempt has been made to start a new process before...）。把执行代码放进 if __name__ == "__main__":。
4. 吞掉 CancelledError 让 worker 无法被取消——queue.join() 之后 worker.cancel() 永远不生效，程序卡住。捕获后必须重新 raise（或根本不捕获），清理放 finally；并且 task_done() 应放在 finally 里保证一定调用。
5. 它验证了一个空库存的总价是 0——mock 根本没被用到（没有 add 任何 sku），RealPriceService 也没被替换为注入的对象（patch 的是类上的方法，但 Inventory 不一定走那条路径）。覆盖率 ≠ 断言强度。正确做法：注入假的 PriceService，add 几件商品，断言总价等于数量 × 单价，并断言 price_of 每个 sku 各调一次。

三、写代码（参考）¶

16.

import asyncio

async def gather_with_limit(coros, limit: int) -> list:
    sem = asyncio.Semaphore(limit)
    async def guarded(c):
        async with sem:
            return await c
    return list(await asyncio.gather(*(guarded(c) for c in coros)))

import asyncio, httpx

async def fetch_json(client: httpx.AsyncClient, url: str, timeout: float = 3.0) -> dict | None:
    try:
        async with asyncio.timeout(timeout):
            response = await client.get(url)
    except TimeoutError:
        return None
    response.raise_for_status()
    return response.json()

# 测试
import pytest

def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/ok":
        return httpx.Response(200, json={"id": 1})
    return httpx.Response(500, json={"detail": "boom"})

async def test_fetch_json():
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="http://t") as c:
        assert await fetch_json(c, "/ok") == {"id": 1}
        with pytest.raises(httpx.HTTPStatusError):
            await fetch_json(c, "/bad")

import subprocess

class CommandTimeout(Exception): ...

def run_command(cmd: list[str], timeout: float = 10) -> tuple[int, str]:
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", timeout=timeout)
    except subprocess.TimeoutExpired as e:
        raise CommandTimeout(f"超时 {timeout}s：{cmd}") from e
    return r.returncode, r.stdout

import logging, pytest

def test_remove_reduces(inventory):
    inventory.add("pen", 5)
    assert inventory.remove("pen", 2).quantity == 3

def test_remove_too_many(inventory):
    inventory.add("pen", 1)
    with pytest.raises(ValueError, match="库存不足"):
        inventory.remove("pen", 2)

def test_remove_logs_low_stock(inventory, caplog):
    inventory.add("pen", 4)
    with caplog.at_level(logging.WARNING, logger="inventory"):
        inventory.remove("pen", 2)          # 剩 2 <= 阈值 3
    assert any("库存偏低" in r.message for r in caplog.records)

import time

def benchmark(fn, *args, repeat: int = 5) -> float:
    if repeat <= 0:
        raise ValueError("repeat 必须为正")
    best = float("inf")
    for _ in range(repeat):
        t = time.perf_counter(); fn(*args)
        best = min(best, time.perf_counter() - t)
    return best