第 3 周 练习骨架（只读展示）¶

"""Week 3 · M13 · w3_02 的测试（已写好，不要改）

用法：uv run pytest exercises/week3/test_w3_02_threads_processes_subprocess.py -q
读输出：. 通过，F 断言失败，E 测试里抛了异常（比如还没实现的 NotImplementedError）。
说明：所有"网络请求"都用 time.sleep 桩代替，整个文件不联网、不到 2 秒跑完。
"""

import sys
import time

import pytest
from w3_02_threads_processes_subprocess import (
    CommandTimeout,
    count_primes_in_range,
    count_primes_parallel,
    fetch_all_threaded,
    producer_consumer_threads,
    python_version_via_subprocess,
    run_command,
    safe_counter,
)

SLEEP = 0.05
ITEMS = [f"item-{i}" for i in range(8)]


def slow_fetch(item: str) -> str:
    """假的"网络请求"：睡 0.05 秒再返回大写结果。"""
    time.sleep(SLEEP)
    return item.upper()


def boom_fetch(item: str) -> str:
    """对某一个 item 抛异常，用来验证异常会在 result() 时冒出来。"""
    if item == "item-3":
        raise RuntimeError("模拟请求失败")
    return item.upper()


def test_fetch_all_threaded_keeps_item_to_result_mapping() -> None:
    got = fetch_all_threaded(["a", "b", "c"], lambda s: s.upper())
    assert got == {"a": "A", "b": "B", "c": "C"}


def test_fetch_all_threaded_is_much_faster_than_sequential() -> None:
    start = time.perf_counter()
    got = fetch_all_threaded(ITEMS, slow_fetch, max_workers=8)
    elapsed = time.perf_counter() - start
    assert len(got) == len(ITEMS)
    assert elapsed < len(ITEMS) * SLEEP / 2


def test_fetch_all_threaded_propagates_worker_exception() -> None:
    with pytest.raises(RuntimeError, match="模拟请求失败"):
        fetch_all_threaded(ITEMS, boom_fetch)


@pytest.mark.parametrize(
    ("n_threads", "increments"),
    [(2, 1000), (8, 5000)],
    ids=["2线程", "8线程"],
)
def test_safe_counter_is_exact(n_threads: int, increments: int) -> None:
    assert safe_counter(n_threads, increments) == n_threads * increments


def test_count_primes_in_range() -> None:
    assert count_primes_in_range((0, 10)) == 4
    assert count_primes_in_range((10, 20)) == 4
    assert count_primes_in_range((0, 2)) == 0


def test_count_primes_parallel_matches_sequential() -> None:
    ranges = [(0, 2000), (2000, 4000), (4000, 6000)]
    expected = sum(count_primes_in_range(bounds) for bounds in ranges)
    assert count_primes_parallel(ranges) == expected


def test_run_command_captures_stdout() -> None:
    result = run_command([sys.executable, "-X", "utf8", "-c", "print('你好')"])
    assert result.returncode == 0
    assert result.stdout.strip() == "你好"
    assert result.duration >= 0


def test_run_command_nonzero_exit_does_not_raise() -> None:
    result = run_command([sys.executable, "-c", "import sys; sys.exit(3)"])
    assert result.returncode == 3
    assert result.stdout == ""


def test_run_command_writes_stderr() -> None:
    code = "import sys; print('出错了', file=sys.stderr); sys.exit(1)"
    result = run_command([sys.executable, "-X", "utf8", "-c", code])
    assert result.returncode == 1
    assert "出错了" in result.stderr


def test_run_command_timeout_raises_command_timeout() -> None:
    with pytest.raises(CommandTimeout):
        run_command([sys.executable, "-c", "import time; time.sleep(5)"], timeout=0.3)


def test_python_version_via_subprocess() -> None:
    version = python_version_via_subprocess()
    assert version.count(".") == 2
    assert version.startswith("3.")


def test_producer_consumer_threads_processes_all_items() -> None:
    assert producer_consumer_threads(["a", "b", "c"]) == ["A", "B", "C"]


def test_producer_consumer_threads_handles_empty_input() -> None:
    assert producer_consumer_threads([]) == []

"""Week 3 · M14a · w3_03 的测试（已写好，不要改）

用法：uv run pytest exercises/week3/test_w3_03_asyncio_basics.py -q
说明：pyproject.toml 里配了 asyncio_mode = "auto"，所以 async def test_x() 直接就能跑，
      不需要 @pytest.mark.asyncio；测试里的"网络"全是 asyncio.sleep。
"""

