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

Merge remote-tracking branch 'prgs/master' into feat/issue-337-create-issue-verifier

Browse Source
This commit is contained in:
sysadmin
2026-07-07 04:44:03 -04:00
parent 14f5c865b3 fd6f7ff22c
commit 6e024db9eb
11 changed files with 1031 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_health.py
+93 -5
View File
@@ -1,9 +1,12 @@
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
import capability_stop_terminal
import role_session_router
from role_session_router import python_bytes_have_conflict_markers
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
@@ -117,7 +120,7 @@ class TestMCPHealth(unittest.TestCase):
stderr_text,
)
@patch("role_session_router.check_mid_merge", return_value=True)
@patch("role_session_router.assess_infra_stop")
@patch(
"mcp_server.get_profile",
return_value={
@@ -125,12 +128,23 @@ class TestMCPHealth(unittest.TestCase):
"allowed_operations": ["gitea.pr.review"],
},
)
def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
def test_route_task_session_blocks_during_merge(self, mock_profile, mock_assess):
mock_assess.return_value = {
"infra_stop": True,
"project_root": "/repo",
"conflict_file": None,
"mid_merge": True,
"infra_stop_reasons": [
"mid-merge markers detected in /repo/.git/MERGE_HEAD (project_root=/repo)"
],
}
res = gitea_route_task_session("review_pr")
self.assertEqual(res["route_result"], "infra_stop")
self.assertIn("infra_stop", res["message"])
self.assertTrue(res.get("infra_stop_assessment", {}).get("infra_stop"))
mock_assess.assert_called()
@patch("role_session_router.check_mid_merge", return_value=True)
@patch("role_session_router.assess_infra_stop")
@patch(
"mcp_server.get_profile",
return_value={
@@ -139,9 +153,83 @@ class TestMCPHealth(unittest.TestCase):
},
)
def test_resolve_task_capability_blocks_during_merge(
self, mock_profile, mock_check
self, mock_profile, mock_assess
):
mock_assess.return_value = {
"infra_stop": True,
"project_root": "/repo",
"conflict_file": "/repo/gitea_mcp_server.py",
"mid_merge": False,
"infra_stop_reasons": [
"conflict markers detected in /repo/gitea_mcp_server.py (project_root=/repo)"
],
}
res = gitea_resolve_task_capability("review_pr")
self.assertTrue(res["infra_stop"])
self.assertFalse(res["allowed_in_current_session"])
self.assertIn("infra_stop", res["exact_safe_next_action"])
self.assertIn("infra_stop", res["exact_safe_next_action"])
self.assertEqual(
res["infra_stop_assessment"]["conflict_file"],
"/repo/gitea_mcp_server.py",
)
@patch("role_session_router.assess_infra_stop")
@patch("mcp_server._authenticated_username", return_value="reviewer-user")
@patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-reviewer",
"allowed_operations": ["gitea.pr.review", "gitea.pr.approve"],
},
)
def test_resolve_task_capability_clears_stale_terminal_when_infra_clean(
self, mock_profile, _username, mock_assess
):
mock_assess.return_value = {
"infra_stop": False,
"project_root": "/repo",
"conflict_file": None,
"mid_merge": False,
"infra_stop_reasons": [],
}
capability_stop_terminal.enter_from_capability_result({
"requested_task": "review_pr",
"required_role_kind": "reviewer",
"stop_required": True,
"active_profile": "prgs-reviewer",
"active_identity": "reviewer-user",
"exact_safe_next_action": "blocked",
})
self.assertTrue(capability_stop_terminal.is_active())
res = gitea_resolve_task_capability("review_pr", remote="prgs")
self.assertFalse(capability_stop_terminal.is_active())
self.assertTrue(res["allowed_in_current_session"])
class TestInfraStopLiveAssessment(unittest.TestCase):
def setUp(self):
self.tempdir = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.tempdir, ignore_errors=True)
def test_clean_project_root_is_not_infra_stop(self):
infra = role_session_router.assess_infra_stop(self.tempdir)
self.assertFalse(infra["infra_stop"])
self.assertIsNone(infra["conflict_file"])
def test_conflict_present_then_resolved_recomputes_allowed(self):
conflict_path = os.path.join(self.tempdir, "conflicted.py")
with open(conflict_path, "wb") as handle:
handle.write(b"<<<<<<< HEAD\n")
blocked = role_session_router.assess_infra_stop(self.tempdir)
self.assertTrue(blocked["infra_stop"])
self.assertEqual(
os.path.realpath(blocked["conflict_file"]),
os.path.realpath(conflict_path),
)
os.remove(conflict_path)
cleared = role_session_router.assess_infra_stop(self.tempdir)
self.assertFalse(cleared["infra_stop"])
self.assertIsNone(cleared["conflict_file"])
tests/test_mcp_server.py
+55
View File
@@ -24,6 +24,7 @@ from mcp_server import ( # noqa: E402
gitea_merge_pr,
gitea_review_pr,
gitea_delete_branch,
gitea_reconcile_merged_cleanups,
gitea_edit_pr,
gitea_get_file,
gitea_commit_files,
@@ -990,6 +991,60 @@ class TestDeleteBranch(unittest.TestCase):
self.assertIn("feat%2Fbranch", url)
# ---------------------------------------------------------------------------
# Reconcile merged cleanups (#269)
# ---------------------------------------------------------------------------
READ_ENV = {
"GITEA_ALLOWED_OPERATIONS": "gitea.read",
}
class TestReconcileMergedCleanups(unittest.TestCase):
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_dry_run_builds_report_without_mutations(self, _auth, mock_api_get_all):
mock_api_get_all.side_effect = [
[
{
"number": 50,
"title": "merged feature",
"body": "Closes #50",
"merged": True,
"merged_at": "2026-07-06T12:00:00Z",
"merge_commit_sha": "deadbeef",
"head": {"ref": "feat/issue-50-example", "sha": "cafebabe"},
}
],
[],
]
with patch.dict(os.environ, READ_ENV, clear=True):
with patch(
"mcp_server._remote_branch_exists",
return_value=True,
):
with patch(
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
return_value=True,
):
result = gitea_reconcile_merged_cleanups(
dry_run=True,
remote="prgs",
limit=10,
)
self.assertTrue(result["success"])
self.assertTrue(result["dry_run"])
self.assertFalse(result["executed"])
self.assertEqual(result["merged_pr_count"], 1)
self.assertEqual(result["entries"][0]["issue_number"], 50)
def test_execute_requires_confirmation(self):
with patch.dict(os.environ, READ_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_reconcile_merged_cleanups(dry_run=False, execute_confirmed=False)
self.assertIn("execute_confirmed", str(ctx.exception))
# ---------------------------------------------------------------------------
# Edit PR
# ---------------------------------------------------------------------------
tests/test_merged_cleanup_reconcile.py
+159
View File
@@ -0,0 +1,159 @@
"""Tests for merged PR cleanup reconciliation (#269)."""
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import merged_cleanup_reconcile as mcr # noqa: E402
class TestMergedCleanupAssessment(unittest.TestCase):
def test_extract_linked_issue_from_closes(self):
issue = mcr.extract_linked_issue(
"feat: cleanup (Closes #269)",
"No other refs",
)
self.assertEqual(issue, 269)
def test_merged_remote_branch_safe_when_gates_pass(self):
result = mcr.assess_remote_branch_cleanup(
pr_number=42,
head_branch="feat/issue-269-merged-pr-cleanup-reconcile",
merged=True,
remote_branch_exists=True,
open_pr_heads=set(),
head_on_master=True,
delete_capability_allowed=True,
active_lock=False,
)
self.assertTrue(result["safe_to_delete_remote"])
self.assertEqual(result["block_reasons"], [])
def test_open_pr_blocks_remote_delete(self):
result = mcr.assess_remote_branch_cleanup(
pr_number=42,
head_branch="feat/issue-9-example",
merged=True,
remote_branch_exists=True,
open_pr_heads={"feat/issue-9-example"},
head_on_master=True,
delete_capability_allowed=True,
active_lock=False,
)
self.assertFalse(result["safe_to_delete_remote"])
self.assertIn("open PR", result["block_reasons"][0])
def test_protected_branch_blocks_remote_delete(self):
result = mcr.assess_remote_branch_cleanup(
pr_number=1,
head_branch="master",
merged=True,
remote_branch_exists=True,
open_pr_heads=set(),
head_on_master=True,
delete_capability_allowed=True,
active_lock=False,
)
self.assertFalse(result["safe_to_delete_remote"])
self.assertIn("protected", result["block_reasons"][0])
def test_active_lock_blocks_remote_and_worktree_cleanup(self):
remote = mcr.assess_remote_branch_cleanup(
pr_number=42,
head_branch="feat/issue-9-example",
merged=True,
remote_branch_exists=True,
open_pr_heads=set(),
head_on_master=True,
delete_capability_allowed=True,
active_lock=True,
)
local = mcr.assess_local_worktree_cleanup(
pr_number=42,
head_branch="feat/issue-9-example",
merged=True,
worktree_state={
"exists": True,
"clean": True,
"current_branch": "feat/issue-9-example",
"worktree_path": "/repo/branches/feat-issue-9-example",
},
active_lock=True,
)
self.assertFalse(remote["safe_to_delete_remote"])
self.assertFalse(local["safe_to_remove_worktree"])
self.assertIn("active issue lock", remote["block_reasons"][0])
def test_dirty_worktree_blocks_local_cleanup(self):
result = mcr.assess_local_worktree_cleanup(
pr_number=42,
head_branch="feat/issue-9-example",
merged=True,
worktree_state={
"exists": True,
"clean": False,
"dirty_files": ["gitea_mcp_server.py"],
"current_branch": "feat/issue-9-example",
"worktree_path": "/repo/branches/feat-issue-9-example",
},
active_lock=False,
)
self.assertFalse(result["safe_to_remove_worktree"])
self.assertIn("tracked edits", result["block_reasons"][0])
class TestMergedCleanupReport(unittest.TestCase):
def test_build_report_skips_unmerged_closed_prs(self):
with tempfile.TemporaryDirectory() as tmp:
report = mcr.build_reconciliation_report(
project_root=tmp,
closed_prs=[
{
"number": 10,
"title": "closed not merged",
"body": "Closes #10",
"head": {"ref": "feat/issue-10-x"},
"merged_at": None,
},
{
"number": 11,
"title": "merged",
"body": "Closes #11",
"head": {"ref": "feat/issue-11-x"},
"merged_at": "2026-07-06T12:00:00Z",
"merge_commit_sha": "abc123",
},
],
open_prs=[],
remote_branch_exists={"feat/issue-11-x": True},
head_on_master={11: True},
delete_capability_allowed=False,
)
self.assertEqual(report["merged_pr_count"], 1)
entry = report["entries"][0]
self.assertEqual(entry["pr_number"], 11)
self.assertEqual(entry["issue_number"], 11)
self.assertFalse(entry["remote_branch"]["safe_to_delete_remote"])
self.assertIn("delete_branch capability", entry["remote_branch"]["block_reasons"][0])
def test_has_active_issue_lock_reads_lock_file(self):
with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False) as handle:
handle.write('{"branch_name": "feat/issue-9-example"}')
lock_path = handle.name
try:
self.assertTrue(
mcr.has_active_issue_lock("feat/issue-9-example", lock_path=lock_path)
)
self.assertFalse(
mcr.has_active_issue_lock("feat/other-branch", lock_path=lock_path)
)
finally:
os.remove(lock_path)
if __name__ == "__main__":
unittest.main()
tests/test_operator_guide.py
+2 -1
View File
@@ -134,7 +134,8 @@ class TestControlPlaneGuide(GuideTestBase):
"merge_confirmation", "redaction", "separation",
"profile_switching", "identity_verification",
"work_selection", "global_worktree",
"shell_spawn_hard_stop"):
"shell_spawn_hard_stop", "subagent_tool_budget",
"subagent_delegation"):
self.assertIn(key, rules)
self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"]))
self.assertTrue(rules["hard_stops"])
tests/test_subagent_tool_budget_docs.py
+89
View File
@@ -0,0 +1,89 @@
"""Doc-contract checks for subagent tool-budget guardrails (#259).
Single-step MCP tasks must not expand into 100+ tool-call retry spirals with
WebFetch/Playwright/manual-encoding fallbacks. These tests pin budget defaults,
native-MCP-first rules, and forbidden detours in the runbooks, portable skill,
and operator guide.
"""
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
SKILL = REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md"
def _normalize(text: str) -> str:
"""Collapse whitespace so phrase checks survive markdown line wrapping."""
return " ".join(text.split())
def _runbook_text() -> str:
return _normalize(RUNBOOK.read_text(encoding="utf-8"))
def _skill_text() -> str:
return _normalize(SKILL.read_text(encoding="utf-8"))
def test_runbook_has_subagent_tool_budget_section():
text = _runbook_text()
assert "## Subagent Tool-Budget Guardrails" in text
for phrase in (
"15",
"5 minutes",
"40",
"15 minutes",
"gitea_commit_files",
"WebFetch",
"Playwright",
"no retry spirals",
):
assert phrase.lower() in text.lower(), f"runbook missing phrase: {phrase!r}"
def test_runbook_forbids_subagent_commit_delegation():
text = _runbook_text()
for phrase in (
"do not delegate commit authority to a subagent",
"main session first",
"native MCP before fallback",
):
assert phrase.lower() in text.lower(), f"runbook missing: {phrase!r}"
def test_runbook_rejects_fallback_detours_when_mcp_commit_available():
lower = _runbook_text().lower()
for forbidden in ("webfetch", "playwright", "manual llm-generated base64"):
assert forbidden in lower, f"runbook must name forbidden detour: {forbidden!r}"
assert "_encode_" in lower
assert "gitea_commit_files" in lower
def test_skill_doc_declares_subagent_tool_budget_guardrails():
text = _skill_text()
assert "## Subagent Tool-Budget Guardrails" in text
for phrase in (
"15 tool calls",
"5 minutes",
"gitea_commit_files",
"WebFetch",
"never resume a failed subagent",
):
assert phrase.lower() in text.lower(), f"SKILL.md missing phrase: {phrase!r}"
def test_operator_guide_rules_include_subagent_tool_budget():
import sys
sys.path.insert(0, str(REPO_ROOT))
import gitea_mcp_server
rule = gitea_mcp_server._GUIDE_RULES["subagent_tool_budget"]
for phrase in (
"15 tool calls",
"gitea_commit_files",
"WebFetch",
"retry loop",
):
assert phrase in rule, f"operator guide rule missing: {phrase!r}"