Gitea-Tools/tests/test_issue_683_workflow_scope_guards.py

"""#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", "[email protected]"],
                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()