from __future__ import annotations
import importlib.util
import json
import subprocess
import sys
import unittest
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
REPO = Path(__file__).resolve().parents[2]
MODULE_PATH = REPO / "scripts/ci/run_sensitive_implement_gate.py"
SPEC = importlib.util.spec_from_file_location("sensitive_gate", MODULE_PATH)
assert SPEC is not None and SPEC.loader is not None
gate = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = gate
SPEC.loader.exec_module(gate)
NOW = datetime(2026, 7, 21, 12, 0, tzinfo=timezone.utc)
HEAD = "a" * 40
BASE = "b" * 40
REPO_NAME = "majiayu000/remem"
HEAD_REF = "codex/gh813-implementation"
def completed(argv: list[str], payload: Any = None, *, stdout: str | None = None) -> subprocess.CompletedProcess[str]:
text = stdout if stdout is not None else json.dumps(payload)
return subprocess.CompletedProcess(argv, 0, text, "")
class FakeRunner:
def __init__(self) -> None:
self.origin = "git@github.com:majiayu000/remem.git"
self.pr = {
"number": 908,
"state": "OPEN",
"draft": True,
"body": "Refs #813\nenforcement_sensitive: true",
"html_url": "https://github.com/majiayu000/remem/pull/908",
"changed_files": 1,
"head": {
"ref": HEAD_REF,
"sha": HEAD,
"repo": {"full_name": REPO_NAME, "fork": False},
},
"base": {
"ref": "main",
"sha": BASE,
"repo": {"full_name": REPO_NAME, "fork": False},
},
}
self.final_pr: dict[str, Any] | None = None
self.pr_query_count = 0
self.remote_head_sha = HEAD
self.current_issue = {
"number": 813,
"state": "OPEN",
"labels": [{"name": "ready_to_implement"}],
}
self.events: Any = [[{
"id": 17,
"event": "labeled",
"created_at": "2026-07-21T11:59:00Z",
"label": {"name": "ready_to_implement"},
"actor": {"login": "majiayu000", "type": "User"},
}]]
self.final_events: Any | None = None
self.event_query_count = 0
self.changed_files_pages: Any = [[
{"filename": "workflow.yaml", "status": "modified"}
]]
self.final_changed_files_pages: Any | None = None
self.changed_files_query_count = 0
self.issue = {
"issue": 813,
"repository": REPO_NAME,
"state": "ready_to_implement",
"state_source": "label",
"state_trusted": True,
"default_base_sha": BASE,
}
self.duplicate = {
"issue": 813,
"collected_at": "2026-07-21T11:59:30Z",
"open_prs_complete": True,
"open_pr_limit": 100,
"open_prs": [{"number": 908, "head_ref": HEAD_REF, "references_issue": True}],
"remote_branches": [HEAD_REF, "main"],
}
self.final_duplicate: dict[str, Any] | None = None
self.duplicate_query_count = 0
self.permission = {"permission": "maintain", "user": {"type": "User"}}
self.comment: dict[str, Any] | None = None
self.mutate_route_input = False
self.commands: list[list[str]] = []
def __call__(self, argv: list[str]) -> subprocess.CompletedProcess[str]:
self.commands.append(list(argv))
if argv[:4] == ["git", "-C", str(REPO), "remote"]:
return completed(argv, stdout=self.origin + "\n")
if argv[:4] == ["git", "-C", str(REPO), "check-ref-format"]:
return completed(argv, stdout="")
if argv[:4] == ["git", "-C", str(REPO), "rev-parse"]:
return completed(argv, stdout=self.remote_head_sha + "\n")
if argv[:2] == ["gh", "api"] and argv[2] == f"repos/{REPO_NAME}/pulls/908":
self.pr_query_count += 1
return completed(argv, self.final_pr if self.pr_query_count > 1 and self.final_pr is not None else self.pr)
if argv[:3] == ["gh", "issue", "view"]:
return completed(argv, self.current_issue)
if argv[:4] == ["gh", "api", "--paginate", "--slurp"] and "/files?" in argv[4]:
self.changed_files_query_count += 1
payload = self.changed_files_pages
if self.changed_files_query_count > 1 and self.final_changed_files_pages is not None:
payload = self.final_changed_files_pages
return completed(argv, payload)
if argv[:4] == ["gh", "api", "--paginate", "--slurp"]:
self.event_query_count += 1
payload = self.events
if self.event_query_count > 1 and self.final_events is not None:
payload = self.final_events
return completed(argv, payload)
if argv[:2] == ["gh", "api"] and "/collaborators/" in argv[2]:
return completed(argv, self.permission)
if argv[:2] == ["gh", "api"] and "/issues/comments/" in argv[2]:
return completed(argv, self.comment)
if argv[0] == sys.executable and argv[1].endswith("github_issue_evidence.py"):
return completed(argv, self.issue)
if argv[0] == sys.executable and argv[1].endswith("github_duplicate_evidence.py"):
self.duplicate_query_count += 1
payload = self.duplicate
if self.duplicate_query_count > 1 and self.final_duplicate is not None:
payload = self.final_duplicate
return completed(argv, payload)
if argv[0] == sys.executable and argv[1].endswith("route_gate.py"):
evidence_path = Path(argv[argv.index("--evidence") + 1])
duplicate_path = Path(argv[argv.index("--duplicate-evidence") + 1])
duplicate = json.loads(duplicate_path.read_text())
blocked = any(item["references_issue"] for item in duplicate["open_prs"])
matching = gate.matching_branches(REPO, 813, duplicate["remote_branches"])
if self.mutate_route_input:
evidence_path.write_text("{}\n", encoding="utf-8")
result = {
"decision": "blocked" if blocked or matching else "allowed",
"missing": ["duplicate_work"] if blocked or matching else [],
"current_state": "ready_to_implement",
"satisfied": ["state provided by evidence: ready_to_implement (label)"],
"sensitive_classification": {"enforcement_sensitive": True},
}
return completed(argv, result)
if argv[:4] == ["git", "-C", str(REPO), "show"]:
return completed(
argv,
stdout=(REPO / "specs/GH813/tech.md").read_text(encoding="utf-8"),
)
raise AssertionError(f"unexpected command: {argv}")
def config() -> Any:
return gate.Config(REPO, REPO_NAME, 813, 908, HEAD, REPO / ".specrail/unused.json")
class SensitiveImplementGateTests(unittest.TestCase):
def execute(self, runner: FakeRunner) -> dict[str, Any]:
return gate.execute(config(), runner=runner, now=lambda: NOW)
def assert_blocked(self, runner: FakeRunner, pattern: str) -> None:
with self.assertRaisesRegex(gate.GateError, pattern):
self.execute(runner)
def test_schema_valid_current_head_positive(self) -> None:
runner = FakeRunner()
result = self.execute(runner)
self.assertEqual(result["decision"], "allowed")
self.assertEqual(result["authorization"], "advisory_only")
self.assertFalse(result["ordinary_pr_ci_is_final_authorization"])
self.assertEqual(result["pr"]["head_sha"], HEAD)
self.assertEqual(result["label_authority"]["role"], "owner")
self.assertEqual(result["artifact_hashes"]["issue_evidence_pre"], result["artifact_hashes"]["issue_evidence_post"])
self.assertEqual(result["artifact_hashes"]["duplicate_filtered_pre"], result["artifact_hashes"]["duplicate_filtered_post"])
route = next(command for command in runner.commands if command[1].endswith("route_gate.py"))
self.assertNotIn("--state", route)
self.assertNotIn("--label", route)
self.assertEqual(result["artifacts"]["duplicate_original"]["open_prs"][0]["number"], 908)
self.assertEqual(result["artifacts"]["duplicate_filtered"]["open_prs"], [])
def test_wrong_local_origin_fails(self) -> None:
runner = FakeRunner()
runner.origin = "https://github.com/other/remem.git"
self.assert_blocked(runner, "origin does not match")
def test_wrong_pr_head_fails(self) -> None:
runner = FakeRunner()
runner.pr["head"]["sha"] = "c" * 40
self.assert_blocked(runner, "head does not match")
def test_fork_pr_fails_even_with_matching_head_sha(self) -> None:
runner = FakeRunner()
runner.pr["head"]["repo"] = {
"full_name": REPO_NAME,
"fork": True,
}
self.assert_blocked(runner, "must not be a fork")
def test_same_named_remote_branch_at_different_sha_fails(self) -> None:
runner = FakeRunner()
runner.remote_head_sha = "b" * 40
self.assert_blocked(runner, "remote-tracking head ref does not resolve")
def test_closed_pr_fails(self) -> None:
runner = FakeRunner()
runner.pr["state"] = "MERGED"
self.assert_blocked(runner, "not the requested open PR")
def test_missing_head_repository_fails(self) -> None:
runner = FakeRunner()
runner.pr["head"]["repo"] = None
self.assert_blocked(runner, "head repository")
def test_base_repository_mismatch_fails(self) -> None:
runner = FakeRunner()
runner.pr["base"]["repo"]["full_name"] = "other/remem"
self.assert_blocked(runner, "base repository")
def test_unlinked_issue_fails(self) -> None:
runner = FakeRunner()
runner.pr["body"] = "No linked work\nenforcement_sensitive: true"
self.assert_blocked(runner, "does not reference")
def test_bot_label_actor_fails(self) -> None:
runner = FakeRunner()
runner.events[0][0]["actor"] = {"login": "release[bot]", "type": "Bot"}
self.assert_blocked(runner, "non-bot")
def test_agent_named_user_fails(self) -> None:
runner = FakeRunner()
runner.events[0][0]["actor"] = {"login": "merge-agent", "type": "User"}
self.assert_blocked(runner, "non-bot")
def test_non_maintainer_label_actor_fails(self) -> None:
runner = FakeRunner()
runner.events[0][0]["actor"] = {"login": "contributor", "type": "User"}
runner.permission["permission"] = "write"
self.assert_blocked(runner, "lacks live admin or maintain")
def test_missing_label_event_actor_fails(self) -> None:
runner = FakeRunner()
del runner.events[0][0]["actor"]
self.assert_blocked(runner, "label event actor")
def test_untrusted_issue_label_state_fails(self) -> None:
runner = FakeRunner()
runner.issue["state_trusted"] = False
self.assert_blocked(runner, "not trusted label-derived")
def test_ready_label_removed_before_allowed_output_fails(self) -> None:
runner = FakeRunner()
runner.current_issue["labels"] = []
self.assert_blocked(runner, "label was removed")
def test_final_pr_head_drift_fails(self) -> None:
runner = FakeRunner()
runner.final_pr = deepcopy(runner.pr)
runner.final_pr["head"]["sha"] = "c" * 40
self.assert_blocked(runner, "head does not match")
def test_final_pr_state_drift_fails(self) -> None:
runner = FakeRunner()
runner.final_pr = deepcopy(runner.pr)
runner.final_pr["state"] = "MERGED"
self.assert_blocked(runner, "not the requested open PR")
def test_final_pr_body_drift_fails(self) -> None:
runner = FakeRunner()
runner.final_pr = deepcopy(runner.pr)
runner.final_pr["body"] += "\nchanged"
self.assert_blocked(runner, "body changed")
def test_final_readiness_event_drift_fails(self) -> None:
runner = FakeRunner()
runner.final_events = deepcopy(runner.events)
runner.final_events[0][0]["id"] = 18
self.assert_blocked(runner, "readiness label event or authority changed")
def test_changed_file_count_mismatch_fails(self) -> None:
runner = FakeRunner()
runner.pr["changed_files"] = 2
self.assert_blocked(runner, "count does not match")
def test_complete_multi_page_changed_files_pass(self) -> None:
runner = FakeRunner()
runner.pr["changed_files"] = 2
runner.changed_files_pages = [
[{"filename": "workflow.yaml", "status": "modified"}],
[{"filename": "scripts/ci/check_pr_tier.py", "status": "modified"}],
]
result = self.execute(runner)
self.assertEqual(result["changed_files"]["collected_count"], 2)
def test_rename_preserves_old_and_new_paths(self) -> None:
runner = FakeRunner()
runner.changed_files_pages = [[{
"filename": "scripts/ci/check_pr_tier.py",
"previous_filename": "workflow.yaml",
"status": "renamed",
}]]
result = self.execute(runner)
self.assertEqual(
result["changed_files"]["classification_paths"],
["scripts/ci/check_pr_tier.py", "workflow.yaml"],
)
def test_unplanned_changed_file_fails(self) -> None:
runner = FakeRunner()
runner.changed_files_pages = [[{"filename": "unplanned.txt", "status": "added"}]]
self.assert_blocked(runner, "exceed the approved planned-change manifest")
def test_changed_file_drift_before_output_fails(self) -> None:
runner = FakeRunner()
runner.final_changed_files_pages = [[
{"filename": "scripts/ci/check_pr_tier.py", "status": "modified"}
]]
self.assert_blocked(runner, "changed-file evidence drifted")
def test_stale_duplicate_evidence_fails(self) -> None:
runner = FakeRunner()
runner.duplicate["collected_at"] = "2026-07-21T11:54:59Z"
self.assert_blocked(runner, "older than 300")
def test_future_duplicate_evidence_fails(self) -> None:
runner = FakeRunner()
runner.duplicate["collected_at"] = "2026-07-21T12:00:01Z"
self.assert_blocked(runner, "future-dated")
def test_evidence_collected_after_wrapper_start_is_fresh(self) -> None:
runner = FakeRunner()
runner.duplicate["collected_at"] = "2026-07-21T12:00:00.500000Z"
clock = iter([
datetime(2026, 7, 21, 12, 0, 0, tzinfo=timezone.utc),
datetime(2026, 7, 21, 12, 0, 1, tzinfo=timezone.utc),
datetime(2026, 7, 21, 12, 0, 2, tzinfo=timezone.utc),
])
result = gate.execute(config(), runner=runner, now=lambda: next(clock))
self.assertEqual(result["evidence_trust"]["duplicate_age_seconds"], 0.5)
def test_incomplete_pr_collection_fails(self) -> None:
runner = FakeRunner()
runner.duplicate["open_prs_complete"] = False
self.assert_blocked(runner, "incomplete open PR")
def test_other_referencing_pr_remains_blocking(self) -> None:
runner = FakeRunner()
runner.duplicate["open_prs"].append({"number": 777, "head_ref": "human/other", "references_issue": True})
self.assert_blocked(runner, "nested route gate is not allowed")
def test_other_matching_branch_requires_decision(self) -> None:
runner = FakeRunner()
runner.duplicate["remote_branches"].append("human/gh813-existing")
self.assert_blocked(runner, "exactly one explicit SpecRail")
def test_human_branch_decision_filters_only_listed_branch(self) -> None:
runner = FakeRunner()
runner.duplicate["remote_branches"].append("human/gh813-existing")
runner.pr["body"] += "\nhttps://github.com/majiayu000/remem/issues/813#issuecomment-12345"
runner.comment = {
"body": "\n".join([
"SpecRail branch ownership decision",
"Repository: majiayu000/remem", "Issue: 813", "PR: 908", f"Head-SHA: {HEAD}",
"Decision: continue_existing_work", "Rationale: maintainer confirms this is the same implementation lane",
"Branches:", "- human/gh813-existing",
]),
"created_at": "2026-07-21T11:58:00Z",
"html_url": "https://github.com/majiayu000/remem/issues/813#issuecomment-12345",
"user": {"login": "majiayu000", "type": "User"},
}
result = self.execute(runner)
self.assertEqual(result["branch_ownership_decision"]["branches"], ["human/gh813-existing"])
self.assertNotIn("human/gh813-existing", result["artifacts"]["duplicate_filtered"]["remote_branches"])
self.assertIn("main", result["artifacts"]["duplicate_filtered"]["remote_branches"])
def test_cleanup_completed_cannot_hide_live_branch(self) -> None:
runner = FakeRunner()
runner.duplicate["remote_branches"].append("human/gh813-existing")
runner.pr["body"] += "\nhttps://github.com/majiayu000/remem/issues/813#issuecomment-12345"
runner.comment = {
"body": "\n".join([
"SpecRail branch ownership decision",
"Repository: majiayu000/remem", "Issue: 813", "PR: 908",
f"Head-SHA: {HEAD}", "Decision: cleanup_completed",
"Rationale: cleanup was requested", "Branches: human/gh813-existing",
]),
"created_at": "2026-07-21T11:58:00Z",
"html_url": "https://github.com/majiayu000/remem/issues/813#issuecomment-12345",
"user": {"login": "majiayu000", "type": "User"},
}
self.assert_blocked(runner, "still names live remote branches")
def test_cleanup_completed_requires_fresh_absence(self) -> None:
runner = FakeRunner()
runner.duplicate["remote_branches"].append("human/gh813-existing")
runner.final_duplicate = deepcopy(runner.duplicate)
runner.final_duplicate["remote_branches"].remove("human/gh813-existing")
runner.pr["body"] += "\nhttps://github.com/majiayu000/remem/issues/813#issuecomment-12345"
runner.comment = {
"body": "\n".join([
"SpecRail branch ownership decision",
"Repository: majiayu000/remem", "Issue: 813", "PR: 908",
f"Head-SHA: {HEAD}", "Decision: cleanup_completed",
"Rationale: branch was removed by a maintainer",
"Branches: human/gh813-existing",
]),
"created_at": "2026-07-21T11:58:00Z",
"html_url": "https://github.com/majiayu000/remem/issues/813#issuecomment-12345",
"user": {"login": "majiayu000", "type": "User"},
}
result = self.execute(runner)
self.assertNotIn(
"human/gh813-existing",
result["artifacts"]["duplicate_after_decision"]["remote_branches"],
)
def test_decision_must_cover_every_conflicting_branch(self) -> None:
runner = FakeRunner()
runner.duplicate["remote_branches"] += ["human/gh813-one", "human/gh813-two"]
runner.pr["body"] += "\nhttps://github.com/majiayu000/remem/issues/813#issuecomment-12345"
runner.comment = {
"body": "\n".join([
"SpecRail branch ownership decision",
"Repository: majiayu000/remem", "Issue: 813", "PR: 908", f"Head-SHA: {HEAD}",
"Decision: continue_existing_work", "Rationale: only one branch was reviewed", "Branches: human/gh813-one",
]),
"created_at": "2026-07-21T11:58:00Z",
"html_url": "https://github.com/majiayu000/remem/issues/813#issuecomment-12345",
"user": {"login": "majiayu000", "type": "User"},
}
self.assert_blocked(runner, "does not cover all conflicting")
def test_sha_authorization_comment_can_coexist_with_ownership_decision(self) -> None:
runner = FakeRunner()
runner.duplicate["remote_branches"].append("human/gh813-existing")
runner.pr["body"] += "\n".join([
"\nhttps://github.com/majiayu000/remem/issues/813#issuecomment-99999",
"https://github.com/majiayu000/remem/issues/813#issuecomment-12345",
])
ownership = {
"body": "\n".join([
"SpecRail branch ownership decision",
"Repository: majiayu000/remem", "Issue: 813", "PR: 908", f"Head-SHA: {HEAD}",
"Decision: continue_existing_work", "Rationale: one implementation lane is retained",
"Branches: human/gh813-existing",
]),
"created_at": "2026-07-21T11:58:00Z",
"html_url": "https://github.com/majiayu000/remem/issues/813#issuecomment-12345",
"user": {"login": "majiayu000", "type": "User"},
}
authorization = {
"body": "Exact-SHA authorization only",
"created_at": "2026-07-21T11:57:00Z",
"html_url": "https://github.com/majiayu000/remem/issues/813#issuecomment-99999",
"user": {"login": "majiayu000", "type": "User"},
}
original = runner.__call__
def comments(argv: list[str]) -> subprocess.CompletedProcess[str]:
if argv[:2] == ["gh", "api"] and argv[2].endswith("/99999"):
return completed(argv, authorization)
if argv[:2] == ["gh", "api"] and argv[2].endswith("/12345"):
return completed(argv, ownership)
return original(argv)
result = gate.execute(config(), runner=comments, now=lambda: NOW)
self.assertEqual(result["branch_ownership_decision"]["comment_id"], 12345)
def test_ci_invokes_wrapper_and_never_bare_route_gate(self) -> None:
workflow = (REPO / ".github/workflows/ci.yml").read_text(encoding="utf-8")
self.assertIn("scripts/ci/test_run_sensitive_implement_gate.py", workflow)
self.assertIn("scripts/ci/run_sensitive_implement_gate.py", workflow)
self.assertIn("scripts/ci/extract_nonclosing_issue.py", workflow)
self.assertIn("steps.linked_issue.outputs.number", workflow)
self.assertNotIn("sensitive implementation PR must close exactly one issue", workflow)
self.assertNotIn("python3 checks/route_gate.py", workflow)
self.assertIn("1f67531098f8a3fb96a34153593f427280a7e5be", workflow)
self.assertNotIn("e4867a0517df0d0e8487ac20d702d7a5444c321a", workflow)
self.assertNotIn("bbc762b6cf6c323e8ea1996e8c7ca44bc41ca9e5", workflow)
def test_ci_reuses_the_closing_aware_issue_for_sensitive_wrapper(self) -> None:
workflow = (REPO / ".github/workflows/ci.yml").read_text(encoding="utf-8")
resolve = workflow.index("- name: Resolve linked issue for classification")
gate = workflow.index("- name: Run advisory sensitive implementation gate")
resolve_block = workflow[resolve:gate]
gate_block = workflow[gate:]
self.assertIn("--allow-closing", resolve_block)
self.assertIn('ISSUE_NUMBER: ${{ steps.linked_issue.outputs.number }}', gate_block)
self.assertNotIn("strict_issue", gate_block)
def test_ci_pins_trusted_default_branch_before_wrapper(self) -> None:
workflow = (REPO / ".github/workflows/ci.yml").read_text(encoding="utf-8")
pin = workflow.index("- name: Pin trusted default branch")
gate = workflow.index("- name: Run advisory sensitive implementation gate")
self.assertLess(pin, gate)
trusted_block = workflow[pin:gate]
self.assertIn(
"DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}",
trusted_block,
)
self.assertIn("git check-ref-format --branch", trusted_block)
self.assertIn("refs/remotes/origin/HEAD", trusted_block)
self.assertIn("refs/remotes/origin/$DEFAULT_BRANCH", trusted_block)
self.assertNotIn("contains(github.event.pull_request.body", workflow)
self.assertIn("steps.pr_tier.outputs.enforcement_sensitive == 'true'", workflow)
def test_gate_input_hash_drift_fails(self) -> None:
runner = FakeRunner()
runner.mutate_route_input = True
self.assert_blocked(runner, "evidence changed during gate")
def test_duplicate_schema_extra_field_fails(self) -> None:
runner = FakeRunner()
runner.duplicate["repository"] = REPO_NAME
self.assert_blocked(runner, "fields do not match")
if __name__ == "__main__":
unittest.main()