ersatztv/scripts/tests/test_decisions_validate.py

import subprocess
from pathlib import Path
from typing import Any

import scripts.decisions_lib as dl
import scripts.decisions_validate as dv


def _rec(**kw: Any) -> dl.Record:
    base: dict[str, Any] = dict(
        heading="H",
        source=Path("x"),
        lineno=1,
        status="active",
        since="2026-01-01",
        supersedes="none",
        superseded_by="none",
        signals="concept · paths: a/b.py · issues: #1",
    )
    base.update(kw)
    return dl.Record(**base)


def _v(recs: list[dl.Record], **kw: Any) -> list[str]:
    args: dict[str, Any] = dict(
        archive_keys=set(),
        catalog_ok=True,
        budget_ok=True,
        removed=[],
        rewritten=[],
        archive_records=[],
        demoted=[],
    )
    args.update(kw)
    return dv.validate(recs, **args)


def test_two_active_same_key_fails():
    assert any("more than one active" in e for e in _v([_rec(key="a.b"), _rec(key="a.b")]))


def test_bad_key_format_fails():
    assert any("key format" in e for e in _v([_rec(key="BadKey")]))


def test_dangling_superseded_by_fails():
    recs = [_rec(key="a.b", status="superseded", superseded_by="a.c@2026-07-01")]
    assert any("superseded-by" in e and "a.c" in e for e in _v(recs))


def test_superseded_by_resolves_to_archive_key_passes():
    # successor lives in archive → known via archive_keys, no dangling error
    recs = [_rec(key="a.b", status="superseded", superseded_by="a.c@2026-07-01")]
    assert not any("superseded-by" in e for e in _v(recs, archive_keys={"a.c"}))


def test_reciprocal_superseded_by_with_no_back_link_fails():
    # B exists (has a record) but its `supersedes` does not point back to A → dangling back-link.
    a = _rec(heading="A", key="a.b", status="superseded", superseded_by="a.c@2026-07-01")
    b = _rec(heading="B", key="a.c", status="active", supersedes="none")
    assert any("does not point back" in e and "a.c" in e for e in _v([a], archive_records=[b], archive_keys={"a.c"}))


def test_reciprocal_supersession_correct_pair_passes():
    # active supersedes archived; archived is superseded-by active → reciprocal, no errors.
    active = _rec(heading="Active", key="a.b", status="active", supersedes="a.c@2026-01-01")
    archived = _rec(heading="Archived", key="a.c", status="superseded", superseded_by="a.b@2026-07-01")
    errs = _v([active], archive_records=[archived], archive_keys={"a.c"})
    assert errs == []


def test_archive_to_archive_reciprocal_pair_passes():
    # A twice-reversed decision: both records live in the archive wing, pointing at each other.
    a = _rec(heading="A", key="a.b", status="superseded", superseded_by="a.c@2026-07-01")
    b = _rec(heading="B", key="a.c", status="retired", supersedes="a.b@2026-07-01")
    errs = _v([], archive_records=[a, b], archive_keys={"a.b", "a.c"})
    assert errs == []


def test_archive_to_archive_broken_back_link_fails():
    # A.superseded-by=B but B.supersedes=none → the back-link is broken, and neither side is active
    # (so this could previously only be caught by iterating archive_records, not decision_recs).
    a = _rec(heading="A", key="a.b", status="superseded", superseded_by="a.c@2026-07-01")
    b = _rec(heading="B", key="a.c", status="retired", supersedes="none")
    errs = _v([], archive_records=[a, b], archive_keys={"a.b", "a.c"})
    assert any("does not point back" in e and "a.c" in e for e in errs)


def test_active_record_with_superseded_by_fails():
    recs = [_rec(key="a.b", status="active", superseded_by="a.c@2026-07-01")]
    assert any("active record cannot already be superseded" in e for e in _v(recs, archive_keys={"a.c"}))


def test_removed_active_not_in_archive_fails():
    assert any("removed from the active set" in e for e in _v([], removed=["2026-01-01 — Gone (#9)"]))


def test_rewritten_rationale_without_token_fails():
    assert any(
        "rationale" in e and "decisions-edit" in e
        for e in _v([_rec(key="a.b")], rewritten=["2026-01-01 — Reworded (#9)"])
    )


