#!/usr/bin/env python3
"""Offline behavior checks for classify-changes.py; no git repo or runner."""

import importlib.util
import io
import os
from pathlib import Path
import re
import tempfile
import unittest

_spec = importlib.util.spec_from_file_location(
    "classify_changes", Path(__file__).with_name("classify-changes.py"))
cc = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(cc)

BEFORE = "b" * 40
SHA = "c" * 40


class FakeGit:
    """Answers git calls from a table; records every call."""

    def __init__(self, diffs=None, reachable_after=0, ancestor=True, raise_on=None,
                 ancestor_after_fetches=0, shallow=True):
        self.diffs = diffs or {}
        self.reachable_after = reachable_after  # cat-file succeeds on the Nth probe (0 = never)
        self.ancestor = ancestor
        # is-ancestor answers False until this many deepen fetches have run: the
        # stale-object shallow graft, where the OBJECT is present but the
        # graph from sha does not reach it yet (measured on kitsubito, IR-147).
        self.ancestor_after_fetches = ancestor_after_fetches
        self.shallow = shallow
        self.raise_on = raise_on
        self.calls = []
        self.probes = 0
        self.fetches = 0

    def run(self, *args):
        self.calls.append(args)
        if self.raise_on and args[0] == self.raise_on:
            raise OSError("boom")
        if args[0] == "cat-file":
            self.probes += 1
            ok = self.reachable_after and self.probes >= self.reachable_after
            return (0 if ok else 1), ""
        if args[0] == "merge-base":
            linked = self.fetches >= self.ancestor_after_fetches
            return (0 if (self.ancestor and linked) else 1), ""
        if args[0] == "fetch":
            self.fetches += 1
            return 0, ""
        if args[0] == "rev-parse":
            return 0, ("true" if self.shallow else "false") + "\n"
        if args[0] == "diff":
            key = (args[2], args[3])
            if key not in self.diffs:
                return 128, ""
            return 0, "\n".join(self.diffs[key]) + "\n"
        raise AssertionError(f"unexpected git call {args}")


def run(git, **env):
    out = io.StringIO()
    with tempfile.TemporaryDirectory() as d:
        target = os.path.join(d, "out")
        env = {"GITHUB_OUTPUT": target, **env}
        cc.main(git=git, env=env, out=out)
        with open(target, encoding="utf-8") as fh:
            written = fh.read()
    line = [l for l in out.getvalue().splitlines() if l.startswith("CLASSIFY ")]
    assert len(line) == 1, out.getvalue()
    return line[0], written


def push(git, before=BEFORE, forced="false"):
    return run(git, GITHUB_EVENT_NAME="push", GITHUB_SHA=SHA,
               CLASSIFY_BEFORE=before, CLASSIFY_FORCED=forced)


# [unit->REQ-CI-PUSH-DOCS-ONLY-THIN]
class PullRequestArm(unittest.TestCase):
    def pr(self, files):
        return run(FakeGit({("HEAD^1", "HEAD"): files}), GITHUB_EVENT_NAME="pull_request")

    def test_docs_only_pr_is_thin(self):
        line, out = self.pr(["docs/RELEASE-RUNBOOK.md", "traceable-reqs.toml"])
        self.assertEqual(line, "CLASSIFY code=false reason=docs-only")
        self.assertEqual(out, "code=false\ndocs-drift=false\n")

    def test_code_and_docs_site_prs_run_units(self):
        self.assertIn("code=true reason=code-path:crates/x.rs", self.pr(["README.md", "crates/x.rs"])[0])
        self.assertIn("code=true reason=code-path:docs-site/src/a.md", self.pr(["docs-site/src/a.md"])[0])


