state-hub/tests/test_c25_source_revision.py

"""C-25 JSON binds to the scanned source commit (STATE-WP-0094)."""
from __future__ import annotations

import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))

from consistency_check import (  # noqa: E402
    ConsistencyReport, report_to_dict, scan_source_revision,
)


def _git(repo: Path, *args: str) -> str:
    return subprocess.run(
        ["git", "-c", "user.name=t", "-c", "user.email=t@example.invalid", *args],
        cwd=repo, check=True, capture_output=True, text=True,
    ).stdout.strip()


def _repo(tmp_path: Path) -> Path:
    repo = tmp_path / "repo"
    (repo / "workplans").mkdir(parents=True)
    (repo / "workplans" / "DEMO-WP-0001-x.md").write_text("---\nid: DEMO-WP-0001\n---\n")
    (repo / "README.md").write_text("demo\n")
    _git(repo, "init", "-q")
    _git(repo, "add", ".")
    _git(repo, "commit", "-qm", "init")
    return repo


def test_clean_tree_reports_full_head_sha(tmp_path):
    repo = _repo(tmp_path)
    sha, reason = scan_source_revision(repo)
    assert sha == _git(repo, "rev-parse", "HEAD")
    assert len(sha) == 40 and sha == sha.lower()
    assert reason is None


def test_dirt_outside_workplans_does_not_unbind(tmp_path):
    repo = _repo(tmp_path)
    (repo / "README.md").write_text("changed\n")
    (repo / "scratch.txt").write_text("x\n")
    assert scan_source_revision(repo) == (_git(repo, "rev-parse", "HEAD"), None)


def test_modified_staged_or_untracked_workplans_are_not_head(tmp_path):
    repo = _repo(tmp_path)
    plan = repo / "workplans" / "DEMO-WP-0001-x.md"
    plan.write_text("---\nid: DEMO-WP-0001\nstatus: active\n---\n")
    assert scan_source_revision(repo) == (None, "dirty_workplans")
    _git(repo, "add", ".")
    assert scan_source_revision(repo) == (None, "dirty_workplans")
    _git(repo, "commit", "-qm", "two")
    assert scan_source_revision(repo)[1] is None
    (repo / "workplans" / "archived").mkdir()
    (repo / "workplans" / "archived" / "new.md").write_text("---\n---\n")
    assert scan_source_revision(repo) == (None, "dirty_workplans")


def test_ignored_workplan_file_is_not_head(tmp_path):
    repo = _repo(tmp_path)
    (repo / ".gitignore").write_text("*.local.md\n")
    _git(repo, "add", ".")
    _git(repo, "commit", "-qm", "ignore")
    (repo / "workplans" / "secret.local.md").write_text("---\n---\n")
    assert scan_source_revision(repo) == (None, "dirty_workplans")


def test_non_git_path_and_missing_head(tmp_path):
    plain = tmp_path / "plain"
    (plain / "workplans").mkdir(parents=True)
    assert scan_source_revision(plain) == (None, "not_a_git_tree")
    assert scan_source_revision(tmp_path / "absent") == (None, "not_a_git_tree")
    empty = tmp_path / "empty"
    empty.mkdir()
    _git(empty, "init", "-q")
    assert scan_source_revision(empty) == (None, "no_head_commit")


def _report(sha, reason):
    report = ConsistencyReport(repo_slug="demo", repo_path="/x",
                               source_revision=sha, source_revision_reason=reason)
    report.add(severity="WARN", check_id="C-25", message="m", file_path="workplans/p.md#T01",
               file_value="message-from:provider",
               _fix_context={"task_record_id": "DEMO-WP-0001-T01", "message_ids": ["a"]})
    report.add(severity="WARN", check_id="C-25", message="legacy", file_value="x")
    report.add(severity="WARN", check_id="C-24", message="other")
    return report_to_dict(report)


def test_json_report_and_c25_findings_carry_revision():
    sha = "0123456789abcdef0123456789abcdef01234567"
    data = _report(sha, None)
    assert (data["repo_slug"], data["source_revision"], data["source_revision_reason"]) == ("demo", sha, None)
    c25 = [i for i in data["issues"] if i["check_id"] == "C-25"]
    assert len(c25) == 2
    for issue in c25:
        assert (issue["repo_slug"], issue["source_revision"], issue["source_revision_reason"]) == ("demo", sha, None)
    other = next(i for i in data["issues"] if i["check_id"] == "C-24")
    assert "source_revision" not in other and "repo_slug" not in other


def test_json_unknown_revision_is_null_with_reason():
    data = _report(None, "dirty_workplans")
    assert data["source_revision"] is None
    assert data["source_revision_reason"] == "dirty_workplans"
    assert all(i["source_revision"] is None for i in data["issues"] if i["check_id"] == "C-25")


def test_unscanned_report_defaults_to_null():
    data = report_to_dict(ConsistencyReport(repo_slug="demo", repo_path="(unknown)"))
    assert (data["source_revision"], data["source_revision_reason"]) == (None, "not_scanned")


def _check(repo, monkeypatch, sweep=None):
    import socket

    import consistency_check

    def fake_get(_api_base, path, params=None, **_kwargs):
        if path == "/repos/demo":
            return {"id": "repo-1", "slug": "demo", "local_path": str(repo),
                    "host_paths": {socket.gethostname(): str(repo)}, "domain_slug": "infotech"}
        return []

    monkeypatch.setattr("consistency_check.load_classification_file", lambda _d: ({}, [], []))
    monkeypatch.setattr("consistency_check._api_get", fake_get)
    if sweep:
        monkeypatch.setattr("consistency_check._check_blocked_unblock_sweep", sweep)
    return consistency_check.check_repo("http://unused", "demo")


def test_check_repo_binds_clean_scan_to_head(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    report = _check(repo, monkeypatch)
    assert (report.source_revision, report.source_revision_reason) == (_git(repo, "rev-parse", "HEAD"), None)


def test_check_repo_dirty_workplans_is_null(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    (repo / "workplans" / "DEMO-WP-0001-x.md").write_text("---\nid: DEMO-WP-0001\nstatus: active\n---\n")
    report = _check(repo, monkeypatch)
    assert (report.source_revision, report.source_revision_reason) == (None, "dirty_workplans")


def test_check_repo_tree_changed_during_scan_is_null(tmp_path, monkeypatch):
    repo = _repo(tmp_path)

    def sweep(_api, _slug, repo_dir, _infos, _report):
        (repo_dir / "workplans" / "late.md").write_text("---\n---\n")

    report = _check(repo, monkeypatch, sweep)
    assert (report.source_revision, report.source_revision_reason) == (None, "changed_during_scan")