def test_clean_corpus_passes():
    assert _v([_rec(key="a.b"), _rec(key="c.d")]) == []


def test_superseded_record_in_active_fails():
    recs = [_rec(key="a.b", status="superseded", superseded_by="a.c@2026-01-01")]
    assert any("relocate to docs/decisions/archive/" in e for e in _v(recs, archive_keys={"a.c"}))


def test_active_record_in_archive_fails():
    assert any(
        "must not live under docs/decisions/archive/" in e
        for e in _v([], archive_records=[_rec(key="a.b", status="active")])
    )


def test_missing_since_fails():
    rec = dl.Record(
        heading="H",
        source=Path("x"),
        lineno=1,
        key="a.b",
        status="active",
        since=None,
        supersedes="none",
        superseded_by="none",
    )
    assert any("missing required metadata since" in e for e in _v([rec]))


def test_missing_signals_fails():
    # `**Signals:**` is required (ersatztv#545): it is what MemPalace's keyword recall matches on, so
    # a record without it ingests with weak metadata and under-surfaces ("no convention exists").
    assert any("missing required metadata signals" in e for e in _v([_rec(key="a.b", signals=None)]))


def test_empty_signals_fails():
    assert any("missing required metadata signals" in e for e in _v([_rec(key="a.b", signals="")]))


def test_signals_present_passes():
    assert not any("signals" in e for e in _v([_rec(key="a.b", signals="concept · paths: x.py · issues: #1")]))


def test_demoted_heading_fails():
    assert any(
        "demoted to legacy-unmigrated" in e for e in _v([_rec(key="a.b")], demoted=["2026-01-01 — Some decision (#1)"])
    )


def test_duplicate_metadata_block_fails():
    # A migration bug left records with two stacked metadata blocks; the parser only reads the
    # first, so only an explicit body scan (metadata_line_count) can catch the leftover second block.
    body = (
        "`key: a.b` · `status: active` · `since: 2026-01-01` · `supersedes: none` · `superseded-by: none`\n"
        "**Rule:** first block rule.\n"
        "\n"
        "`key: a.b` · `status: active` · `since: 2026-01-01` · `supersedes: none` · `superseded-by: none`\n"
        "**Rule:** second block rule.\n"
        "\n"
        "Rationale prose goes here."
    )
    rec = _rec(key="a.b", body=body)
    assert dl.metadata_line_count(rec) == 2
    assert any("duplicate metadata block" in e and "2 metadata blocks" in e for e in _v([rec]))


def test_single_metadata_block_passes():
    body = (
        "`key: a.b` · `status: active` · `since: 2026-01-01` · `supersedes: none` · `superseded-by: none`\n"
        "**Rule:** the only rule.\n"
        "\n"
        "Rationale prose goes here."
    )
    rec = _rec(key="a.b", body=body)
    assert dl.metadata_line_count(rec) == 1
    assert not any("duplicate metadata block" in e for e in _v([rec]))


def test_contents_heading_is_skipped():
    # docs/decisions/*.md topic files use "## Contents" as their index heading (the analog of
    # decisions.md's "## Index") — it carries no metadata and must not be miscounted as a
    # legacy-unmigrated record, same as "Index" already isn't.
    assert "Contents" in dv.SKIP_HEADINGS

    rec = dl.Record(
        heading="Contents",
        source=Path("x"),
        lineno=1,
        status="legacy-unmigrated",
        since=None,
        supersedes=None,
        superseded_by=None,
    )
    # validate() itself excludes SKIP_HEADINGS from decision_recs, so a bare Contents record
    # produces no errors (it isn't checked for required metadata, bad status, etc.)
    assert _v([rec]) == []
    # mirrors main()'s legacy-unmigrated count filter
    unmigrated = [r for r in [rec] if r.status == "legacy-unmigrated" and r.heading not in dv.SKIP_HEADINGS]
    assert unmigrated == []


def _git(cwd: Path, *args: str) -> None:
    subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True, text=True)


def _write_decisions(path: Path, *, status: str, since: str, rationale: str) -> None:
    path.write_text(
        "## 2026-01-01 — Some decision (#1)\n"
        f"`key: a.b` · `status: {status}` · `since: {since}` · `supersedes: none` · `superseded-by: none`\n"
        "**Rule:** one-line current rule.\n"
        "**Signals:** concept · paths: a/b.py · issues: #1\n"
        "**Mechanics:** docs/foo.md\n"
        "\n"
        f"{rationale}\n",
        encoding="utf-8",
    )