# [unit->REQ-CI-PUSH-DOCS-ONLY-THIN]
class PushArm(unittest.TestCase):
    def test_docs_only_push_range_is_thin(self):
        g = FakeGit({(BEFORE, SHA): ["docs/INFRA-REGISTER.md"]}, reachable_after=1)
        line, out = push(g)
        self.assertEqual(line, "CLASSIFY code=false reason=docs-only")
        self.assertEqual(out, "code=false\ndocs-drift=false\n")

    def test_multi_commit_push_reads_the_whole_range_not_head_parent(self):
        # The IR-147 false negative: the LAST commit of the lane is docs-only,
        # an EARLIER one is code. HEAD^1..HEAD would say docs-only; the push
        # range must say code.
        g = FakeGit({
            (BEFORE, SHA): ["crates/xtask/src/bundle.rs", "docs/INFRA-REGISTER.md"],
            ("HEAD^1", "HEAD"): ["docs/INFRA-REGISTER.md"],
        }, reachable_after=1)
        line, out = push(g)
        self.assertEqual(line, "CLASSIFY code=true reason=code-path:crates/xtask/src/bundle.rs")
        self.assertEqual(out, "code=true\ndocs-drift=true\n")
        self.assertNotIn(("diff", "--name-only", "HEAD^1", "HEAD"), g.calls)

    def test_before_zero_runs_units(self):
        self.assertEqual(push(FakeGit(), before="0" * 40)[0], "CLASSIFY code=true reason=before-zero")
        self.assertEqual(push(FakeGit(), before="")[0], "CLASSIFY code=true reason=before-zero")

    def test_forced_push_runs_units(self):
        g = FakeGit({(BEFORE, SHA): ["a.md"]}, reachable_after=1)
        self.assertEqual(push(g, forced="true")[0], "CLASSIFY code=true reason=forced-push")

    def test_before_reached_only_after_deepening(self):
        g = FakeGit({(BEFORE, SHA): ["a.md"]}, reachable_after=2)
        self.assertEqual(push(g)[0], "CLASSIFY code=false reason=docs-only")
        fetches = [c for c in g.calls if c[0] == "fetch"]
        self.assertEqual(fetches, [("fetch", "--no-tags", "--deepen=50", "origin", SHA)])

    def test_before_unreachable_after_bounded_deepening_runs_units(self):
        g = FakeGit({(BEFORE, SHA): ["a.md"]}, reachable_after=0)
        self.assertEqual(push(g)[0], "CLASSIFY code=true reason=before-unreachable")
        fetches = [c[2] for c in g.calls if c[0] == "fetch"]
        self.assertEqual(fetches, ["--deepen=50", "--deepen=500"], "deepening is bounded")

    def test_before_not_ancestor_on_a_complete_graph_runs_units(self):
        g = FakeGit({(BEFORE, SHA): ["a.md"]}, reachable_after=1, ancestor=False, shallow=False)
        self.assertEqual(push(g)[0], "CLASSIFY code=true reason=before-not-ancestor")

    def test_present_object_on_a_shallow_graft_still_deepens(self):
        # IR-147 field defect (main run 35998753683): a STALE before object is
        # present, so a presence probe passes at step 0, but the depth-2 graph
        # from sha is grafted short of it and is-ancestor says no. Presence
        # must not stop the deepening; connectivity must.
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=1, ancestor_after_fetches=1)
        self.assertEqual(push(g)[0], "CLASSIFY code=false reason=docs-only")
        self.assertEqual([c[2] for c in g.calls if c[0] == "fetch"], ["--deepen=50"])

    def test_a_shallow_graph_that_never_decides_is_undecidable_not_non_ancestor(self):
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=1, ancestor=False, shallow=True)
        self.assertEqual(push(g)[0], "CLASSIFY code=true reason=before-undecidable")
        self.assertEqual([c[2] for c in g.calls if c[0] == "fetch"], ["--deepen=50", "--deepen=500"])

    def test_diff_failure_runs_units(self):
        g = FakeGit({}, reachable_after=1)
        self.assertEqual(push(g)[0], "CLASSIFY code=true reason=diff-failed")

    def test_empty_range_runs_units(self):
        g = FakeGit({(BEFORE, SHA): []}, reachable_after=1)
        self.assertEqual(push(g)[0], "CLASSIFY code=true reason=empty-range")


