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

"""Week 1 小项目 · 文本处理管线（骨架）

用生成器管线统计一个目录下全部 .md 文件的词频、标题数、代码块数。
用法：uv run python exercises/week1/project_pipeline/pipeline.py <目录> [--top 20]
测试：uv run pytest exercises/week1/project_pipeline -q
做法：Registry/Timer/@timer/Pipeline 已给出；
      实现 7 个 TODO（5 个步骤 + summarize + Pipeline.stream）。

设计要点：
- 每个步骤都是"接收可迭代、产出迭代器"的生成器函数，用 @register_step 注册；
- 出错的文件不中断管线：Record.error 记录原因，后续步骤跳过它，summarize 统计错误数；
- Pipeline 按名字从注册表取步骤并串起来；Timer/@timer 负责计时。
"""

from __future__ import annotations

import argparse
import functools
import re
import time
from collections import Counter  # noqa: F401  (summarize 会用到)
from collections.abc import Callable, Iterable, Iterator
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Protocol

# ---------- 数据模型 ----------


@dataclass
class Record:
    """管线里流动的一条记录：一个文件。"""

    path: Path
    text: str = ""
    words: list[str] = field(default_factory=list)
    headings: int = 0
    code_blocks: int = 0
    error: str | None = None


@dataclass(frozen=True)
class Stats:
    """最终统计结果（不可变）。"""

    files: int
    errors: int
    headings: int
    code_blocks: int
    top_words: tuple[tuple[str, int], ...]


class PipelineError(Exception):
    """管线内部错误的基类。"""


# ---------- 注册表 ----------


class Step(Protocol):
    """一个步骤：接收可迭代对象，返回迭代器（通常是生成器函数）。"""

    def __call__(self, items: Iterable[Any]) -> Iterator[Any]: ...


class Registry:
    """名字 → 步骤 的注册表；`@registry.register("name")` 或 `@registry.register`。"""

    def __init__(self) -> None:
        self._steps: dict[str, Step] = {}

    def register(self, name: str | Callable | None = None) -> Any:
        def decorator(fn: Step) -> Step:
            key = fn.__name__ if name is None or callable(name) else name
            if key in self._steps:
                raise ValueError(f"步骤重名：{key}")
            self._steps[key] = fn
            return fn

        if callable(name):  # 不带括号使用：@registry.register
            return decorator(name)
        return decorator

    def get(self, name: str) -> Step:
        try:
            return self._steps[name]
        except KeyError as e:
            raise PipelineError(f"未知步骤：{name}") from e

    def names(self) -> list[str]:
        return list(self._steps)


registry = Registry()
register_step = registry.register

# ---------- 计时 ----------


class Timer:
    """上下文管理器：`with Timer() as t: ...; t.elapsed`。"""

    def __enter__(self) -> Timer:
        self._start = time.perf_counter()
        self.elapsed = 0.0
        return self

    def __exit__(self, *exc: object) -> None:
        self.elapsed = time.perf_counter() - self._start


def timer[**P, R](fn: Callable[P, R]) -> Callable[P, R]:
    """装饰器：把耗时（秒）记到 fn.last_elapsed 上。"""

    @functools.wraps(fn)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        with Timer() as t:
            result = fn(*args, **kwargs)
        wrapper.last_elapsed = t.elapsed  # type: ignore[attr-defined]
        return result

    wrapper.last_elapsed = 0.0  # type: ignore[attr-defined]
    return wrapper


# ---------- 步骤 ----------

CODE_BLOCK = re.compile(r"```.*?```", re.DOTALL)
HEADING = re.compile(r"^#{1,6}\s+\S", re.MULTILINE)
WORD = re.compile(r"[A-Za-z][A-Za-z'-]+|[\u4e00-\u9fff]+")


def iter_markdown(root: Path) -> Iterator[Path]:
    """惰性遍历目录下全部 .md 文件（按路径排序保证可重复）。"""
    yield from sorted(p for p in root.rglob("*.md") if p.is_file())


@register_step("read_files")
def read_files(paths: Iterable[Path]) -> Iterator[Record]:
    """读文件成 Record；读不了的文件不中断，记入 Record.error。"""
    # TODO: 在这里实现
    raise NotImplementedError