import time

import pytest
from w3_03_asyncio_basics import (
    fetch_fake,
    first_n_done,
    gather_all,
    run_group,
    sequential_vs_concurrent,
    with_thread,
)


async def test_fetch_fake_returns_key_ok() -> None:
    assert await fetch_fake("a", 0.001) == "a:ok"


async def test_gather_all_keeps_input_order() -> None:
    assert await gather_all(["a", "b", "c"]) == ["a:ok", "b:ok", "c:ok"]


async def test_gather_all_is_concurrent() -> None:
    keys = [f"k{i}" for i in range(10)]
    start = time.perf_counter()
    got = await gather_all(keys)
    elapsed = time.perf_counter() - start
    assert len(got) == 10
    assert elapsed < 0.15


async def test_gather_all_empty_list() -> None:
    assert await gather_all([]) == []


async def test_run_group_keeps_input_order() -> None:
    assert await run_group(["x", "y", "z"]) == ["x:ok", "y:ok", "z:ok"]


async def test_run_group_is_concurrent() -> None:
    keys = [f"k{i}" for i in range(10)]
    start = time.perf_counter()
    await run_group(keys)
    assert time.perf_counter() - start < 0.15


async def test_first_n_done_returns_fastest_first() -> None:
    coros = [fetch_fake("slow", 0.5), fetch_fake("fast", 0.01)]
    start = time.perf_counter()
    got = await first_n_done(coros, 1)
    assert got == ["fast:ok"]
    assert time.perf_counter() - start < 0.3


async def test_first_n_done_rejects_bad_n() -> None:
    coros = [fetch_fake("a", 0.001)]
    with pytest.raises(ValueError, match="n 必须"):
        await first_n_done(coros, 5)


async def test_with_thread_runs_blocking_function() -> None:
    start = time.perf_counter()
    assert await with_thread(str.upper, "abc") == "ABC"
    await with_thread(time.sleep, 0.05)
    assert time.perf_counter() - start >= 0.05


async def test_sequential_vs_concurrent() -> None:
    seq, conc = await sequential_vs_concurrent(5, 0.02)
    assert seq >= 5 * 0.02 * 0.8
    assert conc < seq / 2

"""Week 3 · M16 · w3_04 的测试（已写好，不要改）

用法：uv run pytest exercises/week3/test_w3_04_profiling.py -q
说明：性能断言只要求 fast 不慢于 slow（不写"快 10 倍"这种会随机器抖动的断言），
      正确性断言才是主角：fast_* 必须和 slow_* 逐个输入结果一致。
"""

import pytest
from w3_04_profiling import (
    benchmark,
    fast_count_matches,
    fast_fib,
    fast_join,
    fast_unique,
    peak_memory,
    profile_top,
    slow_count_matches,
    slow_fib,
    slow_join,
    slow_unique,
)

BIG_ITEMS = [i % 500 for i in range(3000)]
TEXTS = [f"line-{i}" if i % 3 else "no digits here" for i in range(2000)]


@pytest.mark.parametrize(
    "items",
    [[], [1], [1, 1, 1], [3, 1, 3, 2, 1], list(range(20)), BIG_ITEMS],
    ids=["空", "单个", "全重复", "混合", "无重复", "大输入"],
)
def test_fast_unique_matches_slow(items: list[int]) -> None:
    assert fast_unique(items) == slow_unique(items)


@pytest.mark.parametrize("n", [0, 1, 3, 100], ids=["0", "1", "3", "100"])
def test_fast_join_matches_slow(n: int) -> None:
    assert fast_join(n) == slow_join(n)


@pytest.mark.parametrize(
    ("pattern", "texts"),
    [(r"\d+", ["a1", "bb", "c3"]), (r"^no", TEXTS[:50]), (r"zzz", ["a", "b"])],
    ids=["数字", "行首", "无命中"],
)
def test_fast_count_matches_matches_slow(pattern: str, texts: list[str]) -> None:
    assert fast_count_matches(pattern, texts) == slow_count_matches(pattern, texts)


@pytest.mark.parametrize("n", [0, 1, 2, 10, 25], ids=["0", "1", "2", "10", "25"])
def test_fast_fib_matches_slow(n: int) -> None:
    assert fast_fib(n) == slow_fib(n)