# [unit->REQ-CI-PUSH-DOCS-ONLY-THIN]
class DocsSetIsPositive(unittest.TestCase):
    def test_docs_set(self):
        for p in ["README.md", "CHANGELOG.md", "docs/INFRA-REGISTER.md",
                  "docs/adr/0050-golden-ci-merge-integration.md", "traceable-reqs.toml"]:
            self.assertTrue(cc.is_docs_path(p), p)

    def test_everything_else_is_code(self):
        for p in [".github/workflows/ci.yml", ".github/ci/classify-changes.py",
                  ".github/PULL_REQUEST_TEMPLATE.md", "docs-site/src/a.md",
                  "docs/ER-SKELETON.md", "crates/spt-daemon/tests/fixtures/enlyzeam/README.md",
                  "releases-repo/SYNC.md", "docs/instruments/ir21/bindeps-probe/Cargo.toml",
                  "Cargo.toml", "Cargo.lock", "crates/spt/src/cli.rs", "notes.txt"]:
            self.assertFalse(cc.is_docs_path(p), p)

    def test_a_ci_only_push_runs_units(self):
        g = FakeGit({(BEFORE, SHA): [".github/workflows/ci.yml", "docs/a.md"]}, reachable_after=1)
        self.assertEqual(push(g)[0], "CLASSIFY code=true reason=code-path:.github/workflows/ci.yml")

    def test_compiled_markdown_matches_the_tree(self):
        # Every .md the Rust tree compiles in must be in COMPILED_MARKDOWN,
        # and nothing else may be: the list is derived, not remembered.
        root = Path(__file__).resolve().parents[2]
        pat = re.compile(r'include_(?:str|bytes)!\(\s*"([^"]+\.md)"')
        found = set()
        for rs in root.rglob("*.rs"):
            parts = rs.relative_to(root).parts
            if parts[0] in ("target", ".worktrees") or "target" in parts:
                continue
            for rel in pat.findall(rs.read_text(encoding="utf-8", errors="replace")):
                found.add((rs.parent / rel).resolve().relative_to(root).as_posix())
        self.assertTrue(found, "control: the tree compiles in at least one .md today")
        self.assertEqual(found, set(cc.COMPILED_MARKDOWN))


def lines(git, **env):
    """(CLASSIFY_PROBE line, CLASSIFY line) from one real main() call."""
    out = io.StringIO()
    cc.main(git=git, env=env, out=out)
    text = out.getvalue().splitlines()
    probe = [l for l in text if l.startswith("CLASSIFY_PROBE ")]
    verdict = [l for l in text if l.startswith("CLASSIFY ")]
    assert len(probe) == 1 and len(verdict) == 1, text
    return probe[0], verdict[0]


def push_lines(git, before=BEFORE, forced="false"):
    return lines(git, GITHUB_EVENT_NAME="push", GITHUB_SHA=SHA,
                 CLASSIFY_BEFORE=before, CLASSIFY_FORCED=forced)