@register_step("strip_code_blocks")
def strip_code_blocks(records: Iterable[Record]) -> Iterator[Record]:
    """统计并删除 ``` 代码块，避免代码里的词污染词频。"""
    # TODO: 在这里实现
    raise NotImplementedError


@register_step("count_headings")
def count_headings(records: Iterable[Record]) -> Iterator[Record]:
    """统计 Markdown 标题行数。"""
    # TODO: 在这里实现
    raise NotImplementedError


@register_step("split_words")
def split_words(records: Iterable[Record]) -> Iterator[Record]:
    """切词：英文按单词、中文按连续汉字串。"""
    # TODO: 在这里实现
    raise NotImplementedError


@register_step("normalize")
def normalize(records: Iterable[Record]) -> Iterator[Record]:
    """小写化并去掉 1 个字符的词。"""
    # TODO: 在这里实现
    raise NotImplementedError


def summarize(records: Iterable[Record], top: int = 20) -> Stats:
    """终点：把记录流聚合成 Stats（这一步会消耗整个迭代器）。"""
    # TODO: 在这里实现
    raise NotImplementedError


# ---------- 管线 ----------

DEFAULT_STEPS = [
    "read_files",
    "strip_code_blocks",
    "count_headings",
    "split_words",
    "normalize",
]


class Pipeline:
    """按名字串起若干步骤：source → step1 → step2 → ... → summarize。"""

    def __init__(
        self, steps: Iterable[str] = DEFAULT_STEPS, registry: Registry = registry
    ) -> None:
        self.step_names = list(steps)
        self._steps = [registry.get(name) for name in self.step_names]

    def __len__(self) -> int:
        return len(self._steps)

    def __iter__(self) -> Iterator[str]:
        return iter(self.step_names)

    def __repr__(self) -> str:
        return f"Pipeline({' -> '.join(self.step_names)})"

    def stream(self, source: Iterable[Any]) -> Iterator[Any]:
        """只串联步骤、不聚合，仍是惰性的。"""
        # TODO: 在这里实现
        raise NotImplementedError

    @timer
    def run(self, source: Iterable[Any], top: int = 20) -> Stats:
        return summarize(self.stream(source), top=top)


