Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

fix: resolve conflicts for PR #416

Browse Source 
Merge current prgs/master into feat/issue-399-conflict-fix-leases.
Preserve reviewer PR lease gate (#407) before conflict-fix stale-head
gate (#399); document both as §26B and §26C. Align MCP and validator
tests with mandatory expected_head_sha and stale-head proof fields.
This commit is contained in:
sysadmin
2026-07-07 17:38:45 -04:00
parent 87beb44394 f061fee18d
commit 55b3658d93
38 changed files with 4767 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_agent_temp_artifacts.py
+4 -1
View File
@@ -80,7 +80,10 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
def tearDown(self):
self._env_patcher.stop()
@patch("mcp_server.api_get_all", return_value=[])
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server._auth", return_value="token x")
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
@patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())
tests/test_audit.py
+29 -3
View File
@@ -286,6 +286,21 @@ class TestSimpleToolAudit(_AuditWiringBase):
class TestGatedToolAudit(_AuditWiringBase):
def setUp(self):
super().setUp()
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {"user": {"login": author}, "state": state,
"head": {"sha": sha}, "mergeable": mergeable}
@@ -344,11 +359,22 @@ class TestGatedToolAudit(_AuditWiringBase):
GITEA_ALLOWED_OPERATIONS="read,review,approve")
with patch.dict(os.environ, env, clear=True):
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
r = gitea_submit_pr_review(pr_number=8, action="approve",
body="LGTM", remote="prgs",
final_review_decision_ready=True)
lease_patch = _install_owned_reviewer_lease(8)
lease_patch.start()
try:
r = gitea_submit_pr_review(
pr_number=8, action="approve",
body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
finally:
lease_patch.stop()
reviewer_pr_lease.clear_session_lease()
self.assertTrue(r["performed"])
recs = self._records()
self.assertEqual(len(recs), 1)
tests/test_commit_files_capability.py
+7
View File
@@ -76,7 +76,14 @@ class CommitFilesCapabilityBase(unittest.TestCase):
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
self._dup_fetcher_patcher = patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
self._dup_fetcher_patcher.start()
def tearDown(self):
self._dup_fetcher_patcher.stop()
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
tests/test_commit_payloads.py
+21
View File
@@ -67,6 +67,15 @@ class TestCommitPayloads(unittest.TestCase):
self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
self.lock_file_path = "/tmp/gitea_issue_lock.json"
import issue_lock_provenance
work_lease = {
"operation_type": "author_issue_work",
"issue_number": 263,
"branch": "feat/issue-263-native-commit-payloads",
"claimant": {"username": "test-user", "profile": "test-author"},
"expires_at": "2999-01-01T00:00:00Z",
}
self.lock_data = {
"issue_number": 263,
"branch_name": "feat/issue-263-native-commit-payloads",
@@ -74,6 +83,11 @@ class TestCommitPayloads(unittest.TestCase):
"org": "Example-Org",
"repo": "Example-Repo",
"worktree_path": self.locked_worktree_path,
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
}
with open(self.lock_file_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(self.lock_data))
@@ -84,7 +98,14 @@ class TestCommitPayloads(unittest.TestCase):
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
self._dup_fetcher_patcher = patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
self._dup_fetcher_patcher.start()
def tearDown(self):
self._dup_fetcher_patcher.stop()
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
tests/test_create_issue_workspace_guard.py
+145
View File
@@ -0,0 +1,145 @@
import os
import sys
import unittest
from pathlib import Path
from unittest.mock import patch, MagicMock
# Ensure we import from the repo root
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import gitea_mcp_server as srv
FAKE_AUTH = {"Authorization": "token test-token"}
# Stable control checkout (parent of branches/), not the MCP server worktree root.
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
PROJECT_ROOT = srv.PROJECT_ROOT
class TestCreateIssueWorkspaceGuard(unittest.TestCase):
def setUp(self):
# Reset preflight flags
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_resolved_role = "author"
srv._preflight_whoami_violation = False
srv._preflight_capability_violation = False
# Disable early return in verify_preflight_purity for testing
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
@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.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
def test_create_issue_stable_checkout_rejected(self, _git, _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.
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(title="Test issue", body="body text")
self.assertIn("stable control checkout", str(ctx.exception))
@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.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
@patch("os.path.exists", return_value=True)
@patch("os.path.isdir", return_value=True)
@patch("subprocess.run")
def test_create_issue_valid_worktree_succeeds(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
# Mock subprocess.run for git --git-common-dir to return PROJECT_ROOT/.git
mock_res = MagicMock()
mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
mock_run.return_value = mock_res
mock_api.return_value = {"number": 42, "html_url": "https://gitea.example.com/issues/42"}
# Provide a valid branches path under the control checkout root
valid_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
res = srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=valid_path
)
self.assertEqual(res["number"], 42)
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.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
def test_create_issue_missing_worktree_fails_closed(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
# Path under branches/ but doesn't exist
missing_path = os.path.join(
CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-worktree-path-999"
)
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=missing_path
)
self.assertIn("does not exist (fail closed)", str(ctx.exception))
@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.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
@patch("os.path.exists", return_value=True)
@patch("os.path.isdir", return_value=True)
@patch("subprocess.run")
def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
# Mock subprocess.run for git --git-common-dir to return a different path
mock_res = MagicMock()
mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
mock_run.return_value = mock_res
wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=wrong_repo_path
)
self.assertIn("does not belong to the target repository", str(ctx.exception))
@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)
def test_create_issue_fails_without_whoami_preflight(self, _ns, _prof, _auth):
srv._preflight_whoami_called = False
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(title="Test issue", body="body")
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
@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)
def test_create_issue_fails_without_capability_preflight(self, _ns, _prof, _auth):
srv._preflight_capability_called = False
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(title="Test issue", body="body")
self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx.exception))
if __name__ == "__main__":
unittest.main()
tests/test_final_report_validator.py
+4
View File
@@ -41,6 +41,10 @@ def _review_handoff(**overrides):
"- Selected PR: #203",
"- Reviewer eligibility: eligible",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Push occurred during validation: no",
"- Worktree path: branches/review-203",
"- Worktree dirty: clean",
"- Scratch worktree used: yes (branches/review-203)",
tests/test_issue_lock_provenance.py
+130
View File
@@ -0,0 +1,130 @@
"""Tests for issue-lock provenance and external-state disclosure (#447)."""
from __future__ import annotations
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_provenance as ilp # noqa: E402
from final_report_validator import assess_final_report_validator # noqa: E402
def _sanctioned_lock(**overrides):
work_lease = {
"operation_type": "author_issue_work",
"issue_number": 447,
"branch": "feat/issue-447-lock-provenance",
"claimant": {"username": "jcwalker3", "profile": "prgs-author"},
"expires_at": "2999-01-01T00:00:00Z",
}
data = {
"issue_number": 447,
"branch_name": "feat/issue-447-lock-provenance",
"work_lease": work_lease,
"lock_provenance": ilp.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease["claimant"],
),
}
data.update(overrides)
return data
class TestLockProvenanceForCreatePr(unittest.TestCase):
def test_sanctioned_lock_passes(self):
result = ilp.assess_lock_file_for_create_pr(_sanctioned_lock())
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_manual_seed_without_provenance_blocked(self):
result = ilp.assess_lock_file_for_create_pr(
{"issue_number": 420, "branch_name": "feat/x", "work_lease": {}}
)
self.assertTrue(result["block"])
self.assertIn("lock_provenance", result["reasons"][0])
def test_operator_override_requires_reason(self):
result = ilp.assess_lock_file_for_create_pr(
_sanctioned_lock(
lock_provenance=ilp.build_sanctioned_lock_provenance(
tool="operator_override",
source=ilp.SOURCE_OPERATOR_OVERRIDE,
)
)
)
self.assertTrue(result["block"])
class TestExternalStateReportRules(unittest.TestCase):
def test_seed_with_external_none_blocked(self):
report = (
"Restored /tmp/gitea_issue_lock.json to unblock PR creation.\n"
"- External-state mutations: none\n"
)
result = ilp.assess_issue_lock_external_state_report(report)
self.assertTrue(result["block"])
def test_seed_with_disclosure_passes(self):
report = (
"Restored /tmp/gitea_issue_lock.json after MCP restart.\n"
"- External-state mutations: wrote /tmp/gitea_issue_lock.json\n"
)
result = ilp.assess_issue_lock_external_state_report(report)
self.assertTrue(result["proven"])
def test_remove_claimed_as_cleanup_only_blocked(self):
report = (
"rm /tmp/gitea_issue_lock.json after PR creation.\n"
"- Cleanup mutations: lock removed\n"
"- External-state mutations: none\n"
)
result = ilp.assess_issue_lock_external_state_report(report)
self.assertTrue(result["block"])
def test_manual_lock_pr_without_override_blocked(self):
report = (
"Programmatically seeded gitea_issue_lock.json then gitea_create_pr.\n"
"PR #444 created.\n"
)
result = ilp.assess_manual_lock_pr_without_override(report)
self.assertTrue(result["block"])
def test_author_reviewer_same_run_blocked(self):
report = (
"gitea_create_pr opened PR #444.\n"
"Submitted approve review on PR #444.\n"
)
result = ilp.assess_author_reviewer_same_run_report(report)
self.assertTrue(result["block"])
class TestFinalReportValidatorIntegration(unittest.TestCase):
def test_work_issue_blocks_hidden_lock_mutation(self):
report = (
"## Controller Handoff\n"
"- Task: work issue #420\n"
"- External-state mutations: none\n"
"Restored /tmp/gitea_issue_lock.json before PR creation.\n"
)
result = assess_final_report_validator(report, "work_issue")
rule_ids = {f["rule_id"] for f in result["findings"]}
self.assertIn("shared.issue_lock_external_state", rule_ids)
self.assertTrue(result["blocked"])
def test_review_pr_blocks_create_and_approve(self):
report = (
"## Controller Handoff\n"
"- Task: review PR #444\n"
"- Review decision: approve\n"
"Created PR #444 via gitea_create_pr earlier in this run.\n"
)
result = assess_final_report_validator(report, "review_pr")
rule_ids = {f["rule_id"] for f in result["findings"]}
self.assertIn("shared.author_reviewer_same_run", rule_ids)
if __name__ == "__main__":
unittest.main()
tests/test_issue_work_duplicate_gate.py
+264
View File
@@ -0,0 +1,264 @@
"""Tests for early duplicate-work detection (#400)."""
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_provenance
import issue_work_duplicate_gate as dup_gate
import mcp_server
from issue_work_duplicate_gate import (
OUTCOME_DUPLICATE_BRANCH_PREVENTED,
OUTCOME_DUPLICATE_COMMIT_PREVENTED,
OUTCOME_DUPLICATE_PR_PREVENTED,
OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
PHASE_COMMIT,
PHASE_CREATE_PR,
PHASE_LOCK,
assess_work_issue_duplicate_gate,
assess_work_issue_duplicate_report,
)
class TestDuplicateGateAssessment(unittest.TestCase):
def test_clear_issue_passes(self):
result = assess_work_issue_duplicate_gate(
400,
open_prs=[],
branch_names=["feat/other-issue-99"],
claim_entry={"status": "not_claimed"},
locked_branch="feat/issue-400-duplicate-work-preflight",
phase=PHASE_LOCK,
)
self.assertFalse(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
def test_open_pr_blocks(self):
prs = [{
"number": 397,
"title": "feat: handoff",
"body": "Closes #395",
"head": {"ref": "feat/issue-395-proof-backed-review-handoff"},
}]
result = assess_work_issue_duplicate_gate(
395,
open_prs=prs,
branch_names=[],
phase=PHASE_LOCK,
)
self.assertTrue(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
def test_conflicting_remote_branch_blocks(self):
result = assess_work_issue_duplicate_gate(
395,
open_prs=[],
branch_names=["feat/issue-395-proof-backed-handoff-claims"],
locked_branch="feat/issue-395-new-attempt",
phase=PHASE_LOCK,
)
self.assertTrue(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_BRANCH_PREVENTED)
def test_active_claim_on_other_branch_blocks(self):
result = assess_work_issue_duplicate_gate(
398,
open_prs=[],
branch_names=[],
claim_entry={
"status": "active",
"latest_heartbeat": {
"branch": "feat/issue-398-validation-cwd-proof",
},
},
locked_branch="feat/issue-398-other-branch",
phase=PHASE_LOCK,
)
self.assertTrue(result["block"])
def test_commit_phase_maps_to_commit_outcome(self):
prs = [{
"number": 411,
"title": "x",
"body": "Closes #398",
"head": {"ref": "feat/issue-398-validation-cwd-proof"},
}]
result = assess_work_issue_duplicate_gate(
398,
open_prs=prs,
branch_names=[],
locked_branch="feat/issue-398-alt",
phase=PHASE_COMMIT,
)
self.assertTrue(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_COMMIT_PREVENTED)
def test_stale_claim_does_not_block_by_status_alone(self):
result = assess_work_issue_duplicate_gate(
400,
open_prs=[],
branch_names=[],
claim_entry={"status": "reclaimable", "reasons": ["stale"]},
locked_branch="feat/issue-400-duplicate-work-preflight",
phase=PHASE_LOCK,
)
self.assertFalse(result["block"])
class TestDuplicateReportOutcome(unittest.TestCase):
def test_requires_exactly_one_outcome(self):
bad = assess_work_issue_duplicate_report("work finished")
self.assertFalse(bad["complete"])
good = assess_work_issue_duplicate_report(
"Duplicate work not prevented for issue #400."
)
self.assertTrue(good["complete"])
self.assertEqual(good["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
class TestInjectableDuplicateFetcher(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value="token x")
def test_lock_issue_uses_injected_fetcher(self, _auth):
seen = {}
def fetcher(h, o, r, auth, issue_number):
seen["issue_number"] = issue_number
return [], [], {"status": "not_claimed"}
with patch(
"mcp_server.issue_duplicate_context_fetcher",
side_effect=fetcher,
), patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "master",
"porcelain_status": "",
"base_equivalent": True,
},
), patch.dict(os.environ, {
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
}, clear=True):
with patch.object(mcp_server, "ISSUE_LOCK_FILE", tempfile.mktemp()):
mcp_server.gitea_lock_issue(
issue_number=400,
branch_name="feat/issue-400-duplicate-work-preflight",
remote="prgs",
)
self.assertEqual(seen["issue_number"], 400)
class TestMcpDuplicateRecheck(unittest.TestCase):
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
self._lock_patch = patch.object(
mcp_server, "ISSUE_LOCK_FILE", self.lock_path
)
self._lock_patch.start()
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
"repo": "Example-Repo"},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
def tearDown(self):
patch.stopall()
self._dir.cleanup()
def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
work_lease = {
"operation_type": "author_issue_work",
"issue_number": issue_number,
"branch": branch,
"claimant": {"username": "test-user", "profile": "test-author"},
"expires_at": "2999-01-01T00:00:00Z",
}
with open(self.lock_path, "w", encoding="utf-8") as fh:
json.dump({
"issue_number": issue_number,
"branch_name": branch,
"remote": "prgs",
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
}, fh)
@patch("mcp_server._assess_issue_duplicate_gate")
@patch("mcp_server.get_profile", return_value={
"profile_name": "test-author",
"allowed_operations": ["gitea.read", "gitea.repo.commit"],
"forbidden_operations": [],
"audit_label": "test-author",
})
@patch("mcp_server.get_auth_header", return_value="token x")
def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate):
self._write_lock()
mock_gate.return_value = {
"block": True,
"reasons": ["open PR #412 already covers issue #400"],
"outcome": OUTCOME_DUPLICATE_COMMIT_PREVENTED,
"safe_next_action": "stop",
}
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with patch(
"mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
):
result = mcp_server.gitea_commit_files(
files=[{
"operation": "create",
"path": "a.txt",
"content_plain": "hi",
}],
message="test",
remote="prgs",
)
self.assertFalse(result["success"])
self.assertIn("duplicate_gate", result)
@patch("mcp_server._assess_issue_duplicate_gate")
@patch("mcp_server.get_profile", return_value={
"profile_name": "test-author",
"allowed_operations": ["gitea.read", "gitea.pr.create"],
"forbidden_operations": [],
"audit_label": "test-author",
})
@patch("mcp_server.get_auth_header", return_value="token x")
def test_create_pr_returns_handoff_on_duplicate(self, _auth, _profile, mock_gate):
self._write_lock()
mock_gate.return_value = {
"block": True,
"reasons": ["open PR #412 already covers issue #400"],
"outcome": OUTCOME_DUPLICATE_PR_PREVENTED,
"safe_next_action": "reconciliation handoff",
}
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with patch(
"mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
):
result = mcp_server.gitea_create_pr(
title="feat: x (Closes #400)",
head="feat/issue-400-x",
base="master",
body="Closes #400",
remote="prgs",
)
self.assertFalse(result["success"])
self.assertIsNone(result.get("number"))
self.assertIn("duplicate_gate", result)
if __name__ == "__main__":
unittest.main()
tests/test_mcp_server.py
+291 -74
View File
@@ -60,6 +60,7 @@ def _mark_request_changes_ready(pr_number=8, **kwargs):
suppression feedback fetch stubbed to 'no existing blocker'."""
with patch("mcp_server.gitea_get_pr_review_feedback",
return_value=dict(_NO_BLOCKER_FEEDBACK)):
kwargs.setdefault("expected_head_sha", "abc123")
return gitea_mark_final_review_decision(
pr_number, "request_changes", **kwargs)
@@ -79,6 +80,65 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
return [_formal_review(reviewer, "APPROVED", sha=sha)]
_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
def _reviewer_lease_comment(
pr_number,
*,
session_id=_DEFAULT_LEASE_SESSION,
head_sha="abc123",
reviewer="reviewer-bot",
):
from datetime import datetime, timezone
import reviewer_pr_lease
body = reviewer_pr_lease.format_lease_body(
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=pr_number,
issue_number=407,
reviewer_identity=reviewer,
profile="gitea-reviewer",
session_id=session_id,
worktree="branches/review-test",
phase="claimed",
candidate_head=head_sha,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=datetime.now(timezone.utc),
)
return {"id": 9001, "body": body, "user": {"login": reviewer}}
def _install_owned_reviewer_lease(
pr_number,
*,
session_id=_DEFAULT_LEASE_SESSION,
head_sha="abc123",
):
import reviewer_pr_lease
reviewer_pr_lease.clear_session_lease()
reviewer_pr_lease.record_session_lease({
"pr_number": pr_number,
"session_id": session_id,
"candidate_head": head_sha,
"target_branch": "master",
})
return patch(
"mcp_server._fetch_pr_comments",
return_value=[
_reviewer_lease_comment(
pr_number,
session_id=session_id,
head_sha=head_sha,
)
],
)
# Issue-write tools are profile-gated (#69).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
@@ -101,17 +161,36 @@ ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
import issue_lock_provenance
work_lease = {
"operation_type": "author_issue_work",
"issue_number": issue_number,
"branch": branch_name,
"claimant": {"username": "test-user", "profile": "test-author"},
"expires_at": "2999-01-01T00:00:00Z",
}
record = {
"issue_number": issue_number,
"branch_name": branch_name,
"remote": "dadeschools",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
}
record.update(overrides)
return record
def _clear_duplicate_context_fetcher(*_args, **_kwargs):
"""Default injectable duplicate-work context for lock/create_pr tests."""
return [], [], {"status": "not_claimed"}
# ---------------------------------------------------------------------------
# Create Issue
# ---------------------------------------------------------------------------
@@ -166,13 +245,17 @@ class TestCreateIssue(unittest.TestCase):
# ---------------------------------------------------------------------------
class TestCreatePR(unittest.TestCase):
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
@@ -187,13 +270,17 @@ class TestCreatePR(unittest.TestCase):
self.assertEqual(payload["base"], "main")
self.assertIn("Closes #123", payload["title"])
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
@@ -512,6 +599,30 @@ class TestViewPR(unittest.TestCase):
class TestMergePR(unittest.TestCase):
"""Gated merge workflow (#16). gitea_merge_pr is the only merge path."""
def setUp(self):
import reviewer_pr_lease
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
self._pr_lease_comments_patch = patch(
"mcp_server._list_pr_lease_comments", return_value=[]
)
self._pr_lease_comments_patch.start()
self.addCleanup(self._pr_lease_comments_patch.stop)
self._pr_work_lease_patch = patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
)
self._pr_work_lease_patch.start()
self.addCleanup(self._pr_work_lease_patch.stop)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {
"user": {"login": author},
@@ -584,7 +695,8 @@ class TestMergePR(unittest.TestCase):
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_changed_files=["b.py", "a.py"], remote="prgs")
expected_changed_files=["b.py", "a.py"],
expected_head_sha="abc123", remote="prgs")
self.assertTrue(r["performed"])
# -- read-back / cleanup surfacing (#98) -----------------------------------
@@ -602,8 +714,10 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation=self._confirm(8),
remote="prgs")
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
# The merge itself is still reported performed/successful.
self.assertTrue(r["performed"])
self.assertEqual(r["merge_result"], "PR #8 merged via 'merge'.")
@@ -631,8 +745,10 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation=self._confirm(8),
remote="prgs")
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertTrue(r["performed"])
self.assertEqual(r["merge_commit"], "c9")
self.assertTrue(r["cleanup_status"].startswith("skipped (cleanup error:"))
@@ -646,7 +762,7 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation="", remote="prgs")
r = gitea_merge_pr(pr_number=8, confirmation="", expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"]))
mock_api.assert_not_called()
@@ -657,7 +773,7 @@ class TestMergePR(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", remote="prgs")
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
mock_api.assert_not_called()
@@ -772,9 +888,11 @@ class TestMergePR(unittest.TestCase):
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="deadbeef", remote="prgs")
self.assertFalse(r["performed"])
self.assertIn(
"expected head SHA does not match current PR head (fail closed)",
r["reasons"])
self.assertTrue(any(
"expected head SHA does not match current PR head (fail closed)" in reason
or "PR head changed during lease" in reason
for reason in r["reasons"]
))
self._assert_no_merge_call(mock_api)
@patch("mcp_server.api_request")
@@ -789,7 +907,8 @@ class TestMergePR(unittest.TestCase):
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8),
expected_changed_files=["a.py", "b.py"], remote="prgs")
expected_changed_files=["a.py", "b.py"],
expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
self.assertIn(
"PR changed files do not match expected_changed_files (fail closed)",
@@ -802,7 +921,7 @@ class TestMergePR(unittest.TestCase):
def test_invalid_merge_method_rejected(self, mock_api):
with patch.dict(os.environ, {}, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation="MERGE PR 8", do="octopus", remote="prgs")
pr_number=8, confirmation="MERGE PR 8", do="octopus", expected_head_sha="abc123", remote="prgs")
self.assertFalse(r["performed"])
self.assertTrue(any("unknown merge method" in x for x in r["reasons"]))
mock_api.assert_not_called()
@@ -820,7 +939,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_TOKEN": "super-secret-token"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
blob = repr(r).lower()
for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
self.assertNotIn(secret, blob)
@@ -837,7 +958,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertFalse(r["performed"])
blob = repr(r)
self.assertIn("[REDACTED]", blob)
@@ -855,7 +978,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertFalse(r["performed"])
self.assertFalse(r.get("approval_visible"))
self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
@@ -876,7 +1001,9 @@ class TestMergePR(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="abc123", remote="prgs",
)
self.assertFalse(r["performed"])
self.assertTrue(r.get("has_blocking_change_requests"))
self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
@@ -986,7 +1113,7 @@ class TestReviewPR(unittest.TestCase):
]
from mcp_server import init_review_decision_lock
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
gitea_mark_final_review_decision(1, "approve", remote="prgs", expected_head_sha="abc123")
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
@@ -1653,7 +1780,26 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
}
def setUp(self):
import reviewer_pr_lease
init_review_decision_lock("prgs", "review_pr")
self._lease_patch = _install_owned_reviewer_lease(
self.PR, head_sha=self.SHA,
)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
self._pr_lease_comments_patch = patch(
"mcp_server._list_pr_lease_comments", return_value=[]
)
self._pr_lease_comments_patch.start()
self.addCleanup(self._pr_lease_comments_patch.stop)
def _env(self):
return patch.dict(os.environ, {
@@ -1748,8 +1894,32 @@ class TestSubmitPrReview(unittest.TestCase):
"""Gated review-mutation tool (#15)."""
def setUp(self):
import reviewer_pr_lease
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self._pr_lease_comments_patch = patch(
"mcp_server._list_pr_lease_comments", return_value=[]
)
self._pr_lease_comments_patch.start()
self._pr_work_lease_patch = patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
)
self._pr_work_lease_patch.start()
gitea_mark_final_review_decision(
8, "approve", remote="prgs", expected_head_sha="abc123",
)
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(self._pr_lease_comments_patch.stop)
self.addCleanup(self._pr_work_lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {
@@ -1909,10 +2079,11 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "comment", remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
]
gitea_mark_final_review_decision(8, "comment", remote="prgs", expected_head_sha="abc123")
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review"}
with patch.dict(os.environ, env, clear=True):
@@ -1928,12 +2099,15 @@ class TestSubmitPrReview(unittest.TestCase):
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [
{"login": "jcwalker3"}, self._pr("jcwalker3"),
{"login": "jcwalker3"}, self._pr("jcwalker3"), {"id": 4},
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(8, "comment", remote="prgs")
gitea_mark_final_review_decision(
8, "comment", remote="prgs", expected_head_sha="abc123",
)
r = gitea_submit_pr_review(
pr_number=8, action="comment", body="note", remote="prgs",
final_review_decision_ready=True,
@@ -1976,20 +2150,22 @@ class TestSubmitPrReview(unittest.TestCase):
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
{"login": "reviewer-bot"}, self._pr("author-bot", sha="deadbeef"),
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve",
expected_head_sha="deadbeef", remote="prgs",
expected_head_sha="abc123", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertIn(
"expected head SHA does not match current PR head (fail closed)",
r["reasons"])
self.assertTrue(any(
"expected head SHA does not match current PR head (fail closed)" in reason
or "PR head changed during lease" in reason
for reason in r["reasons"]
))
self._assert_no_mutation(mock_api)
def test_head_sha_match_allows(self):
@@ -2032,7 +2208,7 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review,approve",
"GITEA_TOKEN": "super-secret-token"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(5, "approve", remote="prgs")
gitea_mark_final_review_decision(5, "approve", remote="prgs", expected_head_sha="abc123")
r = gitea_submit_pr_review(
pr_number=5, action="approve", remote="prgs",
final_review_decision_ready=True,
@@ -2052,9 +2228,8 @@ class TestSubmitPrReview(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(5, "approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=5, action="approve", remote="prgs",
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
@@ -2142,7 +2317,7 @@ class TestSubmitPrReview(unittest.TestCase):
def test_mark_final_decision_rejects_remote_mismatch(self):
init_review_decision_lock("prgs", "review_pr")
r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools")
r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools", expected_head_sha="abc123")
self.assertFalse(r["marked_ready"])
self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
@@ -2154,28 +2329,30 @@ class TestSubmitPrReview(unittest.TestCase):
{"id": 42, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=42)],
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 43, "state": "REQUEST_CHANGES"},
[_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
gitea_mark_final_review_decision(8, "approve", remote="prgs")
with patch.dict(os.environ, env, clear=True):
first = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
auth = gitea_authorize_review_correction(
prior_review_id=42,
prior_review_state="approve",
reason="operator approved correcting mistaken approve",
operator_authorized=True,
)
_mark_request_changes_ready(remote="prgs")
second = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
)
with patch("mcp_server.gitea_get_pr_review_feedback",
return_value=dict(_NO_BLOCKER_FEEDBACK)):
with patch.dict(os.environ, env, clear=True):
first = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
auth = gitea_authorize_review_correction(
prior_review_id=42,
prior_review_state="approve",
reason="operator approved correcting mistaken approve",
operator_authorized=True,
)
_mark_request_changes_ready(remote="prgs")
second = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(first["performed"])
self.assertTrue(auth["authorized"])
self.assertTrue(second["performed"])
@@ -2187,7 +2364,7 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
# Mark decision for specific remote, org, repo
gitea_mark_final_review_decision(8, "approve", remote="prgs", org="MyOrg", repo="MyRepo")
gitea_mark_final_review_decision(8, "approve", remote="prgs", expected_head_sha="abc123", org="MyOrg", repo="MyRepo")
# Mismatched remote
r = gitea_submit_pr_review(
@@ -2273,6 +2450,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
self.assertEqual(res["cleanup_status"].get(1), "not present")
def test_merge_pr_with_closes_removes_label(self):
import reviewer_pr_lease
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
@@ -2307,6 +2491,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
self.assertEqual(res["cleanup_status"].get(123), "released")
def test_merge_pr_with_branch_name_removes_label(self):
import reviewer_pr_lease
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
@@ -2988,10 +3179,12 @@ class TestVerifyMutationAuthority(unittest.TestCase):
# profile; the active profile resolves as reviewer — side-channel
# override rejected even with a matching in-process authority.
self._authority()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
with patch("mcp_server.gitea_config.is_runtime_switching_enabled",
return_value=False):
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
self.assertIn("side-channel override rejected", str(ctx.exception))
def test_foreign_pid_authority_is_not_trusted(self):
@@ -3044,8 +3237,14 @@ class TestIssueLocking(unittest.TestCase):
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._env_patcher.start()
self._dup_fetcher_patcher = patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
self.mock_dup_fetcher = self._dup_fetcher_patcher.start()
def tearDown(self):
self._dup_fetcher_patcher.stop()
self._env_patcher.stop()
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
@@ -3054,10 +3253,8 @@ class TestIssueLocking(unittest.TestCase):
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_success(self, _auth, mock_api, _git_state):
mock_api.return_value = [] # no open PRs
def test_lock_issue_success(self, _auth, _git_state):
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
@@ -3082,50 +3279,48 @@ class TestIssueLocking(unittest.TestCase):
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api, _git_state):
mock_api.return_value = [{
def test_lock_issue_reused_by_open_pr_branch(self, _auth, _git_state):
self.mock_dup_fetcher.return_value = ([{
"number": 200,
"head": {"ref": "feat/issue-196-boundary"},
"title": "Some PR",
"body": "No closes ref"
}]
"body": "No closes ref",
}], [], {"status": "not_claimed"})
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already tied to an open PR", str(ctx.exception))
self.assertIn("open PR #200 already covers issue", str(ctx.exception))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state):
mock_api.return_value = [{
def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, _git_state):
self.mock_dup_fetcher.return_value = ([{
"number": 200,
"head": {"ref": "feat/other-branch"},
"title": "Some PR",
"body": "fixes #196"
}]
"body": "fixes #196",
}], [], {"status": "not_claimed"})
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already tied to an open PR", str(ctx.exception))
self.assertIn("open PR #200 already covers issue", str(ctx.exception))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_remote_branch(self, _auth, mock_api, _git_state):
mock_api.side_effect = [
def test_lock_issue_reused_by_remote_branch(self, _auth, _git_state):
self.mock_dup_fetcher.return_value = (
[],
[{"name": "feat/issue-196-existing-work"}],
]
["feat/issue-196-existing-work"],
{"status": "not_claimed"},
)
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already has matching branch", str(ctx.exception))
self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))
def test_lock_issue_blocks_active_same_operation_lease(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
@@ -3284,6 +3479,28 @@ class TestIssueLocking(unittest.TestCase):
)
self.assertIn("does not match locked worktree", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(
_sample_issue_lock(
issue_number=447,
branch_name="feat/issue-447-lock-provenance",
lock_provenance=None,
),
f,
)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(
title="feat: lock provenance Closes #447",
head="feat/issue-447-lock-provenance",
remote="prgs",
)
self.assertIn("lock provenance", str(ctx.exception).lower())
@patch("mcp_server.api_request")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
tests/test_pr_queue_inventory.py
+16 -3
View File
@@ -156,9 +156,22 @@ class TestPRQueueInventory(unittest.TestCase):
]
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True)
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
with patch("mcp_server._authenticated_username", return_value="reviewer1"):
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
lease_patch = _install_owned_reviewer_lease(
1, head_sha="abc1", session_id="inventory-review-lease",
)
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
result = gitea_review_pr(
pr_number=1, event="APPROVE", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(result["success"])
self.assertIn("=== PR Queue Inventory ===", result["message"])
self.assertIn("Repository:", result["message"])
tests/test_review_proofs.py
+1
View File
@@ -2346,6 +2346,7 @@ class TestWorkIssueFinalReport(unittest.TestCase):
"- Safe next action: open PR",
"- Next: open PR",
"- Safety statement: no review/merge",
"- Duplicate work outcome: duplicate work not prevented",
])
def test_complete_work_issue_report_earns_a(self):
tests/test_reviewer_pr_lease.py
+193
View File
@@ -0,0 +1,193 @@
"""Tests for per-PR reviewer leases (#407)."""
import sys
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import reviewer_pr_lease as leases
def _lease_comment(
pr_number: int,
session_id: str,
*,
phase: str = "claimed",
minutes_ago: int = 0,
candidate_head: str = "a" * 40,
) -> dict:
now = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
body = leases.format_lease_body(
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=pr_number,
issue_number=295,
reviewer_identity="rev1",
profile="prgs-reviewer",
session_id=session_id,
worktree="branches/review-pr382",
phase=phase,
candidate_head=candidate_head,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=now,
)
return {"id": 1, "body": body, "user": {"login": "rev1"}}
class TestReviewerLeaseAcquire(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_two_reviewers_cannot_lease_same_pr(self):
comments = [_lease_comment(382, "session-a")]
result = leases.assess_acquire_lease(
comments,
pr_number=382,
reviewer_identity="rev2",
profile="prgs-reviewer",
session_id="session-b",
repo="Scaled-Tech-Consulting/Gitea-Tools",
issue_number=295,
worktree="branches/review-pr382-b",
candidate_head="c" * 40,
target_branch="master",
target_branch_sha="d" * 40,
)
self.assertFalse(result["acquire_allowed"])
self.assertTrue(any("already has active" in r for r in result["reasons"]))
def test_two_reviewers_can_lease_different_prs(self):
comments = [_lease_comment(382, "session-a")]
result = leases.assess_acquire_lease(
comments,
pr_number=383,
reviewer_identity="rev2",
profile="prgs-reviewer",
session_id="session-b",
repo="Scaled-Tech-Consulting/Gitea-Tools",
issue_number=296,
worktree="branches/review-pr383",
candidate_head="c" * 40,
target_branch="master",
target_branch_sha="d" * 40,
)
self.assertTrue(result["acquire_allowed"])
self.assertIsNotNone(result["lease_body"])
class TestReviewerLeaseFreshness(unittest.TestCase):
def test_stale_warning_after_30_minutes(self):
lease = leases.parse_lease_comment(
_lease_comment(382, "session-a", minutes_ago=35)["body"]
)
self.assertEqual(
leases.classify_lease_freshness(lease),
"stale_warning",
)
def test_reclaimable_after_60_minutes(self):
lease = leases.parse_lease_comment(
_lease_comment(382, "session-a", minutes_ago=65)["body"]
)
self.assertEqual(
leases.classify_lease_freshness(lease),
"reclaimable",
)
class TestReviewerLeaseMutationGate(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_reviewer_without_lease_cannot_mutate(self):
head = "f" * 40
comments = [_lease_comment(382, "other-session", candidate_head=head)]
result = leases.assess_mutation_lease_gate(
pr_number=382,
comments=comments,
reviewer_identity="rev1",
session_id="my-session",
mutation="approve",
live_head_sha=head,
pinned_head_sha=head,
)
self.assertTrue(result["block"])
def test_owned_lease_allows_mutation(self):
head = "f" * 40
comments = [_lease_comment(382, "my-session", candidate_head=head)]
leases.record_session_lease({
"pr_number": 382,
"session_id": "my-session",
"candidate_head": head,
"target_branch": "master",
})
result = leases.assess_mutation_lease_gate(
pr_number=382,
comments=comments,
reviewer_identity="rev1",
session_id="my-session",
mutation="approve",
live_head_sha=head,
pinned_head_sha=head,
)
self.assertFalse(result["block"])
def test_head_change_invalidates_lease(self):
reviewed = "f" * 40
live = "e" * 40
comments = [_lease_comment(382, "my-session", candidate_head=reviewed)]
leases.record_session_lease({
"pr_number": 382,
"session_id": "my-session",
"candidate_head": reviewed,
})
result = leases.assess_mutation_lease_gate(
pr_number=382,
comments=comments,
reviewer_identity="rev1",
session_id="my-session",
mutation="merge",
live_head_sha=live,
pinned_head_sha=reviewed,
)
self.assertTrue(result["block"])
self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
class TestReviewerLeaseMcpGate(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
patch("mcp_server.verify_preflight_purity").start()
patch("gitea_audit.audit_enabled", return_value=False).start()
mcp_server = __import__("mcp_server")
mcp_server._IDENTITY_CACHE.clear()
mcp_server.init_review_decision_lock("prgs", "review_pr")
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", "reviewer")
def tearDown(self):
patch.stopall()
leases.clear_session_lease()
def test_reviewer_pr_lease_gate_helper_blocks_without_session(self):
import mcp_server
head = "a" * 40
with patch("mcp_server._fetch_pr_comments", return_value=[]):
reasons = mcp_server._reviewer_pr_lease_gate(
pr_number=382,
remote="prgs",
host=None,
org=None,
repo=None,
mutation="approve",
live_head_sha=head,
pinned_head_sha=head,
)
self.assertTrue(any("lease" in r.lower() for r in reasons))
if __name__ == "__main__":
unittest.main()
tests/test_webui_project_registry.py
+111
View File
@@ -0,0 +1,111 @@
"""Tests for web UI project registry (#427)."""
import json
import sys
import tempfile
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.project_registry import (
default_registry_path,
load_registry,
project_to_dict,
)
class TestProjectRegistryLoader(unittest.TestCase):
def test_default_registry_loads_gitea_tools(self):
registry = load_registry()
self.assertEqual(registry.version, 1)
self.assertEqual(len(registry.projects), 1)
project = registry.projects[0]
self.assertEqual(project.id, "gitea-tools")
self.assertEqual(project.repo_name, "Gitea-Tools")
self.assertEqual(project.gitea_owner, "Scaled-Tech-Consulting")
self.assertEqual(project.remote_host, "https://gitea.prgs.cc")
self.assertEqual(project.profiles["author"], "prgs-author")
self.assertEqual(project.profiles["reviewer"], "prgs-reviewer")
self.assertEqual(project.profiles["reconciler"], "prgs-reconciler")
self.assertIn("skill", project.workflow_paths)
self.assertGreaterEqual(len(project.onboarding_checklist), 4)
def test_registry_rejects_credential_keys(self):
payload = {
"version": 1,
"projects": [
{
"id": "bad",
"repo_name": "Bad",
"gitea_owner": "Org",
"remote_host": "https://gitea.example.invalid",
"default_branch": "main",
"local_checkout_path": ".",
"profiles": {
"author": "a",
"reviewer": "r",
"reconciler": "c",
},
"workflow_paths": {"skill": "skills/x.md"},
"api_token": "secret",
}
],
}
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
json.dump(payload, handle)
path = Path(handle.name)
try:
with self.assertRaises(ValueError):
load_registry(path)
finally:
path.unlink(missing_ok=True)
def test_default_registry_path_points_at_packaged_data(self):
path = default_registry_path()
self.assertTrue(path.name == "projects.registry.json")
self.assertTrue(path.parent.name == "data")
class TestProjectRegistryRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_projects_page_lists_gitea_tools(self):
response = self.client.get("/projects")
self.assertEqual(response.status_code, 200)
self.assertIn("Gitea-Tools", response.text)
self.assertIn("Scaled-Tech-Consulting", response.text)
self.assertIn("prgs-author", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_project_detail_renders_checklist(self):
response = self.client.get("/projects/gitea-tools")
self.assertEqual(response.status_code, 200)
self.assertIn("Onboarding checklist", response.text)
self.assertIn("Configure execution profiles", response.text)
self.assertIn("branches/", response.text)
def test_project_detail_404(self):
response = self.client.get("/projects/unknown-repo")
self.assertEqual(response.status_code, 404)
def test_api_projects_json(self):
response = self.client.get("/api/projects")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["version"], 1)
self.assertEqual(len(data["projects"]), 1)
self.assertEqual(data["projects"][0]["id"], "gitea-tools")
self.assertIn("onboarding_checklist", data["projects"][0])
def test_project_to_dict_is_json_safe(self):
registry = load_registry()
encoded = json.dumps(project_to_dict(registry.projects[0]))
self.assertIn("gitea-tools", encoded)
if __name__ == "__main__":
unittest.main()
tests/test_webui_prompt_library.py
+90
View File
@@ -0,0 +1,90 @@
"""Tests for web UI prompt library (#428)."""
import json
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.prompt_library import find_prompt, library_to_dict, load_prompt_library, prompt_to_dict
REQUIRED_PROMPT_SLUGS = frozenset({
"review-pr",
"work-issue",
"create-issue",
"comment-issue",
"cleanup",
"audit",
"onboarding",
})
class TestPromptLibraryLoader(unittest.TestCase):
def test_library_loads_required_prompts(self):
entries = load_prompt_library()
slugs = {entry.slug for entry in entries}
self.assertEqual(slugs, REQUIRED_PROMPT_SLUGS)
def test_workflow_hashes_present(self):
review = find_prompt("review-pr")
self.assertIsNotNone(review)
assert review is not None
self.assertTrue(review.workflow_hash)
self.assertEqual(len(review.workflow_hash), 64)
def test_prompt_text_is_short(self):
for entry in load_prompt_library():
self.assertLess(len(entry.prompt_text), 400)
self.assertNotIn("Do not improvise around the gates", entry.prompt_text)
class TestPromptLibraryRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_prompts_page_lists_all_entries(self):
response = self.client.get("/prompts")
self.assertEqual(response.status_code, 200)
for label in (
"Review PR",
"Work issue",
"Create issue",
"Comment on issue",
"Post-merge cleanup",
"Reconciliation audit",
"Project onboarding",
):
self.assertIn(label, response.text)
self.assertIn("Copy prompt", response.text)
self.assertIn("sha256:", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_prompt_detail_route(self):
response = self.client.get("/prompts/work-issue")
self.assertEqual(response.status_code, 200)
self.assertIn("work-issue.md", response.text)
self.assertIn("Copy prompt", response.text)
def test_prompt_detail_404(self):
self.assertEqual(self.client.get("/prompts/missing").status_code, 404)
def test_api_prompts_json(self):
response = self.client.get("/api/prompts")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["count"], 7)
review = next(item for item in data["prompts"] if item["slug"] == "review-pr")
self.assertIn("workflow_hash", review)
self.assertIn("review-merge-pr.md", review["workflow_path"])
def test_prompt_to_dict_roundtrip(self):
entry = load_prompt_library()[0]
encoded = json.dumps(prompt_to_dict(entry))
self.assertIn("workflow_path", encoded)
if __name__ == "__main__":
unittest.main()
tests/test_webui_queue_dashboard.py
+288
View File
@@ -0,0 +1,288 @@
"""Tests for web UI live queue dashboard (#429)."""
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.queue_loader import (
PaginationMeta,
QueueSnapshot,
_classify_issue,
_classify_pr,
_extract_linked_issue,
load_queue_snapshot,
snapshot_to_dict,
)
from webui.queue_views import render_queue_page
_RECENT = datetime.now(timezone.utc).isoformat()
_STALE = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
_SAMPLE_PRS = [
{
"number": 100,
"title": "feat: queue dashboard (Closes #429)",
"body": "",
"mergeable": True,
"updated_at": _RECENT,
"head": {"ref": "feat/issue-429", "sha": "abc123def456"},
"base": {"ref": "master"},
},
{
"number": 99,
"title": "fix: conflict",
"body": "Closes #50",
"mergeable": False,
"updated_at": _STALE,
"head": {"ref": "fix/issue-50", "sha": "deadbeef0001"},
"base": {"ref": "master"},
},
{
"number": 98,
"title": "feat: duplicate A (Closes #60)",
"body": "",
"mergeable": True,
"updated_at": _RECENT,
"head": {"ref": "feat/issue-60-a", "sha": "111111111111"},
"base": {"ref": "master"},
},
{
"number": 97,
"title": "feat: duplicate B (Closes #60)",
"body": "",
"mergeable": True,
"updated_at": _RECENT,
"head": {"ref": "feat/issue-60-b", "sha": "222222222222"},
"base": {"ref": "master"},
},
]
_SAMPLE_ISSUES = [
{
"number": 429,
"title": "Web UI queue dashboard",
"state": "open",
"labels": [{"name": "status:in-progress"}],
"assignee": None,
"updated_at": _RECENT,
},
{
"number": 60,
"title": "Duplicate PR target",
"state": "open",
"labels": [],
"assignee": None,
"updated_at": _RECENT,
},
{
"number": 50,
"title": "Blocked PR target",
"state": "open",
"labels": [],
"assignee": None,
"updated_at": _STALE,
},
]
def _mock_pagination(count: int) -> PaginationMeta:
return PaginationMeta(
page=1,
per_page=50,
returned_count=count,
has_more=False,
is_final_page=True,
inventory_complete=True,
pages_fetched=1,
)
def _mock_fetch_prs(*_args, **_kwargs):
return list(_SAMPLE_PRS), _mock_pagination(len(_SAMPLE_PRS))
def _mock_fetch_issues(*_args, **_kwargs):
return list(_SAMPLE_ISSUES), _mock_pagination(len(_SAMPLE_ISSUES))
class TestQueueClassification(unittest.TestCase):
def test_extract_linked_issue_from_title(self):
self.assertEqual(
_extract_linked_issue("feat: X (Closes #429)", ""),
429,
)
def test_classify_pr_blocked_and_stale(self):
pr = _SAMPLE_PRS[1]
badges = _classify_pr(
pr,
issue_claimed=set(),
issue_to_prs={50: [99]},
)
self.assertIn("blocked", badges)
self.assertIn("stale", badges)
def test_classify_pr_duplicate(self):
badges = _classify_pr(
_SAMPLE_PRS[2],
issue_claimed=set(),
issue_to_prs={60: [98, 97]},
)
self.assertIn("duplicate", badges)
self.assertIn("in-review", badges)
def test_classify_issue_claimed_and_duplicate(self):
claimed = _classify_issue(_SAMPLE_ISSUES[0], linked_prs=[])
self.assertIn("claimed", claimed)
duplicate = _classify_issue(_SAMPLE_ISSUES[1], linked_prs=[98, 97])
self.assertIn("duplicate", duplicate)
class TestQueueLoader(unittest.TestCase):
def test_snapshot_with_mock_fetch(self):
snapshot = load_queue_snapshot(
fetch_prs=_mock_fetch_prs,
fetch_issues=_mock_fetch_issues,
)
self.assertEqual(snapshot.project_id, "gitea-tools")
self.assertIsNone(snapshot.fetch_error)
self.assertEqual(len(snapshot.prs), 4)
self.assertEqual(len(snapshot.issues), 3)
self.assertTrue(snapshot.pr_pagination.inventory_complete)
self.assertTrue(snapshot.issue_pagination.inventory_complete)
pr100 = next(p for p in snapshot.prs if p.number == 100)
self.assertEqual(pr100.extra["linked_issue"], "429")
self.assertIn("claimed", pr100.badges)
def test_snapshot_dict_export(self):
snapshot = load_queue_snapshot(
fetch_prs=_mock_fetch_prs,
fetch_issues=_mock_fetch_issues,
)
data = snapshot_to_dict(snapshot)
self.assertEqual(data["project_id"], "gitea-tools")
self.assertIsNone(data["fetch_error"])
self.assertEqual(len(data["prs"]), 4)
self.assertTrue(data["pagination"]["prs"]["inventory_complete"])
class TestQueueRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
self._patch = mock.patch(
"webui.app.load_queue_snapshot",
return_value=load_queue_snapshot(
fetch_prs=_mock_fetch_prs,
fetch_issues=_mock_fetch_issues,
),
)
self._patch.start()
def tearDown(self):
self._patch.stop()
def test_queue_page_renders_tables_and_pagination(self):
response = self.client.get("/queue")
self.assertEqual(response.status_code, 200)
self.assertIn("Live queue", response.text)
self.assertIn("Open pull requests", response.text)
self.assertIn("Open issues", response.text)
self.assertIn("pages_fetched", response.text)
self.assertIn("Gitea-Tools", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_api_queue_json(self):
response = self.client.get("/api/queue")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["repo"], "Scaled-Tech-Consulting/Gitea-Tools")
self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
def test_queue_fail_closed_without_credentials(self):
with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
snapshot = load_queue_snapshot()
self.assertIsNotNone(snapshot.fetch_error)
self.assertEqual(len(snapshot.prs), 0)
def _empty_pagination() -> PaginationMeta:
return PaginationMeta(
page=1,
per_page=50,
returned_count=0,
has_more=False,
is_final_page=True,
inventory_complete=True,
pages_fetched=1,
)
def _empty_fetch(*_args, **_kwargs):
return [], _empty_pagination()
class TestQueueFailClosedUx(unittest.TestCase):
"""Regression tests for #458 fail-closed empty-state copy."""
def test_fail_closed_view_suppresses_empty_queue_copy(self):
snapshot = QueueSnapshot(
project_id="gitea-tools",
repo_label="Scaled-Tech-Consulting/Gitea-Tools",
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
)
html = render_queue_page(snapshot)
self.assertIn("Queue unavailable", html)
self.assertIn("Not loaded", html)
self.assertNotIn("No open items.", html)
self.assertIn("pagination:</strong> unavailable", html)
def test_fail_closed_route_does_not_show_empty_queue_copy(self):
client = TestClient(create_app())
snapshot = load_queue_snapshot(
fetch_prs=_empty_fetch,
fetch_issues=_empty_fetch,
)
snapshot = QueueSnapshot(
project_id=snapshot.project_id,
repo_label=snapshot.repo_label,
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
)
with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
response = client.get("/queue")
self.assertEqual(response.status_code, 200)
self.assertIn("Queue unavailable", response.text)
self.assertNotIn("No open items.", response.text)
def test_successful_empty_inventory_shows_empty_copy(self):
snapshot = load_queue_snapshot(
fetch_prs=_empty_fetch,
fetch_issues=_empty_fetch,
)
self.assertIsNone(snapshot.fetch_error)
self.assertEqual(len(snapshot.prs), 0)
self.assertEqual(len(snapshot.issues), 0)
self.assertTrue(snapshot.pr_pagination.inventory_complete)
html = render_queue_page(snapshot)
self.assertNotIn("Queue unavailable", html)
self.assertEqual(html.count("No open items."), 2)
self.assertIn("pagination (complete)", html)
if __name__ == "__main__":
unittest.main()
tests/test_webui_skeleton.py
+73
View File
@@ -0,0 +1,73 @@
"""Tests for internal web UI skeleton (#426)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
class TestWebuiSkeleton(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_health_returns_json(self):
response = self.client.get("/health")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["status"], "ok")
self.assertEqual(data["service"], "mcp-control-plane-webui")
self.assertEqual(data["mode"], "read-only-mvp")
self.assertIn("timestamp", data)
def test_home_renders(self):
response = self.client.get("/")
self.assertEqual(response.status_code, 200)
self.assertIn("Operator console", response.text)
self.assertIn("Read-only MVP", response.text)
def test_route_stubs_render(self):
for path in ("/runtime", "/audit"):
with self.subTest(path=path):
response = self.client.get(path)
self.assertEqual(response.status_code, 200)
self.assertIn("child issue", response.text.lower())
def test_prompts_is_implemented(self):
response = self.client.get("/prompts")
self.assertEqual(response.status_code, 200)
self.assertIn("Prompt library", response.text)
def test_projects_is_implemented(self):
response = self.client.get("/projects")
self.assertEqual(response.status_code, 200)
self.assertIn("Gitea-Tools", response.text)
def test_extra_stub_routes(self):
for path in ("/worktrees", "/leases"):
with self.subTest(path=path):
self.assertEqual(self.client.get(path).status_code, 200)
def test_post_is_rejected(self):
response = self.client.post("/health")
self.assertEqual(response.status_code, 405)
self.assertEqual(response.json()["error"], "read-only-mvp")
def test_queue_route_renders(self):
response = self.client.get("/queue")
self.assertEqual(response.status_code, 200)
self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
with self.subTest(path=path):
text = self.client.get(path).text
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
self.assertIn(f'href="{href}"', text)
if __name__ == "__main__":
unittest.main()