# [unit->REQ-CI-CLASSIFY-PROBE-WITNESS]
class ProbeWitnessesTheMechanism(unittest.TestCase):
    """IR-148: the log names WHICH step connected before..sha and whether the
    graph was shallow AT DECISION TIME, so a dead deepen cannot hide behind
    a correct-looking verdict on a warm workspace."""

    def test_graft_arm_names_the_deepen_that_connected(self):
        # The #258 field shape: before PRESENT at step 0 but grafted off; only
        # the first deepen connects it. The probe must say deepen50, not step0.
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=1, ancestor_after_fetches=1)
        probe, verdict = push_lines(g)
        self.assertEqual(probe, "CLASSIFY_PROBE connected_at=deepen50 shallow=true")
        self.assertEqual(verdict, "CLASSIFY code=false reason=docs-only")

    def test_already_connected_names_step0(self):
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=1)
        self.assertEqual(push_lines(g)[0], "CLASSIFY_PROBE connected_at=step0 shallow=true")

    def test_second_deepen_is_named(self):
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=1, ancestor_after_fetches=2)
        self.assertEqual(push_lines(g)[0], "CLASSIFY_PROBE connected_at=deepen500 shallow=true")

    def test_undecidable_and_unreachable_name_none(self):
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=1, ancestor=False, shallow=True)
        self.assertEqual(push_lines(g), ("CLASSIFY_PROBE connected_at=none shallow=true",
                                         "CLASSIFY code=true reason=before-undecidable"))
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=0)
        self.assertEqual(push_lines(g), ("CLASSIFY_PROBE connected_at=none shallow=true",
                                         "CLASSIFY code=true reason=before-unreachable"))

    def test_complete_graph_non_ancestor_reports_shallow_false(self):
        g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, reachable_after=1, ancestor=False, shallow=False)
        self.assertEqual(push_lines(g), ("CLASSIFY_PROBE connected_at=none shallow=false",
                                         "CLASSIFY code=true reason=before-not-ancestor"))

    def test_shallow_is_read_once_at_decision_time(self):
        # One read feeds both the probe and the undecidable/not-ancestor
        # split; a second, later read would be a different measurement.
        for kw in ({"reachable_after": 1}, {"reachable_after": 1, "ancestor": False},
                   {"reachable_after": 1, "ancestor_after_fetches": 1}):
            g = FakeGit({(BEFORE, SHA): ["docs/a.md"]}, **kw)
            push_lines(g)
            reads = [c for c in g.calls if c[0] == "rev-parse"]
            self.assertEqual(reads, [("rev-parse", "--is-shallow-repository")], kw)

    def test_paths_that_never_probe_say_so(self):
        self.assertEqual(push_lines(FakeGit(), before="0" * 40)[0],
                         "CLASSIFY_PROBE connected_at=not-probed shallow=n/a")
        self.assertEqual(push_lines(FakeGit(), forced="true")[0],
                         "CLASSIFY_PROBE connected_at=not-probed shallow=n/a")
        pr = FakeGit({("HEAD^1", "HEAD"): ["docs/a.md"]})
        self.assertEqual(lines(pr, GITHUB_EVENT_NAME="pull_request")[0],
                         "CLASSIFY_PROBE connected_at=not-probed shallow=n/a")
        self.assertEqual(lines(FakeGit(raise_on="cat-file"), GITHUB_EVENT_NAME="push",
                               GITHUB_SHA=SHA, CLASSIFY_BEFORE=BEFORE, CLASSIFY_FORCED="false"),
                         ("CLASSIFY_PROBE connected_at=not-probed shallow=n/a",
                          "CLASSIFY code=true reason=classifier-error:OSError"))


# [unit->REQ-CI-PUSH-DOCS-ONLY-THIN]
class OtherEventsAndErrors(unittest.TestCase):
    def test_dispatch_runs_units(self):
        line, _ = run(FakeGit(), GITHUB_EVENT_NAME="workflow_dispatch")
        self.assertEqual(line, "CLASSIFY code=true reason=event-workflow_dispatch")

    def test_unexpected_exception_runs_units(self):
        g = FakeGit(raise_on="cat-file")
        line, out = push(g)
        self.assertEqual(line, "CLASSIFY code=true reason=classifier-error:OSError")
        self.assertEqual(out, "code=true\ndocs-drift=true\n")


def drift(git, **env):
    """(CLASSIFY line, CLASSIFY_DOCS_DRIFT line, GITHUB_OUTPUT) from one main() call."""
    out = io.StringIO()
    with tempfile.TemporaryDirectory() as d:
        target = os.path.join(d, "out")
        cc.main(git=git, env={"GITHUB_OUTPUT": target, **env}, out=out)
        with open(target, encoding="utf-8") as fh:
            written = fh.read()
    text = out.getvalue().splitlines()
    verdict = [l for l in text if l.startswith("CLASSIFY ")]
    leg = [l for l in text if l.startswith("CLASSIFY_DOCS_DRIFT ")]
    assert len(verdict) == 1 and len(leg) == 1, text
    return verdict[0], leg[0], written


