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

fix: resolve conflicts for PR #397

Browse Source 
Merge prgs/master into feat/issue-395-proof-backed-review-handoff; keep
both proof-backed handoff (#395) and already-landed classification (#295)
downgrade gates in review_proofs.py.
This commit is contained in:
sysadmin
2026-07-07 11:21:22 -04:00
parent c1d85b621a fcb9944378
commit 6220304f8c
22 changed files with 2584 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_author_mutation_worktree.py
+109
View File
@@ -0,0 +1,109 @@
"""Tests for branches-only author mutation worktree guard (#274)."""
from __future__ import annotations
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import author_mutation_worktree as amw # noqa: E402
class TestPathUnderBranches(unittest.TestCase):
ROOT = "/repo/Gitea-Tools"
def test_branches_subdirectory_passes(self):
self.assertTrue(
amw.is_path_under_branches(f"{self.ROOT}/branches/issue-274", self.ROOT)
)
def test_control_checkout_fails(self):
self.assertFalse(amw.is_path_under_branches(self.ROOT, self.ROOT))
def test_sibling_directory_fails(self):
self.assertFalse(
amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
)
class TestAssessAuthorMutationWorktree(unittest.TestCase):
ROOT = "/repo/Gitea-Tools"
def test_branches_worktree_passes(self):
result = amw.assess_author_mutation_worktree(
workspace_path=f"{self.ROOT}/branches/issue-274",
project_root=self.ROOT,
current_branch="feat/issue-274-branches-only-worktrees",
)
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_control_checkout_blocks(self):
result = amw.assess_author_mutation_worktree(
workspace_path=self.ROOT,
project_root=self.ROOT,
current_branch="master",
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertIn("control checkout", result["reasons"][0])
def test_drifted_control_branch_blocks(self):
result = amw.assess_author_mutation_worktree(
workspace_path=self.ROOT,
project_root=self.ROOT,
current_branch="feat/some-task",
)
self.assertTrue(result["block"])
self.assertTrue(any("drift" in r for r in result["reasons"]))
def test_branches_worktree_as_project_root_passes(self):
"""MCP launched from a branches/ worktree uses that path as PROJECT_ROOT."""
worktree_root = f"{self.ROOT}/branches/issue-274"
result = amw.assess_author_mutation_worktree(
workspace_path=worktree_root,
project_root=worktree_root,
current_branch="feat/issue-274-branches-only-worktrees",
)
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
class TestPreflightIntegration(unittest.TestCase):
def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self):
import mcp_server
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "author"
control_root = "/repo/Gitea-Tools"
with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root):
with mock.patch.dict(
"os.environ",
{"GITEA_TEST_PORCELAIN": ""},
clear=False,
):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Branches-only mutation guard", str(ctx.exception))
def test_verify_preflight_allows_branches_worktree(self):
import mcp_server
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "author"
worktree = "/repo/Gitea-Tools/branches/issue-274"
with mock.patch.dict(
"os.environ",
{"GITEA_TEST_PORCELAIN": ""},
clear=False,
):
mcp_server.verify_preflight_purity(worktree_path=worktree)
if __name__ == "__main__":
unittest.main()
tests/test_commit_payloads.py
+10 -1
View File
@@ -55,7 +55,15 @@ class TestCommitPayloads(unittest.TestCase):
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
self.locked_worktree_dir = tempfile.TemporaryDirectory()
repo_root = os.path.realpath(
os.path.join(os.path.dirname(__file__), os.pardir)
)
branches_root = os.path.join(repo_root, "branches")
os.makedirs(branches_root, exist_ok=True)
self.locked_worktree_dir = tempfile.TemporaryDirectory(
dir=branches_root,
prefix="test-commit-payloads-",
)
self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
self.lock_file_path = "/tmp/gitea_issue_lock.json"
@@ -94,6 +102,7 @@ class TestCommitPayloads(unittest.TestCase):
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TEST_PORCELAIN": "",
"GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
}
@patch("mcp_server.api_request")
tests/test_final_report_validator.py
+194 -7
View File
@@ -57,26 +57,47 @@ def _review_handoff(**overrides):
return text
def _reconcile_handoff(**overrides):
def _reconcile_handoff(drop=(), **overrides):
lines = [
"## Controller Handoff",
"",
"- Task: reconcile already-landed PR #99",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: reconciler",
"- Identity: sysadmin / prgs-author",
"- Role/profile: reconciler / prgs-reconciler",
"- Identity: sysadmin",
"- Selected PR: #99",
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
"- PR live state: open",
"- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Target branch: master",
"- Target branch SHA: 5e023dc71b0e2b813a0b1eee6b0f42630c47a95c",
"- Ancestor proof: candidate head is ancestor of target branch SHA",
"- Linked issue: #98",
"- Linked issue live status: not verified in this session",
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
"- Capabilities proven: gitea.read, gitea.pr.comment",
"- Missing capabilities: gitea.pr.close",
"- PR comments posted: 1 reconciliation comment on PR #99",
"- Issue comments posted: none",
"- PRs closed: none",
"- Issues closed: none",
"- File edits by reconciler: none",
"- Git ref mutations: git fetch prgs master",
"- Worktree mutations: none",
"- MCP/Gitea mutations: PR comment posted",
"- External-state mutations: none",
"- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
"- Reconciliation mutations: PR comment posted",
"- Current status: reconciliation complete",
"- Safe next action: none",
"- Blocker: missing gitea.pr.close capability",
"- Safe next action: hand off to a profile with gitea.pr.close",
"- No review/merge confirmation: confirmed",
]
text = "\n".join(lines)
kept = [
line
for line in lines
if not any(line.startswith(f"- {name}:") for name in drop)
]
text = "\n".join(kept)
for key, value in overrides.items():
if f"- {key}:" in text:
text = text.replace(f"- {key}:", f"- {key}: {value}")
@@ -194,6 +215,20 @@ class TestReviewPrRules(unittest.TestCase):
)
)
def test_already_landed_oldest_eligible_pr_blocks(self):
report = (
_review_handoff()
+ "\nAlready landed.\nOldest eligible PR: PR #278."
)
result = assess_final_report_validator(report, "review_pr")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reviewer.already_landed_eligible_wording"
for f in result["findings"]
)
)
def test_already_landed_gate_blocks_review_terminal_states(self):
cases = {
"APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
@@ -242,7 +277,6 @@ class TestReviewPrRules(unittest.TestCase):
)
result = assess_final_report_validator(report, "review_pr")
self.assertEqual(result["grade"], "A")
def test_git_fetch_must_not_be_readonly_only(self):
report = (
_review_handoff()
@@ -307,6 +341,159 @@ class TestReconciliationRules(unittest.TestCase):
)
class TestCanonicalReconcileSchema(unittest.TestCase):
"""Issue #307: reconciliation workflows emit only the canonical schema."""
def test_comment_only_reconciliation_passes(self):
result = assess_final_report_validator(
_reconcile_handoff(),
"reconcile_already_landed",
)
self.assertEqual(result["grade"], "A")
self.assertEqual(result["findings"], [])
def test_successful_close_reconciliation_passes(self):
report = _reconcile_handoff().replace(
"- Capabilities proven: gitea.read, gitea.pr.comment",
"- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
).replace(
"- Missing capabilities: gitea.pr.close",
"- Missing capabilities: none",
).replace(
"- PRs closed: none",
"- PRs closed: #99",
).replace(
"- Blocker: missing gitea.pr.close capability",
"- Blocker: none",
)
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "A")
def test_blocked_reconciliation_passes(self):
report = _reconcile_handoff().replace(
"- Capabilities proven: gitea.read, gitea.pr.comment",
"- Capabilities proven: gitea.read",
).replace(
"- Missing capabilities: gitea.pr.close",
"- Missing capabilities: gitea.pr.close, gitea.pr.comment",
).replace(
"- PR comments posted: 1 reconciliation comment on PR #99",
"- PR comments posted: none",
).replace(
"- MCP/Gitea mutations: PR comment posted",
"- MCP/Gitea mutations: none",
).replace(
"- Reconciliation mutations: PR comment posted",
"- Reconciliation mutations: none",
)
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "A")
def test_missing_capabilities_field_required(self):
report = _reconcile_handoff(drop=("Missing capabilities",))
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "downgraded")
self.assertTrue(
any(
f["rule_id"] == "reconcile.controller_handoff"
and "Missing capabilities" in f["reason"]
for f in result["findings"]
)
)
def test_missing_ancestor_proof_and_candidate_sha_downgrade(self):
report = _reconcile_handoff(drop=("Ancestor proof", "Candidate head SHA"))
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "downgraded")
reasons = " ".join(f["reason"] for f in result["findings"])
self.assertIn("Ancestor proof", reasons)
self.assertIn("Candidate head SHA", reasons)
def test_stale_issue_lock_proof_blocks(self):
report = _reconcile_handoff() + "\n- Issue lock proof: locked before diff"
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "issue lock proof"
for f in result["findings"]
)
)
def test_stale_claim_comment_status_blocks(self):
report = _reconcile_handoff() + "\n- Claim/comment status: claimed"
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "claim/comment status"
for f in result["findings"]
)
)
def test_stale_workspace_mutations_blocks_without_observed_mutations(self):
report = _reconcile_handoff() + "\n- Workspace mutations: None"
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "workspace mutations"
for f in result["findings"]
)
)
def test_pr_number_opened_allowed_when_session_opened_pr(self):
report = _reconcile_handoff() + "\n- PR number opened: #12"
result = assess_final_report_validator(
report,
"reconcile_already_landed",
session_pr_opened=True,
)
self.assertFalse(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "pr number opened"
for f in result["findings"]
)
)
def test_pr_number_opened_still_blocked_without_session_proof(self):
report = _reconcile_handoff() + "\n- PR number opened: #12"
result = assess_final_report_validator(
report,
"reconcile_already_landed",
session_pr_opened=False,
)
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "pr number opened"
for f in result["findings"]
)
)
def test_action_log_fetch_requires_git_ref_mutation_entry(self):
report = _reconcile_handoff().replace(
"- Git ref mutations: git fetch prgs master",
"- Git ref mutations: none",
)
result = assess_final_report_validator(
report,
"reconcile_already_landed",
action_log=[{"command": "git fetch prgs master", "performed": True}],
)
self.assertTrue(
any(
f["rule_id"] == "reviewer.git_fetch_readonly"
for f in result["findings"]
)
)
class TestEntryPoint(unittest.TestCase):
def test_unknown_task_kind_blocks(self):
result = assess_final_report_validator("report", "unknown_mode")
tests/test_llm_workflow_split.py
+28
View File
@@ -139,6 +139,16 @@ def test_start_issue_template_references_work_issue_workflow():
assert "workflows/work-issue.md" in text
def test_reconcile_skill_registered_in_mcp_server():
from mcp_server import _PROJECT_SKILLS
assert "gitea-reconcile-landed-pr" in _PROJECT_SKILLS
steps = _PROJECT_SKILLS["gitea-reconcile-landed-pr"]["steps"]
joined = " ".join(steps).lower()
assert "gitea_scan_already_landed_open_prs" in joined
assert "gitea_assess_already_landed_reconciliation" in joined
def test_pr_queue_cleanup_template_references_workflow():
text = (SKILL_DIR / "templates" / "pr-queue-cleanup.md").read_text(
encoding="utf-8"
@@ -165,6 +175,12 @@ def test_final_report_validator_exported():
assert callable(assess_final_report_validator)
def test_review_final_report_schema_verifier_exported():
from review_proofs import assess_review_final_report_schema
assert callable(assess_review_final_report_schema)
def test_create_issue_final_report_verifier_exported():
from review_proofs import assess_create_issue_final_report
@@ -201,6 +217,12 @@ def test_validation_worktree_edit_verifier_exported():
assert callable(assess_validation_worktree_edit_report)
def test_reconcile_inventory_verifier_exported():
from review_proofs import assess_reconcile_inventory_report
assert callable(assess_reconcile_inventory_report)
def test_reconcile_linked_issue_verifier_exported():
from review_proofs import assess_reconcile_linked_issue_report
@@ -243,6 +265,12 @@ def test_worktree_ownership_verifier_exported():
assert callable(assess_worktree_ownership_report)
def test_already_landed_classification_verifier_exported():
from review_proofs import assess_already_landed_classification_report
assert callable(assess_already_landed_classification_report)
def test_pr_queue_cleanup_verifier_exported():
from review_proofs import assess_pr_queue_cleanup_report
tests/test_mcp_server.py
+53
View File
@@ -3050,10 +3050,18 @@ 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.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
self.assertEqual(res["work_lease"]["issue_number"], 196)
self.assertEqual(res["work_lease"]["branch"], "feat/issue-196-mutations")
self.assertIn("created_at", res["work_lease"])
self.assertIn("expires_at", res["work_lease"])
self.assertIn("last_heartbeat_at", res["work_lease"])
self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
lock = json.load(f)
self.assertIn("worktree_path", lock)
self.assertIn("work_lease", lock)
def test_lock_issue_mismatch_branch_fails(self):
with self.assertRaises(ValueError) as ctx:
@@ -3094,6 +3102,51 @@ class TestIssueLocking(unittest.TestCase):
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))
@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 = [
[],
[{"name": "feat/issue-196-existing-work"}],
]
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))
def test_lock_issue_blocks_active_same_operation_lease(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}, f)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already has an active author_issue_work lease", str(ctx.exception))
def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2000-01-01T00:00:00Z",
},
}, f)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("Recovery review is required before takeover", str(ctx.exception))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_from_clean_scratch_worktree(self, _auth, _api):
tests/test_reconciliation_mcp_tools.py
+106
View File
@@ -0,0 +1,106 @@
"""Tests for reconciliation MCP assessment tools (#301)."""
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
from mcp_server import (
gitea_assess_already_landed_reconciliation,
gitea_scan_already_landed_open_prs,
)
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
],
"forbidden_operations": ["gitea.pr.close"],
"audit_label": "prgs-author",
}
OPEN_PR = {
"number": 278,
"title": "Landed (Closes #263)",
"body": "",
"state": "open",
"head": {"ref": "feat/x", "sha": "a" * 40},
"base": {"ref": "master"},
}
class TestReconciliationMcpTools(unittest.TestCase):
def setUp(self):
self.mock_api = patch("mcp_server.api_request").start()
self.mock_auth = patch(
"mcp_server.get_auth_header", return_value="token test"
).start()
patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE).start()
mcp_server._IDENTITY_CACHE.clear()
def tearDown(self):
patch.stopall()
mcp_server._IDENTITY_CACHE.clear()
def test_assess_returns_plan_without_mutations(self):
self.mock_api.return_value = OPEN_PR
with patch(
"mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
return_value={
"reconciliation_allowed": True,
"eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"candidate_head_sha": OPEN_PR["head"]["sha"],
"target_branch_sha": "deadbeef",
"linked_issue": 263,
"review_merge_allowed": False,
},
):
res = gitea_assess_already_landed_reconciliation(
pr_number=278, remote="prgs"
)
self.assertTrue(res["success"])
self.assertFalse(res["performed"])
self.assertEqual(res["plan"]["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
self.assertFalse(res["review_merge_allowed"])
patch_calls = [
c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
]
self.assertEqual(patch_calls, [])
def test_scan_returns_candidates_with_pagination(self):
with patch(
"mcp_server.api_fetch_page",
return_value=([OPEN_PR], {
"page": 1,
"per_page": 50,
"is_final_page": True,
"has_more": False,
"next_page": None,
}),
), patch(
"mcp_server.reconciliation_workflow.fetch_target_branch",
return_value={
"success": True,
"target_branch_sha": "deadbeef",
"git_fetch_command": "git fetch prgs master",
},
), patch(
"mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
return_value={
"pr_number": 278,
"eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"reconciliation_allowed": True,
},
):
res = gitea_scan_already_landed_open_prs(remote="prgs", limit=50)
self.assertTrue(res["success"])
self.assertEqual(res["candidate_count"], 1)
self.assertTrue(res["pagination"]["inventory_complete"])
if __name__ == "__main__":
unittest.main()
tests/test_reconciliation_workflow.py
+149
View File
@@ -0,0 +1,149 @@
"""Tests for already-landed reconciliation workflow helpers (#301)."""
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import reconciliation_workflow
from review_proofs import assess_reconcile_workflow_source
class TestReconciliationAssessment(unittest.TestCase):
def test_already_landed_open_pr(self):
with patch(
"reconciliation_workflow.is_head_ancestor_of_ref",
return_value=True,
):
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr={
"number": 278,
"state": "open",
"title": "Fix (Closes #263)",
"body": "",
"head": {"ref": "feat/x", "sha": "abc" * 13 + "a"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": True,
"target_ref": "prgs/master",
"target_branch_sha": "deadbeef",
"git_fetch_command": "git fetch prgs master",
},
)
self.assertTrue(assessment["reconciliation_allowed"])
self.assertEqual(
assessment["eligibility_class"],
reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
)
def test_not_landed_pr(self):
with patch(
"reconciliation_workflow.is_head_ancestor_of_ref",
return_value=False,
):
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr={
"number": 1,
"state": "open",
"title": "WIP",
"body": "",
"head": {"ref": "feat/y", "sha": "fff" * 13 + "f"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": True,
"target_ref": "prgs/master",
"target_branch_sha": "deadbeef",
},
)
self.assertFalse(assessment["reconciliation_allowed"])
class TestReconciliationPlan(unittest.TestCase):
def test_full_close_when_close_pr_allowed(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": True,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
},
capabilities={
"close_pr": True,
"comment_pr": True,
"close_issue": True,
"comment_issue": True,
},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_FULL_RECONCILE,
)
self.assertTrue(plan["close_pr_allowed"])
self.assertFalse(plan["review_merge_allowed"])
def test_comment_then_stop_without_close(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": True,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
},
capabilities={
"close_pr": False,
"comment_pr": True,
},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_PARTIAL_COMMENT,
)
def test_recovery_handoff_without_comment_or_close(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": True,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
},
capabilities={"close_pr": False, "comment_pr": False},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_RECOVERY_HANDOFF,
)
def test_not_landed_no_action(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": False,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_NOT_LANDED,
"reasons": ["not ancestor"],
},
capabilities={"close_pr": True},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_NOT_LANDED,
)
class TestReconcileWorkflowSource(unittest.TestCase):
def test_valid_report_passes(self):
report = (
"Task mode: reconcile-landed-pr\n"
"Workflow source: workflows/reconcile-landed-pr.md\n"
)
result = assess_reconcile_workflow_source(report)
self.assertTrue(result["complete"])
def test_missing_workflow_fails(self):
result = assess_reconcile_workflow_source("Task mode: reconcile-landed-pr")
self.assertFalse(result["complete"])
if __name__ == "__main__":
unittest.main()
tests/test_review_final_report_schema.py
+158
View File
@@ -0,0 +1,158 @@
"""Tests for reviewer final-report schema verification (#391)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from review_final_report_schema import assess_review_final_report_schema # noqa: E402
def _minimal_review_report(**overrides):
lines = [
"## Controller Handoff",
"",
"- Task: review PR #203",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: reviewer",
"- Identity: sysadmin / prgs-reviewer",
"- Issue/PR: #182 / PR #203",
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Files changed: review_proofs.py",
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
"- Mutations: review only",
"- File edits by reviewer: none",
"- Worktree/index mutations: none",
"- Git ref mutations: git fetch prgs master",
"- MCP/Gitea mutations: review submitted",
"- Review mutations: request_changes submitted",
"- Merge mutations: none",
"- Cleanup mutations: none",
"- External-state mutations: none",
"- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
"- Current status: PR open",
"- Blockers: none",
"- Next: await author",
"- Safety: no self-review; no self-merge",
"- Selected PR: #203",
"- Reviewer eligibility: eligible",
"- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Worktree path: branches/review-203",
"- Worktree dirty: clean",
"- Unrelated local mutations: none",
"- Review decision: request_changes",
"- Merge result: not attempted",
"- Linked issue: #182",
"- Linked issue live status: open (gitea_view_issue)",
"- Cleanup status: none",
]
text = "\n".join(lines)
for key, value in overrides.items():
prefix = f"- {key}:"
replaced = False
new_lines = []
for line in text.splitlines():
if line.strip().startswith(prefix):
new_lines.append(f"{prefix} {value}")
replaced = True
else:
new_lines.append(line)
text = "\n".join(new_lines)
if not replaced:
text += f"\n{prefix} {value}"
return text
class TestReviewFinalReportSchema(unittest.TestCase):
def test_clean_report_passes(self):
result = assess_review_final_report_schema(
_minimal_review_report(),
linked_issue_lock={"issue_number": 182, "pr_number": 203},
)
self.assertFalse(result["blocked"])
self.assertIn(result["grade"], ("A", "downgraded"))
def test_legacy_pinned_reviewed_head_blocks(self):
report = _minimal_review_report(**{"Pinned reviewed head": "abc123"})
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.legacy_stale_field", rule_ids)
def test_reviewed_head_without_validation_blocks(self):
report = _minimal_review_report().replace(
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
"- Validation: not run",
)
report += "\n- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.reviewed_head_without_validation", rule_ids)
def test_merged_without_proof_blocks(self):
report = _minimal_review_report() + "\n\nPR #203 merged successfully."
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.merged_without_proof", rule_ids)
def test_issue_closed_without_live_proof_blocks(self):
report = _minimal_review_report(
**{
"Linked issue live status": "closed",
"Read-only diagnostics": "gitea_view_pr only",
}
)
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.issue_closed_without_live_proof", rule_ids)
def test_full_suite_pass_with_ignored_tests_blocks(self):
report = (
_minimal_review_report()
+ "\nFull test suite passed with 0 failed; some tests ignored."
)
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.full_suite_pass_ignored_tests", rule_ids)
def test_blocked_handoff_replay_blocks(self):
report = (
_minimal_review_report(**{"Blockers": "capability stop"})
+ "\nSafe next action: run gitea_merge_pr to finish."
)
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.blocked_handoff_replay", rule_ids)
def test_narrative_handoff_drift_blocks(self):
report = (
_minimal_review_report()
+ "\n## Summary\nVerdict: APPROVE this PR."
)
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.narrative_handoff_drift", rule_ids)
def test_pending_vs_approved_blocks(self):
report = _minimal_review_report(
**{"Review decision": "PENDING official review submitted approve"}
)
result = assess_review_final_report_schema(report)
rule_ids = [f["rule_id"] for f in result["findings"]]
self.assertIn("reviewer.pending_vs_approved", rule_ids)
def test_exported_from_review_proofs(self):
from review_proofs import assess_review_final_report_schema as exported
self.assertTrue(callable(exported))
class TestMcpValidateReviewFinalReport(unittest.TestCase):
def test_mcp_tool_callable(self):
import gitea_mcp_server
result = gitea_mcp_server.gitea_validate_review_final_report(
_minimal_review_report()
)
self.assertIn("grade", result)
self.assertIn("findings", result)
self.assertFalse(result["blocked"])
tests/test_reviewer_already_landed_classification.py
+81
View File
@@ -0,0 +1,81 @@
import unittest
from reviewer_already_landed_classification import ( # noqa: E402
ALREADY_LANDED_ELIGIBILITY_CLASS,
assess_already_landed_classification_report,
)
class TestAlreadyLandedClassification(unittest.TestCase):
def test_non_landed_report_passes(self):
result = assess_already_landed_classification_report(
"Selected the next eligible PR: PR #286."
)
self.assertTrue(result["proven"])
self.assertFalse(result["landed_context"])
def test_oldest_eligible_pr_on_landed_blocks(self):
result = assess_already_landed_classification_report(
"PR is already landed.\n"
"Oldest eligible PR: PR #278."
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_next_eligible_pr_on_landed_blocks(self):
result = assess_already_landed_classification_report(
"Ancestor proof: passed\n"
"Selected the next eligible PR: PR #278."
)
self.assertFalse(result["proven"])
def test_eligible_for_review_on_landed_blocks(self):
result = assess_already_landed_classification_report(
"Already landed on master and eligible for review next."
)
self.assertFalse(result["proven"])
def test_proper_landed_wording_passes(self):
result = assess_already_landed_classification_report(
"Oldest open PR requiring action: PR #278\n"
f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
"Candidate head SHA: abc123\n"
"Reviewed head SHA: none\n"
"Queue blocked by already-landed PR requiring reconciliation",
selected_pr_already_landed=True,
)
self.assertTrue(result["proven"])
def test_pinned_reviewed_head_without_validation_blocks(self):
result = assess_already_landed_classification_report(
"Already landed.\n"
"Pinned reviewed head: abc123def456",
selected_pr_already_landed=True,
)
self.assertFalse(result["proven"])
def test_wrong_eligibility_class_blocks(self):
result = assess_already_landed_classification_report(
"Already landed.\n"
"Eligibility class: REVIEW_ELIGIBLE",
selected_pr_already_landed=True,
)
self.assertFalse(result["proven"])
def test_missing_queue_blocker_blocks(self):
result = assess_already_landed_classification_report(
f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
"Candidate head SHA: abc123\n"
"Reviewed head SHA: none",
selected_pr_already_landed=True,
)
self.assertFalse(result["proven"])
def test_exported_from_review_proofs(self):
from review_proofs import assess_already_landed_classification_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_reconcile_inventory.py
+102
View File
@@ -0,0 +1,102 @@
"""Tests for reconciliation inventory pagination verifier (#308)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_reconcile_inventory import assess_reconcile_inventory_report # noqa: E402
def _good_report() -> str:
return "\n".join([
"## Already-landed reconciliation",
"Open PR inventory:",
"- Requested page size: 50",
"- Pages fetched: 2",
"- Returned PR count per page: page1=50, page2=6",
"- Inventory pagination proof: is_final_page: true, has_more: false",
"All already-landed PRs in the scanned open queue were found.",
"",
"## Controller Handoff",
"- Selected PR: #278",
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
])
class TestReconcileInventoryPagination(unittest.TestCase):
def test_complete_metadata_passes(self):
result = assess_reconcile_inventory_report(_good_report())
self.assertTrue(result["proven"], result["reasons"])
self.assertTrue(result["pagination_proven"])
def test_no_inventory_claim_skips_checks(self):
report = "## Controller Handoff\n- Selected PR: #278"
result = assess_reconcile_inventory_report(report)
self.assertTrue(result["proven"])
self.assertFalse(result["inventory_claimed"])
def test_first_page_below_limit_with_finality_passes(self):
report = "\n".join([
"Open PRs listed for reconciliation.",
"- Requested page size: 50",
"- Pages fetched: 1",
"- Returned PR count per page: page1=6",
"- Inventory pagination proof: is_final_page: true",
])
result = assess_reconcile_inventory_report(report)
self.assertTrue(result["proven"], result["reasons"])
def test_exactly_full_first_page_without_finality_blocks(self):
report = "\n".join([
"Complete queue scan of open PRs.",
"- Requested page size: 50",
"- Pages fetched: 1",
"- Returned PR count per page: page1=50",
"Listed 50 open PRs on the first page.",
])
result = assess_reconcile_inventory_report(
report,
inventory_session={"requested_page_size": 50, "returned_page_size": 50},
)
self.assertFalse(result["proven"])
self.assertTrue(any("full first" in r.lower() for r in result["reasons"]))
def test_missing_pagination_metadata_blocks(self):
report = "All open PRs were inventoried. Inventory is complete."
result = assess_reconcile_inventory_report(report)
self.assertFalse(result["proven"])
self.assertTrue(any("metadata" in r.lower() for r in result["reasons"]))
def test_default_page_size_assumption_blocks(self):
report = (
"Exhaustive PR inventory: fewer than the default Gitea page size returned."
)
result = assess_reconcile_inventory_report(report)
self.assertFalse(result["proven"])
self.assertTrue(any("page-size" in r.lower() or "page size" in r.lower()
for r in result["reasons"]))
def test_session_pagination_complete_passes(self):
report = "All already-landed PRs were found after scanning open PRs."
result = assess_reconcile_inventory_report(
report,
inventory_session={
"pagination_complete": True,
"requested_page_size": 50,
"pages_fetched": 2,
"returned_per_page": [50, 3],
},
)
self.assertTrue(result["proven"], result["reasons"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_reconcile_inventory_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()