def test_fast_fib_handles_big_n() -> None:
    assert fast_fib(60) == 1548008755920


def test_fast_unique_not_slower_than_slow() -> None:
    slow = benchmark(slow_unique, BIG_ITEMS, repeat=3)
    fast = benchmark(fast_unique, BIG_ITEMS, repeat=3)
    assert fast <= slow


def test_fast_join_not_slower_than_slow() -> None:
    slow = benchmark(slow_join, 20000, repeat=3)
    fast = benchmark(fast_join, 20000, repeat=3)
    assert fast <= slow


def test_fast_count_matches_not_slower_than_slow() -> None:
    slow = benchmark(slow_count_matches, r"\d+", TEXTS, repeat=3)
    fast = benchmark(fast_count_matches, r"\d+", TEXTS, repeat=3)
    assert fast <= slow


def test_fast_fib_not_slower_than_slow() -> None:
    slow = benchmark(slow_fib, 22, repeat=3)
    fast = benchmark(fast_fib, 22, repeat=3)
    assert fast <= slow


def test_benchmark_returns_positive_seconds() -> None:
    elapsed = benchmark(slow_fib, 15, repeat=2)
    assert 0 < elapsed < 5


def test_benchmark_rejects_non_positive_repeat() -> None:
    with pytest.raises(ValueError, match="repeat 必须为正"):
        benchmark(slow_fib, 5, repeat=0)


def test_profile_top_lists_hot_functions() -> None:
    names = profile_top(lambda: slow_fib(18), n=3)
    assert len(names) == 3
    assert all(isinstance(name, str) for name in names)
    assert "slow_fib" in names


def test_peak_memory_grows_with_allocation() -> None:
    small = peak_memory(lambda: [0] * 1000)
    big = peak_memory(lambda: [0] * 1_000_000)
    assert big > small
    assert big > 1_000_000

"""Week 3 · M14b · w3_05 的测试（已写好，不要改）

用法：uv run pytest exercises/week3/test_w3_05_asyncio_advanced.py -q
说明：全部离线；"慢请求"用 asyncio.sleep 模拟，超时/取消都是真的走 asyncio 机制。
"""

import asyncio
import time

import pytest
from w3_05_asyncio_advanced import (
    aticker,
    fetch_all_with_timeout,
    first_completed,
    gather_with_limit,
    producer_consumer,
    retry_async,
    run_group_collect_errors,
    timed_section,
)


async def work(key: str, delay: float = 0.01) -> str:
    """成功的假请求。"""
    await asyncio.sleep(delay)
    return f"{key}:ok"


async def boom(delay: float = 0.05) -> str:
    """失败的假请求（比成功的慢一点，好让 TaskGroup 的行为确定）。"""
    await asyncio.sleep(delay)
    raise ValueError("模拟失败")


async def test_gather_with_limit_keeps_order() -> None:
    got = await gather_with_limit([work("a"), work("b"), work("c")], limit=2)
    assert got == ["a:ok", "b:ok", "c:ok"]


async def test_gather_with_limit_actually_limits_concurrency() -> None:
    coros = [work(str(i), 0.05) for i in range(4)]
    start = time.perf_counter()
    await gather_with_limit(coros, limit=2)
    elapsed = time.perf_counter() - start
    assert elapsed >= 0.09
    assert elapsed < 0.19


async def test_gather_with_limit_rejects_bad_limit() -> None:
    coros = [work("a")]
    with pytest.raises(ValueError, match="limit 必须为正"):
        await gather_with_limit(coros, limit=0)


async def test_fetch_all_with_timeout_marks_slow_keys_none() -> None:
    async def fetch(key: str) -> str:
        await asyncio.sleep(1.0 if key == "slow" else 0.01)
        return f"{key}:ok"

    got = await fetch_all_with_timeout(fetch, ["slow", "fast"], timeout=0.1)
    assert got == {"slow": None, "fast": "fast:ok"}


async def test_fetch_all_with_timeout_all_ok() -> None:
    async def fetch(key: str) -> str:
        await asyncio.sleep(0.01)
        return f"{key}:ok"

    got = await fetch_all_with_timeout(fetch, ["a", "b"], timeout=0.5)
    assert got == {"a": "a:ok", "b": "b:ok"}


async def test_producer_consumer_processes_every_item() -> None:
    got = await producer_consumer(5, 2)
    assert got == [f"item-{i}:done" for i in range(5)]