def format_report(stats: Stats, elapsed: float) -> str:
    head = f"文件 {stats.files}（失败 {stats.errors}）"
    head += f"  标题 {stats.headings}  代码块 {stats.code_blocks}"
    lines = [head, f"耗时 {elapsed * 1000:.1f} ms", "词频："]
    lines += [f"  {w:<20}{n:>6}" for w, n in stats.top_words]
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="统计目录下 Markdown 的词频/标题/代码块"
    )
    parser.add_argument("root", type=Path)
    parser.add_argument("--top", type=int, default=20)
    args = parser.parse_args(argv)
    if not args.root.is_dir():
        parser.error(f"{args.root} 不是目录")
    pipeline = Pipeline()
    stats = pipeline.run(iter_markdown(args.root), top=args.top)
    elapsed: float = pipeline.run.last_elapsed  # type: ignore[attr-defined]
    print(format_report(stats, elapsed))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""文本处理管线的测试。运行：uv run pytest exercises/week1/project_pipeline -q"""

import itertools
from pathlib import Path

import pytest
from pipeline import (
    Pipeline,
    PipelineError,
    Record,
    Registry,
    Stats,
    Timer,
    count_headings,
    iter_markdown,
    normalize,
    read_files,
    split_words,
    strip_code_blocks,
    summarize,
)


@pytest.fixture
def docs(tmp_path: Path) -> Path:
    (tmp_path / "a.md").write_text(
        "# 标题一\n\nPython python 很好用。\n", encoding="utf-8"
    )
    (tmp_path / "sub").mkdir()
    (tmp_path / "sub" / "b.md").write_text(
        "## 标题二\n```python\nprint('code words here')\n```\n"
        "### 标题三\nhello Hello world\n",
        encoding="utf-8",
    )
    (tmp_path / "c.txt").write_text("不是 markdown", encoding="utf-8")
    (tmp_path / "bad.md").write_bytes(b"\xff\xfe\xfa bad bytes")
    return tmp_path


def test_iter_markdown_only_md_sorted(docs: Path):
    names = [p.name for p in iter_markdown(docs)]
    assert names == ["a.md", "bad.md", "b.md"]


def test_read_files_records_errors_instead_of_raising(docs: Path):
    records = list(read_files(iter_markdown(docs)))
    by_name = {r.path.name: r for r in records}
    assert by_name["a.md"].text.startswith("# 标题一")
    assert (
        by_name["bad.md"].error is not None
        and "UnicodeDecodeError" in by_name["bad.md"].error
    )


def test_steps_are_lazy():
    counter = itertools.count()

    def infinite_records():
        for i in counter:
            yield Record(path=Path(f"{i}.md"), text="# h\nword word")

    stream = normalize(
        split_words(count_headings(strip_code_blocks(infinite_records())))
    )
    first = next(stream)
    assert first.headings == 1 and first.words == ["word", "word"]
    assert next(counter) < 5  # 没有把无限源耗尽


def test_strip_code_blocks_and_headings(docs: Path):
    records = list(count_headings(strip_code_blocks(read_files(iter_markdown(docs)))))
    b = next(r for r in records if r.path.name == "b.md")
    assert b.code_blocks == 1 and "print" not in b.text
    assert b.headings == 2


def test_split_and_normalize():
    rec = Record(path=Path("x.md"), text="Python python A 很好用 hello-world")
    (out,) = list(normalize(split_words(iter([rec]))))
    assert out.words == ["python", "python", "很好用", "hello-world"]


def test_registry_rejects_duplicates_and_unknown():
    reg = Registry()

    @reg.register("s")
    def s(items):
        yield from items

    with pytest.raises(ValueError):
        reg.register("s")(s)
    with pytest.raises(PipelineError):
        reg.get("nope")
    assert reg.names() == ["s"]


def test_pipeline_run_returns_stats(docs: Path):
    p = Pipeline()
    assert len(p) == 5 and list(p)[0] == "read_files" and "read_files" in repr(p)
    stats = p.run(iter_markdown(docs), top=3)
    assert isinstance(stats, Stats)
    assert stats.files == 3 and stats.errors == 1
    assert stats.headings == 3 and stats.code_blocks == 1
    assert stats.top_words[0] in {("python", 2), ("hello", 2)}
    assert p.run.last_elapsed >= 0


def test_summarize_and_timer():
    stats = summarize(iter([Record(path=Path("x"), words=["a", "a", "b"])]))
    assert stats.top_words == (("a", 2), ("b", 1))
    with Timer() as t:
        sum(range(1000))
    assert t.elapsed >= 0

"""Week 1 · M1a · 生成器练习的测试（已写好，不要改）

