"""Regression coverage for #690: cross-role profile activation invalidation. A mid-run profile switch (e.g. reviewer → author → reviewer) must invalidate workflow-load proof, reviewer lease binding, review decision lock, live namespace health, and preflight identity/capability stamps, and must require a full reviewer preflight before any formal verdict. Namespace provenance must be reported and fail closed on task/namespace mismatch. """ import json import os import sys import tempfile import unittest from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import gitea_config import mcp_namespace_health import mcp_server import mcp_session_state import review_workflow_load import reviewer_pr_lease from tests.test_runtime_clarity import CONFIG_SWITCHING_ENABLED class TestProfileSwitchReviewGuard(unittest.TestCase): def setUp(self): self._remotes_patch = patch.dict(mcp_server.REMOTES, { "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}, "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}, }) self._remotes_patch.start() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None self._dir = tempfile.TemporaryDirectory() self.config_path = os.path.join(self._dir.name, "profiles.json") with open(self.config_path, "w", encoding="utf-8") as fh: fh.write(json.dumps(CONFIG_SWITCHING_ENABLED)) def tearDown(self): self._remotes_patch.stop() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None self._dir.cleanup() def _env(self, profile="reviewer-profile"): return { "GITEA_MCP_CONFIG": self.config_path, "GITEA_MCP_PROFILE": profile, "GITEA_TOKEN_AUTHOR": "author-pass", "GITEA_TOKEN_REVIEWER": "reviewer-pass", "GITEA_TOKEN_MERGER": "merger-pass", } def _seed_contaminated_review_state(self): """Simulate an in-flight reviewer run under reviewer-profile.""" mcp_server._preflight_whoami_called = True mcp_server._preflight_capability_called = True mcp_server._preflight_resolved_role = "reviewer" mcp_server._preflight_resolved_task = "review_pr" review_workflow_load._REVIEW_WORKFLOW_LOAD = {"loaded": True} mcp_server._REVIEW_DECISION_LOCK = { "session_profile": "reviewer-profile", "final_review_decision_ready": True, "ready_pr_number": 688, } reviewer_pr_lease.record_session_lease( {"session_id": "lease-session-1", "pr_number": 688} ) mcp_server._LIVE_NAMESPACE_HEALTH["gitea-reviewer"] = { "namespace": "gitea-reviewer", "healthy": True, "ide_namespace_proven": True, } # Durable records keyed by the reviewer identity must also be cleared. mcp_session_state.save_state( kind=mcp_session_state.KIND_WORKFLOW_LOAD, payload={"loaded": True}, profile_identity="reviewer-profile", ) mcp_session_state.save_state( kind=mcp_session_state.KIND_DECISION_LOCK, payload={"final_review_decision_ready": True, "ready_pr_number": 688}, profile_identity="reviewer-profile", ) def _activate(self, target, logins): with patch.object( mcp_server, "get_auth_header", side_effect=[f"token p" for _ in logins] ), patch.object( mcp_server, "api_request", side_effect=[{"login": l} for l in logins] ), patch.object( mcp_server, "_workspace_repository_slug", return_value="Example-Org/Example-Repo", ), patch.object( mcp_server, "_canonical_repository_slug", return_value=(None, []) ): return mcp_server.gitea_activate_profile(profile_name=target) # ----------------------------------------------------------------- # AC1/AC2/AC3: switch invalidates review state; re-preflight required # ----------------------------------------------------------------- def test_switch_invalidates_review_state_and_blocks_verdict(self): with patch.dict(os.environ, self._env("reviewer-profile"), clear=True): self._seed_contaminated_review_state() res = self._activate("author-profile", ["reviewer-user", "author-user"]) self.assertTrue(res["success"]) self.assertTrue(res["re_preflight_required"]) inv = res["review_state_invalidation"] self.assertEqual(inv["from_profile"], "reviewer-profile") self.assertEqual(inv["to_profile"], "author-profile") for item in ( "preflight_identity_capability", "review_workflow_load", "review_decision_lock", "reviewer_session_lease", "live_namespace_health", ): self.assertIn(item, inv["invalidated"]) # In-memory state cleared. self.assertFalse(mcp_server._preflight_whoami_called) self.assertFalse(mcp_server._preflight_capability_called) self.assertIsNone(mcp_server._preflight_resolved_task) self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD) self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK) self.assertIsNone(reviewer_pr_lease.get_session_lease()) self.assertEqual(mcp_server._LIVE_NAMESPACE_HEALTH, {}) self.assertIsNotNone(mcp_server._PROFILE_SWITCH_INVALIDATION) # Durable records keyed by the reviewer identity are gone. self.assertIsNone( mcp_session_state.load_state( kind=mcp_session_state.KIND_WORKFLOW_LOAD, profile_identity="reviewer-profile", ) ) self.assertIsNone( mcp_session_state.load_state( kind=mcp_session_state.KIND_DECISION_LOCK, profile_identity="reviewer-profile", ) ) # A formal verdict without re-preflight fails closed. reasons = mcp_server.check_review_decision_gate( 688, "APPROVE", final_review_decision_ready=True ) self.assertTrue(reasons) def test_switch_back_cannot_resurrect_stale_review_run(self): with patch.dict(os.environ, self._env("reviewer-profile"), clear=True): self._seed_contaminated_review_state() self._activate("author-profile", ["reviewer-user", "author-user"]) res = self._activate("reviewer-profile", ["author-user", "reviewer-user"]) self.assertTrue(res["success"]) # The pre-switch review run must not reappear. self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK) self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD) self.assertIsNone(reviewer_pr_lease.get_session_lease()) status = review_workflow_load.workflow_load_status() self.assertFalse(status["workflow_load_valid"]) reasons = mcp_server.check_review_decision_gate( 688, "APPROVE", final_review_decision_ready=True ) self.assertTrue(reasons) def test_same_profile_reactivation_keeps_state(self): with patch.dict(os.environ, self._env("reviewer-profile"), clear=True): self._seed_contaminated_review_state() res = self._activate("reviewer-profile", ["reviewer-user", "reviewer-user"]) self.assertTrue(res["success"], res) self.assertNotIn("review_state_invalidation", res) self.assertIsNotNone(mcp_server._REVIEW_DECISION_LOCK) self.assertTrue(mcp_server._preflight_whoami_called) def test_clean_repreflight_after_switch_allows_gate(self): with patch.dict(os.environ, self._env("reviewer-profile"), clear=True): self._seed_contaminated_review_state() self._activate("author-profile", ["reviewer-user", "author-user"]) self._activate("reviewer-profile", ["author-user", "reviewer-user"]) # Re-establish the full reviewer preflight under the new profile. mcp_server.record_preflight_check("whoami") mcp_server.record_preflight_check( "capability", resolved_role="reviewer", resolved_task="review_pr" ) mcp_server.init_review_decision_lock("dadeschools", "review_pr") lock = mcp_server._load_review_decision_lock() self.assertIsNotNone(lock) lock.update( { "final_review_decision_ready": True, "ready_pr_number": 688, "ready_action": "APPROVE", "ready_remote": "dadeschools", "ready_org": "Example-Org", "ready_repo": "Example-Repo", } ) mcp_server._save_review_decision_lock(lock) with patch.object( mcp_server, "_review_workflow_load_gate_reasons", return_value=[] ): reasons = mcp_server.check_review_decision_gate( 688, "APPROVE", final_review_decision_ready=True, remote="dadeschools", ) self.assertEqual(reasons, []) # ----------------------------------------------------------------- # AC4: namespace provenance reporting + fail-closed mismatch # ----------------------------------------------------------------- def test_namespace_provenance_mismatch_detection(self): prov = mcp_namespace_health.namespace_provenance( task="review_pr", active_profile="reviewer-profile", env={"GITEA_MCP_NAMESPACE": "gitea-author"}, ) self.assertTrue(prov["mismatch"]) self.assertEqual(prov["required_namespace"], "gitea-reviewer") prov_ok = mcp_namespace_health.namespace_provenance( task="review_pr", active_profile="reviewer-profile", env={"GITEA_MCP_NAMESPACE": "gitea-reviewer"}, ) self.assertFalse(prov_ok["mismatch"]) prov_unknown = mcp_namespace_health.namespace_provenance( task="review_pr", active_profile="reviewer-profile", env={} ) self.assertIsNone(prov_unknown["configured_namespace"]) self.assertFalse(prov_unknown["mismatch"]) self.assertEqual(prov_unknown["namespace_source"], "unknown") @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_whoami_reports_namespace_provenance(self, _auth, _api): env = self._env("reviewer-profile") env["GITEA_MCP_NAMESPACE"] = "gitea-reviewer" with patch.dict(os.environ, env, clear=True): res = mcp_server.gitea_whoami(remote="dadeschools") prov = res["namespace_provenance"] self.assertEqual(prov["configured_namespace"], "gitea-reviewer") self.assertEqual(prov["active_profile"], "reviewer-profile") self.assertFalse(prov["mismatch"]) @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_resolve_fails_closed_on_namespace_mismatch(self, _auth, _api): env = self._env("reviewer-profile") env["GITEA_MCP_NAMESPACE"] = "gitea-author" with patch.dict(os.environ, env, clear=True): res = mcp_server.gitea_resolve_task_capability( task="review_pr", kwargs="{}", remote="dadeschools" ) self.assertFalse(res["allowed_in_current_session"]) self.assertTrue(res["namespace_provenance"]["mismatch"]) self.assertTrue( any("namespace" in g for g in res["task_role_guidance"]) ) if __name__ == "__main__": unittest.main()