async def test_producer_consumer_with_zero_items() -> None:
    assert await producer_consumer(0, 2) == []


async def test_aticker_yields_count_values() -> None:
    got = [i async for i in aticker(0.01, 3)]
    assert got == [0, 1, 2]


async def test_aticker_takes_at_least_interval_times_count() -> None:
    start = time.perf_counter()
    async for _ in aticker(0.02, 3):
        pass
    assert time.perf_counter() - start >= 0.05


async def test_retry_async_succeeds_after_failures() -> None:
    calls = 0

    async def flaky() -> str:
        nonlocal calls
        calls += 1
        if calls < 3:
            raise RuntimeError("暂时失败")
        return "ok"

    assert await retry_async(flaky, attempts=3, backoff=0.001) == "ok"
    assert calls == 3


async def test_retry_async_raises_last_error() -> None:
    async def always_fail() -> str:
        raise RuntimeError("一直失败")

    with pytest.raises(RuntimeError, match="一直失败"):
        await retry_async(always_fail, attempts=2, backoff=0.001)


async def test_retry_async_rejects_bad_attempts() -> None:
    async def ok() -> str:
        return "ok"

    with pytest.raises(ValueError, match="attempts 必须为正"):
        await retry_async(ok, attempts=0)


async def test_first_completed_returns_fastest() -> None:
    start = time.perf_counter()
    got = await first_completed([work("slow", 1.0), work("fast", 0.01)])
    assert got == "fast:ok"
    assert time.perf_counter() - start < 0.5


async def test_first_completed_rejects_empty() -> None:
    with pytest.raises(ValueError, match="coros 不能为空"):
        await first_completed([])


async def test_timed_section_records_elapsed() -> None:
    log: list[tuple[str, float]] = []
    async with timed_section("fetch", log):
        await asyncio.sleep(0.03)
    assert len(log) == 1
    label, elapsed = log[0]
    assert label == "fetch"
    assert elapsed >= 0.03


async def test_timed_section_records_even_on_error() -> None:
    log: list[tuple[str, float]] = []
    with pytest.raises(RuntimeError):
        async with timed_section("bad", log):
            raise RuntimeError("炸了")
    assert [label for label, _ in log] == ["bad"]


async def test_run_group_collect_errors_all_ok() -> None:
    results, errors = await run_group_collect_errors([work("a"), work("b")])
    assert sorted(results) == ["a:ok", "b:ok"]
    assert errors == []


async def test_run_group_collect_errors_collects_failure() -> None:
    coros = [work("a", 0.01), work("b", 0.01), boom(0.05)]
    results, errors = await run_group_collect_errors(coros)
    assert sorted(results) == ["a:ok", "b:ok"]
    assert len(errors) == 1
    assert isinstance(errors[0], ValueError)

"""Week 3 · M15 · w3_06 的测试（已写好，不要改）

用法：uv run pytest exercises/week3/test_w3_06_httpx.py -q
说明：全部用 httpx.MockTransport(handler) 假装服务器，**一个字节都不走网络**。
      handler 是个普通函数：收到 httpx.Request，返回 httpx.Response。
"""

import asyncio
import time
from pathlib import Path

import httpx
import pytest
from w3_06_httpx import (
    classify_error,
    download,
    fetch_many,
    get_json,
    get_json_retrying,
    make_client,
    stream_lines,
)

BASE_URL = "http://stub.local"


def handler(request: httpx.Request) -> httpx.Response:
    """假服务器：按路径返回不同响应。"""
    path = request.url.path
    if path.startswith("/items/"):
        item_id = path.rsplit("/", 1)[-1]
        return httpx.Response(200, json={"id": int(item_id), "name": f"物品{item_id}"})
    if path == "/lines":
        return httpx.Response(200, text="第一行\n\n第二行\n")
    if path == "/big.bin":
        return httpx.Response(200, content=b"x" * 1024)
    if path == "/not-found":
        return httpx.Response(404, json={"detail": "没有这个东西"})
    if path == "/boom":
        return httpx.Response(500, text="服务器内部错误")
    if path == "/slow":
        raise httpx.ReadTimeout("读超时", request=request)
    return httpx.Response(404, text="unknown stub path")


def make_test_client() -> httpx.Client:
    """给同步测试用的客户端（MockTransport 版）。"""
    return make_client(BASE_URL, timeout=1.0, transport=httpx.MockTransport(handler))


