"""Tests for the stable-control runtime mode gates (#615). Covers acceptance criteria 6-11: runtime mode + SHA reporting, the fail-closed mutation gates (dev-test targeting production, unknown runtime, dirty stable checkout, dev-worktree launch, unsafe alignment), per-namespace post-flap re-proving, promotion-record completeness, and the policy statements that keep normal sessions from restarting the stable MCP runtime. """ import os import sys import unittest from pathlib import Path from unittest.mock import patch REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT)) import stable_control_runtime as scr # noqa: E402 SHA_A = "a" * 40 SHA_B = "b" * 40 STABLE_ROOT = "/Users/dev/Development/Gitea-Tools" DEV_WORKTREE_ROOT = "/Users/dev/Development/Gitea-Tools/branches/issue-615-work" def stable_report(**overrides): """A healthy stable-control runtime report, overridable per test.""" base = dict( process_root=STABLE_ROOT, checkout_branch="master", runtime_head=SHA_A, active_task_workspace=STABLE_ROOT, canonical_repository_root=STABLE_ROOT, repository_slug="Scaled-Tech-Consulting/Gitea-Tools", profile="prgs-author", authenticated_identity="jcwalker3", dirty_files=[], workspace_roots_aligned=True, ) base.update(overrides) return scr.build_runtime_report(**base) class TestClassifyRuntimeMode(unittest.TestCase): def test_stable_branch_checkout_is_stable_control(self): res = scr.classify_runtime_mode( process_root=STABLE_ROOT, checkout_branch="master") self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_STABLE) self.assertFalse(res["dev_worktree_launched"]) def test_main_and_dev_are_also_stable(self): for branch in ("main", "dev"): res = scr.classify_runtime_mode( process_root=STABLE_ROOT, checkout_branch=branch) self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_STABLE, branch) def test_branches_worktree_launch_is_dev_test(self): res = scr.classify_runtime_mode( process_root=DEV_WORKTREE_ROOT, checkout_branch="feat/issue-615-runtime-mode-enforcement", ) self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_DEV_TEST) self.assertTrue(res["dev_worktree_launched"]) def test_feature_branch_outside_branches_is_still_dev_test(self): res = scr.classify_runtime_mode( process_root="/Users/dev/Development/scratch-clone", checkout_branch="feat/experiment", ) self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_DEV_TEST) self.assertFalse(res["dev_worktree_launched"]) def test_unresolvable_root_is_unknown(self): res = scr.classify_runtime_mode(process_root=None, checkout_branch=None) self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_UNKNOWN) def test_non_git_root_is_unknown(self): res = scr.classify_runtime_mode( process_root="/opt/gitea-tools-release", checkout_branch=None, is_git_checkout=False, ) self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_UNKNOWN) def test_detached_head_is_unknown(self): res = scr.classify_runtime_mode( process_root=STABLE_ROOT, checkout_branch=None) self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_UNKNOWN) def test_operator_declaration_wins_over_inference(self): res = scr.classify_runtime_mode( process_root="/opt/gitea-tools-release", checkout_branch=None, is_git_checkout=False, declared_mode=scr.RUNTIME_MODE_STABLE, ) self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_STABLE) self.assertTrue(res["declared"]) def test_invalid_declaration_is_ignored(self): with patch.dict(os.environ, {scr.ENV_RUNTIME_MODE: "production-ish"}): self.assertIsNone(scr.declared_runtime_mode()) def test_valid_declaration_is_read_from_env(self): with patch.dict(os.environ, {scr.ENV_RUNTIME_MODE: "dev-test"}): self.assertEqual(scr.declared_runtime_mode(), scr.RUNTIME_MODE_DEV_TEST) class TestRuntimeReport(unittest.TestCase): """Acceptance criterion 6: runtime mode and SHA reporting.""" def test_report_carries_every_required_field(self): report = stable_report() for field in ( "runtime_mode", "runtime_git_sha", "runtime_branch", "runtime_checkout_path", "mcp_process_root", "active_task_workspace", "repository_slug", "profile", "authenticated_identity", "dirty_files", "workspace_roots_aligned", "real_mutations_allowed", ): self.assertIn(field, report, field) def test_report_records_the_runtime_sha(self): self.assertEqual(stable_report()["runtime_git_sha"], SHA_A) def test_format_summarises_mode_sha_and_branch(self): summary = scr.format_runtime_mode(stable_report()) self.assertIn(scr.RUNTIME_MODE_STABLE, summary) self.assertIn(SHA_A[:12], summary) self.assertIn("master", summary) class TestMutationGate(unittest.TestCase): """Acceptance criterion 7: fail-closed mutation gates.""" def test_stable_healthy_runtime_allows_real_mutations(self): report = stable_report() gate = scr.assess_runtime_mutation_gate(report) self.assertFalse(gate["block"]) self.assertEqual(gate["reasons"], []) self.assertTrue(report["real_mutations_allowed"]) def test_dev_test_runtime_blocks_real_production_mutations(self): report = stable_report( process_root=DEV_WORKTREE_ROOT, checkout_branch="feat/issue-615-runtime-mode-enforcement", active_task_workspace=DEV_WORKTREE_ROOT, ) gate = scr.assess_runtime_mutation_gate(report) self.assertTrue(gate["block"]) self.assertIn(scr.BLOCKER_DEV_TEST_PRODUCTION, gate["blocker_kinds"]) self.assertFalse(report["real_mutations_allowed"]) def test_dev_test_runtime_may_mutate_a_non_production_target(self): report = stable_report( process_root=DEV_WORKTREE_ROOT, checkout_branch="feat/issue-615-runtime-mode-enforcement", ) gate = scr.assess_runtime_mutation_gate( report, target_is_production=False) self.assertFalse(gate["block"]) def test_unknown_runtime_blocks_mutations(self): report = stable_report(checkout_branch=None) gate = scr.assess_runtime_mutation_gate(report) self.assertTrue(gate["block"]) self.assertIn(scr.BLOCKER_UNKNOWN_RUNTIME, gate["blocker_kinds"]) def test_unknown_runtime_blocks_even_a_non_production_target(self): report = stable_report(checkout_branch=None) gate = scr.assess_runtime_mutation_gate( report, target_is_production=False) self.assertTrue(gate["block"]) def test_dirty_stable_runtime_blocks_mutations(self): report = stable_report(dirty_files=["gitea_mcp_server.py"]) gate = scr.assess_runtime_mutation_gate(report) self.assertTrue(gate["block"]) self.assertIn(scr.BLOCKER_DIRTY_STABLE_RUNTIME, gate["blocker_kinds"]) self.assertTrue( any("dirty" in reason for reason in gate["reasons"])) def test_dev_worktree_launch_is_reported_as_its_own_blocker(self): report = stable_report( process_root=DEV_WORKTREE_ROOT, checkout_branch="feat/issue-615-runtime-mode-enforcement", ) gate = scr.assess_runtime_mutation_gate(report) self.assertIn(scr.BLOCKER_DEV_WORKTREE_LAUNCH, gate["blocker_kinds"]) def test_unsafe_workspace_alignment_blocks_mutations(self): report = stable_report(workspace_roots_aligned=False) gate = scr.assess_runtime_mutation_gate(report) self.assertTrue(gate["block"]) self.assertIn(scr.BLOCKER_UNSAFE_ALIGNMENT, gate["blocker_kinds"]) def test_unknown_alignment_does_not_block(self): report = stable_report(workspace_roots_aligned=None) self.assertFalse(scr.assess_runtime_mutation_gate(report)["block"]) def test_env_escape_hatch_disables_the_gate(self): report = stable_report(checkout_branch=None) with patch.dict(os.environ, {scr.ENV_DISABLE: "1"}): gate = scr.assess_runtime_mutation_gate(report) self.assertFalse(gate["block"]) self.assertTrue(gate["gate_disabled"]) def test_block_reasons_helper_matches_the_gate(self): report = stable_report(checkout_branch=None) self.assertEqual( scr.runtime_block_reasons(report), scr.assess_runtime_mutation_gate(report)["reasons"], ) def test_block_payload_names_the_operator_recovery_path(self): report = stable_report(checkout_branch=None) payload = scr.runtime_report_payload(report) self.assertEqual(payload["kind"], "runtime_mode_block") self.assertEqual(payload["blocker_kind"], scr.BLOCKER_UNKNOWN_RUNTIME) self.assertTrue( any("promotion-runbook" in line for line in payload["recovery"])) class TestPostFlapReproving(unittest.TestCase): """Acceptance criterion 8: per-namespace post-flap re-proving.""" def test_no_flap_means_no_reproof_required(self): state = scr.new_reproof_state() res = scr.assess_namespace_reproof(state, "reviewer") self.assertFalse(res["reproof_required"]) self.assertTrue(res["proven"]) def test_transport_recovery_requires_namespace_specific_reproving(self): state = scr.record_transport_flap( scr.new_reproof_state(), at="2026-07-20T14:00:00Z") res = scr.assess_namespace_reproof(state, "reviewer") self.assertTrue(res["reproof_required"]) self.assertFalse(res["proven"]) self.assertEqual( res["missing_steps"], list(scr.REQUIRED_NAMESPACE_PROOF_STEPS)) def test_author_proof_does_not_imply_other_namespaces(self): state = scr.record_transport_flap( scr.new_reproof_state(), at="2026-07-20T14:00:00Z") state = scr.record_namespace_proof( state, "author", at="2026-07-20T14:05:00Z", whoami=True, runtime_context=True, capability_resolved=True, ) self.assertTrue(scr.assess_namespace_reproof(state, "author")["proven"]) for other in ("reviewer", "merger", "reconciler"): assessment = scr.assess_namespace_reproof(state, other) self.assertFalse(assessment["proven"], other) self.assertTrue( any("does not transfer" in reason for reason in assessment["reasons"]), other, ) self.assertEqual( scr.unproven_namespaces(state), ["reviewer", "merger", "reconciler"], ) def test_proof_recorded_before_the_flap_does_not_count(self): state = scr.record_namespace_proof( scr.new_reproof_state(), "merger", at="2026-07-20T13:00:00Z", whoami=True, runtime_context=True, capability_resolved=True, ) state = scr.record_transport_flap(state, at="2026-07-20T14:00:00Z") res = scr.assess_namespace_reproof(state, "merger") self.assertFalse(res["proven"]) self.assertTrue(any("predates" in reason for reason in res["reasons"])) def test_incomplete_proof_lists_the_missing_steps(self): state = scr.record_transport_flap( scr.new_reproof_state(), at="2026-07-20T14:00:00Z") state = scr.record_namespace_proof( state, "reviewer", at="2026-07-20T14:05:00Z", whoami=True) res = scr.assess_namespace_reproof(state, "reviewer") self.assertFalse(res["proven"]) self.assertEqual( res["missing_steps"], ["runtime_context", "capability_resolved"]) def test_stale_runtime_report_keeps_the_namespace_unproven(self): state = scr.record_transport_flap( scr.new_reproof_state(), at="2026-07-20T14:00:00Z") state = scr.record_namespace_proof( state, "reviewer", at="2026-07-20T14:05:00Z", whoami=True, runtime_context=True, capability_resolved=True, stale_runtime_reported=True, ) res = scr.assess_namespace_reproof(state, "reviewer") self.assertFalse(res["proven"]) self.assertTrue( any("stale-runtime" in reason for reason in res["reasons"])) def test_unproven_namespace_blocks_the_mutation_gate(self): state = scr.record_transport_flap( scr.new_reproof_state(), at="2026-07-20T14:00:00Z") gate = scr.assess_runtime_mutation_gate( stable_report(), namespace="reviewer", namespace_reproof=state) self.assertTrue(gate["block"]) self.assertIn(scr.BLOCKER_NAMESPACE_NOT_REPROVEN, gate["blocker_kinds"]) def test_reproven_namespace_clears_the_mutation_gate(self): state = scr.record_transport_flap( scr.new_reproof_state(), at="2026-07-20T14:00:00Z") state = scr.record_namespace_proof( state, "reviewer", at="2026-07-20T14:05:00Z", whoami=True, runtime_context=True, capability_resolved=True, ) gate = scr.assess_runtime_mutation_gate( stable_report(), namespace="reviewer", namespace_reproof=state) self.assertFalse(gate["block"]) class TestPromotionRecord(unittest.TestCase): """Acceptance criteria 4 / 10: promotion records previous and promoted SHAs.""" def complete_record(self, **overrides): record = { "previous_runtime_sha": SHA_A, "promoted_runtime_sha": SHA_B, "source_branch": "feat/issue-615-runtime-mode-enforcement", "source_pr": "770", "restart_method": "operator reload of the stable control runtime", "health_check_proof": "gitea_assess_mcp_namespace_health: healthy", "identity_proof": "gitea_whoami: sysadmin / prgs-reviewer", "profile_proof": "runtime context: prgs-reviewer", "workspace_proof": "process root == canonical root, clean", "mutation_capability_proof": "resolve review_pr: allowed", "rollback_instructions": "re-promote " + SHA_A, } record.update(overrides) return record def test_complete_record_is_valid(self): res = scr.assess_promotion_record(self.complete_record()) self.assertTrue(res["valid"]) self.assertEqual(res["missing_fields"], []) def test_promotion_records_previous_and_promoted_shas(self): res = scr.assess_promotion_record( self.complete_record(previous_runtime_sha="", promoted_runtime_sha="")) self.assertFalse(res["valid"]) self.assertIn("previous_runtime_sha", res["missing_fields"]) self.assertIn("promoted_runtime_sha", res["missing_fields"]) def test_identical_shas_are_not_a_promotion(self): res = scr.assess_promotion_record( self.complete_record(promoted_runtime_sha=SHA_A)) self.assertFalse(res["valid"]) self.assertTrue( any("nothing was promoted" in reason for reason in res["reasons"])) def test_missing_rollback_instructions_fail_closed(self): res = scr.assess_promotion_record( self.complete_record(rollback_instructions="")) self.assertFalse(res["valid"]) self.assertIn("rollback_instructions", res["missing_fields"]) def test_empty_record_is_invalid(self): self.assertFalse(scr.assess_promotion_record(None)["valid"]) class TestNormalSessionsCannotRestartStableRuntime(unittest.TestCase): """Acceptance criterion 3: normal sessions do not restart the stable MCP.""" def test_adr_forbids_kill_restart_and_relaunch(self): adr = ( REPO_ROOT / "docs" / "architecture" / "mcp-stable-control-runtime-policy-adr.md" ).read_text() for phrase in ("Kill the running MCP server process", "Restart / relaunch the MCP server process", "Relaunch MCP from a development worktree"): self.assertIn(phrase, adr, phrase) def test_promotion_runbook_exists_and_lists_every_record_field(self): runbook = ( REPO_ROOT / "docs" / "stable-runtime-promotion-runbook.md" ).read_text() for field in scr.PROMOTION_REQUIRED_FIELDS: self.assertIn(field, runbook, field) def test_no_mcp_tool_offers_a_runtime_restart(self): server = (REPO_ROOT / "gitea_mcp_server.py").read_text() for forbidden in ("def gitea_restart_", "def gitea_kill_"): self.assertNotIn(forbidden, server, forbidden) class TestServerWiring(unittest.TestCase): """The gate is wired into the server's mutation permission path.""" def setUp(self): import gitea_mcp_server as srv # imported lazily: heavy module self.srv = srv def test_reads_are_never_blocked_by_runtime_mode(self): with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}): self.assertEqual(self.srv._runtime_mode_block("gitea.read"), []) def test_gate_is_skipped_under_pure_unit_test_isolation(self): # The suite itself runs from a branches/ worktree (dev-test by design); # without forced production guards the gate must not fire. self.assertEqual(self.srv._runtime_mode_block("gitea.pr.create"), []) def test_dev_worktree_runtime_blocks_mutations_when_guards_forced(self): report = stable_report( process_root=DEV_WORKTREE_ROOT, checkout_branch="feat/issue-615-runtime-mode-enforcement", ) with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}), \ patch.object( self.srv, "_current_runtime_mode_report", return_value=report): reasons = self.srv._runtime_mode_block("gitea.pr.create") self.assertTrue(reasons) self.assertTrue(any("dev-test" in reason for reason in reasons)) def test_stable_runtime_allows_mutations_when_guards_forced(self): with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}), \ patch.object( self.srv, "_current_runtime_mode_report", return_value=stable_report()): self.assertEqual(self.srv._runtime_mode_block("gitea.pr.create"), []) def test_unassessable_runtime_fails_closed(self): with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}), \ patch.object( self.srv, "_current_runtime_mode_report", side_effect=RuntimeError("boom")): reasons = self.srv._runtime_mode_block("gitea.pr.create") self.assertTrue(reasons) self.assertTrue(any("fail closed" in reason for reason in reasons)) def test_live_report_describes_this_checkout(self): report = self.srv._current_runtime_mode_report() self.assertIn(report["runtime_mode"], scr.VALID_RUNTIME_MODES) self.assertEqual(report["mcp_process_root"], self.srv.PROJECT_ROOT) if __name__ == "__main__": unittest.main()