用法：uv run pytest exercises/week1/test_w1_01_generators.py -q
读输出：. 通过，F 断言失败，E 测试里抛了异常（骨架还没实现时全是 E）。
惰性相关的测试都用 itertools.count() 这类无限迭代器：
如果你的实现先把输入读完，测试会挂住而不是失败——那也说明写错了。
"""

import inspect
from itertools import count, islice
from pathlib import Path

import pytest
from w1_01_generators import (
    Countdown,
    countdown,
    flatten,
    parse_records,
    read_chunks,
    running_mean,
    take,
)


def test_countdown_values() -> None:
    assert list(countdown(3)) == [3, 2, 1, 0]
    assert list(countdown(0)) == [0]


def test_countdown_is_generator() -> None:
    gen = countdown(3)
    assert inspect.isgenerator(gen)
    assert next(gen) == 3


def test_countdown_exhausted_after_one_pass() -> None:
    gen = countdown(2)
    assert list(gen) == [2, 1, 0]
    assert list(gen) == []


def test_read_chunks_joins_back(tmp_path: Path) -> None:
    text = "中文和 ASCII 混排\n第二行\n"
    path = tmp_path / "sample.txt"
    path.write_text(text, encoding="utf-8")
    assert "".join(read_chunks(path, 3)) == text


def test_read_chunks_sizes(tmp_path: Path) -> None:
    path = tmp_path / "abc.txt"
    path.write_text("abcde", encoding="utf-8")
    assert list(read_chunks(path, 2)) == ["ab", "cd", "e"]


def test_read_chunks_rejects_bad_size(tmp_path: Path) -> None:
    path = tmp_path / "abc.txt"
    path.write_text("abc", encoding="utf-8")
    with pytest.raises(ValueError):
        list(read_chunks(path, 0))


def test_parse_records_skips_blank_and_comments() -> None:
    lines = ["alice,90", "", "  ", "# 注释", "bob,80"]
    assert list(parse_records(lines)) == [
        {"name": "alice", "score": 90},
        {"name": "bob", "score": 80},
    ]


def test_parse_records_is_lazy() -> None:
    infinite = (f"user{i},{i}" for i in count())
    assert take(2, parse_records(infinite)) == [
        {"name": "user0", "score": 0},
        {"name": "user1", "score": 1},
    ]


def test_parse_records_reports_line_number() -> None:
    with pytest.raises(ValueError, match="3"):
        list(parse_records(["a,1", "b,2", "c,oops"]))


def test_parse_records_rejects_wrong_field_count() -> None:
    with pytest.raises(ValueError):
        list(parse_records(["a,1,2"]))


def test_take_finite() -> None:
    assert take(2, [1, 2, 3]) == [1, 2]
    assert take(10, [1, 2]) == [1, 2]


def test_take_infinite() -> None:
    assert take(3, count()) == [0, 1, 2]
    assert take(0, count()) == []


def test_take_rejects_negative() -> None:
    with pytest.raises(ValueError):
        take(-1, [1, 2])


def test_running_mean_values() -> None:
    assert list(running_mean([2, 4, 6])) == [2.0, 3.0, 4.0]
    assert list(running_mean([])) == []


def test_running_mean_is_lazy() -> None:
    assert list(islice(running_mean(count(1)), 3)) == [1.0, 1.5, 2.0]


def test_flatten_keeps_strings_whole() -> None:
    assert list(flatten([1, [2, [3, "ab"]]])) == [1, 2, 3, "ab"]


def test_flatten_tuples_and_empty() -> None:
    assert list(flatten([(1, 2), [], [[[3]]]])) == [1, 2, 3]
    assert list(flatten([])) == []


def test_countdown_class_iterates() -> None:
    assert list(Countdown(2)) == [2, 1, 0]
    collected: list[int] = []
    for value in Countdown(1):
        collected.append(value)
    assert collected == [1, 0]


def test_countdown_class_is_its_own_iterator() -> None:
    c = Countdown(1)
    assert iter(c) is c
    assert list(c) == [1, 0]
    assert list(c) == []

"""Week 1 · M1b · itertools 练习的测试（已写好，不要改）

