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")]) == []


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)


# ---- edit-token scoping (ersatztv#609) ----


def test_token_armed_by_subject_the_established_form():
    """Every historical use appends it to the subject; that must keep working."""
    for subj in (
        "fix(460): write null LastScan on disable-sync [decisions-edit]",
        "docs(434): [decisions-edit] update field-values decision",
        "fix(529): address review [DECISIONS-EDIT]",
        "Merge pull request 'fix(460): ... [decisions-edit]' (#601) from x into main",
    ):
        assert dv.token_armed_in(subj, ""), subj


def test_token_armed_by_trailer():
    assert dv.token_armed_in("docs: correct a stale figure", "corrected the 2026-07-17 load number")


def test_token_NOT_armed_by_body_prose():
    """The #609 defect: a commit DESCRIBING the token silently disarmed the whole guard."""
    for subj in (
        "docs(603): correct the record's own no-backfill claim",
        "fix(609): scope the edit token to the subject line",
    ):
        assert not dv.token_armed_in(subj, ""), subj


def test_token_not_armed_by_empty_or_whitespace_trailer():
    assert not dv.token_armed_in("docs: something", "")
    assert not dv.token_armed_in("docs: something", "   \n  ")


# ---- _token_armed INTEGRATION against real git (ersatztv#609 review round 2) ----
#
# The unit tests above only exercise `token_armed_in`. Both false-arm defects found in review
# (git's `%s` folding the first PARAGRAPH, and an injectable in-band separator) lived in the git
# plumbing and passed those tests untouched. These drive the real command.


def _repo(tmp_path, messages: list[str]) -> tuple[str, str]:
    """Build a throwaway repo; return (base_sha, head_sha)."""
    import subprocess

    def g(*a, **kw):
        return subprocess.run(["git", *a], cwd=tmp_path, capture_output=True, text=True, **kw)

    g("init", "-q", ".")
    g("config", "user.email", "t@e")
    g("config", "user.name", "t")
    (tmp_path / "f").write_text("base\n")
    g("add", "-A")
    g("commit", "-qm", "base")
    base = g("rev-parse", "HEAD").stdout.strip()
    for i, msg in enumerate(messages):
        (tmp_path / "f").write_text(f"{i}\n")
        g("add", "-A")
        g("commit", "-q", "-F", "-", input=msg)
    return base, g("rev-parse", "HEAD").stdout.strip()


def _armed(tmp_path, message: str) -> bool:
    import os

    base, head = _repo(tmp_path, [message])
    cwd = os.getcwd()
    os.chdir(tmp_path)
    try:
        return dv._token_armed(base, head)
    finally:
        os.chdir(cwd)


def test_integration_subject_token_arms(tmp_path):
    assert _armed(tmp_path, "fix(460): write null LastScan [decisions-edit]\n\nbody\n")


def test_integration_trailer_arms(tmp_path):
    assert _armed(tmp_path, "docs: correct a figure\n\nwhy\n\nDecisions-Edit: the load number was wrong\n")


def test_integration_body_prose_does_not_arm(tmp_path):
    assert not _armed(tmp_path, "docs: explain it\n\nThe check matches [decisions-edit] as a substring.\n")


def test_integration_second_line_of_first_paragraph_does_not_arm(tmp_path):
    """git's %s is the first PARAGRAPH, not the first line — it folds line 2 in with a space.

    Using %s directly, this message arms the token. It must not: line 2 is body prose.
    """
    assert not _armed(tmp_path, "fix: harmless subject\nThis explains [decisions-edit] on line two.\n\nbody\n")


def test_integration_separator_in_subject_cannot_inject(tmp_path):
    """An in-band \\x1f separator was injectable; NUL cannot appear in a commit message."""
    assert not _armed(tmp_path, "docs: harmless\x1fsuffix\n\nbody\n")
    assert not _armed(tmp_path, "docs: harmless\x1e\x1fsuffix\n\nbody\n")


def test_integration_empty_trailer_does_not_arm(tmp_path):
    assert not _armed(tmp_path, "docs: something\n\nbody\n\nDecisions-Edit:\n")


def test_integration_only_a_later_commit_carries_it(tmp_path):
    import os

    base, head = _repo(tmp_path, ["chore: unrelated\n", "fix: real correction [decisions-edit]\n"])
    cwd = os.getcwd()
    os.chdir(tmp_path)
    try:
        assert dv._token_armed(base, head)
    finally:
        os.chdir(cwd)


def test_integration_unresolvable_refs_do_not_arm(tmp_path):
    """Guard must still RUN when git can't answer; arming on failure would disable it."""
    assert not _armed(tmp_path, "docs: x\n") or True  # build a repo first
    import os

    cwd = os.getcwd()
    os.chdir(tmp_path)
    try:
        assert dv._token_armed("nope1", "nope2") is False
    finally:
        os.chdir(cwd)