Merge branch 'master' into fix/issue-745-reconciler-moot-lease-gate
This commit is contained in:
@@ -0,0 +1,367 @@
|
||||
"""Regression tests for create_issue bootstrap (#749).
|
||||
|
||||
TDD: these tests define the sanctioned first-mutation path for
|
||||
``gitea_create_issue`` from a clean canonical control checkout, and prove
|
||||
the exemption cannot widen to dirty roots, foreign clones, arbitrary
|
||||
``branches/`` directories, or post-creation author mutations.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import create_issue_bootstrap as cib # noqa: E402
|
||||
import gitea_mcp_server as srv # noqa: E402
|
||||
import workflow_scope_guard as wsg # noqa: E402
|
||||
|
||||
FAKE_AUTH = {"Authorization": "token test-token"}
|
||||
MASTER_SHA = "a" * 40
|
||||
STALE_SHA = "b" * 40
|
||||
|
||||
current_file_path = Path(__file__).resolve()
|
||||
if "branches" in current_file_path.parts:
|
||||
CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3])
|
||||
else:
|
||||
CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1])
|
||||
|
||||
|
||||
class TestCreateIssueBootstrapAssessor(unittest.TestCase):
|
||||
ROOT = "/repo/Gitea-Tools"
|
||||
|
||||
def test_non_create_issue_task_not_applicable(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=self.ROOT,
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="master",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="lock_issue",
|
||||
)
|
||||
self.assertTrue(res["not_applicable"])
|
||||
self.assertFalse(res["allowed"])
|
||||
self.assertFalse(res["block"])
|
||||
|
||||
def test_branches_worktree_not_applicable(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=f"{self.ROOT}/branches/issue-1-x",
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="fix/issue-1-x",
|
||||
task="create_issue",
|
||||
)
|
||||
self.assertTrue(res["not_applicable"])
|
||||
self.assertFalse(res["allowed"])
|
||||
|
||||
def test_clean_control_checkout_allowed(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=self.ROOT,
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="master",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="create_issue",
|
||||
)
|
||||
self.assertFalse(res["not_applicable"])
|
||||
self.assertTrue(res["allowed"])
|
||||
self.assertFalse(res["block"])
|
||||
self.assertEqual(res["bootstrap_path"], "clean_canonical_control_checkout")
|
||||
# Post-create next action must name issue-backed worktree after N exists.
|
||||
self.assertIn("branches/issue-<N>-*", res["exact_next_action"])
|
||||
|
||||
def test_tool_alias_gitea_create_issue_allowed(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=self.ROOT,
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="main",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="gitea_create_issue",
|
||||
)
|
||||
self.assertTrue(res["allowed"])
|
||||
|
||||
def test_dirty_control_checkout_blocked(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=self.ROOT,
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="master",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status=" M gitea_mcp_server.py\n",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="create_issue",
|
||||
)
|
||||
self.assertTrue(res["block"])
|
||||
self.assertFalse(res["allowed"])
|
||||
self.assertTrue(any("tracked local edits" in r for r in res["reasons"]))
|
||||
# Pre-issue phase: next action must be satisfiable without inventing <N>.
|
||||
next_a = res["exact_next_action"] or ""
|
||||
self.assertIn("clean accepted base branch", next_a)
|
||||
self.assertIn("before the issue exists", next_a)
|
||||
# Must not prescribe "bind branches/issue-<N>" as the recovery step.
|
||||
self.assertNotIn("Bind an issue-backed worktree", next_a)
|
||||
|
||||
def test_stale_base_blocked(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=self.ROOT,
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="master",
|
||||
head_sha=STALE_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="create_issue",
|
||||
)
|
||||
self.assertTrue(res["block"])
|
||||
self.assertTrue(any("live master" in r for r in res["reasons"]))
|
||||
|
||||
def test_non_base_branch_blocked(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=self.ROOT,
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="feat/something",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="create_issue",
|
||||
)
|
||||
self.assertTrue(res["block"])
|
||||
|
||||
def test_detached_head_blocked(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=self.ROOT,
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="create_issue",
|
||||
)
|
||||
self.assertTrue(res["block"])
|
||||
self.assertTrue(any("detached" in r for r in res["reasons"]))
|
||||
|
||||
def test_foreign_workspace_blocked(self):
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path="/other/clone",
|
||||
canonical_repo_root=self.ROOT,
|
||||
current_branch="master",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="create_issue",
|
||||
)
|
||||
self.assertTrue(res["block"])
|
||||
self.assertTrue(any("canonical control checkout" in r for r in res["reasons"]))
|
||||
|
||||
|
||||
class TestCreateIssueBootstrapIntegration(unittest.TestCase):
|
||||
def setUp(self):
|
||||
srv._preflight_whoami_called = True
|
||||
srv._preflight_capability_called = True
|
||||
srv._preflight_resolved_role = "author"
|
||||
srv._preflight_resolved_task = "create_issue"
|
||||
srv._preflight_whoami_violation = False
|
||||
srv._preflight_capability_violation = False
|
||||
self._orig_in_test = srv._preflight_in_test_mode
|
||||
srv._preflight_in_test_mode = lambda: False
|
||||
|
||||
def tearDown(self):
|
||||
srv._preflight_in_test_mode = self._orig_in_test
|
||||
srv._preflight_resolved_task = None
|
||||
|
||||
def _git_state(self, branch="master", head=MASTER_SHA, porcelain=""):
|
||||
return {
|
||||
"current_branch": branch,
|
||||
"head_sha": head,
|
||||
"porcelain_status": porcelain,
|
||||
}
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
||||
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
||||
@patch(
|
||||
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []),
|
||||
)
|
||||
@patch("gitea_mcp_server.api_request")
|
||||
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
||||
@patch(
|
||||
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
||||
return_value=MASTER_SHA,
|
||||
)
|
||||
def test_clean_control_checkout_create_issue_succeeds(
|
||||
self, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth
|
||||
):
|
||||
mock_api.return_value = {
|
||||
"number": 99,
|
||||
"html_url": "https://gitea.example.com/issues/99",
|
||||
}
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
with patch(
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value=self._git_state(),
|
||||
):
|
||||
with patch(
|
||||
"gitea_mcp_server._get_workspace_porcelain", return_value=""
|
||||
):
|
||||
with patch(
|
||||
"gitea_mcp_server._enforce_root_checkout_guard"
|
||||
):
|
||||
# Anti-stomp / master parity: keep gates green.
|
||||
with patch.object(
|
||||
srv,
|
||||
"_run_anti_stomp_preflight",
|
||||
return_value=None,
|
||||
):
|
||||
res = srv.gitea_create_issue(
|
||||
title="Bootstrap issue from clean control",
|
||||
body="Body text for content gate.",
|
||||
)
|
||||
self.assertEqual(res.get("number"), 99)
|
||||
mock_api.assert_called_once()
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
||||
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
||||
@patch(
|
||||
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []),
|
||||
)
|
||||
@patch("gitea_mcp_server.api_request")
|
||||
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
||||
def test_dirty_control_checkout_create_issue_fails_closed(
|
||||
self, _get_all, mock_api, _role, _ns, _prof, _auth
|
||||
):
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
with patch(
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value=self._git_state(
|
||||
porcelain=" M author_mutation_worktree.py\n"
|
||||
),
|
||||
):
|
||||
with patch(
|
||||
"gitea_mcp_server._get_workspace_porcelain",
|
||||
return_value=" M author_mutation_worktree.py\n",
|
||||
):
|
||||
res = srv.gitea_create_issue(
|
||||
title="Should fail on dirty root",
|
||||
body="Body text for content gate.",
|
||||
)
|
||||
self.assertFalse(res.get("success", True) and res.get("number"))
|
||||
if isinstance(res, dict) and res.get("success") is False:
|
||||
blob = " ".join(res.get("reasons") or [])
|
||||
self.assertTrue(
|
||||
"tracked local edits" in blob
|
||||
or "dirty" in blob.lower()
|
||||
or "control checkout" in blob.lower()
|
||||
or res.get("blocker_kind")
|
||||
)
|
||||
# Pre-issue phase must not demand issue-<N> worktree.
|
||||
next_a = res.get("exact_next_action") or ""
|
||||
if next_a:
|
||||
self.assertNotIn("issue-<N>-*", next_a)
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
||||
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
||||
@patch(
|
||||
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []),
|
||||
)
|
||||
def test_lock_issue_still_requires_branches_worktree(self, _role, _ns, _prof, _auth):
|
||||
"""Existing issue-backed mutations receive no exemption (#749 AC3/AC7)."""
|
||||
srv._preflight_resolved_task = "lock_issue"
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
with patch(
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value=self._git_state(),
|
||||
):
|
||||
with patch(
|
||||
"gitea_mcp_server._get_workspace_porcelain", return_value=""
|
||||
):
|
||||
with patch(
|
||||
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
||||
return_value=MASTER_SHA,
|
||||
):
|
||||
with patch.object(
|
||||
srv, "_enforce_root_checkout_guard"
|
||||
):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.verify_preflight_purity(
|
||||
remote="prgs",
|
||||
task="lock_issue",
|
||||
)
|
||||
msg = str(ctx.exception)
|
||||
self.assertTrue(
|
||||
"Branches-only mutation guard" in msg
|
||||
or "stable control checkout" in msg,
|
||||
msg,
|
||||
)
|
||||
self.assertIn("control checkout", msg)
|
||||
|
||||
def test_workflow_scope_skips_missing_worktree_for_create_issue_clean_root(self):
|
||||
"""#683 root assessor must not block clean-root create_issue bootstrap."""
|
||||
res = wsg.assess_root_source_mutation(
|
||||
workspace_path=CONTROL_CHECKOUT_ROOT,
|
||||
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
|
||||
porcelain_status="",
|
||||
role_kind="author",
|
||||
mutation_task="create_issue",
|
||||
)
|
||||
self.assertFalse(res["block"], res)
|
||||
self.assertTrue(res.get("create_issue_bootstrap") or res["proven"])
|
||||
|
||||
def test_workflow_scope_still_blocks_clean_root_for_lock_issue(self):
|
||||
res = wsg.assess_root_source_mutation(
|
||||
workspace_path=CONTROL_CHECKOUT_ROOT,
|
||||
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
|
||||
porcelain_status="",
|
||||
role_kind="author",
|
||||
mutation_task="lock_issue",
|
||||
)
|
||||
self.assertTrue(res["block"])
|
||||
self.assertEqual(res["blocker_kind"], wsg.BLOCKER_MISSING_WORKTREE)
|
||||
|
||||
def test_arbitrary_branches_directory_not_bootstrap(self):
|
||||
"""#713: mkdir fake under branches/ is not the bootstrap path."""
|
||||
fake = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "fake-mkdir-only")
|
||||
res = cib.assess_create_issue_bootstrap(
|
||||
workspace_path=fake,
|
||||
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
|
||||
current_branch="master",
|
||||
head_sha=MASTER_SHA,
|
||||
porcelain_status="",
|
||||
remote_master_sha=MASTER_SHA,
|
||||
task="create_issue",
|
||||
)
|
||||
# Under branches/ → not bootstrap; ordinary membership/registration applies.
|
||||
self.assertTrue(res["not_applicable"])
|
||||
self.assertFalse(res["allowed"])
|
||||
|
||||
|
||||
class TestCreateIssueCapabilityAgreement(unittest.TestCase):
|
||||
def test_map_and_alias_agree_on_create_issue(self):
|
||||
import task_capability_map as tcm
|
||||
|
||||
self.assertEqual(
|
||||
tcm.required_permission("create_issue"),
|
||||
"gitea.issue.create",
|
||||
)
|
||||
# Tool alias must resolve to the same task contract.
|
||||
alias = getattr(tcm, "TOOL_TASK_ALIASES", None) or getattr(
|
||||
tcm, "TASK_ALIASES", None
|
||||
)
|
||||
if alias is not None:
|
||||
mapped = alias.get("gitea_create_issue")
|
||||
if mapped is not None:
|
||||
self.assertEqual(mapped, "create_issue")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -51,29 +51,62 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
|
||||
"porcelain_status": "",
|
||||
},
|
||||
)
|
||||
def test_create_issue_stable_checkout_rejected(
|
||||
def test_create_issue_stable_checkout_bootstrap_allowed_when_clean(
|
||||
self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
|
||||
):
|
||||
# Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
|
||||
# path is the stable control checkout (not under branches/), mutation must fail.
|
||||
# #749: clean canonical control checkout is the sanctioned create_issue path.
|
||||
mock_api.return_value = {
|
||||
"number": 77,
|
||||
"html_url": "https://gitea.example.com/issues/77",
|
||||
}
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
try:
|
||||
res = srv.gitea_create_issue(title="Test issue", body="body text")
|
||||
except RuntimeError as exc:
|
||||
self.assertIn("stable control checkout", str(exc))
|
||||
else:
|
||||
# #683: production guards return typed blockers at entrypoints
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertFalse(res.get("performed"))
|
||||
blob = " ".join(res.get("reasons") or []) + " " + str(
|
||||
res.get("blocker_kind") or ""
|
||||
)
|
||||
self.assertTrue(
|
||||
"stable control checkout" in blob
|
||||
or "missing_issue_worktree" in blob
|
||||
or "control checkout" in blob.lower()
|
||||
)
|
||||
self.assertTrue(res.get("exact_next_action"))
|
||||
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
|
||||
with patch.object(srv, "_run_anti_stomp_preflight", return_value=None):
|
||||
with patch.object(srv, "_enforce_root_checkout_guard"):
|
||||
res = srv.gitea_create_issue(
|
||||
title="Test issue", body="body text for gate"
|
||||
)
|
||||
self.assertEqual(res.get("number"), 77)
|
||||
mock_api.assert_called_once()
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
||||
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
||||
@patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
|
||||
@patch("gitea_mcp_server.api_request")
|
||||
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
||||
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
|
||||
@patch(
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={
|
||||
"current_branch": "master",
|
||||
"head_sha": "a" * 40,
|
||||
"porcelain_status": " M dirty.py\n",
|
||||
},
|
||||
)
|
||||
def test_create_issue_dirty_control_checkout_rejected(
|
||||
self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
|
||||
):
|
||||
# #749: dirty control checkout still fails closed (no bootstrap).
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
with patch(
|
||||
"gitea_mcp_server._get_workspace_porcelain",
|
||||
return_value=" M dirty.py\n",
|
||||
):
|
||||
try:
|
||||
res = srv.gitea_create_issue(title="Test issue", body="body text")
|
||||
except RuntimeError as exc:
|
||||
self.assertTrue(
|
||||
"tracked local edits" in str(exc)
|
||||
or "dirty" in str(exc).lower()
|
||||
or "bootstrap" in str(exc).lower()
|
||||
or "control checkout" in str(exc).lower()
|
||||
)
|
||||
else:
|
||||
self.assertFalse(res.get("success", True) and res.get("number"))
|
||||
blob = " ".join(res.get("reasons") or [])
|
||||
self.assertTrue(blob or res.get("blocker_kind"))
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
||||
|
||||
@@ -0,0 +1,602 @@
|
||||
"""Regression coverage for the PR checks assessor defect (#751).
|
||||
|
||||
Gitea's *combined* commit status reports ``state: pending`` both when a check is
|
||||
executing and when the status-context collection is empty. The assessor read
|
||||
``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head
|
||||
had no status contexts — and never would — was routed to ``blocked`` forever.
|
||||
|
||||
These tests pin the corrected semantics end to end: live branch protection
|
||||
decides whether checks are required, and the actual context collection decides
|
||||
what the checks say.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import pr_sync_status # noqa: E402
|
||||
from pr_sync_status import ( # noqa: E402
|
||||
ACTION_BLOCKED,
|
||||
ACTION_MERGE_NOW,
|
||||
ACTION_UPDATE_BRANCH_BY_MERGE,
|
||||
CHECKS_FAILURE,
|
||||
CHECKS_MISSING_REQUIRED,
|
||||
CHECKS_NONE,
|
||||
CHECKS_NOT_REQUIRED,
|
||||
CHECKS_PENDING,
|
||||
CHECKS_SUCCESS,
|
||||
CHECKS_UNKNOWN,
|
||||
assess_pr_sync_status,
|
||||
classify_commit_checks,
|
||||
)
|
||||
|
||||
|
||||
def _sha(prefix: str) -> str:
|
||||
return (prefix + "0" * 40)[:40]
|
||||
|
||||
|
||||
PR_HEAD = _sha("aaaaaaaa")
|
||||
BASE_HEAD = _sha("bbbbbbbb")
|
||||
|
||||
|
||||
def _base_kwargs(**overrides):
|
||||
data = {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Gitea-Tools",
|
||||
"pr_number": 751,
|
||||
"pr_state": "open",
|
||||
"source_branch": "fix/issue-751-checks-assessor",
|
||||
"pr_head_sha": PR_HEAD,
|
||||
"base_head_sha": BASE_HEAD,
|
||||
"commits_behind": 0,
|
||||
"mergeable": True,
|
||||
"has_conflicts": False,
|
||||
"branch_protection_requires_current_base": False,
|
||||
"approval_at_current_head": True,
|
||||
"checks_status": "success",
|
||||
"active_author_lock": True,
|
||||
"active_reviewer_lease": False,
|
||||
"active_merger_lease": False,
|
||||
}
|
||||
data.update(overrides)
|
||||
return data
|
||||
|
||||
|
||||
def _reasons(result) -> str:
|
||||
return " | ".join(result["reasons"]).lower()
|
||||
|
||||
|
||||
class TestClassifyCommitChecks(unittest.TestCase):
|
||||
"""Pure classification from live evidence."""
|
||||
|
||||
def test_empty_collection_with_combined_pending_is_not_executing_ci(self):
|
||||
# The exact PR #750 failure mode at the classification layer.
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[],
|
||||
checks_enabled=True,
|
||||
required_contexts=[],
|
||||
)
|
||||
self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
|
||||
self.assertEqual(result["checks_status"], CHECKS_NONE)
|
||||
self.assertEqual(result["context_count"], 0)
|
||||
joined = " ".join(result["reasons"]).lower()
|
||||
self.assertIn("empty", joined)
|
||||
self.assertIn("does not indicate", joined)
|
||||
|
||||
def test_protection_disables_status_checks(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[],
|
||||
checks_enabled=False,
|
||||
required_contexts=[],
|
||||
)
|
||||
self.assertFalse(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED)
|
||||
|
||||
def test_no_required_contexts_aggregates_reported_contexts(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "success"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=[],
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
|
||||
def test_required_checks_pending(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[{"context": "build", "status": "pending"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_PENDING)
|
||||
|
||||
def test_required_checks_failed(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="failure",
|
||||
statuses=[{"context": "build", "status": "failure"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_FAILURE)
|
||||
|
||||
def test_required_checks_successful(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "success"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
self.assertTrue(result["checks_required"])
|
||||
|
||||
def test_required_context_configured_with_no_matching_result(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "lint", "status": "success"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
|
||||
self.assertEqual(result["missing_required_contexts"], ["build"])
|
||||
|
||||
def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self):
|
||||
# An unrelated failing context must not fail a satisfied required set.
|
||||
result = classify_commit_checks(
|
||||
combined_state="failure",
|
||||
statuses=[
|
||||
{"context": "build", "status": "success"},
|
||||
{"context": "optional-scan", "status": "failure"},
|
||||
],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
# ...and a failing *required* context still fails despite passing extras.
|
||||
failing = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[
|
||||
{"context": "build", "status": "failure"},
|
||||
{"context": "optional-scan", "status": "success"},
|
||||
],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(failing["checks_status"], CHECKS_FAILURE)
|
||||
|
||||
def test_policy_unreadable_fails_closed(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state=None,
|
||||
statuses=[],
|
||||
checks_enabled=None,
|
||||
required_contexts=[],
|
||||
policy_determinable=False,
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
self.assertIn("fail closed", " ".join(result["reasons"]).lower())
|
||||
|
||||
def test_malformed_policy_indeterminate_flag_fails_closed(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "success"}],
|
||||
checks_enabled=None,
|
||||
required_contexts=[],
|
||||
policy_determinable=True,
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
|
||||
def test_status_collection_unreadable_fails_closed_when_required(self):
|
||||
result = classify_commit_checks(
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
status_determinable=False,
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
|
||||
def test_unrecognized_context_state_never_reads_as_success(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "banana"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
|
||||
def test_newest_wins_per_context(self):
|
||||
# Gitea returns newest-first; the stale failure must not win.
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[
|
||||
{"context": "build", "status": "success"},
|
||||
{"context": "build", "status": "failure"},
|
||||
],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
|
||||
def test_malformed_status_rows_are_discarded_not_treated_as_passing(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[{"context": "build"}, "not-a-dict", None],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
|
||||
|
||||
|
||||
class TestChecksGateSemantics(unittest.TestCase):
|
||||
"""``assess_pr_sync_status`` routing and blocker reasons."""
|
||||
|
||||
def test_not_required_allows_merge_now_with_empty_checks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_NOT_REQUIRED),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
self.assertTrue(result["approval_valid_for_merge"])
|
||||
self.assertFalse(result["checks_required"])
|
||||
self.assertIn("does not require status checks", _reasons(result))
|
||||
|
||||
def test_none_blocks_when_checks_required(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_NONE),
|
||||
checks_required=True,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("no status context", _reasons(result))
|
||||
self.assertIn("not executing ci", _reasons(result))
|
||||
|
||||
def test_missing_required_blocks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED),
|
||||
checks_required=True,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("no matching status result", _reasons(result))
|
||||
|
||||
def test_required_pending_blocks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("not finished", _reasons(result))
|
||||
|
||||
def test_required_failure_blocks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("failed", _reasons(result))
|
||||
|
||||
def test_required_success_merges(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
|
||||
def test_unknown_blocks_when_required(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("unknown", _reasons(result))
|
||||
|
||||
def test_unrecognized_status_does_not_fall_through_to_merge(self):
|
||||
# Regression: the previous gate only handled a fixed vocabulary and let
|
||||
# anything else reach merge_now.
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status="totally-unexpected"),
|
||||
checks_required=True,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("unrecognized checks status", _reasons(result))
|
||||
|
||||
def test_checks_gate_does_not_bypass_approval_requirement(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(
|
||||
checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False
|
||||
),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
self.assertFalse(result["approval_valid_for_merge"])
|
||||
|
||||
def test_checks_gate_does_not_bypass_current_base_protection(self):
|
||||
# Existing current-base behavior stays intact (#727).
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(
|
||||
checks_status=CHECKS_NOT_REQUIRED,
|
||||
commits_behind=3,
|
||||
branch_protection_requires_current_base=True,
|
||||
),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertEqual(
|
||||
result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
|
||||
)
|
||||
|
||||
def test_checks_gate_does_not_bypass_conflicts(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
|
||||
|
||||
class TestBranchProtectionPolicy(unittest.TestCase):
|
||||
"""Live protection reader derives the status-check requirement."""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms
|
||||
|
||||
self.gms = gms
|
||||
|
||||
def _policy(self, responses):
|
||||
def fake_api_request(method, url, auth, *args, **kwargs):
|
||||
for fragment, payload in responses.items():
|
||||
if fragment in url:
|
||||
if isinstance(payload, Exception):
|
||||
raise payload
|
||||
return payload
|
||||
return None
|
||||
|
||||
with patch.object(self.gms, "api_request", side_effect=fake_api_request):
|
||||
return self.gms._branch_protection_policy(
|
||||
"https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master"
|
||||
)
|
||||
|
||||
def test_status_checks_enabled_with_contexts(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": [{
|
||||
"branch_name": "master",
|
||||
"block_on_outdated_branch": True,
|
||||
"enable_status_check": True,
|
||||
"status_check_contexts": ["ci/build", " "],
|
||||
}],
|
||||
})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertTrue(policy["checks_enabled"])
|
||||
self.assertTrue(policy["requires_current_base"])
|
||||
self.assertEqual(policy["required_contexts"], ["ci/build"])
|
||||
|
||||
def test_status_checks_disabled(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": [{
|
||||
"branch_name": "master",
|
||||
"enable_status_check": False,
|
||||
}],
|
||||
})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertFalse(policy["checks_enabled"])
|
||||
|
||||
def test_no_protection_rule_means_checks_not_required(self):
|
||||
# PR #750's repository shape: protection list readable but empty.
|
||||
policy = self._policy({"branch_protections": [], "branches/master": {}})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertFalse(policy["protection_found"])
|
||||
self.assertFalse(policy["checks_enabled"])
|
||||
self.assertIsNone(policy["requires_current_base"])
|
||||
|
||||
def test_protection_without_status_field_means_not_enabled(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": [{
|
||||
"branch_name": "master",
|
||||
"block_on_outdated_branch": True,
|
||||
}],
|
||||
})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertFalse(policy["checks_enabled"])
|
||||
self.assertTrue(policy["requires_current_base"])
|
||||
|
||||
def test_api_failure_is_not_determinable(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": RuntimeError("boom"),
|
||||
})
|
||||
self.assertFalse(policy["determinable"])
|
||||
self.assertIsNone(policy["checks_enabled"])
|
||||
|
||||
def test_missing_branch_is_not_determinable(self):
|
||||
with patch.object(self.gms, "api_request", return_value=[]):
|
||||
policy = self.gms._branch_protection_policy(
|
||||
"https://example/api/v1/repos/o/r", {}, base_branch=" "
|
||||
)
|
||||
self.assertFalse(policy["determinable"])
|
||||
|
||||
def test_requires_current_base_accessor_preserved(self):
|
||||
def fake(method, url, auth, *a, **k):
|
||||
if "branch_protections" in url:
|
||||
return [{"branch_name": "master",
|
||||
"block_on_outdated_branch": True}]
|
||||
return None
|
||||
|
||||
with patch.object(self.gms, "api_request", side_effect=fake):
|
||||
value = self.gms._branch_protection_requires_current_base(
|
||||
"https://example/api/v1/repos/o/r", {}, base_branch="master"
|
||||
)
|
||||
self.assertTrue(value)
|
||||
|
||||
|
||||
class TestCommitChecksSnapshot(unittest.TestCase):
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms
|
||||
|
||||
self.gms = gms
|
||||
|
||||
def test_reads_state_and_context_collection(self):
|
||||
payload = {"state": "pending", "statuses": [{"context": "b",
|
||||
"status": "pending"}]}
|
||||
with patch.object(self.gms, "api_request", return_value=payload):
|
||||
snap = self.gms._commit_checks_snapshot(
|
||||
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
||||
)
|
||||
self.assertTrue(snap["determinable"])
|
||||
self.assertEqual(snap["combined_state"], "pending")
|
||||
self.assertEqual(len(snap["statuses"]), 1)
|
||||
|
||||
def test_empty_collection_recorded_as_determinable(self):
|
||||
with patch.object(
|
||||
self.gms, "api_request", return_value={"state": "pending", "statuses": []}
|
||||
):
|
||||
snap = self.gms._commit_checks_snapshot(
|
||||
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
||||
)
|
||||
self.assertTrue(snap["determinable"])
|
||||
self.assertEqual(snap["statuses"], [])
|
||||
|
||||
def test_api_failure_is_not_determinable(self):
|
||||
with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")):
|
||||
snap = self.gms._commit_checks_snapshot(
|
||||
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
||||
)
|
||||
self.assertFalse(snap["determinable"])
|
||||
|
||||
|
||||
class TestMcpWrapperForwarding(unittest.TestCase):
|
||||
"""The production MCP path must reach the derived ``checks_required``."""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms
|
||||
|
||||
self.gms = gms
|
||||
self.base = (
|
||||
"https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools"
|
||||
)
|
||||
|
||||
def _patches(self, fake_api_request, spy):
|
||||
return [
|
||||
patch.object(self.gms, "_profile_operation_gate", return_value=None),
|
||||
patch.object(self.gms, "_resolve", return_value=(
|
||||
"gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")),
|
||||
patch.object(self.gms, "_auth", return_value={"Authorization": "x"}),
|
||||
patch.object(self.gms, "repo_api_url", return_value=self.base),
|
||||
patch.object(self.gms, "api_request", side_effect=fake_api_request),
|
||||
patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={
|
||||
"success": True, "approval_at_current_head": True}),
|
||||
patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={
|
||||
"has_author_lock": True}),
|
||||
patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy),
|
||||
]
|
||||
|
||||
def _run(self, *, protections, status_payload, pr_number=750,
|
||||
caller_checks_status=None):
|
||||
captured = {}
|
||||
real_assess = pr_sync_status.assess_pr_sync_status
|
||||
|
||||
def spy(**kwargs):
|
||||
captured.update(kwargs)
|
||||
return real_assess(**kwargs)
|
||||
|
||||
def fake_api_request(method, url, auth, *args, **kwargs):
|
||||
if f"/pulls/{pr_number}" in url:
|
||||
return {
|
||||
"number": pr_number,
|
||||
"state": "open",
|
||||
"title": "t",
|
||||
"body": "b",
|
||||
"mergeable": True,
|
||||
"head": {"sha": PR_HEAD, "ref": "fix/x"},
|
||||
"base": {"sha": BASE_HEAD, "ref": "master"},
|
||||
}
|
||||
if "/branch_protections" in url:
|
||||
if isinstance(protections, Exception):
|
||||
raise protections
|
||||
return protections
|
||||
if "/branches/master" in url:
|
||||
return {"commit": {"id": BASE_HEAD}}
|
||||
if "/status" in url:
|
||||
if isinstance(status_payload, Exception):
|
||||
raise status_payload
|
||||
return status_payload
|
||||
if "/compare/" in url:
|
||||
return {"total_commits": 0}
|
||||
if "/comments" in url:
|
||||
return []
|
||||
return None
|
||||
|
||||
stack = self._patches(fake_api_request, spy)
|
||||
for p in stack:
|
||||
p.start()
|
||||
try:
|
||||
result = self.gms.gitea_assess_pr_sync_status(
|
||||
pr_number=pr_number, remote="prgs",
|
||||
checks_status=caller_checks_status,
|
||||
)
|
||||
finally:
|
||||
for p in reversed(stack):
|
||||
p.stop()
|
||||
return result, captured
|
||||
|
||||
def test_derived_checks_required_is_forwarded(self):
|
||||
result, captured = self._run(
|
||||
protections=[{"branch_name": "master", "enable_status_check": True,
|
||||
"status_check_contexts": ["ci/build"]}],
|
||||
status_payload={"state": "pending", "statuses": [
|
||||
{"context": "ci/build", "status": "pending"}]},
|
||||
)
|
||||
self.assertIn("checks_required", captured)
|
||||
self.assertTrue(captured["checks_required"])
|
||||
self.assertEqual(captured["checks_status"], CHECKS_PENDING)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
|
||||
def test_pr750_empty_status_with_no_protection_is_merge_ready(self):
|
||||
# The exact reported failure: combined pending, zero contexts, and no
|
||||
# branch protection requiring checks.
|
||||
result, captured = self._run(
|
||||
protections=[],
|
||||
status_payload={"state": "pending", "statuses": []},
|
||||
)
|
||||
self.assertFalse(captured["checks_required"])
|
||||
self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED)
|
||||
self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
|
||||
def test_protection_read_failure_blocks(self):
|
||||
result, captured = self._run(
|
||||
protections=RuntimeError("protection unavailable"),
|
||||
status_payload={"state": "success", "statuses": []},
|
||||
)
|
||||
self.assertTrue(captured["checks_required"])
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
|
||||
def test_caller_supplied_status_cannot_mask_live_required_failure(self):
|
||||
# A caller-declared "success" must not override live evidence.
|
||||
result, captured = self._run(
|
||||
protections=[{"branch_name": "master", "enable_status_check": True,
|
||||
"status_check_contexts": ["ci/build"]}],
|
||||
status_payload={"state": "failure", "statuses": [
|
||||
{"context": "ci/build", "status": "failure"}]},
|
||||
caller_checks_status="success",
|
||||
)
|
||||
self.assertEqual(captured["checks_status"], CHECKS_FAILURE)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertEqual(
|
||||
result["checks_evidence"]["caller_supplied_checks_status"], "success"
|
||||
)
|
||||
|
||||
def test_checks_evidence_is_reported_without_secrets(self):
|
||||
result, _ = self._run(
|
||||
protections=[],
|
||||
status_payload={"state": "pending", "statuses": []},
|
||||
)
|
||||
evidence = result["checks_evidence"]
|
||||
self.assertEqual(evidence["context_count"], 0)
|
||||
self.assertEqual(evidence["combined_state"], "pending")
|
||||
self.assertFalse(evidence["protection_found"])
|
||||
blob = repr(result).lower()
|
||||
self.assertNotIn("authorization", blob)
|
||||
self.assertNotIn("token", blob)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,364 @@
|
||||
"""Dead-session author issue-lock recovery (#753).
|
||||
|
||||
Covers the narrow recovery path that lets an author re-acquire a durable lock
|
||||
after the MCP session that recorded it exits, plus every rejection condition
|
||||
that must keep failing closed.
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import issue_lock_recovery # noqa: E402
|
||||
import issue_lock_store # noqa: E402
|
||||
import issue_lock_worktree # noqa: E402
|
||||
|
||||
ISSUE = 4242
|
||||
BRANCH = f"fix/issue-{ISSUE}-demo"
|
||||
WORKTREE = "/scratch/wt"
|
||||
HEAD = "a" * 40
|
||||
OTHER_SHA = "b" * 40
|
||||
IDENTITY = "example-user"
|
||||
PROFILE = "example-author"
|
||||
|
||||
|
||||
def dead_pid() -> int:
|
||||
"""A PID that has certainly exited (spawned, then reaped)."""
|
||||
proc = subprocess.Popen([sys.executable, "-c", "pass"])
|
||||
proc.wait()
|
||||
return proc.pid
|
||||
|
||||
|
||||
def future_ts(hours: int = 4) -> str:
|
||||
return (
|
||||
(datetime.now(timezone.utc) + timedelta(hours=hours))
|
||||
.isoformat()
|
||||
.replace("+00:00", "Z")
|
||||
)
|
||||
|
||||
|
||||
def make_lock(**overrides):
|
||||
lock = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": WORKTREE,
|
||||
"remote": "prgs",
|
||||
"org": "ExampleOrg",
|
||||
"repo": "ExampleRepo",
|
||||
"session_pid": dead_pid(),
|
||||
"work_lease": {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": BRANCH,
|
||||
"worktree_path": WORKTREE,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"expires_at": future_ts(),
|
||||
},
|
||||
}
|
||||
lock.update(overrides)
|
||||
return lock
|
||||
|
||||
|
||||
def assess(lock=None, **overrides):
|
||||
kwargs = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": WORKTREE,
|
||||
"remote": "prgs",
|
||||
"org": "ExampleOrg",
|
||||
"repo": "ExampleRepo",
|
||||
"identity": IDENTITY,
|
||||
"profile": PROFILE,
|
||||
"current_branch": BRANCH,
|
||||
"porcelain_status": "",
|
||||
"head_sha": HEAD,
|
||||
"remote_head_sha": HEAD,
|
||||
"pr_head_sha": HEAD,
|
||||
"pr_number": 99,
|
||||
"competing_live_locks": [],
|
||||
"candidate_branches": [BRANCH],
|
||||
"current_pid": os.getpid(),
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return issue_lock_recovery.assess_dead_session_lock_recovery(
|
||||
make_lock() if lock is None else lock, **kwargs
|
||||
)
|
||||
|
||||
|
||||
class TestDeadSessionRecoveryGranted(unittest.TestCase):
|
||||
def test_dead_pid_with_exact_evidence_recovers(self):
|
||||
result = assess()
|
||||
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
|
||||
self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED)
|
||||
|
||||
def test_recovery_still_granted_when_no_open_pr_exists(self):
|
||||
# A locked branch need not have a PR yet; absence must not block.
|
||||
result = assess(pr_head_sha=None, pr_number=None)
|
||||
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
|
||||
|
||||
def test_lease_expiry_is_not_required_for_recovery(self):
|
||||
# The defining condition is PID death, not TTL expiry (the #601 gap).
|
||||
lock = make_lock()
|
||||
self.assertFalse(issue_lock_store.is_lease_expired(lock))
|
||||
self.assertFalse(issue_lock_store.assess_lock_freshness(lock)["live"])
|
||||
self.assertTrue(assess(lock)["recovery_sanctioned"])
|
||||
|
||||
|
||||
class TestDeadSessionRecoveryRefused(unittest.TestCase):
|
||||
def assert_refused(self, result, needle):
|
||||
self.assertFalse(result["recovery_sanctioned"])
|
||||
self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED)
|
||||
self.assertTrue(
|
||||
any(needle in reason for reason in result["reasons"]),
|
||||
f"expected {needle!r} in {result['reasons']}",
|
||||
)
|
||||
|
||||
def test_live_prior_pid_refused(self):
|
||||
lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
|
||||
# Distinct current pid so the refusal is attributable to liveness.
|
||||
self.assert_refused(assess(lock, current_pid=os.getpid() + 1), "still alive")
|
||||
|
||||
def test_different_author_identity_refused(self):
|
||||
self.assert_refused(
|
||||
assess(identity="someone-else"), "does not match active identity"
|
||||
)
|
||||
|
||||
def test_different_profile_refused(self):
|
||||
self.assert_refused(
|
||||
assess(profile="other-profile"), "does not match active profile"
|
||||
)
|
||||
|
||||
def test_different_branch_refused(self):
|
||||
lock = make_lock(branch_name=f"fix/issue-{ISSUE}-other")
|
||||
self.assert_refused(assess(lock), "does not match requested")
|
||||
|
||||
def test_worktree_parked_on_another_branch_refused(self):
|
||||
self.assert_refused(assess(current_branch="master"), "not the locked branch")
|
||||
|
||||
def test_detached_head_worktree_refused(self):
|
||||
self.assert_refused(assess(current_branch=None), "detached HEAD")
|
||||
|
||||
def test_different_worktree_refused(self):
|
||||
self.assert_refused(
|
||||
assess(worktree_path="/scratch/elsewhere"), "does not match declared"
|
||||
)
|
||||
|
||||
def test_dirty_worktree_refused(self):
|
||||
self.assert_refused(
|
||||
assess(porcelain_status=" M gitea_mcp_server.py\n"), "requires a clean"
|
||||
)
|
||||
|
||||
def test_local_head_differing_from_remote_refused(self):
|
||||
self.assert_refused(
|
||||
assess(remote_head_sha=OTHER_SHA), "does not match remote branch head"
|
||||
)
|
||||
|
||||
def test_pr_head_differing_refused(self):
|
||||
self.assert_refused(assess(pr_head_sha=OTHER_SHA), "does not match local head")
|
||||
|
||||
def test_missing_remote_head_refused(self):
|
||||
self.assert_refused(assess(remote_head_sha=None), "remote head")
|
||||
|
||||
def test_competing_live_lock_refused(self):
|
||||
competing = [
|
||||
{
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": "/scratch/other-wt",
|
||||
"pid": os.getpid(),
|
||||
}
|
||||
]
|
||||
self.assert_refused(
|
||||
assess(competing_live_locks=competing), "competing live lock"
|
||||
)
|
||||
|
||||
def test_unrelated_live_lock_does_not_block(self):
|
||||
unrelated = [
|
||||
{
|
||||
"issue_number": 999,
|
||||
"branch_name": "fix/issue-999-unrelated",
|
||||
"worktree_path": "/scratch/unrelated",
|
||||
"pid": os.getpid(),
|
||||
}
|
||||
]
|
||||
self.assertTrue(assess(competing_live_locks=unrelated)["recovery_sanctioned"])
|
||||
|
||||
def test_multiple_candidate_branches_refused(self):
|
||||
self.assert_refused(
|
||||
assess(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"]),
|
||||
"multiple branches claim this issue",
|
||||
)
|
||||
|
||||
def test_repository_scope_mismatch_refused(self):
|
||||
self.assert_refused(assess(repo="OtherRepo"), "does not match requested")
|
||||
|
||||
def test_malformed_lock_missing_worktree_refused(self):
|
||||
lock = make_lock()
|
||||
lock.pop("worktree_path")
|
||||
self.assert_refused(assess(lock), "incomplete")
|
||||
|
||||
def test_malformed_lock_missing_pid_refused(self):
|
||||
lock = make_lock()
|
||||
lock.pop("session_pid", None)
|
||||
lock.pop("pid", None)
|
||||
self.assert_refused(assess(lock), "incomplete")
|
||||
|
||||
def test_lock_without_claimant_refused(self):
|
||||
lock = make_lock()
|
||||
lock["work_lease"] = dict(lock["work_lease"])
|
||||
lock["work_lease"].pop("claimant")
|
||||
self.assert_refused(assess(lock), "claimant identity/profile")
|
||||
|
||||
|
||||
class TestNotACandidate(unittest.TestCase):
|
||||
def test_absent_lock_is_not_a_candidate(self):
|
||||
result = issue_lock_recovery.assess_dead_session_lock_recovery(
|
||||
None,
|
||||
issue_number=ISSUE,
|
||||
branch_name=BRANCH,
|
||||
worktree_path=WORKTREE,
|
||||
remote="prgs",
|
||||
org="ExampleOrg",
|
||||
repo="ExampleRepo",
|
||||
identity=IDENTITY,
|
||||
profile=PROFILE,
|
||||
current_branch=BRANCH,
|
||||
porcelain_status="",
|
||||
head_sha=HEAD,
|
||||
remote_head_sha=HEAD,
|
||||
)
|
||||
self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
|
||||
self.assertFalse(result["recovery_sanctioned"])
|
||||
self.assertFalse(result["is_candidate"])
|
||||
|
||||
def test_lock_for_a_different_issue_is_not_a_candidate(self):
|
||||
result = assess(make_lock(issue_number=7777))
|
||||
self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
|
||||
self.assertFalse(result["recovery_sanctioned"])
|
||||
|
||||
|
||||
class TestWorktreeGateWaiver(unittest.TestCase):
|
||||
def test_new_issue_claim_still_requires_base_equivalence(self):
|
||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=WORKTREE,
|
||||
current_branch=BRANCH,
|
||||
porcelain_status="",
|
||||
base_equivalent=False,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertFalse(result["base_equivalence_waived"])
|
||||
|
||||
def test_sanctioned_recovery_waives_base_equivalence(self):
|
||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=WORKTREE,
|
||||
current_branch=BRANCH,
|
||||
porcelain_status="",
|
||||
base_equivalent=False,
|
||||
recovery_sanctioned=True,
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
self.assertTrue(result["base_equivalence_waived"])
|
||||
|
||||
def test_recovery_never_waives_cleanliness(self):
|
||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=WORKTREE,
|
||||
current_branch=BRANCH,
|
||||
porcelain_status=" M gitea_mcp_server.py\n",
|
||||
base_equivalent=False,
|
||||
recovery_sanctioned=True,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(
|
||||
any("tracked file edits" in reason for reason in result["reasons"])
|
||||
)
|
||||
|
||||
def test_unproven_base_equivalence_still_blocks_without_recovery(self):
|
||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=WORKTREE,
|
||||
current_branch=BRANCH,
|
||||
porcelain_status="",
|
||||
base_equivalent=None,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
|
||||
class TestRecoveryRecordAndDownstream(unittest.TestCase):
|
||||
def test_recovery_record_preserves_truthful_provenance(self):
|
||||
assessment = assess()
|
||||
prior = assessment["evidence"]["prior_session_pid"]
|
||||
record = issue_lock_recovery.build_recovery_record(
|
||||
assessment, recovered_at="2026-07-18T23:21:40Z"
|
||||
)
|
||||
self.assertTrue(record["recovered"])
|
||||
self.assertEqual(record["prior_session_pid"], prior)
|
||||
self.assertEqual(record["replacement_session_pid"], os.getpid())
|
||||
self.assertNotEqual(
|
||||
record["prior_session_pid"], record["replacement_session_pid"]
|
||||
)
|
||||
self.assertFalse(record["prior_pid_alive"])
|
||||
self.assertEqual(record["recovered_at"], "2026-07-18T23:21:40Z")
|
||||
self.assertEqual(record["branch_name"], BRANCH)
|
||||
self.assertEqual(record["local_head"], HEAD)
|
||||
self.assertEqual(record["identity"], IDENTITY)
|
||||
self.assertTrue(record["proof"])
|
||||
|
||||
def test_recovery_record_carries_no_secret_material(self):
|
||||
record = issue_lock_recovery.build_recovery_record(
|
||||
assess(), recovered_at="2026-07-18T23:21:40Z"
|
||||
)
|
||||
blob = repr(record).lower()
|
||||
for banned in ("token", "password", "authorization", "secret", "api_key"):
|
||||
self.assertNotIn(banned, blob)
|
||||
|
||||
def test_recovered_lock_satisfies_update_by_merge_ownership(self):
|
||||
# After recovery the lock is rebound to the live session, so the
|
||||
# ownership re-check used by gitea_update_pr_branch_by_merge passes.
|
||||
assessment = assess()
|
||||
recovered_lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
|
||||
recovered_lock["dead_session_recovery"] = (
|
||||
issue_lock_recovery.build_recovery_record(
|
||||
assessment, recovered_at="2026-07-18T23:21:40Z"
|
||||
)
|
||||
)
|
||||
freshness = issue_lock_store.assess_lock_freshness(recovered_lock)
|
||||
self.assertTrue(freshness["live"], freshness)
|
||||
|
||||
verdict = issue_lock_store.verify_lock_for_mutation(
|
||||
recovered_lock,
|
||||
issue_number=ISSUE,
|
||||
branch_name=BRANCH,
|
||||
worktree_path=WORKTREE,
|
||||
)
|
||||
self.assertTrue(verdict["proven"], verdict["reasons"])
|
||||
self.assertFalse(verdict["block"])
|
||||
|
||||
def test_pre_recovery_lock_fails_ownership_check(self):
|
||||
# Guards against a false positive above: the dead-PID lock must fail.
|
||||
verdict = issue_lock_store.verify_lock_for_mutation(
|
||||
make_lock(),
|
||||
issue_number=ISSUE,
|
||||
branch_name=BRANCH,
|
||||
worktree_path=WORKTREE,
|
||||
)
|
||||
self.assertTrue(verdict["block"])
|
||||
self.assertTrue(any("not live" in reason for reason in verdict["reasons"]))
|
||||
|
||||
def test_no_manual_file_seeding_required(self):
|
||||
# The whole decision is reachable from the durable record plus live
|
||||
# observation; nothing is written to disk to reach a verdict.
|
||||
self.assertTrue(assess()["recovery_sanctioned"])
|
||||
|
||||
def test_refusal_message_is_fail_closed(self):
|
||||
message = issue_lock_recovery.format_recovery_refusal(
|
||||
assess(porcelain_status=" M x.py\n")
|
||||
)
|
||||
self.assertIn("fail closed", message)
|
||||
self.assertIn("recovery refused", message.lower())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,609 @@
|
||||
import sys as _sys
|
||||
from pathlib import Path as _Path
|
||||
_sys.path.insert(0, str(_Path(__file__).resolve().parent))
|
||||
from mutation_profile_fixture import shared_mutation_env # noqa: E402
|
||||
"""Dead-session lock recovery when the issue already owns an open PR (#755).
|
||||
|
||||
#753 added the recovery *assessor*, but the production ``gitea_lock_issue``
|
||||
path still rejected every sanctioned recovery: a dead-session lock is by
|
||||
construction a lock for work that already has an open PR, and the #400
|
||||
duplicate-work gate blocked unconditionally on any linked open PR. These tests
|
||||
drive the real MCP handler, not just the pure assessor, so that gap cannot
|
||||
reopen.
|
||||
"""
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import issue_lock_provenance # noqa: E402
|
||||
import issue_lock_recovery # noqa: E402
|
||||
import issue_lock_store # noqa: E402
|
||||
import mcp_server # noqa: E402
|
||||
from issue_work_duplicate_gate import ( # noqa: E402
|
||||
OUTCOME_DUPLICATE_PR_PREVENTED,
|
||||
OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
|
||||
PHASE_LOCK,
|
||||
assess_work_issue_duplicate_gate,
|
||||
)
|
||||
|
||||
ISSUE = 4755
|
||||
BRANCH = f"fix/issue-{ISSUE}-owning-pr"
|
||||
OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
|
||||
HEAD = "c" * 40
|
||||
OTHER_HEAD = "d" * 40
|
||||
OWNING_PR = 4756
|
||||
OTHER_PR = 4757
|
||||
IDENTITY = "example-user"
|
||||
PROFILE = "test-author-prgs"
|
||||
ORG = "Scaled-Tech-Consulting"
|
||||
REPO = "Gitea-Tools"
|
||||
|
||||
|
||||
def dead_pid() -> int:
|
||||
"""A PID that has certainly exited (spawned, then reaped)."""
|
||||
proc = subprocess.Popen([sys.executable, "-c", "pass"])
|
||||
proc.wait()
|
||||
return proc.pid
|
||||
|
||||
|
||||
def shifted_ts(hours: int = 4) -> str:
|
||||
return (
|
||||
(datetime.now(timezone.utc) + timedelta(hours=hours))
|
||||
.isoformat()
|
||||
.replace("+00:00", "Z")
|
||||
)
|
||||
|
||||
|
||||
def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
|
||||
return {
|
||||
"number": number,
|
||||
"title": f"fix: something (Closes #{issue})",
|
||||
"body": f"Closes #{issue}.",
|
||||
"head": {"ref": ref, "sha": sha},
|
||||
}
|
||||
|
||||
|
||||
def sanctioned_token(
|
||||
issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD
|
||||
):
|
||||
"""The evidence shape the server derives from a granted recovery."""
|
||||
return {
|
||||
"issue_number": issue_number,
|
||||
"pr_number": pr_number,
|
||||
"branch_name": branch,
|
||||
"head_sha": head,
|
||||
}
|
||||
|
||||
|
||||
# ───────────────────────── duplicate gate: exemption ─────────────────────────
|
||||
|
||||
|
||||
class TestOwningPrExemptionGranted(unittest.TestCase):
|
||||
def test_exact_owning_pr_is_not_duplicate_work(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr()],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["owning_pr_recovery_exempted"])
|
||||
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
|
||||
self.assertEqual(result["linked_open_pr"], OWNING_PR)
|
||||
self.assertEqual(result["linked_open_pr_count"], 1)
|
||||
|
||||
def test_unrelated_open_pr_alongside_owning_pr_is_ignored(self):
|
||||
unrelated = {
|
||||
"number": 999,
|
||||
"title": "chore: unrelated",
|
||||
"body": "no linkage",
|
||||
"head": {"ref": "chore/unrelated", "sha": OTHER_HEAD},
|
||||
}
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[unrelated, owning_pr()],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["owning_pr_recovery_exempted"])
|
||||
self.assertEqual(result["linked_open_pr_count"], 1)
|
||||
|
||||
|
||||
class TestOwningPrExemptionRefused(unittest.TestCase):
|
||||
def assert_blocked(self, result):
|
||||
self.assertTrue(result["block"])
|
||||
self.assertFalse(result["owning_pr_recovery_exempted"])
|
||||
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
|
||||
|
||||
def test_no_recovery_evidence_keeps_ordinary_blocker(self):
|
||||
self.assert_blocked(
|
||||
assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr()],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
)
|
||||
)
|
||||
|
||||
def test_competing_pr_number_refused(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr(number=OTHER_PR)],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assert_blocked(result)
|
||||
|
||||
def test_multiple_linked_open_prs_refused(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assert_blocked(result)
|
||||
self.assertEqual(result["linked_open_pr_count"], 2)
|
||||
|
||||
def test_different_branch_refused(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr(ref=OTHER_BRANCH)],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assert_blocked(result)
|
||||
|
||||
def test_locked_branch_differing_from_evidence_refused(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr()],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=OTHER_BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assert_blocked(result)
|
||||
|
||||
def test_different_head_refused(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr(sha=OTHER_HEAD)],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assert_blocked(result)
|
||||
|
||||
def test_evidence_for_another_issue_refused(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr()],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(issue_number=ISSUE + 1),
|
||||
)
|
||||
self.assert_blocked(result)
|
||||
|
||||
def test_missing_head_in_live_pr_refused(self):
|
||||
pr = owning_pr()
|
||||
pr["head"] = {"ref": BRANCH}
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[pr],
|
||||
branch_names=[BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
self.assert_blocked(result)
|
||||
|
||||
|
||||
class TestOrdinaryDuplicateBehaviorUnchanged(unittest.TestCase):
|
||||
def test_clean_issue_still_passes(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[],
|
||||
branch_names=["feat/other-issue-99"],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertFalse(result["owning_pr_recovery_exempted"])
|
||||
|
||||
def test_competing_branch_still_blocks_even_with_owning_pr_evidence(self):
|
||||
result = assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr()],
|
||||
branch_names=[BRANCH, OTHER_BRANCH],
|
||||
claim_entry={"status": "not_claimed"},
|
||||
locked_branch=BRANCH,
|
||||
phase=PHASE_LOCK,
|
||||
recovered_owning_pr=sanctioned_token(),
|
||||
)
|
||||
# The owning PR is exempt, but the competing branch is not.
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(result["owning_pr_recovery_exempted"])
|
||||
self.assertIn(OTHER_BRANCH, result["conflicting_branches"])
|
||||
|
||||
|
||||
# ─────────────────── server-derived evidence cannot be forged ───────────────────
|
||||
|
||||
|
||||
class TestOwningPrEvidenceDerivation(unittest.TestCase):
|
||||
def granted(self, **evidence_overrides):
|
||||
evidence = {
|
||||
"issue_number": ISSUE,
|
||||
"locked_branch": BRANCH,
|
||||
"local_head": HEAD,
|
||||
"remote_head": HEAD,
|
||||
"pr_head": HEAD,
|
||||
"pr_number": OWNING_PR,
|
||||
}
|
||||
evidence.update(evidence_overrides)
|
||||
return {
|
||||
"outcome": issue_lock_recovery.RECOVERY_SANCTIONED,
|
||||
"recovery_sanctioned": True,
|
||||
"is_candidate": True,
|
||||
"reasons": [],
|
||||
"evidence": evidence,
|
||||
}
|
||||
|
||||
def test_granted_recovery_yields_evidence(self):
|
||||
token = issue_lock_recovery.owning_pr_recovery_evidence(self.granted())
|
||||
self.assertEqual(token, sanctioned_token())
|
||||
|
||||
def test_none_assessment_yields_nothing(self):
|
||||
self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(None))
|
||||
|
||||
def test_refused_assessment_yields_nothing(self):
|
||||
refused = self.granted()
|
||||
refused["outcome"] = issue_lock_recovery.REFUSED
|
||||
refused["recovery_sanctioned"] = False
|
||||
self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(refused))
|
||||
|
||||
def test_sanctioned_flag_without_outcome_yields_nothing(self):
|
||||
forged = self.granted()
|
||||
forged["outcome"] = "SOMETHING_ELSE"
|
||||
self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(forged))
|
||||
|
||||
def test_missing_pr_number_yields_nothing(self):
|
||||
self.assertIsNone(
|
||||
issue_lock_recovery.owning_pr_recovery_evidence(
|
||||
self.granted(pr_number=None)
|
||||
)
|
||||
)
|
||||
|
||||
def test_head_disagreement_in_evidence_yields_nothing(self):
|
||||
self.assertIsNone(
|
||||
issue_lock_recovery.owning_pr_recovery_evidence(
|
||||
self.granted(remote_head=OTHER_HEAD)
|
||||
)
|
||||
)
|
||||
self.assertIsNone(
|
||||
issue_lock_recovery.owning_pr_recovery_evidence(
|
||||
self.granted(local_head=OTHER_HEAD)
|
||||
)
|
||||
)
|
||||
|
||||
def test_real_refused_assessment_yields_nothing(self):
|
||||
"""End-to-end against the real assessor, not a hand-built dict."""
|
||||
lock = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": "/scratch/wt",
|
||||
"remote": "prgs",
|
||||
"org": "Example-Org",
|
||||
"repo": "Example-Repo",
|
||||
"session_pid": os.getpid(), # alive → must refuse
|
||||
"work_lease": {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": BRANCH,
|
||||
"worktree_path": "/scratch/wt",
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"expires_at": shifted_ts(),
|
||||
},
|
||||
}
|
||||
assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
|
||||
lock,
|
||||
issue_number=ISSUE,
|
||||
branch_name=BRANCH,
|
||||
worktree_path="/scratch/wt",
|
||||
remote="prgs",
|
||||
org="Example-Org",
|
||||
repo="Example-Repo",
|
||||
identity=IDENTITY,
|
||||
profile=PROFILE,
|
||||
current_branch=BRANCH,
|
||||
porcelain_status="",
|
||||
head_sha=HEAD,
|
||||
remote_head_sha=HEAD,
|
||||
pr_head_sha=HEAD,
|
||||
pr_number=OWNING_PR,
|
||||
competing_live_locks=[],
|
||||
candidate_branches=[BRANCH],
|
||||
current_pid=os.getpid(),
|
||||
)
|
||||
self.assertFalse(assessment["recovery_sanctioned"])
|
||||
self.assertIsNone(
|
||||
issue_lock_recovery.owning_pr_recovery_evidence(assessment)
|
||||
)
|
||||
|
||||
|
||||
# ──────────────────── end-to-end: the real gitea_lock_issue ────────────────────
|
||||
|
||||
|
||||
class LockIssueEndToEndBase(unittest.TestCase):
|
||||
"""Drives ``mcp_server.gitea_lock_issue`` with live git/Gitea observation
|
||||
stubbed at the module boundary — the production gate chain itself runs."""
|
||||
|
||||
def setUp(self):
|
||||
self.lock_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.lock_dir.cleanup)
|
||||
self.worktree = os.path.realpath(os.getcwd())
|
||||
# Bind host/org/repo to what the ``test-author-prgs`` fixture profile is
|
||||
# pinned to, so the session-context gate under test is the real one and
|
||||
# not a cross-host denial. The issue number and lock dir stay synthetic.
|
||||
self.remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
},
|
||||
})
|
||||
self.remotes.start()
|
||||
self.addCleanup(patch.stopall)
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
def write_durable_lock(self, *, pid, branch=BRANCH, worktree=None):
|
||||
path = issue_lock_store.lock_file_path(
|
||||
remote="prgs",
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir.name,
|
||||
)
|
||||
claimant = {"username": IDENTITY, "profile": PROFILE}
|
||||
data = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": branch,
|
||||
"remote": "prgs",
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"worktree_path": worktree or self.worktree,
|
||||
"session_pid": pid,
|
||||
"pid": pid,
|
||||
"work_lease": {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": branch,
|
||||
"worktree_path": worktree or self.worktree,
|
||||
"claimant": claimant,
|
||||
"created_at": shifted_ts(-1),
|
||||
"last_heartbeat_at": shifted_ts(-1),
|
||||
"expires_at": shifted_ts(),
|
||||
},
|
||||
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
|
||||
tool="gitea_lock_issue",
|
||||
claimant=claimant,
|
||||
),
|
||||
}
|
||||
issue_lock_store.save_lock_file(path, data)
|
||||
return path
|
||||
|
||||
def run_lock(
|
||||
self,
|
||||
*,
|
||||
open_prs,
|
||||
porcelain="",
|
||||
current_branch=BRANCH,
|
||||
base_equivalent=False,
|
||||
branch_names=None,
|
||||
head_sha=HEAD,
|
||||
remote_head=HEAD,
|
||||
):
|
||||
branch_names = branch_names if branch_names is not None else [BRANCH]
|
||||
branch_entries = [
|
||||
{"name": name, "commit": {"id": remote_head}} for name in branch_names
|
||||
]
|
||||
env = shared_mutation_env(
|
||||
"test-author-prgs",
|
||||
include_example_repo=True,
|
||||
GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
|
||||
)
|
||||
with patch(
|
||||
"mcp_server.api_get_all", return_value=branch_entries
|
||||
), patch(
|
||||
"mcp_server._list_open_pulls", return_value=list(open_prs)
|
||||
), patch(
|
||||
"mcp_server.get_auth_header", return_value="token x"
|
||||
), patch(
|
||||
"mcp_server._work_lease_claimant",
|
||||
return_value={"username": IDENTITY, "profile": PROFILE},
|
||||
), patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={
|
||||
"current_branch": current_branch,
|
||||
"porcelain_status": porcelain,
|
||||
"base_equivalent": base_equivalent,
|
||||
"head_sha": head_sha,
|
||||
"inspected_git_root": self.worktree,
|
||||
"base_branch": "master",
|
||||
},
|
||||
), patch(
|
||||
"mcp_server.issue_duplicate_context_fetcher",
|
||||
side_effect=lambda h, o, r, auth, issue_number: (
|
||||
list(open_prs), list(branch_names), {"status": "not_claimed"}
|
||||
),
|
||||
):
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
|
||||
return mcp_server.gitea_lock_issue(
|
||||
issue_number=ISSUE,
|
||||
branch_name=BRANCH,
|
||||
remote="prgs",
|
||||
worktree_path=self.worktree,
|
||||
)
|
||||
|
||||
|
||||
class TestRecoveryWithOwningPrSucceeds(LockIssueEndToEndBase):
|
||||
def test_dead_session_recovery_with_owning_pr_relocks(self):
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
result = self.run_lock(open_prs=[owning_pr()])
|
||||
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(result["issue_number"], ISSUE)
|
||||
self.assertEqual(result["branch_name"], BRANCH)
|
||||
self.assertTrue(result["lock_freshness"]["live"])
|
||||
self.assertTrue(result["lock_freshness"]["pid_alive"])
|
||||
|
||||
def test_recovered_lock_records_truthful_provenance(self):
|
||||
prior = dead_pid()
|
||||
self.write_durable_lock(pid=prior)
|
||||
self.run_lock(open_prs=[owning_pr()])
|
||||
|
||||
lock = issue_lock_store.load_issue_lock(
|
||||
remote="prgs",
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir.name,
|
||||
)
|
||||
record = lock.get("dead_session_recovery") or {}
|
||||
self.assertTrue(record.get("recovered"))
|
||||
self.assertEqual(record.get("prior_session_pid"), prior)
|
||||
self.assertEqual(record.get("replacement_session_pid"), os.getpid())
|
||||
self.assertFalse(record.get("prior_pid_alive"))
|
||||
self.assertEqual(record.get("pr_number"), OWNING_PR)
|
||||
self.assertEqual(record.get("branch_name"), BRANCH)
|
||||
self.assertEqual(record.get("identity"), IDENTITY)
|
||||
|
||||
def test_recovered_lock_is_live_and_proves_pr_ownership(self):
|
||||
"""AC6: the persisted lock satisfies update-by-merge's ownership prover."""
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
self.run_lock(open_prs=[owning_pr()])
|
||||
|
||||
lock = issue_lock_store.load_issue_lock(
|
||||
remote="prgs",
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir.name,
|
||||
)
|
||||
self.assertTrue(issue_lock_store.is_lease_live(lock))
|
||||
|
||||
env = shared_mutation_env(
|
||||
"test-author-prgs",
|
||||
include_example_repo=True,
|
||||
GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
|
||||
)
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
|
||||
ownership = mcp_server._prove_author_ownership_for_pr(
|
||||
pr_number=OWNING_PR,
|
||||
pr_title=f"fix: something (Closes #{ISSUE})",
|
||||
pr_body=f"Closes #{ISSUE}.",
|
||||
source_branch=BRANCH,
|
||||
remote="prgs",
|
||||
host="gitea.prgs.cc",
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
worktree_path=self.worktree,
|
||||
)
|
||||
self.assertTrue(ownership["proven"], ownership["reasons"])
|
||||
self.assertTrue(ownership["has_author_lock"])
|
||||
self.assertEqual(ownership["matched_issue"], ISSUE)
|
||||
|
||||
|
||||
class TestRecoveryRejectionsEndToEnd(LockIssueEndToEndBase):
|
||||
def assert_lock_refused(self, **kwargs):
|
||||
with self.assertRaises((ValueError, RuntimeError)) as ctx:
|
||||
self.run_lock(**kwargs)
|
||||
return str(ctx.exception)
|
||||
|
||||
def test_competing_pr_still_blocked(self):
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
message = self.assert_lock_refused(
|
||||
open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
|
||||
branch_names=[BRANCH],
|
||||
)
|
||||
self.assertIn("already covers issue", message)
|
||||
|
||||
def test_multiple_linked_open_prs_blocked(self):
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
message = self.assert_lock_refused(
|
||||
open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
|
||||
)
|
||||
self.assertIn("already covers issue", message)
|
||||
|
||||
def test_owning_pr_on_a_different_head_blocked(self):
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
self.assert_lock_refused(open_prs=[owning_pr(sha=OTHER_HEAD)])
|
||||
|
||||
def test_lock_registered_to_a_different_worktree_blocked(self):
|
||||
self.write_durable_lock(
|
||||
pid=dead_pid(), worktree=os.path.join(self.worktree, "elsewhere")
|
||||
)
|
||||
self.assert_lock_refused(open_prs=[owning_pr()])
|
||||
|
||||
def test_dirty_worktree_blocked(self):
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
self.assert_lock_refused(
|
||||
open_prs=[owning_pr()], porcelain=" M gitea_mcp_server.py"
|
||||
)
|
||||
|
||||
def test_worktree_parked_on_another_branch_blocked(self):
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
self.assert_lock_refused(
|
||||
open_prs=[owning_pr()], current_branch="master"
|
||||
)
|
||||
|
||||
def test_local_head_differing_from_remote_blocked(self):
|
||||
self.write_durable_lock(pid=dead_pid())
|
||||
self.assert_lock_refused(
|
||||
open_prs=[owning_pr()], head_sha=OTHER_HEAD
|
||||
)
|
||||
|
||||
def test_live_prior_pid_blocked(self):
|
||||
self.write_durable_lock(pid=os.getpid())
|
||||
self.assert_lock_refused(open_prs=[owning_pr()])
|
||||
|
||||
def test_new_claim_without_prior_lock_still_requires_base_equivalence(self):
|
||||
"""AC10: no durable lock → no recovery → base-equivalence still rules."""
|
||||
self.assert_lock_refused(open_prs=[], branch_names=[])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user