用法：uv run pytest exercises/week1/test_w1_02_itertools.py -q
惰性相关的测试传的是 itertools.count()：先耗尽输入的实现会挂住。
"""

from itertools import count, islice, pairwise

import pytest
from w1_02_itertools import (
    chunked,
    cumulative_max,
    first_true,
    group_by_key,
    interleave,
    memory_report,
    top_k_pairs,
    window,
)


def test_chunked_full_and_partial() -> None:
    assert list(chunked([1, 2, 3, 4], 2)) == [(1, 2), (3, 4)]
    assert list(chunked([1, 2, 3], 2)) == [(1, 2), (3,)]
    assert list(chunked([], 2)) == []


def test_chunked_is_lazy() -> None:
    assert list(islice(chunked(count(), 3), 2)) == [(0, 1, 2), (3, 4, 5)]


def test_chunked_rejects_bad_size() -> None:
    with pytest.raises(ValueError):
        list(chunked([1, 2], 0))
    with pytest.raises(ValueError):
        list(chunked([1, 2], -3))


def test_window_equals_pairwise() -> None:
    data = [1, 2, 3, 4]
    assert list(window(data, 2)) == list(pairwise(data))


def test_window_larger_n_and_too_short() -> None:
    assert list(window([1, 2, 3, 4], 3)) == [(1, 2, 3), (2, 3, 4)]
    assert list(window([1, 2], 3)) == []


def test_window_is_lazy() -> None:
    assert list(islice(window(count(), 2), 3)) == [(0, 1), (1, 2), (2, 3)]


def test_window_rejects_bad_n() -> None:
    with pytest.raises(ValueError):
        list(window([1, 2], 0))


def test_group_by_key_sorts_first() -> None:
    records = [
        {"tag": "b", "n": 1},
        {"tag": "a", "n": 2},
        {"tag": "b", "n": 3},
    ]
    grouped = group_by_key(records, "tag")
    assert set(grouped) == {"a", "b"}
    assert grouped["b"] == [{"tag": "b", "n": 1}, {"tag": "b", "n": 3}]
    assert grouped["a"] == [{"tag": "a", "n": 2}]


def test_group_by_key_returns_plain_dict() -> None:
    grouped = group_by_key([{"t": "x"}], "t")
    assert type(grouped) is dict
    assert grouped == {"x": [{"t": "x"}]}
    assert group_by_key([], "t") == {}


def test_interleave_equal_lengths() -> None:
    assert list(interleave([1, 2], [10, 20])) == [1, 10, 2, 20]


def test_interleave_uneven_lengths() -> None:
    assert list(interleave([1, 2, 3], "ab")) == [1, "a", 2, "b", 3]
    assert list(interleave()) == []


def test_first_true_found() -> None:
    assert first_true([1, 3, 4, 6], lambda x: x % 2 == 0) == 4


def test_first_true_default_and_lazy() -> None:
    assert first_true([1, 3], lambda x: x % 2 == 0) is None
    assert first_true([1, 3], lambda x: x > 100, default=-1) == -1
    assert first_true(count(), lambda x: x == 5) == 5


def test_cumulative_max_values() -> None:
    assert cumulative_max([3, 1, 4, 1, 5]) == [3, 3, 4, 4, 5]
    assert cumulative_max([]) == []


def test_cumulative_max_single_and_decreasing() -> None:
    assert cumulative_max([7]) == [7]
    assert cumulative_max([5, 4, 3]) == [5, 5, 5]


def test_top_k_pairs_counts() -> None:
    words = ["a", "b", "a", "c", "a", "b"]
    assert top_k_pairs(words, 2) == [("a", 3), ("b", 2)]


def test_top_k_pairs_edge_cases() -> None:
    assert top_k_pairs(["a"], 5) == [("a", 1)]
    assert top_k_pairs(["a"], 0) == []
    assert top_k_pairs([], 3) == []


def test_memory_report_keys() -> None:
    report = memory_report()
    assert set(report) == {"list_bytes", "gen_bytes"}
    assert all(isinstance(v, int) for v in report.values())


def test_memory_report_generator_is_much_smaller() -> None:
    report = memory_report()
    assert report["gen_bytes"] * 1000 < report["list_bytes"]

"""Week 1 · M2a · 闭包与装饰器练习的测试（已写好，不要改）

