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

Merge remote-tracking branch 'prgs/master' into feat/issue-514-branch-delete-guard

Browse Source 
# Conflicts:
#	gitea_mcp_server.py
This commit is contained in:
sysadmin
2026-07-09 09:35:55 -04:00
parent 7acd55d2d2 cc4b95839d
commit d7c29afde5
44 changed files with 2932 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+8
View File
@@ -32,6 +32,14 @@ def _reset_mutation_authority(monkeypatch):
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)
monkeypatch.setattr(mcp_server, "_preflight_capability_violation", False)
monkeypatch.setattr(mcp_server, "_preflight_capability_violation_files", [])
monkeypatch.setattr(mcp_server, "_preflight_capability_baseline_porcelain", None)
monkeypatch.setattr(mcp_server, "_preflight_resolved_task", None)
monkeypatch.setattr(mcp_server, "_preflight_reviewer_violation_files", [])
try:
import review_workflow_load
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
tests/test_author_mutation_worktree.py
+15 -13
View File
@@ -81,13 +81,14 @@ class TestPreflightIntegration(unittest.TestCase):
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()
with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"):
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):
@@ -97,12 +98,13 @@ class TestPreflightIntegration(unittest.TestCase):
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)
with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"):
with mock.patch.dict(
"os.environ",
{"GITEA_TEST_PORCELAIN": ""},
clear=False,
):
mcp_server.verify_preflight_purity(worktree_path=worktree)
if __name__ == "__main__":
tests/test_final_report_validator.py
+22
View File
@@ -114,6 +114,28 @@ def _reconcile_handoff(drop=(), **overrides):
return text
class TestTwoCommentWorkflowIntegration(unittest.TestCase):
def test_tagged_pair_produces_no_two_comment_findings(self):
from thread_state_ledger_validator import findings_for_final_report # noqa: E402
from tests.test_thread_state_ledger_validator import ( # noqa: E402
_minimal_handoff,
_minimal_ledger,
)
report = _minimal_handoff() + "\n\n" + _minimal_ledger()
self.assertEqual(findings_for_final_report(report), [])
def test_tagged_handoff_without_ledger_blocks_via_validator(self):
from thread_state_ledger_validator import findings_for_final_report # noqa: E402
from tests.test_thread_state_ledger_validator import _minimal_handoff # noqa: E402
findings = findings_for_final_report(_minimal_handoff())
self.assertTrue(findings)
self.assertTrue(
any(f["severity"] == "block" for f in findings)
)
class TestValidatorFinding(unittest.TestCase):
def test_finding_shape(self):
finding = validator_finding(
tests/test_issue_540_comment_role_poison.py
+4 -1
View File
@@ -198,10 +198,13 @@ class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase):
srv._preflight_capability_baseline_porcelain = ""
self._orig_in_test = srv._preflight_in_test_mode
srv._preflight_in_test_mode = lambda: False
self._env_patch = patch.dict(os.environ, {"GITEA_MCP_DISABLE_PARITY_GATE": "1"}, clear=False)
self._env_patch.start()
def tearDown(self):
srv._preflight_in_test_mode = self._orig_in_test
srv._preflight_resolved_role = None
self._env_patch.stop()
@patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@@ -230,7 +233,7 @@ class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase):
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
os.environ.pop("GITEA_RECONCILER_WORKTREE", None)
result = srv.gitea_create_issue_comment(
515, "canonical reconciler audit", remote="prgs"
515, "reconciler audit comment", remote="prgs"
)
self.assertTrue(result["success"])
self.assertEqual(result["comment_id"], 9001)
tests/test_issue_comment_workspace_guard.py
+6 -6
View File
@@ -107,7 +107,7 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
body="evidence comment",
remote="prgs",
)
self.assertIn("stable control checkout", str(ctx.exception))
@@ -140,7 +140,7 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
result = srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
body="evidence comment",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
@@ -153,7 +153,7 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
method, url, _auth_arg, payload = mock_api.call_args[0]
self.assertEqual(method, "POST")
self.assertIn("/repos/Scaled-Tech-Consulting/Gitea-Tools/issues/557/comments", url)
self.assertEqual(payload, {"body": "canonical evidence"})
self.assertEqual(payload, {"body": "evidence comment"})
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
@@ -183,7 +183,7 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
body="evidence comment",
remote="prgs",
worktree_path=outside_worktree,
)
@@ -198,7 +198,7 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
body="evidence comment",
remote="prgs",
worktree_path=os.path.join(
CONTROL_CHECKOUT_ROOT,
@@ -239,7 +239,7 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
with patch.dict(os.environ, denied_env, clear=True):
result = srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
body="evidence comment",
remote="prgs",
worktree_path=valid_worktree,
)
tests/test_lock_issue_mcp_registration.py
+6 -2
View File
@@ -97,7 +97,7 @@ class TestCreatePrLockRegistrationFlow(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", return_value=[])
@patch("mcp_server.api_get_all", return_value=[{"id": 4, "name": "status:pr-open"}])
@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)
@@ -105,7 +105,11 @@ class TestCreatePrLockRegistrationFlow(unittest.TestCase):
self, _auth, _role, _api, _git_state, mock_api_request
):
worktree = os.path.realpath(os.getcwd())
mock_api_request.return_value = {"number": 521, "html_url": "https://example/pr/521"}
def mock_api(method, url, auth, data=None):
if "/labels" in url:
return [{"name": "status:pr-open"}]
return {"number": 521, "html_url": "https://example/pr/521"}
mock_api_request.side_effect = mock_api
lock_res = gitea_lock_issue(
issue_number=521,
branch_name="feat/issue-521-lock-issue-registration",
tests/test_mcp_server.py
+15 -1
View File
@@ -3196,6 +3196,20 @@ class TestIssueCommentTools(unittest.TestCase):
self.assertFalse(result["success"])
mock_api.assert_not_called()
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_tagged_handoff_missing_mutation_ledger_blocked(
self, _auth, mock_api
):
body = "[CONTROLLER HANDOFF] PR #1 — test\n\nBlockers:\n- none"
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
result = gitea_create_issue_comment(
issue_number=9, body=body, remote="prgs")
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertTrue(result["reasons"])
mock_api.assert_not_called()
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_empty_body_blocked(self, _auth, mock_api):
@@ -4220,7 +4234,7 @@ class TestIssue546Deadlock(unittest.TestCase):
res = mcp_server.gitea_submit_pr_review(
pr_number=550,
action="approve",
body="APPROVED",
body="",
expected_head_sha="abc123",
final_review_decision_ready=True,
remote="prgs",
tests/test_namespace_workspace_binding.py
+3
View File
@@ -103,6 +103,8 @@ class TestNamespaceWorkspaceIntegration(unittest.TestCase):
srv._preflight_in_test_mode = lambda: False
self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False)
self._env_patch.start()
self._guard_patch = mock.patch("gitea_mcp_server._enforce_root_checkout_guard")
self._guard_patch.start()
def tearDown(self):
srv._preflight_whoami_called = self._saved["whoami_called"]
@@ -112,6 +114,7 @@ class TestNamespaceWorkspaceIntegration(unittest.TestCase):
srv._preflight_capability_violation = self._saved["capability_violation"]
srv._preflight_in_test_mode = self._saved["in_test"]
self._env_patch.stop()
self._guard_patch.stop()
for key in (
nwb.AUTHOR_WORKTREE_ENV,
nwb.MERGER_WORKTREE_ENV,
tests/test_operator_guide.py
+4
View File
@@ -56,6 +56,10 @@ EXPECTED_SKILLS = [
"jenkins-mcp",
"glitchtip-mcp",
"release-operator",
# Canonical workflow router (#551) — same skill, multi-runtime names
"gitea-workflow",
"llm-project-workflow",
"git-pr-workflows",
]
tests/test_permission_reports.py
+6 -8
View File
@@ -234,11 +234,10 @@ class TestEligibilityDenialReport(PermissionReportBase):
return {"login": "author-user"}
return PR_PAYLOAD
mock_api.side_effect = fake_api
mcp_server.init_review_decision_lock("prgs", "review_pr")
from tests.test_mcp_server import _seed_ready_review_decision
_seed_ready_review_decision(42, "approve", remote="prgs")
with patch.dict(os.environ, self._env("author-profile")):
mcp_server.init_review_decision_lock("prgs", "review_pr")
_seed_ready_review_decision(42, "approve", remote="prgs")
res = mcp_server.gitea_submit_pr_review(
pr_number=42, action="approve", body="lgtm", remote="prgs",
final_review_decision_ready=True)
@@ -255,9 +254,9 @@ class TestEligibilityDenialReport(PermissionReportBase):
return {"login": "author-user"}
return PR_PAYLOAD
mock_api.side_effect = fake_api
mcp_server.init_review_decision_lock("prgs", "review_pr")
mcp_server.gitea_load_review_workflow()
with patch.dict(os.environ, self._env("author-profile")):
mcp_server.init_review_decision_lock("prgs", "review_pr")
mcp_server.gitea_load_review_workflow()
res = mcp_server.gitea_merge_pr(
pr_number=42, confirmation="MERGE PR 42",
expected_head_sha="abc123", remote="prgs")
@@ -280,11 +279,10 @@ class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
return {"login": "author-user"}
return PR_PAYLOAD
mock_api.side_effect = fake_api
mcp_server.init_review_decision_lock("prgs", "review_pr")
from tests.test_mcp_server import _seed_ready_review_decision
_seed_ready_review_decision(42, "comment", remote="prgs")
with patch.dict(os.environ, self._env("author-profile")):
mcp_server.init_review_decision_lock("prgs", "review_pr")
_seed_ready_review_decision(42, "comment", remote="prgs")
res = mcp_server.gitea_submit_pr_review(
pr_number=42, action="comment", body="finding", remote="prgs",
final_review_decision_ready=True)
tests/test_reconciler_close_workspace_guard.py
+3
View File
@@ -35,9 +35,12 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
srv._preflight_capability_violation = False
self._orig_in_test = srv._preflight_in_test_mode
srv._preflight_in_test_mode = lambda: False
self._env_patch = patch.dict(os.environ, {"GITEA_MCP_DISABLE_PARITY_GATE": "1"}, clear=False)
self._env_patch.start()
def tearDown(self):
srv._preflight_in_test_mode = self._orig_in_test
self._env_patch.stop()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
tests/test_review_final_report_schema.py
+5
View File
@@ -35,6 +35,9 @@ def _minimal_review_report(**overrides):
"- External-state mutations: none",
"- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
"- Current status: PR open",
"- Next actor: author",
"- Next action: Fix",
"- Next prompt: Fix",
"- Blockers: none",
"- Next: await author",
"- Safety: no self-review; no self-merge",
@@ -43,6 +46,8 @@ def _minimal_review_report(**overrides):
"- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Worktree path: branches/review-203",
"- Worktree dirty: clean",
"- Pinned reviewed head: none",
"- Scratch worktree used: none",
"- Unrelated local mutations: none",
"- Review decision: request_changes",
"- Merge result: not attempted",
tests/test_thread_state_ledger_validator.py
+187
View File
@@ -0,0 +1,187 @@
"""Tests for two-comment workflow validation (#507)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from thread_state_ledger_validator import ( # noqa: E402
BLOCKER_CLASSIFICATIONS,
assess_controller_handoff_comment,
assess_thread_state_ledger_comment,
assess_two_comment_workflow,
assess_two_comment_pair,
)
def _minimal_handoff(**overrides):
lines = [
"[CONTROLLER HANDOFF] PR #203 / Issue #182 — review complete",
"",
"Identity/profile:",
"- Active profile: prgs-reviewer",
"- Authenticated identity: sysadmin",
"",
"Server-side mutation ledger:",
"- gitea_submit_pr_review → APPROVED review posted to Gitea",
"",
"Blockers:",
"- none",
]
text = "\n".join(lines)
for key, value in overrides.items():
if f"{key}:" in text:
text = text.replace(f"{key}:", f"{key}: {value}", 1)
return text
def _minimal_ledger(**overrides):
lines = [
"[THREAD STATE LEDGER] PR #203 — APPROVED review posted to Gitea",
"",
"What is true now:",
"- PR state: open",
"- Current head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Server-side decision state: APPROVED review posted to Gitea",
"- Local verdict/state: APPROVE verdict prepared locally",
"- Latest known validation: tests passed locally",
"",
"What changed:",
"- APPROVED review posted to Gitea at pinned head",
"",
"What is blocked:",
"- Blocker classification: no blocker",
"",
"Who/what acts next:",
"- Next actor: merger",
"- Required action: merge on explicit operator command",
"- Do not do: re-post APPROVE",
"- Resume from: PR #203 review feedback",
]
text = "\n".join(lines)
for key, value in overrides.items():
text = text.replace(f"- {key}:", f"- {key}: {value}")
return text
class TestTwoCommentWorkflow(unittest.TestCase):
def test_valid_combined_report_passes(self):
report = _minimal_handoff() + "\n\n" + _minimal_ledger()
result = assess_two_comment_workflow(report)
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_missing_ledger_after_tagged_handoff_blocks(self):
result = assess_two_comment_workflow(_minimal_handoff())
self.assertTrue(result["block"])
self.assertTrue(
any("THREAD STATE LEDGER" in r for r in result["reasons"])
)
def test_legacy_controller_handoff_section_not_required_ledger(self):
legacy = "## Controller Handoff\n\n- Task: work issue\n- Next: continue\n"
result = assess_two_comment_workflow(legacy)
self.assertFalse(result["block"])
def test_ambiguous_approved_blocks(self):
handoff = _minimal_handoff().replace(
"APPROVED review posted to Gitea",
"approved",
)
report = handoff + "\n\n" + _minimal_ledger()
result = assess_two_comment_workflow(report)
self.assertTrue(result["block"])
self.assertTrue(any("approved" in r.lower() for r in result["reasons"]))
def test_ambiguous_merged_blocks(self):
handoff = _minimal_handoff() + "\n\n- Narrative: merged successfully"
ledger = _minimal_ledger()
report = handoff + "\n\n" + ledger
result = assess_two_comment_workflow(report)
self.assertTrue(result["block"])
self.assertTrue(any("merge" in r.lower() for r in result["reasons"]))
def test_blocked_without_gate_blocks(self):
handoff = _minimal_handoff().replace("- none", "- blocked")
ledger = _minimal_ledger().replace(
"no blocker", "blocked but no gate named"
)
report = handoff + "\n\n" + ledger
result = assess_two_comment_workflow(report)
self.assertTrue(result["block"])
def test_mutation_ledger_contradiction_blocks(self):
handoff = _minimal_handoff().replace(
"APPROVED review posted to Gitea",
"none — no server-side state changed",
)
handoff += "\n- Narrative: APPROVED review posted to Gitea"
report = handoff + "\n\n" + _minimal_ledger()
result = assess_two_comment_workflow(report)
self.assertTrue(result["block"])
self.assertTrue(any("contradict" in r.lower() for r in result["reasons"]))
def test_local_verdict_as_server_side_blocks(self):
ledger = _minimal_ledger(
**{
"Server-side decision state": "APPROVE verdict prepared locally",
"Local verdict/state": "none",
}
)
report = _minimal_handoff() + "\n\n" + ledger
result = assess_two_comment_workflow(report)
self.assertTrue(result["block"])
class TestHandoffComment(unittest.TestCase):
def test_handoff_comment_valid(self):
result = assess_controller_handoff_comment(_minimal_handoff())
self.assertTrue(result["proven"])
def test_handoff_without_marker_skips(self):
result = assess_controller_handoff_comment("casual comment")
self.assertTrue(result["proven"])
self.assertFalse(result["performed"])
class TestLedgerComment(unittest.TestCase):
def test_ledger_requires_blocker_classification(self):
ledger = _minimal_ledger().replace(
"Blocker classification: no blocker",
"something unclear",
)
result = assess_thread_state_ledger_comment(ledger)
self.assertTrue(result["block"])
def test_all_blocker_classifications_recognized(self):
self.assertIn("no blocker", BLOCKER_CLASSIFICATIONS)
self.assertIn("environment/tooling blocker", BLOCKER_CLASSIFICATIONS)
class TestPairedComments(unittest.TestCase):
def test_pair_passes(self):
result = assess_two_comment_pair(_minimal_handoff(), _minimal_ledger())
self.assertTrue(result["proven"])
def test_pair_missing_ledger_blocks(self):
result = assess_two_comment_pair(_minimal_handoff(), "")
self.assertTrue(result["block"])
class TestExamples(unittest.TestCase):
"""Smoke-test bundled examples from docs."""
def test_examples_module_loads(self):
from thread_state_ledger_examples import EXAMPLES # noqa: E402
self.assertGreaterEqual(len(EXAMPLES), 8)
for name, handoff, ledger in EXAMPLES:
result = assess_two_comment_pair(handoff, ledger)
self.assertTrue(
result["proven"],
f"example '{name}' failed: {result['reasons']}",
)
if __name__ == "__main__":
unittest.main()
tests/test_webui_audit.py
+94
View File
@@ -0,0 +1,94 @@
"""Tests for web UI report audit paste (#431)."""
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.audit_validator import audit_report, infer_task_kind
_MINIMAL_REVIEW_REPORT = """## Controller Handoff
- Task: review PR #1
- Repo: Scaled-Tech-Consulting/Gitea-Tools
- Role: reviewer
- Identity: sysadmin / prgs-reviewer
- Review decision: approve
"""
class TestAuditValidator(unittest.TestCase):
def test_infer_task_kind_from_review_report(self):
self.assertEqual(infer_task_kind(_MINIMAL_REVIEW_REPORT), "review_pr")
def test_infer_task_kind_explicit_override(self):
self.assertEqual(
infer_task_kind(_MINIMAL_REVIEW_REPORT, explicit="work_issue"),
"work_issue",
)
def test_empty_report_blocked(self):
result = audit_report("")
self.assertTrue(result["blocked"])
self.assertIn("empty", result["reasons"][0].lower())
def test_minimal_report_produces_findings(self):
result = audit_report(_MINIMAL_REVIEW_REPORT)
self.assertIn(result["grade"], {"blocked", "downgraded", "warning", "A"})
self.assertTrue(result["findings"] or result["blocked"] or result["downgraded"])
class TestAuditRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_audit_page_renders_form(self):
response = self.client.get("/audit")
self.assertEqual(response.status_code, 200)
self.assertIn("Report audit", response.text)
self.assertIn('name="report_text"', response.text)
self.assertIn("Run validator preview", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_audit_post_runs_validator(self):
response = self.client.post(
"/audit",
data={"report_text": _MINIMAL_REVIEW_REPORT, "task_kind": "review_pr"},
)
self.assertEqual(response.status_code, 200)
self.assertIn("Validator preview", response.text)
self.assertIn("Safe next action", response.text)
def test_api_audit_post_json(self):
response = self.client.post(
"/api/audit",
json={"report_text": _MINIMAL_REVIEW_REPORT, "task_kind": "review_pr"},
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["task_kind"], "review_pr")
self.assertIn("grade", data)
self.assertIn("findings", data)
self.assertIn("suggested_next_prompt", data)
def test_api_audit_rejects_empty_body(self):
response = self.client.post("/api/audit", json={"report_text": ""})
self.assertEqual(response.status_code, 400)
def test_audit_post_allowed_other_post_still_blocked(self):
self.assertEqual(self.client.post("/health").status_code, 405)
self.assertEqual(
self.client.post(
"/audit",
data={"report_text": _MINIMAL_REVIEW_REPORT},
).status_code,
200,
)
if __name__ == "__main__":
unittest.main()
tests/test_webui_skeleton.py
+13 -10
View File
@@ -34,12 +34,10 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertIn("Runtime health", response.text)
def test_route_stubs_render(self):
for path in ("/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_audit_is_implemented(self):
response = self.client.get("/audit")
self.assertEqual(response.status_code, 200)
self.assertIn("Report audit", response.text)
def test_prompts_is_implemented(self):
response = self.client.get("/prompts")
@@ -51,14 +49,17 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertIn("Gitea-Tools", response.text)
def test_worktrees_is_implemented(self):
response = self.client.get("/worktrees")
self.assertEqual(response.status_code, 200)
self.assertIn("Worktree hygiene", response.text)
def test_leases_is_implemented(self):
response = self.client.get("/leases")
self.assertEqual(response.status_code, 200)
self.assertIn("Leases", response.text)
self.assertIn("Collision warnings", response.text)
def test_extra_stub_routes(self):
self.assertEqual(self.client.get("/worktrees").status_code, 200)
def test_post_is_rejected(self):
response = self.client.post("/health")
@@ -71,10 +72,12 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
for path in paths:
with self.subTest(path=path):
text = self.client.get(path).text
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
for href in hrefs:
self.assertIn(f'href="{href}"', text)
tests/test_webui_worktree_hygiene.py
+114
View File
@@ -0,0 +1,114 @@
"""Tests for web UI worktree hygiene dashboard (#432)."""
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.worktree_scanner import (
classify_entry,
load_hygiene_snapshot,
parse_worktree_list_porcelain,
snapshot_to_dict,
)
class TestWorktreeScanner(unittest.TestCase):
def test_parse_worktree_porcelain(self):
porcelain = (
"worktree /repo/branches/feat-issue-1-foo\n"
"HEAD abcdef0123456789\n"
"branch refs/heads/feat/issue-1-foo\n"
"\n"
)
entries = parse_worktree_list_porcelain(porcelain)
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0]["branch"], "feat/issue-1-foo")
def test_classify_active_pr(self):
classification, _ = classify_entry(
rel_path="branches/feat-issue-99-demo",
worktree_record={"branch": "feat/issue-99-demo"},
state={"exists": True, "clean": True, "dirty_files": []},
open_pr_branches={"feat/issue-99-demo"},
active_lock_branch=None,
)
self.assertEqual(classification, "active-pr")
def test_classify_dirty(self):
classification, _ = classify_entry(
rel_path="branches/feat-issue-2-bar",
worktree_record={"branch": "feat/issue-2-bar"},
state={"exists": True, "clean": False, "dirty_files": ["webui/app.py"]},
open_pr_branches=set(),
active_lock_branch=None,
)
self.assertEqual(classification, "dirty")
def test_missing_lock_worktree_anomaly(self):
with tempfile.TemporaryDirectory() as tmp:
branches = Path(tmp) / "branches" / "other-worktree"
branches.mkdir(parents=True)
lock_path = Path(tmp) / "lock.json"
lock_path.write_text(
json.dumps({
"issue_number": 432,
"branch_name": "feat/issue-432-worktree-hygiene",
"worktree_path": str(Path(tmp) / "branches" / "missing-tree"),
}),
encoding="utf-8",
)
snapshot = load_hygiene_snapshot(
project_root=tmp,
worktree_porcelain="",
open_pr_branches=set(),
issue_lock_path=str(lock_path),
)
self.assertTrue(snapshot.anomalies)
self.assertIn("Missing preserved worktree", snapshot.anomalies[0])
def test_snapshot_to_dict(self):
with tempfile.TemporaryDirectory() as tmp:
(Path(tmp) / "branches").mkdir()
snapshot = load_hygiene_snapshot(
project_root=tmp,
worktree_porcelain="",
open_pr_branches=set(),
issue_lock_path=str(Path(tmp) / "no-lock.json"),
)
data = snapshot_to_dict(snapshot)
self.assertIn("entries", data)
self.assertIn("cleanup_prompt", data)
class TestWorktreeRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_worktrees_page_renders(self):
response = self.client.get("/worktrees")
self.assertEqual(response.status_code, 200)
self.assertIn("Worktree hygiene", response.text)
self.assertIn("Canonical cleanup prompt", response.text)
self.assertIn("Copy cleanup prompt", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_api_worktrees_json(self):
response = self.client.get("/api/worktrees")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIn("entries", data)
self.assertIn("cleanup_prompt", data)
self.assertIn("anomalies", data)
def test_nav_includes_worktrees(self):
self.assertIn('href="/worktrees"', self.client.get("/").text)
if __name__ == "__main__":
unittest.main()
tests/test_workflow_skill.py
+87
View File
@@ -0,0 +1,87 @@
"""Tests for canonical workflow skill naming and preflight (#551)."""
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
import workflow_skill
import mcp_server
class TestWorkflowSkillModule(unittest.TestCase):
def test_canonical_names_include_gitea_workflow(self):
names = workflow_skill.CANONICAL_WORKFLOW_SKILL_NAMES
self.assertIn("gitea-workflow", names)
self.assertIn("llm-project-workflow", names)
self.assertIn("git-pr-workflows", names)
def test_repo_skill_present_in_this_checkout(self):
root = str(Path(__file__).resolve().parent.parent)
result = workflow_skill.assess_workflow_skill_presence(root)
self.assertTrue(result["repo_skill_present"])
self.assertTrue(result["repo_alias_present"])
self.assertTrue(result["workflow_skill_ready"])
self.assertFalse(result["blocked"])
def test_missing_repo_skill_blocks(self):
with tempfile.TemporaryDirectory() as tmp:
result = workflow_skill.assess_workflow_skill_presence(tmp)
self.assertTrue(result["blocked"])
self.assertFalse(result["workflow_skill_ready"])
self.assertTrue(any("missing" in r for r in result["reasons"]))
def test_codex_mount_detection(self):
root = str(Path(__file__).resolve().parent.parent)
with tempfile.TemporaryDirectory() as tmp:
skill_dir = Path(tmp) / "gitea-workflow"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text("# x\n", encoding="utf-8")
result = workflow_skill.assess_workflow_skill_presence(
root, codex_skills_dir=tmp
)
self.assertTrue(result["codex_skill_mounted"])
self.assertTrue(result["codex_skill_mounts"]["gitea-workflow"])
def test_registry_entries_cover_aliases(self):
entries = workflow_skill.workflow_skill_registry_entries()
for name in workflow_skill.CANONICAL_WORKFLOW_SKILL_NAMES:
self.assertIn(name, entries)
self.assertEqual(entries[name]["status"], "available")
class TestWorkflowSkillMcpTools(unittest.TestCase):
def test_list_includes_canonical_names(self):
with patch.dict(
os.environ,
{
"GITEA_PROFILE_NAME": "author-test",
"GITEA_ALLOWED_OPERATIONS": "gitea.read",
"GITEA_FORBIDDEN_OPERATIONS": "",
},
clear=False,
):
result = mcp_server.mcp_list_project_skills()
names = {s["name"] for s in result["skills"]}
for name in workflow_skill.CANONICAL_WORKFLOW_SKILL_NAMES:
self.assertIn(name, names)
def test_get_skill_guide_gitea_workflow(self):
guide = mcp_server.mcp_get_skill_guide("gitea-workflow")
self.assertTrue(guide["success"])
self.assertTrue(guide["steps"])
self.assertIn("workflow", " ".join(guide["steps"]).lower())
def test_preflight_tool_ok_on_repo(self):
result = mcp_server.mcp_check_workflow_skill_preflight()
self.assertTrue(result["success"])
self.assertTrue(result["workflow_skill_ready"])
self.assertFalse(result["blocked"])
self.assertIn("gitea-workflow", result["canonical_names"])
if __name__ == "__main__":
unittest.main()