# [unit->REQ-CI-THIN-DOCS-DRIFT]
class DocsDriftFollowsTheUnion(unittest.TestCase):
    """IR-149: the docs-drift leg runs on code OR a docs path xtask check reads."""

    def pr(self, files):
        return drift(FakeGit({("HEAD^1", "HEAD"): files}), GITHUB_EVENT_NAME="pull_request")

    def test_register_only_docs_diff_skips_the_leg(self):
        v, leg, out = self.pr(["docs/INFRA-REGISTER.md", "traceable-reqs.toml", "README.md"])
        self.assertEqual(v, "CLASSIFY code=false reason=docs-only")
        self.assertEqual(leg, "CLASSIFY_DOCS_DRIFT docs-drift=false reason=no-drift-input")
        self.assertEqual(out, "code=false\ndocs-drift=false\n")

    def test_changelog_only_diff_is_docs_class_yet_runs_the_leg(self):
        # The case that makes grouping the leg under code == 'true' wrong.
        v, leg, out = self.pr(["CHANGELOG.md"])
        self.assertEqual(v, "CLASSIFY code=false reason=docs-only")
        self.assertEqual(leg, "CLASSIFY_DOCS_DRIFT docs-drift=true reason=drift-input:CHANGELOG.md")
        self.assertEqual(out, "code=false\ndocs-drift=true\n")

    def test_changelog_among_other_docs_still_runs_the_leg(self):
        leg = self.pr(["docs/a.md", "CHANGELOG.md", "README.md"])[1]
        self.assertEqual(leg, "CLASSIFY_DOCS_DRIFT docs-drift=true reason=drift-input:CHANGELOG.md")

    def test_code_diff_runs_the_leg(self):
        # W3's red came from a CODE diff: clap help text in cli.rs.
        v, leg, _ = self.pr(["crates/spt/src/cli.rs"])
        self.assertEqual(v, "CLASSIFY code=true reason=code-path:crates/spt/src/cli.rs")
        self.assertEqual(leg, "CLASSIFY_DOCS_DRIFT docs-drift=true reason=code")

    def test_docs_only_push_range_with_changelog_runs_the_leg(self):
        g = FakeGit({(BEFORE, SHA): ["CHANGELOG.md", "docs/RELEASE-RUNBOOK.md"]}, reachable_after=1)
        v, leg, _ = drift(g, GITHUB_EVENT_NAME="push", GITHUB_SHA=SHA,
                          CLASSIFY_BEFORE=BEFORE, CLASSIFY_FORCED="false")
        self.assertEqual(v, "CLASSIFY code=false reason=docs-only")
        self.assertEqual(leg, "CLASSIFY_DOCS_DRIFT docs-drift=true reason=drift-input:CHANGELOG.md")

    def test_every_conservative_exit_runs_the_leg(self):
        cases = [
            drift(FakeGit(), GITHUB_EVENT_NAME="workflow_dispatch"),
            drift(FakeGit(), GITHUB_EVENT_NAME="push", GITHUB_SHA=SHA,
                  CLASSIFY_BEFORE=cc.ZERO_SHA, CLASSIFY_FORCED="false"),
            drift(FakeGit(raise_on="cat-file"), GITHUB_EVENT_NAME="push", GITHUB_SHA=SHA,
                  CLASSIFY_BEFORE=BEFORE, CLASSIFY_FORCED="false"),
            drift(FakeGit({}), GITHUB_EVENT_NAME="pull_request"),
        ]
        for v, leg, out in cases:
            self.assertIn("code=true", v)
            self.assertEqual(leg, "CLASSIFY_DOCS_DRIFT docs-drift=true reason=code")
            self.assertEqual(out, "code=true\ndocs-drift=true\n")

    def test_drift_inputs_are_docs_class(self):
        # A drift input that classified as code would already run the leg,
        # so listing it here would be dead weight that hides a real edit.
        for p in cc.DRIFT_INPUT_DOCS:
            self.assertTrue(cc.is_docs_path(p), p)

    def test_drift_inputs_match_what_xtask_reads(self):
        # Derived, not remembered: every docs-class repo-root path the xtask
        # source reads as root.join("<literal>"). A SPELLING census: a read
        # through another spelling is invisible here, and the control below
        # proves the census can see the one read that motivated it.
        root = Path(__file__).resolve().parents[2]
        pat = re.compile(r'\broot\.join\(\s*"([^"]+)"\s*\)')
        found = set()
        for rs in (root / "crates" / "xtask" / "src").rglob("*.rs"):
            for rel in pat.findall(rs.read_text(encoding="utf-8", errors="replace")):
                if cc.is_docs_path(rel):
                    found.add(rel)
        self.assertIn("CHANGELOG.md", found, 'control: gen_changelog reads root.join("CHANGELOG.md")')
        self.assertEqual(found, set(cc.DRIFT_INPUT_DOCS))


if __name__ == "__main__":
    unittest.main(verbosity=2)