用法：uv run pytest exercises/week1/test_w1_03_decorators.py -q
注意 test_timer_keeps_metadata：忘了 @wraps 的实现会在这里失败。
"""

import pytest
from w1_03_decorators import (
    compose,
    log_calls,
    make_counter,
    once,
    retry,
    timer,
    validate_positive,
)


def test_make_counter_increments() -> None:
    counter = make_counter()
    assert counter() == 1
    assert counter() == 2
    assert counter() == 3


def test_make_counter_instances_are_independent() -> None:
    a = make_counter()
    b = make_counter()
    assert a() == 1
    assert a() == 2
    assert b() == 1


def test_timer_returns_result() -> None:
    @timer
    def add(a: int, b: int) -> int:
        return a + b

    assert add(1, 2) == 3


def test_timer_keeps_metadata() -> None:
    @timer
    def add(a: int, b: int) -> int:
        """加法。"""
        return a + b

    assert add.__name__ == "add"
    assert add.__doc__ == "加法。"


def test_timer_prints_elapsed(capsys: pytest.CaptureFixture[str]) -> None:
    @timer
    def slow() -> str:
        return "ok"

    assert slow() == "ok"
    out = capsys.readouterr().out
    assert "slow" in out
    assert "耗时" in out


def test_log_calls_format() -> None:
    logs: list[str] = []

    @log_calls(logs)
    def add(a: int, b: int) -> int:
        return a + b

    assert add(2, 3) == 5
    assert logs == ["add(2, 3) -> 5"]


def test_log_calls_kwargs_and_order() -> None:
    logs: list[str] = []

    @log_calls(logs)
    def power(base: int, exp: int = 2) -> int:
        return base**exp

    assert power(2, exp=3) == 8
    assert power(3) == 9
    assert logs == ["power(2, exp=3) -> 8", "power(3) -> 9"]


def test_log_calls_keeps_metadata() -> None:
    logs: list[str] = []

    @log_calls(logs)
    def hello() -> None:
        """打招呼。"""

    hello()
    assert hello.__name__ == "hello"
    assert hello.__doc__ == "打招呼。"


def test_retry_succeeds_on_third_call() -> None:
    calls = {"n": 0}

    @retry(times=3, exceptions=(ValueError,), delay=0.0)
    def flaky() -> int:
        calls["n"] += 1
        if calls["n"] < 3:
            raise ValueError("还没好")
        return 42

    assert flaky() == 42
    assert calls["n"] == 3


def test_retry_reraises_after_last_attempt() -> None:
    calls = {"n": 0}

    @retry(times=3, delay=0.0)
    def always_bad() -> None:
        calls["n"] += 1
        raise ValueError("坏了")

    with pytest.raises(ValueError, match="坏了"):
        always_bad()
    assert calls["n"] == 3


def test_retry_does_not_swallow_other_exceptions() -> None:
    calls = {"n": 0}

    @retry(times=3, exceptions=(ValueError,), delay=0.0)
    def wrong_kind() -> None:
        calls["n"] += 1
        raise KeyError("别的错")

    with pytest.raises(KeyError):
        wrong_kind()
    assert calls["n"] == 1


def test_retry_rejects_bad_times() -> None:
    with pytest.raises(ValueError):
        retry(times=0)


def test_validate_positive_allows_positive() -> None:
    @validate_positive
    def area(w: float, h: float) -> float:
        return w * h

    assert area(2, 3) == 6


def test_validate_positive_rejects_non_positive() -> None:
    @validate_positive
    def area(w: float, h: float) -> float:
        return w * h

    with pytest.raises(ValueError):
        area(2, 0)
    with pytest.raises(ValueError):
        area(-1, 3)


def test_once_runs_body_only_once() -> None:
    calls = {"n": 0}

    @once
    def setup(value: int = 1) -> int:
        calls["n"] += 1
        return value

    assert setup(5) == 5
    assert setup(9) == 5
    assert setup() == 5
    assert calls["n"] == 1


def test_once_keeps_metadata() -> None:
    @once
    def setup() -> str:
        """只跑一次。"""
        return "done"

    assert setup.__name__ == "setup"
    assert setup.__doc__ == "只跑一次。"


def test_compose_order() -> None:
    assert compose(str, lambda x: x + 1)(1) == "2"
    assert compose(lambda x: x * 2, lambda x: x + 3)(1) == 8


def test_compose_identity_and_three() -> None:
    assert compose()(7) == 7
    assert compose(str, lambda x: x * 2, lambda x: x + 1)(2) == "6"

"""Week 1 · M2b · 注册表与 inspect 练习的测试（已写好，不要改）

