writer-work-flow/packages/agents/tests/test_prompt_loader.py

"""prompt_loader 单测（Prompt 外置方案A）。

步0 先落「金标准比对」用例（此刻 load_prompt 未实现 → RED）；步2 补齐缓存/
fail-fast/规整/目录完整性/文件层尾换行等用例。

金标准 = packages/agents/tests/fixtures/prompt_hashes.json，取自内置 spec 的
运行时 system_prompt（见 _gen_golden.py）。比对 load_prompt 输出的 sha256 == 金标准，
即可证明 prompt 外置到 .md 后字节级零变化（不变量 #9 / 设计 §6）。
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path

import pytest
from ww_agents import (
    REVIEW_RESERVED_NAMES,
    SCHEMA_CATALOG,
    SPECS,
    continuity_spec,
    output_schema_for,
)
from ww_agents.prompt_loader import (
    _CACHE,
    PROMPTS_DIR,
    PromptNotFoundError,
    load_prompt,
)

FIXTURE = Path(__file__).parent / "fixtures" / "prompt_hashes.json"


def _golden() -> dict[str, str]:
    data: dict[str, str] = json.loads(FIXTURE.read_text(encoding="utf-8"))
    return data


def _sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


@pytest.mark.parametrize("name", sorted(_golden().keys()))
def test_load_prompt_matches_golden(name: str) -> None:
    # Arrange
    expected = _golden()[name]
    # Act
    actual = _sha256(load_prompt(name))
    # Assert
    assert actual == expected, f"prompt {name!r} 字节漂移：load_prompt 输出与金标准不符"


# ---- #1 注册表唯一性 ----
def test_specs_registry_len_is_21() -> None:
    assert len(SPECS) == 21
    # name 即 key，dict 已去重；逐项确认 key == spec.name（无错位）
    assert all(key == spec.name for key, spec in SPECS.items())


# ---- #2 md ↔ spec ↔ catalog 一一对应（覆盖 style.md / character-gen.md 连字符）----
def test_md_spec_catalog_one_to_one() -> None:
    md_stems = {p.stem for p in PROMPTS_DIR.glob("*.md")}
    assert set(SPECS) == md_stems == set(SCHEMA_CATALOG)
    # 易错连字符 / 非变量名命名显式覆盖
    assert "style" in md_stems and "style_drift" not in md_stems
    assert "character-gen" in md_stems
    assert "golden-finger" in md_stems
    assert "book-title" in md_stems
    assert "fine-outline" in md_stems
    assert "de-ai" in md_stems


# ---- #3 load_prompt 正常 + 缓存命中 ----
def test_load_prompt_returns_text_and_caches() -> None:
    # Arrange
    _CACHE.pop("continuity", None)
    # Act
    first = load_prompt("continuity")
    assert "continuity" in _CACHE  # 首次调用后已缓存
    second = load_prompt("continuity")
    # Assert
    assert first and isinstance(first, str)
    assert first is second  # 二次调用命中 _CACHE，返回同一对象


# ---- #4 load_prompt fail-fast ----
def test_load_prompt_unknown_name_raises() -> None:
    with pytest.raises(PromptNotFoundError):
        load_prompt("does-not-exist-spec-name")


# ---- #6 行尾确定性（CRLF/CR → LF）----
def test_load_prompt_normalizes_line_endings(tmp_path: Path) -> None:
    # Arrange — 同一文本的 LF / CRLF / CR 三版
    body = "第一行\n第二行\n第三行"
    (tmp_path / "lf.md").write_text(body + "\n", encoding="utf-8", newline="")
    (tmp_path / "crlf.md").write_bytes((body + "\n").replace("\n", "\r\n").encode("utf-8"))
    (tmp_path / "cr.md").write_bytes((body + "\n").replace("\n", "\r").encode("utf-8"))

    def _load(stem: str) -> str:
        _CACHE.pop(stem, None)
        from ww_agents import prompt_loader

        orig = prompt_loader.PROMPTS_DIR
        prompt_loader.PROMPTS_DIR = tmp_path  # type: ignore[misc]
        try:
            return prompt_loader.load_prompt(stem)
        finally:
            prompt_loader.PROMPTS_DIR = orig  # type: ignore[misc]

    # Act / Assert — 三版规整后 hash 一致
    lf = _load("lf")
    assert _sha256(lf) == _sha256(_load("crlf")) == _sha256(_load("cr"))


# ---- #6b BOM + NFC 规整 ----
def test_load_prompt_strips_bom_and_normalizes_nfc(tmp_path: Path) -> None:
    import unicodedata

    # Arrange — 带 BOM 的 NFD 全角文本 vs 干净 NFC
    text_nfd = unicodedata.normalize("NFD", "全角：测试。")
    (tmp_path / "bom.md").write_bytes("".encode() + (text_nfd + "\n").encode("utf-8"))
    (tmp_path / "clean.md").write_text(
        unicodedata.normalize("NFC", "全角：测试。") + "\n", encoding="utf-8"
    )

    from ww_agents import prompt_loader

    orig = prompt_loader.PROMPTS_DIR
    prompt_loader.PROMPTS_DIR = tmp_path  # type: ignore[misc]
    try:
        _CACHE.pop("bom", None)
        _CACHE.pop("clean", None)
        bom = prompt_loader.load_prompt("bom")
        clean = prompt_loader.load_prompt("clean")
    finally:
        prompt_loader.PROMPTS_DIR = orig  # type: ignore[misc]

    # Assert — 无 BOM 首字节，NFD → NFC 归一后与 clean 一致
    assert not bom.startswith("")
    assert _sha256(bom) == _sha256(clean)


# ---- #9 / #10 output_schema_for + refiner None ----
def test_output_schema_for_builtin_and_refiner() -> None:
    assert output_schema_for("continuity") is SCHEMA_CATALOG["continuity"]
    assert output_schema_for("refiner") is None
    assert SCHEMA_CATALOG["refiner"] is None
    # 精确匹配：拼错近似 name 不误命中连字符版（KeyError，调用方按需 catch）
    with pytest.raises(KeyError):
        output_schema_for("character_gen")


# ---- #13 文件层尾换行契约：每个 .md 磁盘尾部 LF ≤ 1 ----
@pytest.mark.parametrize("md", sorted(PROMPTS_DIR.glob("*.md")), ids=lambda p: p.stem)
def test_md_trailing_lf_at_most_one(md: Path) -> None:
    raw = md.read_bytes()
    trailing = len(raw) - len(raw.rstrip(b"\n"))
    assert trailing <= 1, f"{md.name} 尾部 LF={trailing} > 1"


# ---- #14 同一实例不变量 + 四审保留名安全边界 ----
def test_specs_identity_and_review_reserved_names() -> None:
    assert SPECS["continuity"] is continuity_spec  # 同一实例（兼容期不双真相）
    assert REVIEW_RESERVED_NAMES <= set(SPECS)
    for name in REVIEW_RESERVED_NAMES:
        spec = SPECS[name]
        assert spec.scope == "builtin"
        assert spec.writes == ()


# ---- #15 reads/writes 为不可变 tuple：四审 writes 不可被 .append 旁路（不变量 #3）----
def test_review_specs_writes_immutable_tuple() -> None:
    for name in REVIEW_RESERVED_NAMES:
        spec = SPECS[name]
        assert isinstance(spec.writes, tuple)
        assert isinstance(spec.reads, tuple)
        # tuple 无 append/clear → 任何旁路 writes 守卫的尝试 fail-fast
        with pytest.raises(AttributeError):
            spec.writes.append("INJECTED")  # type: ignore[attr-defined]


# ---- #16 SPECS 注册表运行时只读（MappingProxyType）：单一真相源不可被污染 ----
def test_specs_registry_is_read_only() -> None:
    with pytest.raises(TypeError):
        SPECS["injected"] = continuity_spec  # type: ignore[index]
    with pytest.raises(TypeError):
        del SPECS["outliner"]  # type: ignore[attr-defined]


# ---- #17 .gitattributes 锁 prompts/*.md LF（防 CRLF 漂移，设计 §4/§8）----
def test_gitattributes_locks_prompts_lf() -> None:
    repo_root = Path(__file__).resolve().parents[3]
    gitattributes = repo_root / ".gitattributes"
    assert gitattributes.is_file(), ".gitattributes 缺失（防 CRLF 漂移守卫）"
    content = gitattributes.read_text(encoding="utf-8")
    assert "prompts/*.md text eol=lf" in content