"""#683: block unattributed root WIP; pytest cannot disable production guards. Regression coverage required by issue #683: 1. Session locked to issue A blocks unrelated target issue B until B is selected. 2. Diagnostic source edit on the root checkout is blocked. 3. Same legitimate edit succeeds after issue ownership + isolated worktree bind. 4. Running under pytest does not deactivate production guards when force-on. 5. Dirty tracked Python files remain visible to porcelain consumers. 6. Monkeypatching one helper cannot silently turn the full guard path into a no-op. 7. Real mutation entrypoint proves production guards run before side effects. 8. Same-issue edits in a valid isolated worktree remain unaffected. 9. Blocker includes stable reason + exact recovery action. """ from __future__ import annotations import os import sys import tempfile import textwrap import unittest from pathlib import Path from unittest.mock import MagicMock, patch sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import gitea_mcp_server as mcp_server # noqa: E402 import issue_lock_worktree # noqa: E402 import workflow_scope_guard as wsg # noqa: E402 CONTROL_ROOT = str(Path(__file__).resolve().parent.parent) if "branches" in Path(__file__).resolve().parts: # Running from a worktree under branches/ — parent of branches is control. parts = Path(__file__).resolve().parts idx = parts.index("branches") CONTROL_ROOT = str(Path(*parts[:idx])) if idx > 0 else CONTROL_ROOT class TestProductionGuardsForceOn(unittest.TestCase): def tearDown(self): for key in ( wsg.FORCE_PRODUCTION_GUARDS_ENV, "GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN", "GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE", ): os.environ.pop(key, None) wsg.clear_workflow_failure_ledger() def test_force_on_under_pytest_keeps_production_active(self): self.assertTrue(wsg.production_guards_active(in_test_mode=False)) self.assertFalse(wsg.production_guards_active(in_test_mode=True)) os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" self.assertTrue(wsg.production_guards_active(in_test_mode=True)) self.assertTrue(wsg.production_guards_forced()) def test_no_early_return_in_verify_role_mutation_workspace_source(self): src = Path(mcp_server.__file__).read_text(encoding="utf-8") # Rejected 300a4ca pattern must not exist. self.assertNotIn( "if _preflight_in_test_mode():\n return _resolve_preflight_workspace_path", src, ) # Docstring contract for #683. self.assertIn("#683", src) self.assertIn("must NOT early-return solely because pytest", src) class TestPorcelainIntegrity(unittest.TestCase): def test_read_worktree_git_state_surfaces_dirty_py(self): with tempfile.TemporaryDirectory() as tmp: # Use a real git repo so porcelain is truthful. import subprocess subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True) subprocess.run( ["git", "config", "user.email", "t@example.com"], cwd=tmp, check=True, capture_output=True, ) subprocess.run( ["git", "config", "user.name", "t"], cwd=tmp, check=True, capture_output=True, ) py_path = Path(tmp) / "sample_mod.py" py_path.write_text("x = 1\n", encoding="utf-8") subprocess.run(["git", "add", "sample_mod.py"], cwd=tmp, check=True) subprocess.run( ["git", "commit", "-m", "init"], cwd=tmp, check=True, capture_output=True, ) py_path.write_text("x = 2\n", encoding="utf-8") state = issue_lock_worktree.read_worktree_git_state(tmp) porcelain = state.get("porcelain_status") or "" self.assertIn("sample_mod.py", porcelain) self.assertTrue(any(line.strip().endswith(".py") for line in porcelain.splitlines())) def test_production_reader_source_rejects_pytest_py_filter(self): src = Path(issue_lock_worktree.__file__).read_text(encoding="utf-8") findings = wsg.assert_no_pytest_porcelain_filter(src) self.assertEqual(findings, []) # Negative: the rejected 300a4ca pattern is detected. rejected = textwrap.dedent( """ porcelain = status_res.stdout or "" import sys if "pytest" in sys.modules or "unittest" in sys.modules: porcelain = "\\n".join( line for line in porcelain.splitlines() if not line.strip().endswith(".py") ) """ ) self.assertTrue(wsg.assert_no_pytest_porcelain_filter(rejected)) class TestIssueScopeOwnership(unittest.TestCase): def test_out_of_scope_issue_blocked_until_selected(self): result = wsg.assess_issue_scope_ownership( locked_issue_number=100, target_issue_number=200, branch_name="fix/issue-100-example", role_kind="author", ) self.assertTrue(result["block"]) self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE) self.assertIn("exact_next_action", result) self.assertIn("owning issue", result["exact_next_action"].lower()) self.assertTrue(result["reasons"]) def test_same_issue_scope_allowed(self): result = wsg.assess_issue_scope_ownership( locked_issue_number=100, target_issue_number=100, branch_name="fix/issue-100-example", role_kind="author", ) self.assertFalse(result["block"]) self.assertEqual(result["exact_next_action"], "proceed") def test_missing_lock_when_required(self): result = wsg.assess_issue_scope_ownership( locked_issue_number=None, target_issue_number=None, role_kind="author", require_lock_for_author=True, ) self.assertTrue(result["block"]) self.assertEqual(result["blocker_kind"], wsg.BLOCKER_MISSING_ISSUE_SCOPE) def test_branch_issue_mismatch(self): result = wsg.assess_issue_scope_ownership( locked_issue_number=50, branch_name="fix/issue-99-other", role_kind="author", ) self.assertTrue(result["block"]) self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE) class TestRootDiagnosticEdit(unittest.TestCase): def test_dirty_root_source_blocked(self): result = wsg.assess_root_source_mutation( workspace_path=CONTROL_ROOT, canonical_repo_root=CONTROL_ROOT, porcelain_status=" M gitea_mcp_server.py\n M tests/test_x.py\n", role_kind="author", ) self.assertTrue(result["block"]) self.assertEqual(result["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT) self.assertIn("gitea_mcp_server.py", result["dirty_source_files"]) self.assertIn("exact_next_action", result) self.assertIn("branches/", result["exact_next_action"]) def test_isolated_worktree_same_issue_unaffected(self): wt = f"{CONTROL_ROOT}/branches/issue-100-example" result = wsg.assess_root_source_mutation( workspace_path=wt, canonical_repo_root=CONTROL_ROOT, porcelain_status=" M helper.py\n", current_branch="fix/issue-100-example", locked_issue_number=100, role_kind="author", ) self.assertFalse(result["block"]) self.assertTrue(result["under_branches"]) def test_legitimate_after_ownership_and_worktree(self): wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening" composed = wsg.assess_production_mutation_guards( workspace_path=wt, canonical_repo_root=CONTROL_ROOT, porcelain_status=" M workflow_scope_guard.py\n", current_branch="fix/issue-683-workflow-guard-hardening", locked_issue_number=683, target_issue_number=683, role_kind="author", require_author_lock=True, in_test_mode=True, ) # Force-on required for production path under pytest. os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" try: composed = wsg.assess_production_mutation_guards( workspace_path=wt, canonical_repo_root=CONTROL_ROOT, porcelain_status=" M workflow_scope_guard.py\n", current_branch="fix/issue-683-workflow-guard-hardening", locked_issue_number=683, target_issue_number=683, role_kind="author", require_author_lock=True, in_test_mode=True, ) self.assertFalse(composed["block"]) self.assertFalse(composed.get("skipped")) finally: os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) class TestTypedBlockerResponse(unittest.TestCase): def test_block_response_has_stable_kind_and_next_action(self): assessment = wsg.assess_issue_scope_ownership( locked_issue_number=1, target_issue_number=2, role_kind="author", ) resp = wsg.block_response(assessment) self.assertFalse(resp["success"]) self.assertFalse(resp["performed"]) self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE) self.assertIsInstance(resp["exact_next_action"], str) self.assertTrue(resp["exact_next_action"]) self.assertTrue(resp["reasons"]) def test_production_guard_error_roundtrip(self): err = wsg.ProductionGuardError( "blocked", blocker_kind=wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT, reasons=["dirty root"], ) resp = wsg.block_response(err, issue_number=683) self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT) self.assertEqual(resp["issue_number"], 683) self.assertIn("exact_next_action", resp) class TestDurableFailureRecording(unittest.TestCase): def setUp(self): wsg.clear_workflow_failure_ledger() def tearDown(self): wsg.clear_workflow_failure_ledger() def test_record_before_source_mutation(self): pending = wsg.assess_durable_failure_recorded( require_record=True, pending_source_mutation=True ) self.assertTrue(pending["block"]) self.assertEqual(pending["blocker_kind"], wsg.BLOCKER_UNRECORDED_FAILURE) wsg.record_workflow_failure( kind="transport_eof", detail="EOF during review session (#584 cluster)", issue_number=683, task="comment_issue", ) after = wsg.assess_durable_failure_recorded( require_record=True, pending_source_mutation=True ) self.assertFalse(after["block"]) self.assertEqual(len(wsg.workflow_failure_ledger()), 1) class TestMonkeypatchCannotNoopFullPath(unittest.TestCase): def tearDown(self): os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) def test_patching_branches_only_still_blocks_dirty_root_scope(self): """Monkeypatching branches-only must not silence root diagnostic block.""" os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" with patch.object( mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None ): with patch.object( mcp_server, "_enforce_root_checkout_guard", lambda *a, **k: None ): # Even if both legacy helpers are patched, issue-scope composition # still sees dirty root source via assess_production_mutation_guards. assessment = wsg.assess_production_mutation_guards( workspace_path=CONTROL_ROOT, canonical_repo_root=CONTROL_ROOT, porcelain_status=" M gitea_mcp_server.py\n", role_kind="author", in_test_mode=True, ) self.assertTrue(assessment["block"]) self.assertEqual( assessment["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT ) class TestRealEntrypointProductionGuard(unittest.TestCase): """Real mutation entrypoint: production guard before side effects (#683).""" def setUp(self): os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"): os.environ.pop(key, None) self._orig_whoami = mcp_server._preflight_whoami_called self._orig_cap = mcp_server._preflight_capability_called mcp_server._preflight_whoami_called = False mcp_server._preflight_capability_called = False mcp_server._preflight_resolved_role = None mcp_server._preflight_resolved_task = None def tearDown(self): os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"): os.environ.pop(key, None) mcp_server._preflight_whoami_called = self._orig_whoami mcp_server._preflight_capability_called = self._orig_cap mcp_server._preflight_resolved_role = None mcp_server._preflight_resolved_task = None def test_comment_issue_blocks_dirty_root_before_api(self): api_mock = MagicMock() with patch.object(mcp_server, "api_request", api_mock), patch.object( mcp_server, "_actual_profile_role", return_value="author", ), patch.object( mcp_server, "_effective_workspace_role", return_value="author", ), patch.object( mcp_server, "get_profile", return_value={ "profile_name": "prgs-author", "allowed_operations": [ "gitea.issue.comment", "gitea.read", "gitea.pr.create", "gitea.branch.push", ], "forbidden_operations": [], }, ), patch.object( issue_lock_worktree, "read_worktree_git_state", side_effect=lambda path, **kw: { "current_branch": "master", "porcelain_status": ( " M gitea_mcp_server.py\n" if os.path.realpath(path) == os.path.realpath(CONTROL_ROOT) or path == CONTROL_ROOT else "" ), "head_sha": "a" * 40, "base_equivalent": True, }, ), patch.object( mcp_server, "_resolve_namespace_mutation_context", return_value={ "workspace_path": CONTROL_ROOT, "canonical_repo_root": CONTROL_ROOT, "process_project_root": CONTROL_ROOT, "workspace_role_kind": "author", "workspace_binding_source": "process root", "ignored_bindings": [], }, ), patch.object( mcp_server, "_resolve_author_mutation_context", return_value={ "workspace_path": CONTROL_ROOT, "canonical_repo_root": CONTROL_ROOT, "process_project_root": CONTROL_ROOT, "roots_aligned": True, }, ), patch.object( mcp_server, "_session_locked_issue_number", return_value=None, ): result = mcp_server.gitea_create_issue_comment( issue_number=683, body="diagnostic note", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", worktree_path=CONTROL_ROOT, ) api_mock.assert_not_called() self.assertFalse(result.get("success")) self.assertFalse(result.get("performed")) self.assertIn(result.get("blocker_kind"), wsg.BLOCKER_KINDS) self.assertTrue(result.get("exact_next_action")) self.assertTrue(result.get("reasons")) def test_comment_issue_succeeds_structure_after_worktree_bind(self): """Same-issue isolated worktree is not blocked by root diagnostic path.""" wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening" os.environ["GITEA_AUTHOR_WORKTREE"] = wt assessment = wsg.assess_production_mutation_guards( workspace_path=wt, canonical_repo_root=CONTROL_ROOT, porcelain_status=" M workflow_scope_guard.py\n", current_branch="fix/issue-683-workflow-guard-hardening", locked_issue_number=683, target_issue_number=683, role_kind="author", require_author_lock=True, in_test_mode=True, ) self.assertFalse(assessment["block"], assessment) class TestVerifyPreflightForceOn(unittest.TestCase): def tearDown(self): os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"): os.environ.pop(key, None) def test_force_on_runs_production_guards_under_pytest(self): os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" called = {"root": 0, "branches": 0, "scope": 0} def _root(*a, **k): called["root"] += 1 def _branches(*a, **k): called["branches"] += 1 def _scope(*a, **k): called["scope"] += 1 with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object( mcp_server, "_enforce_branches_only_author_mutation", _branches ), patch.object(mcp_server, "_enforce_issue_scope_guard", _scope): # No whoami/capability — purity-order skipped; production still runs. mcp_server.verify_preflight_purity(task="comment_issue") self.assertEqual(called["root"], 1) self.assertEqual(called["branches"], 1) self.assertEqual(called["scope"], 1) def test_without_force_on_pytest_skips_production_only_for_unit_isolation(self): called = {"root": 0} def _root(*a, **k): called["root"] += 1 with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object( mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None ), patch.object(mcp_server, "_enforce_issue_scope_guard", lambda *a, **k: None): mcp_server.verify_preflight_purity(task="comment_issue") self.assertEqual(called["root"], 0) if __name__ == "__main__": unittest.main()