def make_async_test_client() -> httpx.AsyncClient:
    """给异步测试用的客户端（同一个 handler，MockTransport 两边都能用）。"""
    return httpx.AsyncClient(
        base_url=BASE_URL,
        timeout=httpx.Timeout(1.0),
        transport=httpx.MockTransport(handler),
    )


def test_make_client_sets_base_url_timeout_and_headers() -> None:
    with make_test_client() as client:
        assert str(client.base_url) == BASE_URL
        assert client.timeout.read == 1.0
        assert client.headers["user-agent"] == "aggregator-exercise/1.0"


def test_get_json_returns_parsed_body() -> None:
    with make_test_client() as client:
        assert get_json(client, "/items/7") == {"id": 7, "name": "物品7"}


def test_get_json_raises_on_404() -> None:
    with make_test_client() as client, pytest.raises(httpx.HTTPStatusError):
        get_json(client, "/not-found")


def test_get_json_raises_on_500() -> None:
    with make_test_client() as client, pytest.raises(httpx.HTTPStatusError):
        get_json(client, "/boom")


async def test_fetch_many_returns_results_and_errors() -> None:
    async with make_async_test_client() as aclient:
        got = await fetch_many(aclient, ["/items/1", "/items/2", "/not-found"], limit=2)
    assert got["/items/1"] == {"id": 1, "name": "物品1"}
    assert got["/items/2"] == {"id": 2, "name": "物品2"}
    assert isinstance(got["/not-found"], httpx.HTTPStatusError)


async def test_fetch_many_handles_empty_paths() -> None:
    async with make_async_test_client() as aclient:
        assert await fetch_many(aclient, [], limit=2) == {}


async def test_fetch_many_is_concurrent() -> None:
    async def slow_handler(request: httpx.Request) -> httpx.Response:
        await asyncio.sleep(0.02)
        return httpx.Response(200, json={"path": request.url.path})

    async with httpx.AsyncClient(
        base_url=BASE_URL, transport=httpx.MockTransport(slow_handler)
    ) as aclient:
        paths = [f"/items/{i}" for i in range(8)]
        start = time.perf_counter()
        got = await fetch_many(aclient, paths, limit=8)
    assert len(got) == 8
    assert time.perf_counter() - start < 8 * 0.02


async def test_stream_lines_skips_blank_lines() -> None:
    async with make_async_test_client() as aclient:
        assert await stream_lines(aclient, "/lines") == ["第一行", "第二行"]


async def test_stream_lines_raises_for_status() -> None:
    async with make_async_test_client() as aclient:
        with pytest.raises(httpx.HTTPStatusError):
            await stream_lines(aclient, "/boom")


def test_get_json_retrying_recovers_after_two_503() -> None:
    calls = 0

    def flaky(request: httpx.Request) -> httpx.Response:
        nonlocal calls
        calls += 1
        if calls < 3:
            return httpx.Response(503, text="服务暂时不可用")
        return httpx.Response(200, json={"ok": True, "calls": calls})

    with httpx.Client(
        base_url=BASE_URL, transport=httpx.MockTransport(flaky)
    ) as client:
        assert get_json_retrying(client, "/items/1", attempts=3) == {
            "ok": True,
            "calls": 3,
        }
    assert calls == 3


def test_get_json_retrying_gives_up_and_raises() -> None:
    calls = 0

    def always_503(request: httpx.Request) -> httpx.Response:
        nonlocal calls
        calls += 1
        return httpx.Response(503, text="一直不可用")

    with httpx.Client(
        base_url=BASE_URL, transport=httpx.MockTransport(always_503)
    ) as client:
        with pytest.raises(httpx.HTTPStatusError):
            get_json_retrying(client, "/items/1", attempts=3)
    assert calls == 3


def test_get_json_retrying_does_not_retry_404() -> None:
    calls = 0

    def not_found(request: httpx.Request) -> httpx.Response:
        nonlocal calls
        calls += 1
        return httpx.Response(404, text="没有")

    with httpx.Client(
        base_url=BASE_URL, transport=httpx.MockTransport(not_found)
    ) as client:
        with pytest.raises(httpx.HTTPStatusError):
            get_json_retrying(client, "/items/1", attempts=3)
    assert calls == 1


def test_download_writes_file_and_returns_size(tmp_path: Path) -> None:
    dest = tmp_path / "big.bin"
    with make_test_client() as client:
        size = download(client, "/big.bin", dest)
    assert size == 1024
    assert dest.stat().st_size == 1024