用法：uv run pytest exercises/week1/test_w1_04_registry_inspect.py -q
describe 的期望结果写得很死：先照着 docstring 里的格式实现，再回头看这里。
"""

import time

import pytest
from w1_04_registry_inspect import (
    CALL_COUNTS,
    Registry,
    cached_fib,
    call_with_kwargs,
    describe,
    naive_fib,
    repeat,
    to_str,
)


def test_registry_with_explicit_name() -> None:
    reg = Registry()

    @reg.register("add")
    def add(a: int, b: int) -> int:
        return a + b

    assert reg.names() == ["add"]
    assert reg.get("add")(1, 2) == 3
    assert add(1, 2) == 3  # 装饰器要返回原函数


def test_registry_without_parentheses() -> None:
    reg = Registry()

    @reg.register
    def double(x: int) -> int:
        return x * 2

    @reg.register()
    def triple(x: int) -> int:
        return x * 3

    assert reg.names() == ["double", "triple"]
    assert reg.get("double")(2) == 4
    assert reg.get("triple")(2) == 6


def test_registry_rejects_duplicate_names() -> None:
    reg = Registry()

    @reg.register("add")
    def add(a: int, b: int) -> int:
        return a + b

    with pytest.raises(ValueError, match="add"):

        @reg.register("add")
        def add2(a: int, b: int) -> int:
            return a + b


def test_registry_get_missing_name() -> None:
    reg = Registry()
    assert reg.names() == []
    with pytest.raises(KeyError):
        reg.get("nope")


def test_repeat_collects_results() -> None:
    calls: list[int] = []

    @repeat(3)
    def tick() -> int:
        calls.append(1)
        return len(calls)

    assert tick() == [1, 2, 3]
    assert len(calls) == 3


def test_repeat_metadata_and_bad_n() -> None:
    @repeat(2)
    def hi(name: str) -> str:
        """打招呼。"""
        return f"hi {name}"

    assert hi("a") == ["hi a", "hi a"]
    assert hi.__name__ == "hi"
    assert hi.__doc__ == "打招呼。"
    with pytest.raises(ValueError):
        repeat(0)


def test_cached_fib_values() -> None:
    assert cached_fib(0) == 0
    assert cached_fib(1) == 1
    assert cached_fib(10) == 55


def test_cached_fib_is_fast() -> None:
    start = time.perf_counter()
    assert cached_fib(80) == 23416728348467685
    assert time.perf_counter() - start < 1.0


def test_cached_fib_calls_far_fewer_times_than_naive() -> None:
    CALL_COUNTS["naive_fib"] = 0
    CALL_COUNTS["cached_fib"] = 0
    cached_fib.cache_clear()
    assert naive_fib(20) == cached_fib(20) == 6765
    assert CALL_COUNTS["naive_fib"] > 10_000
    assert CALL_COUNTS["cached_fib"] <= 21


def test_describe_annotated_function() -> None:
    def f(a: int, b: str = "x") -> str:
        """示例"""
        return b * a

    assert describe(f) == {
        "name": "f",
        "doc": "示例",
        "params": [
            {"name": "a", "type": "int", "default": None, "required": True},
            {"name": "b", "type": "str", "default": "x", "required": False},
        ],
        "returns": "str",
    }


def test_describe_without_annotations() -> None:
    def g(x, flag=False):
        return x

    result = describe(g)
    assert result["name"] == "g"
    assert result["doc"] == ""
    assert result["returns"] == "Any"
    assert result["params"] == [
        {"name": "x", "type": "Any", "default": None, "required": True},
        {"name": "flag", "type": "Any", "default": False, "required": False},
    ]


def test_describe_keyword_only_param() -> None:
    def h(a: int, *, verbose: bool = False) -> None:
        """带关键字参数。"""

    params = describe(h)["params"]
    assert [p["name"] for p in params] == ["a", "verbose"]
    assert params[1] == {
        "name": "verbose",
        "type": "bool",
        "default": False,
        "required": False,
    }


def test_call_with_kwargs_picks_and_ignores() -> None:
    def f(a: int, b: int = 2) -> int:
        return a + b

    assert call_with_kwargs(f, {"a": 1, "zzz": 9}) == 3
    assert call_with_kwargs(f, {"a": 1, "b": 10}) == 11


def test_call_with_kwargs_missing_required() -> None:
    def f(a: int, b: int) -> int:
        return a + b

    with pytest.raises(TypeError, match="b"):
        call_with_kwargs(f, {"a": 1})


def test_to_str_dispatch() -> None:
    assert to_str(3) == "整数 3"
    assert to_str([1, 2]) == "列表(2 项)"


def test_to_str_default_branch() -> None:
    assert to_str("a") == "'a'"
    assert to_str({"k": 1}) == "{'k': 1}"
    assert to_str([]) == "列表(0 项)"