"""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")