def test_download_raises_for_status(tmp_path: Path) -> None:
    with make_test_client() as client, pytest.raises(httpx.HTTPStatusError):
        download(client, "/boom", tmp_path / "nope.bin")


@pytest.mark.parametrize(
    ("exc", "expected"),
    [
        (httpx.ConnectTimeout("超时"), "timeout"),
        (httpx.ReadTimeout("超时"), "timeout"),
        (httpx.ConnectError("连不上"), "connect"),
        (httpx.RequestError("其它网络问题"), "network"),
        (
            httpx.HTTPStatusError(
                "404",
                request=httpx.Request("GET", BASE_URL),
                response=httpx.Response(404),
            ),
            "http_4xx",
        ),
        (
            httpx.HTTPStatusError(
                "500",
                request=httpx.Request("GET", BASE_URL),
                response=httpx.Response(500),
            ),
            "http_5xx",
        ),
        (ValueError("别的错"), "unknown"),
    ],
    ids=["连接超时", "读超时", "连接失败", "网络", "4xx", "5xx", "其他"],
)
def test_classify_error(exc: Exception, expected: str) -> None:
    assert classify_error(exc) == expected


def test_timeout_from_transport_is_classified() -> None:
    with make_test_client() as client:
        try:
            get_json(client, "/slow")
        except Exception as exc:
            assert classify_error(exc) == "timeout"
        else:
            raise AssertionError("应该抛超时异常")

"""Week 3 · M12 · 共享 fixture（给你一个范例，其余自己加）

学习目标：知道 conftest.py 里的 fixture 不用 import 就能被同目录测试用；
      会写"假实现"替代注入的依赖，会用 yield 做清理。
用法：不要直接运行本文件；它会被 pytest 自动加载。
做法：照 fixed_clock 的样子，把你常用的假对象（比如 FakePrices、装好货的 Inventory）
      也提成 fixture，测试里直接写在参数列表里就能拿到。
"""

import pytest


class FixedClock:
    """假时钟：now() 永远返回同一个字符串，于是断言时间戳变得可能。"""

    def __init__(self, stamp: str = "2026-09-21T09:00:00+00:00") -> None:
        self.stamp = stamp
        self.calls = 0

    def now(self) -> str:
        """返回固定时间，并记录被问过几次。"""
        self.calls += 1
        return self.stamp


@pytest.fixture
def fixed_clock() -> FixedClock:
    """范例 fixture：默认 function 作用域，每个测试拿到全新的一份。

    想改作用域就写 @pytest.fixture(scope="module")；
    需要清理就把 return 换成 yield，yield 之后的代码在测试结束后执行。
    """
    return FixedClock()

"""Week 3 · M12 · pytest 进阶：被测代码（完整实现，不要改）

学习目标：读懂一段有依赖注入（Clock / PriceService 两个 Protocol）的业务代码，
      为它设计"正常 / 边界 / 错误 / 副作用"四类测试用例。
用法：uv run pytest exercises/week3/w3_01_pytest_advanced -q
      看覆盖率再加：--cov=inventory --cov-report=term-missing
做法：本文件是题面，已经写完，一行都不要改；
      要写的是同目录的 test_inventory.py（≥12 个测试，覆盖率 ≥95%）。
      看不懂某个方法就先在 REPL 里手动调用它，观察返回值和异常。
"""

import csv
import json
import logging
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Protocol

logger = logging.getLogger(__name__)

DEFAULT_LOW_STOCK = 3


class Clock(Protocol):
    """时钟协议：有 now() 就能塞进 Inventory（测试里换成返回固定字符串的假时钟）。"""

    def now(self) -> str:
        """返回 ISO 8601 格式的时间字符串。"""
        ...


class PriceService(Protocol):
    """价格协议：真实实现要查数据库或打网络，测试里用假对象 / mock 替换。"""

    def price_of(self, sku: str) -> float:
        """返回单价；未知 sku 抛 KeyError。"""
        ...


class SystemClock:
    """生产环境用的真时钟。"""

    def now(self) -> str:
        """当前 UTC 时间，秒精度。例：2026-09-21T01:02:03+00:00"""
        return datetime.now(UTC).isoformat(timespec="seconds")


@dataclass(frozen=True, slots=True)
class Item:
    """一条库存记录：库存量与最后一次变动时间。"""

    sku: str
    quantity: int
    updated_at: str


