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

revert: remove #208 from master (land via PR #235)

Browse Source
This commit is contained in:
jcwalker3
2026-07-06 10:19:26 -05:00
parent 8fa94a07a8
commit 056a232ef8
3 changed files with 31 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mcp_server.py
+13 -49
View File
@@ -65,20 +65,6 @@ CREATE_PR_ENV = {
),
}
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
record = {
"issue_number": issue_number,
"branch_name": branch_name,
"remote": "dadeschools",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
}
record.update(overrides)
return record
# ---------------------------------------------------------------------------
# Create Issue
@@ -141,19 +127,15 @@ class TestCreatePR(unittest.TestCase):
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
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_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
payload = mock_api.call_args[0][3]
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
self.assertIn("Closes #123", payload["title"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@@ -162,28 +144,13 @@ class TestCreatePR(unittest.TestCase):
@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):
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_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertIn("pulls/3", result["url"])
@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)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
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
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
self.assertIn("Closes #123", str(ctx.exception))
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
# ---------------------------------------------------------------------------
# Close Issue
@@ -2803,8 +2770,8 @@ class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
def tearDown(self):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2812,7 +2779,7 @@ class TestIssueLocking(unittest.TestCase):
mock_api.return_value = [] # no open PRs
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
def test_lock_issue_mismatch_branch_fails(self):
with self.assertRaises(ValueError) as ctx:
@@ -2849,8 +2816,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_lock_fails(self, _auth, _role):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
@@ -2860,9 +2827,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
@@ -2872,9 +2838,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
for term in ("equivalent to #196", "related to #196", "same as #196"):
with self.assertRaises(ValueError) as ctx:
@@ -2885,9 +2850,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
tests/test_review_proofs.py
+18 -124
View File
@@ -27,7 +27,6 @@ from review_proofs import ( # noqa: E402
assess_controller_handoff,
assess_inventory_completeness,
assess_reviewer_queue_inventory,
assess_empty_queue_report,
assess_live_state_recheck,
assess_review_mutation_final_report,
assess_role_boundary,
@@ -945,46 +944,24 @@ class TestControllerHandoff(unittest.TestCase):
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
def _author_role_fields(self, issue_number=182, pr_number=999):
return [
f"- Selected issue: #{issue_number}",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
f"- PR number opened: #{pr_number}",
"- No review/merge: confirmed",
]
def test_handoff_role_fields_author_includes_issue_lock_proof(self):
from review_proofs import HANDOFF_ROLE_FIELDS
names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
self.assertIn("Issue lock proof", names)
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_requires_issue_lock_proof(self):
without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #182",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
result = assess_controller_handoff(without_lock, role="author")
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_rejects_equivalent_or_multiple_issues(self):
# 1. equivalent reference blocked
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: Issue #194 / #196 equivalent",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -996,7 +973,6 @@ class TestControllerHandoff(unittest.TestCase):
# 2. multiple issues blocked
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #194, #196",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -1008,7 +984,6 @@ class TestControllerHandoff(unittest.TestCase):
def test_author_role_rejects_fuzzy_pr_number(self):
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: PR #203 / #204 equivalent",
"- No review/merge: confirmed",
@@ -1021,10 +996,6 @@ class TestControllerHandoff(unittest.TestCase):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Repositories checked: Gitea-Tools, mcp-control-plane",
"- Open PR counts: 2 / 0",
"- PR inventory trust gate: trusted_nonempty / trusted_empty",
"- Trust gate reasons: none",
"- Trust gate corroborated: true",
"- Inventory profile: prgs-reviewer",
"- Selected PR or reason: none eligible (self-authored)",
"- Inventory completeness: complete, no pagination needed",
])
@@ -1044,17 +1015,23 @@ class TestControllerHandoff(unittest.TestCase):
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
# 1. Workspace mutations: none is rejected when local_edits is True
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
self._author_role_fields(issue_number=196, pr_number=203))
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Workspace mutations", res["missing_fields"])
# 2. Workspace mutations: edited files is allowed when local_edits is True
complete_eq = self.BASE_HANDOFF.replace(
"- Workspace mutations: none",
"- Workspace mutations: edited review_proofs.py",
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
@@ -1285,89 +1262,6 @@ class TestAssessReviewerQueueInventory(unittest.TestCase):
self.assertEqual(result["trust_gates"], {})
class TestAssessEmptyQueueReport(unittest.TestCase):
"""Issue #198: empty-queue reports require formal trust-gate proof."""
def _trusted_report(self, **extra):
lines = [
"Queue inventory complete.",
"Repository: Scaled-Tech-Consulting/Gitea-Tools",
"Open PR count: 0",
"pr_inventory_trust_gate.status: trusted_empty",
"pr_inventory_trust_gate.corroborated: true",
"Inventory profile: prgs-reviewer",
"Workflow correctly stops with nothing to review.",
]
lines.extend(extra)
return "\n".join(lines)
def test_non_empty_report_not_claimed(self):
result = assess_empty_queue_report("Reviewed PR #236 and merged.")
self.assertFalse(result["claimed"])
self.assertTrue(result["proven"])
def test_empty_claim_without_trust_gate_blocked(self):
report = (
"Open PR count: 0\n"
"Pagination complete: yes\n"
"Queue cleared."
)
result = assess_empty_queue_report(report)
self.assertTrue(result["claimed"])
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_trusted_empty_report_with_required_fields_passes(self):
result = assess_empty_queue_report(self._trusted_report())
self.assertTrue(result["proven"])
def test_weak_merge_commit_corroboration_blocked(self):
report = (
"Open PR count: 0\n"
"Master latest commit is merge of PR #79 so queue is empty."
)
result = assess_empty_queue_report(report)
self.assertFalse(result["proven"])
self.assertTrue(
any("weak corroboration" in r for r in result["reasons"])
)
def test_author_session_reviewer_queue_wording_blocked(self):
result = assess_empty_queue_report(
self._trusted_report(),
task_role="author",
)
self.assertFalse(result["proven"])
def test_build_final_report_downgrades_weak_empty_queue(self):
final = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
review_mutation=_good_review_mutation(),
controller_handoff=_good_handoff(),
capability_proof=_good_capability_proof(),
sweep_proof=_good_secret_sweep(),
worktree_proof={
"worktree_path": "/repo/branches/review-pr-1",
"porcelain_status": "",
"scratch_used": True,
"scratch_path": "/repo/branches/review-pr-1",
},
report_text="Open PR count: 0. Queue cleared.",
)
self.assertNotEqual(final["grade"], "A")
self.assertFalse(final["empty_queue_trust_gate_proven"])
class TestCapabilityEvidence(unittest.TestCase):
"""#179 gap 1: capability claims need exact evidence."""