跳转至

第 3 周周测（闭卷，40 分钟，20 题）

不开 REPL、不开 AI。答案见 exercises/solutions/week3/quiz_week3_answers.md。≥16 分通过；错题进下周一的回忆清单。

一、预测输出（10 题）

  1. python import asyncio, time async def work(i): await asyncio.sleep(0.05); return i async def main(): t0 = time.perf_counter() a = await work(1); b = await work(2) seq = time.perf_counter() - t0 t1 = time.perf_counter() c, d = await asyncio.gather(work(3), work(4)) con = time.perf_counter() - t1 print(a + b + c + d, round(seq, 1), round(con, 1)) asyncio.run(main())
  2. python async def main(): coro = work(1) print(type(coro).__name__, asyncio.iscoroutine(coro)) task = asyncio.create_task(work(2)) print(task.done()) await task print(task.done(), task.result()) asyncio.run(main()) 追问：如果 coro 一直不被 await，程序结束时会看到什么？
  3. python async def boom(i): await asyncio.sleep(0.01 * i) if i == 2: raise ValueError("two") return i async def main(): try: async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(boom(i)) for i in range(1, 4)] except* ValueError as eg: print([str(e) for e in eg.exceptions], [t.cancelled() for t in tasks]) asyncio.run(main())
  4. python async def main(): try: async with asyncio.timeout(0.05): await asyncio.sleep(1) except TimeoutError: print("timeout") results = await asyncio.gather(boom(1), boom(2), return_exceptions=True) print([type(r).__name__ if isinstance(r, Exception) else r for r in results]) asyncio.run(main())
  5. python import time from concurrent.futures import ThreadPoolExecutor def square(x): time.sleep(0.05); return x * x with ThreadPoolExecutor(max_workers=4) as pool: print(list(pool.map(square, range(4)))) 追问：总耗时大约是 0.05 秒、0.2 秒还是更多？为什么 map 的结果顺序一定和输入一致？
  6. python from unittest.mock import MagicMock class Svc: def price(self, sku): return 100 m = MagicMock(spec=Svc) m.price.return_value = 5 print(m.price("a"), m.price.call_count, m.price.call_args.args) m.nope
  7. ```python # shop.py import time def now(): return time.time() def stamp(): return round(now())

# test_shop.py from unittest.mock import patch import shop with patch.object(shop, "now", return_value=41.6): print(shop.stamp()) with patch("time.time", return_value=41.6): print(shop.stamp()) 追问：如果 `shop.py` 写的是 `from time import time` 且 `now()` 调 `time()`，第二个 `patch("time.time")` 还有效吗？ 8.python async def main(): async def slow(): time.sleep(0.2) # 注意：不是 asyncio.sleep return "slow" async def fast(): await asyncio.sleep(0.01); return "fast" t0 = time.perf_counter() print(await asyncio.gather(slow(), fast()), round(time.perf_counter() - t0, 1)) asyncio.run(main()) 9.python import subprocess, sys r = subprocess.run([sys.executable, "-c", "import sys; print('out'); print('err', file=sys.stderr); sys.exit(3)"], capture_output=True, text=True) print(r.returncode, r.stdout.strip(), r.stderr.strip()) 追问：加上 `check=True` 会怎样？ 10.python import httpx def handler(request): return httpx.Response(503, json={"detail": "busy"}) client = httpx.Client(transport=httpx.MockTransport(handler)) r = client.get("http://x/items/1") print(r.status_code, r.json()["detail"]) r.raise_for_status() ```

二、找 bug（5 题）

  1. python async def fetch_all(urls): for url in urls: asyncio.create_task(fetch(url)) # 然后函数就返回了
  2. python async def get(url): async with httpx.AsyncClient() as client: # 每次调用都新建 return (await client.get(url)).json() # 没有超时、没有 raise_for_status
  3. python from concurrent.futures import ProcessPoolExecutor def work(n): return n * n with ProcessPoolExecutor() as pool: # 写在模块顶层，没有 __main__ 守卫 print(list(pool.map(work, range(5))))
  4. python async def worker(queue): while True: item = await queue.get() try: await handle(item) except asyncio.CancelledError: pass # 吞掉取消 queue.task_done()
  5. python def test_total(mocker): mocker.patch("inventory.PriceService.price_of", return_value=2.0) inv = Inventory(clock=SystemClock(), prices=RealPriceService()) assert inv.total_value() == 0.0 # 覆盖率工具说这一行跑过了 （测试通过了，但它到底验证了什么？）

三、写代码（5 题）

  1. 写 async def gather_with_limit(coros, limit)：最多同时跑 limit 个，结果顺序与输入一致（Semaphore + gather）。
  2. 写 async def fetch_json(client, url, timeout=3.0) -> dict | None：asyncio.timeout 限时，超时返回 None，4xx/5xx 抛 httpx.HTTPStatusError；再写它的 MockTransport 测试（一个 200、一个 500）。
  3. 写 run_command(cmd: list[str], timeout=10) -> tuple[int, str]：subprocess.run 捕获输出、UTF-8、超时抛自定义 CommandTimeout；解释为什么参数是列表而不是字符串。
  4. 写 pytest：给 Inventory.remove 写三个测试——正常出货、出货量超过库存抛 ValueError（用 match=）、剩余量 ≤ 阈值时 caplog 里出现 WARNING。
  5. 写 benchmark(fn, *args, repeat=5) -> float 返回最快一次耗时，并说明：为什么取最小值不取平均？为什么 fast_unique 用 set 就从 O(n²) 变 O(n)？