class Inventory:
    """库存管理：加货、出货、估值、低库存预警、导出。

    时钟与价格都从构造函数注入，所以测试不需要联网也不需要等真实时间。
    例：
        inv = Inventory(clock=FixedClock("2026-09-21T09:00:00"), prices=FakePrices())
        inv.add("pen", 10)
        inv.total_value()
    """

    def __init__(
        self,
        clock: Clock,
        prices: PriceService,
        low_stock_threshold: int = DEFAULT_LOW_STOCK,
    ) -> None:
        self._clock = clock
        self._prices = prices
        self._threshold = low_stock_threshold
        self._items: dict[str, Item] = {}

    def __len__(self) -> int:
        """当前有几种 sku。"""
        return len(self._items)

    def get(self, sku: str) -> Item | None:
        """查一条记录，没有返回 None。"""
        return self._items.get(sku)

    def add(self, sku: str, quantity: int) -> Item:
        """加货：新 sku 直接建，老 sku 累加，返回加完之后的记录。

        sku 首尾空白会被去掉；空 sku 抛 ValueError("sku 不能为空")；
        quantity <= 0 抛 ValueError("quantity 必须为正，收到 ...")。
        例：add("pen", 3) 两次 -> Item(sku="pen", quantity=6, ...)
        """
        sku = sku.strip()
        if not sku:
            raise ValueError("sku 不能为空")
        if quantity <= 0:
            raise ValueError(f"quantity 必须为正，收到 {quantity}")
        current = self._items.get(sku)
        total = quantity if current is None else current.quantity + quantity
        item = Item(sku=sku, quantity=total, updated_at=self._clock.now())
        self._items[sku] = item
        return item

    def remove(self, sku: str, quantity: int) -> Item:
        """出货：返回出完之后的记录；剩余量 <= 阈值时打一条 WARNING 日志。

        quantity <= 0 抛 ValueError；未知 sku 抛 KeyError("未知 sku：...")；
        库存不足抛 ValueError("... 库存不足：现有 x，要出 y")。
        例：add("pen", 5) 后 remove("pen", 3) -> Item(quantity=2)，并记一条 WARNING
        """
        if quantity <= 0:
            raise ValueError(f"quantity 必须为正，收到 {quantity}")
        current = self._items.get(sku)
        if current is None:
            raise KeyError(f"未知 sku：{sku}")
        if quantity > current.quantity:
            raise ValueError(
                f"{sku} 库存不足：现有 {current.quantity}，要出 {quantity}"
            )
        left = current.quantity - quantity
        item = Item(sku=sku, quantity=left, updated_at=self._clock.now())
        self._items[sku] = item
        if left <= self._threshold:
            logger.warning("库存偏低：%s 只剩 %d 件", sku, left)
        return item

    def total_value(self) -> float:
        """总市值 = Σ 库存量 × 单价，保留 2 位小数；空库存为 0.0。

        单价来自注入的 PriceService，所以测试里可以用 mock 断言"每个 sku 各问了一次"。
        例：pen 单价 2.5、10 支 -> 25.0
        """
        total = 0.0
        for item in self._items.values():
            total += item.quantity * self._prices.price_of(item.sku)
        return round(total, 2)

    def low_stock(self) -> list[str]:
        """库存量 <= 阈值的 sku 列表，按字母排序。例：["ink", "pen"]"""
        return sorted(
            sku for sku, item in self._items.items() if item.quantity <= self._threshold
        )

    def export(self, path: Path) -> int:
        """把库存写到 path，返回写出的记录条数。

        后缀 .json 写 JSON 数组，.csv 写带表头的 CSV（列：sku,quantity,updated_at）；
        其他后缀抛 ValueError("不支持的导出格式：...")。记录按 sku 排序。
        例：export(tmp_path / "stock.csv") -> 2
        """
        rows = [asdict(self._items[sku]) for sku in sorted(self._items)]
        suffix = path.suffix.lower()
        if suffix == ".json":
            path.write_text(
                json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8"
            )
        elif suffix == ".csv":
            with path.open("w", encoding="utf-8", newline="") as f:
                writer = csv.DictWriter(f, fieldnames=["sku", "quantity", "updated_at"])
                writer.writeheader()
                writer.writerows(rows)
        else:
            raise ValueError(f"不支持的导出格式：{path.suffix}")
        return len(rows)