def test_diff_engine_detects_rationale_rewrite_without_token(tmp_path, monkeypatch):
    repo = tmp_path / "repo"
    (repo / "docs").mkdir(parents=True)
    _git(tmp_path, "init", str(repo))
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")

    _write_decisions(repo / "docs" / "decisions.md", status="active", since="2026-01-01", rationale="Original prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "base")

    monkeypatch.chdir(repo)

    # (a) rewrite rationale prose only, no token
    _write_decisions(
        repo / "docs" / "decisions.md",
        status="active",
        since="2026-01-01",
        rationale="**Rule:** this looks like metadata but is prose appended later.",
    )
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "rewrite rationale")

    removed, rewritten, *_rest = dv._diff_findings("HEAD~1", "HEAD")
    assert removed == []
    assert "2026-01-01 — Some decision (#1)" in rewritten


def test_diff_engine_allows_rewrite_with_token(tmp_path, monkeypatch):
    repo = tmp_path / "repo"
    (repo / "docs").mkdir(parents=True)
    _git(tmp_path, "init", str(repo))
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")

    _write_decisions(repo / "docs" / "decisions.md", status="active", since="2026-01-01", rationale="Original prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "base")

    monkeypatch.chdir(repo)

    _write_decisions(repo / "docs" / "decisions.md", status="active", since="2026-01-01", rationale="Reworded prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "rewrite rationale [decisions-edit]")

    removed, rewritten, *_rest = dv._diff_findings("HEAD~1", "HEAD")
    assert removed == []
    assert rewritten == []


def test_diff_engine_detects_appended_smuggled_rationale(tmp_path, monkeypatch):
    repo = tmp_path / "repo"
    (repo / "docs").mkdir(parents=True)
    _git(tmp_path, "init", str(repo))
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")

    _write_decisions(repo / "docs" / "decisions.md", status="active", since="2026-01-01", rationale="Original prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "base")

    monkeypatch.chdir(repo)

    # append-shape exploit: original prose line is PRESERVED, a smuggled **Rule:** paragraph is
    # appended after a blank line, no [decisions-edit] token. The old buggy _rationale() strips
    # ANY `**Rule:**`/`**Signals:**`/`**Mechanics:**`/`` `key:`` line wherever it appears, so it
    # silently strips the appended line too → base==head → bypass succeeds. The bounded strip only
    # removes the contiguous top metadata block, so the appended paragraph survives as prose →
    # base!=head → flagged.
    _write_decisions(
        repo / "docs" / "decisions.md",
        status="active",
        since="2026-01-01",
        rationale="Original prose.\n\n**Rule:** smuggled rewrite that changes the actual meaning.",
    )
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "append smuggled rewrite")

    removed, rewritten, *_rest = dv._diff_findings("HEAD~1", "HEAD")
    assert removed == []
    assert "2026-01-01 — Some decision (#1)" in rewritten


def test_diff_engine_metadata_only_edit_is_free(tmp_path, monkeypatch):
    repo = tmp_path / "repo"
    (repo / "docs").mkdir(parents=True)
    _git(tmp_path, "init", str(repo))
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")

    _write_decisions(repo / "docs" / "decisions.md", status="active", since="2026-01-01", rationale="Original prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "base")

    monkeypatch.chdir(repo)

    # metadata-only edit: status/since change, prose unchanged, no token
    _write_decisions(
        repo / "docs" / "decisions.md", status="superseded", since="2026-02-01", rationale="Original prose."
    )
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "metadata edit")

    removed, rewritten, *_rest = dv._diff_findings("HEAD~1", "HEAD")
    assert removed == []
    assert rewritten == []


def _write_archive(path: Path, *, rationale: str) -> None:
    path.write_text(
        "## 2026-01-01 — Some archived decision (#2)\n"
        "`key: a.z` · `status: superseded` · `since: 2026-01-01` · `supersedes: none` · "
        "`superseded-by: a.b@2026-07-01`\n"
        "**Rule:** one-line historical rule.\n"
        "**Signals:** concept · paths: a/z.py · issues: #2\n"
        "**Mechanics:** docs/foo.md\n"
        "\n"
        f"{rationale}\n",
        encoding="utf-8",
    )


def test_diff_engine_detects_archive_rationale_rewrite_without_token(tmp_path, monkeypatch):
    repo = tmp_path / "repo"
    (repo / "docs" / "decisions" / "archive").mkdir(parents=True)
    _git(tmp_path, "init", str(repo))
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")

    archive_path = repo / "docs" / "decisions" / "archive" / "x.md"
    _write_archive(archive_path, rationale="Original archived prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "base")

    monkeypatch.chdir(repo)

    _write_archive(archive_path, rationale="Rewritten archived prose, no token.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "rewrite archived rationale")

    _removed, rewritten, *_rest = dv._diff_findings("HEAD~1", "HEAD")
    assert "2026-01-01 — Some archived decision (#2)" in rewritten


def test_diff_engine_detects_archive_record_removed(tmp_path, monkeypatch):
    repo = tmp_path / "repo"
    (repo / "docs" / "decisions" / "archive").mkdir(parents=True)
    _git(tmp_path, "init", str(repo))
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")

    archive_path = repo / "docs" / "decisions" / "archive" / "x.md"
    _write_archive(archive_path, rationale="Original archived prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "base")

    monkeypatch.chdir(repo)

    archive_path.unlink()
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "delete archived record")

    removed, _rewritten, *_rest = dv._diff_findings("HEAD~1", "HEAD")
    assert "2026-01-01 — Some archived decision (#2)" in removed


def test_diff_engine_demoted_migrated_to_legacy_unmigrated(tmp_path, monkeypatch):
    repo = tmp_path / "repo"
    (repo / "docs").mkdir(parents=True)
    _git(tmp_path, "init", str(repo))
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")

    decisions_path = repo / "docs" / "decisions.md"
    _write_decisions(decisions_path, status="active", since="2026-01-01", rationale="Original prose.")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "base")

    monkeypatch.chdir(repo)

    # same heading, metadata block stripped: prose survives, migration silently reverted.
    decisions_path.write_text(
        "## 2026-01-01 — Some decision (#1)\n\nOriginal prose.\n",
        encoding="utf-8",
    )
    _git(repo, "add", "-A")
    _git(repo, "commit", "-m", "demote to legacy-unmigrated")

    _removed, _rewritten, demoted = dv._diff_findings("HEAD~1", "HEAD")
    assert "2026-01-01 — Some decision (#1)" in demoted


# ---- optional stale-after / Sources metadata (ersatztv#603) ----


def test_stale_after_absent_is_not_an_error():
    assert _v([_rec(key="a.b")]) == []


def test_wellformed_stale_after_is_not_an_error():
    assert _v([_rec(key="a.b", stale_after="2027-01-01")]) == []


def test_malformed_stale_after_fails():
    for bad in ("2027-13-01", "01-01-2027", "soon", "2027-02-30"):
        errs = _v([_rec(key="a.b", stale_after=bad)])
        assert any("not a YYYY-MM-DD date" in e for e in errs), bad


def test_past_due_record_does_not_fail_validation():
    """A record going stale is the passage of time, not a defect in the commit under test."""
    errs = _v([_rec(key="a.b", stale_after="2020-01-01")])
    assert errs == []


def test_stale_records_reports_only_past_due_active():
    today = dv.date(2026, 7, 25)
    recs = [
        _rec(key="past.due", heading="Past", stale_after="2026-07-24"),
        _rec(key="due.today", heading="Today", stale_after="2026-07-25"),  # today >= date → stale
        _rec(key="not.yet", heading="Future", stale_after="2026-07-26"),
        _rec(key="no.date", heading="Undated"),
        _rec(key="arch.ived", heading="Archived", status="superseded", stale_after="2020-01-01"),
    ]
    assert dv.stale_records(recs, today) == [("Past", "2026-07-24"), ("Today", "2026-07-25")]


def test_stale_records_ignores_malformed_date():
    """A malformed date is caught as a blocking format error; it must not also crash the notice path."""
    assert dv.stale_records([_rec(key="a.b", stale_after="whenever")], dv.date(2026, 7, 25)) == []


def test_sources_is_optional_and_never_required():
    assert _v([_rec(key="a.b")]) == []
    assert _v([_rec(key="a.b", sources="ci run 1234 · docs/ci-cd.md")]) == []