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 _rewrite_repo(tmp_path: Path, monkeypatch, *commit_msg_paragraphs: str) -> tuple[list[str], list[str]]:
    """Seed a repo, rewrite the record's rationale under `commit_msg_paragraphs`, return (removed, rewritten).

    Each paragraph is a separate `-m`, which is how git composes a message with a trailer block: the
    LAST paragraph is the only one git parses for trailers (verified against git 2.55).
    """
    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", *[a for p in commit_msg_paragraphs for a in ("-m", p)])

    removed, rewritten, *_rest = dv._diff_findings("HEAD~1", "HEAD")
    return removed, rewritten


_RECORD = "2026-01-01 — Some decision (#1)"


def test_diff_engine_allows_rewrite_with_trailer(tmp_path, monkeypatch):
    removed, rewritten = _rewrite_repo(tmp_path, monkeypatch, "rewrite rationale", "Decisions-Edit: yes")
    assert removed == []
    assert rewritten == []


def test_diff_engine_trailer_is_case_insensitive(tmp_path, monkeypatch):
    _removed, rewritten = _rewrite_repo(tmp_path, monkeypatch, "rewrite rationale", "Decisions-Edit: YES")
    assert rewritten == []


def test_diff_engine_trailer_coexists_with_coauthor(tmp_path, monkeypatch):
    # the repo's `commit-msg` hook mandates a Co-Authored-By trailer, so the real-world shape of an
    # armed commit is a multi-trailer block — the marker must survive sharing it.
    _removed, rewritten = _rewrite_repo(
        tmp_path,
        monkeypatch,
        "rewrite rationale",
        "Decisions-Edit: yes\nCo-Authored-By: Someone <s@example.com>",
    )
    assert rewritten == []


def test_diff_engine_prose_mention_of_retired_token_does_not_arm(tmp_path, monkeypatch):
    """ersatztv#609's live failure: a message *describing* the marker disarmed the whole body-diff.

    This is the negative control the bare-substring matcher could not pass — under it, `token` went
    True and all three rewrite comparisons were skipped while CI still reported green.
    """
    _removed, rewritten = _rewrite_repo(
        tmp_path,
        monkeypatch,
        "docs: explain the guard",
        "No [decisions-edit] marker is needed here, because this commit only moves a record.",
    )
    assert _RECORD in rewritten


def test_diff_engine_quoted_trailer_mid_body_does_not_arm(tmp_path, monkeypatch):
    """The same class one level up: a commit that QUOTES an armed message as an example.

    Git parses trailers only in the final paragraph, so a `Decisions-Edit: yes` line followed by more
    prose is documentation, not a marker. Docs commits in this repo routinely carry such examples.
    """
    _removed, rewritten = _rewrite_repo(
        tmp_path,
        monkeypatch,
        "docs: document how to arm the guard",
        "Contributors write:",
        "Decisions-Edit: yes",
        "...as the final paragraph of the message, alongside Co-Authored-By.",
    )
    assert _RECORD in rewritten


def test_diff_engine_negative_trailer_value_does_not_arm(tmp_path, monkeypatch):
    # `Decisions-Edit: no` records a deliberate NON-edit; reading it as consent would recreate #609.
    _removed, rewritten = _rewrite_repo(tmp_path, monkeypatch, "rewrite rationale", "Decisions-Edit: no")
    assert _RECORD in rewritten


def test_diff_engine_folded_trailer_continuation_does_not_arm(tmp_path, monkeypatch):
    """A folded value must be judged whole: `no` + continuation ` yes` is the value "no yes".

    Without `unfold`, git emits the continuation as its own line and ` yes`.strip() arms the
    exemption on its own — inverting the author's explicit `no`.
    """
    _removed, rewritten = _rewrite_repo(tmp_path, monkeypatch, "rewrite rationale", "Decisions-Edit: no\n yes")
    assert _RECORD in rewritten


def test_diff_engine_merge_commit_body_does_not_arm(tmp_path, monkeypatch):
    # On a pull_request event, actions/checkout lands on a synthetic merge commit whose body the
    # forge composes (quoting the PR description) — a trailer parsed out of it is not an author's
    # deliberate marker, so merge commits are excluded from the range.
    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)
    base = subprocess.run(
        ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True
    ).stdout.strip()

    _git(repo, "checkout", "-q", "-b", "feature")
    _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")  # the author never armed anything

    # the forge's synthetic merge, whose body quotes a PR description ending in an example marker
    _git(repo, "checkout", "-q", base)
    _git(repo, "merge", "--no-ff", "-q", "feature", "-m", "Merge pull request #1", "-m", "Decisions-Edit: yes")

    removed, rewritten, *_rest = dv._diff_findings(base, "HEAD")
    assert removed == []
    assert _RECORD in rewritten


def test_diff_engine_retired_bracket_token_no_longer_arms(tmp_path, monkeypatch):
    # the pre-#609 habit (token appended to the subject) is retired, not silently honored.
    _removed, rewritten = _rewrite_repo(tmp_path, monkeypatch, "rewrite rationale [decisions-edit]")
    assert _RECORD in 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 edit marker on the commit. 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")]) == []


def test_empty_stale_after_is_malformed_not_absent():
    """A blank/truncated value must not pass as 'absent' — it would silently never fire."""
    errs = _v([_rec(key="a.b", stale_after="")])
    assert any("not a YYYY-MM-DD date" in e for e in errs)


def test_noncanonical_iso_forms_are_rejected():
    """date.fromisoformat accepts these on py>=3.11; the field's contract is YYYY-MM-DD only.

    Accepting them would also break the catalog's Review-due sort, which orders on the raw string.
    """
    for bad in ("20270101", "2027-W01-1", "2027-01-01T00:00:00", " 2027-01-01 extra"):
        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_malformed_stale_after_on_an_archive_record_is_caught():
    arch = _rec(key="a.old", status="superseded", superseded_by="a.b", stale_after="soon")
    errs = _v([_rec(key="a.b", supersedes="a.old")], archive_records=[arch], archive_keys={"a.old"})
    assert any("not a YYYY-MM-DD date" in e for e in errs)