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

Merge remote-tracking branch 'prgs/master' into feat/issue-483-role-boundaries

Browse Source
This commit is contained in:
sysadmin
2026-07-09 08:27:36 -04:00
parent bc16f6652b e2e51290c4
commit ce0fb42bef
11 changed files with 817 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_audit_reconciliation_mode.py
+1 -1
View File
@@ -284,7 +284,7 @@ class TestTaskCapabilityMap(unittest.TestCase):
required_permission("reconciliation_cleanup"),
"gitea.branch.delete",
)
self.assertEqual(required_role("reconciliation_cleanup"), "author")
self.assertEqual(required_role("reconciliation_cleanup"), "reconciler")
if __name__ == "__main__":
tests/test_reconciler_cleanup_integration.py
+263
View File
@@ -0,0 +1,263 @@
"""Integration tests for reconciler merged cleanup (#523)."""
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import task_capability_map
from audit_reconciliation_mode import clear_phase
from task_capability_map import required_role
RECONCILER_PROFILE = {
"profile_name": "prgs-reconciler",
"context": "prgs",
"role": "reconciler",
"username": "sysadmin",
"base_url": "https://gitea.prgs.cc",
"allowed_operations": [
"gitea.read",
"gitea.pr.close",
"gitea.pr.comment",
"gitea.issue.comment",
],
"forbidden_operations": [],
}
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"context": "prgs",
"role": "author",
"username": "jcwalker3",
"base_url": "https://gitea.prgs.cc",
"allowed_operations": [
"gitea.read",
"gitea.pr.create",
"gitea.branch.create",
"gitea.branch.push",
"gitea.repo.commit",
],
"forbidden_operations": [],
}
CONTROL_ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
class TestReconcilerCleanupIntegration(unittest.TestCase):
def setUp(self):
clear_phase()
mcp_server._preflight_whoami_called = False
mcp_server._preflight_capability_called = False
mcp_server._preflight_resolved_role = None
mcp_server._preflight_resolved_task = None
mcp_server._preflight_whoami_violation = False
mcp_server._preflight_capability_violation = False
mcp_server._preflight_whoami_violation_files = []
mcp_server._preflight_capability_violation_files = []
mcp_server._preflight_capability_baseline_porcelain = ""
self.mock_api = patch("gitea_auth.api_request").start()
self.mock_auth = patch(
"mcp_server.get_auth_header", return_value="token test"
).start()
def tearDown(self):
patch.stopall()
clear_phase()
mcp_server._IDENTITY_CACHE.clear()
def test_capability_map_routes_merged_cleanup_to_reconciler(self):
self.assertEqual(required_role("reconcile_merged_cleanups"), "reconciler")
self.assertEqual(required_role("reconciliation_cleanup"), "reconciler")
self.assertEqual(
task_capability_map.required_permission("reconcile_merged_cleanups"),
"gitea.read",
)
@patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
def test_reconciler_can_run_reconcile_merged_cleanups_directly(self, _profile):
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check(
"capability",
resolved_role="reconciler",
resolved_task="reconcile_merged_cleanups",
)
self.mock_api.side_effect = [
[ # closed pulls page
{
"number": 100,
"title": "merged feature",
"body": "Closes #100",
"merged": True,
"merged_at": "2026-07-06T12:00:00Z",
"merge_commit_sha": "deadbeef",
"head": {"ref": "feat/issue-100", "sha": "cafebabe"},
},
{
# closed but NOT merged — must not become a cleanup entry
"number": 101,
"title": "abandoned",
"body": "Closes #101",
"merged": False,
"merged_at": None,
"head": {"ref": "feat/issue-101", "sha": "badcafe"},
},
],
[], # open pulls
]
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 = mcp_server.gitea_reconcile_merged_cleanups(
dry_run=True,
remote="prgs",
)
self.assertTrue(result["success"])
self.assertTrue(result["dry_run"])
entry_numbers = [e["issue_number"] for e in result["entries"]]
self.assertEqual(entry_numbers, [100])
self.assertNotIn(101, entry_numbers)
@patch.dict(
os.environ,
{
"GITEA_PROFILE_NAME": "prgs-reconciler",
# Force preflight purity to evaluate (disable test short-circuit).
"GITEA_TEST_PORCELAIN": "",
},
clear=True,
)
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
def test_reconciler_role_bypasses_branches_only_mutation_guard(self, _profile):
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check(
"capability",
resolved_role="reconciler",
resolved_task="reconcile_merged_cleanups",
)
with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
with patch(
"mcp_server.root_checkout_guard.resolve_remote_master_sha",
return_value="abc123",
):
with patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "master",
"head_sha": "abc123",
"porcelain_status": "",
},
):
try:
mcp_server.verify_preflight_purity(
remote="prgs",
task="reconcile_merged_cleanups",
)
except RuntimeError as e:
self.fail(
"verify_preflight_purity raised RuntimeError "
f"unexpectedly for reconciler: {e}"
)
@patch.dict(
os.environ,
{"GITEA_PROFILE_NAME": "prgs-author", "GITEA_TEST_PORCELAIN": ""},
clear=True,
)
@patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE)
def test_author_role_remains_blocked_on_root_checkout(self, _profile):
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check(
"capability",
resolved_role="author",
resolved_task="reconcile_merged_cleanups",
)
with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
with patch(
"mcp_server.root_checkout_guard.resolve_remote_master_sha",
return_value="abc123",
):
with patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "master",
"head_sha": "abc123",
"porcelain_status": "",
},
):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity(
remote="prgs",
task="reconcile_merged_cleanups",
)
message = str(ctx.exception)
self.assertTrue(
"stable control checkout" in message
or "author mutation blocked" in message
or "branches/" in message,
msg=message,
)
@patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
def test_open_unmerged_pr_heads_are_not_safe_cleanup_targets(self, _profile):
"""AC5: active/unmerged author work must remain blocked from cleanup."""
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check(
"capability",
resolved_role="reconciler",
resolved_task="reconcile_merged_cleanups",
)
open_pr = {
"number": 200,
"title": "active work",
"body": "Closes #200",
"merged": False,
"state": "open",
"head": {"ref": "feat/issue-200", "sha": "livehead1"},
}
# Same head branch still open on another PR while a closed/merged PR
# used the same branch name historically — open head must block delete.
closed_merged = {
"number": 199,
"title": "older merge same branch name",
"body": "Closes #199",
"merged": True,
"merged_at": "2026-07-01T12:00:00Z",
"merge_commit_sha": "deadbeef",
"head": {"ref": "feat/issue-200", "sha": "oldhead99"},
}
self.mock_api.side_effect = [
[closed_merged], # closed
[open_pr], # open
]
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 = mcp_server.gitea_reconcile_merged_cleanups(
dry_run=True,
remote="prgs",
)
self.assertTrue(result["success"])
self.assertEqual(len(result["entries"]), 1)
remote_assessment = result["entries"][0].get("remote_branch") or {}
self.assertFalse(
remote_assessment.get("safe_to_delete_remote"),
msg=f"open head must block remote delete: {remote_assessment}",
)
if __name__ == "__main__":
unittest.main()
tests/test_webui_runtime_health.py
+88
View File
@@ -0,0 +1,88 @@
"""Tests for web UI runtime health dashboard (#430)."""
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.runtime_health import _role_kind, load_runtime_snapshot, snapshot_to_dict
def _mock_identity(_host: str):
return "jcwalker3", None
def _mock_git_sync(_repo: Path, _remote: str, _branch: str):
return "localsha123456", "remoteshaabcdef", 2
class TestRuntimeHealthLoader(unittest.TestCase):
def test_role_kind_author(self):
allowed = ["gitea.pr.create", "gitea.branch.push", "gitea.read"]
forbidden = ["gitea.pr.merge", "gitea.pr.approve"]
self.assertEqual(_role_kind(allowed, forbidden), "author")
def test_snapshot_with_mocks(self):
snapshot = load_runtime_snapshot(
resolve_username=_mock_identity,
git_sync=_mock_git_sync,
)
self.assertEqual(snapshot.project_id, "gitea-tools")
self.assertEqual(snapshot.authenticated_username, "jcwalker3")
self.assertEqual(snapshot.commits_behind_master, 2)
self.assertIsNotNone(snapshot.stale_runtime_warning)
self.assertGreaterEqual(len(snapshot.workflow_hashes), 3)
self.assertGreaterEqual(len(snapshot.schema_hashes), 2)
self.assertIn("shell_use_allowed", snapshot.shell_health)
def test_snapshot_dict_export(self):
snapshot = load_runtime_snapshot(
resolve_username=_mock_identity,
git_sync=_mock_git_sync,
)
data = snapshot_to_dict(snapshot)
self.assertEqual(data["authenticated_username"], "jcwalker3")
self.assertEqual(data["commits_behind_master"], 2)
self.assertTrue(data["stale_runtime_warning"])
class TestRuntimeRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
self.snapshot = load_runtime_snapshot(
resolve_username=_mock_identity,
git_sync=_mock_git_sync,
)
self._patch = mock.patch(
"webui.app.load_runtime_snapshot",
return_value=self.snapshot,
)
self._patch.start()
def tearDown(self):
self._patch.stop()
def test_runtime_page_renders_profile_and_warning(self):
response = self.client.get("/runtime")
self.assertEqual(response.status_code, 200)
self.assertIn("Runtime health", response.text)
self.assertIn("Active profile", response.text)
self.assertIn("Stale runtime warning", response.text)
self.assertIn("Workflow hashes", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_api_runtime_json(self):
response = self.client.get("/api/runtime")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["project_id"], "gitea-tools")
self.assertEqual(data["role_kind"], self.snapshot.role_kind)
self.assertEqual(data["commits_behind_master"], 2)
if __name__ == "__main__":
unittest.main()
tests/test_webui_skeleton.py
+6 -1
View File
@@ -29,8 +29,13 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Operator console", response.text)
self.assertIn("Read-only MVP", response.text)
def test_runtime_is_implemented(self):
response = self.client.get("/runtime")
self.assertEqual(response.status_code, 200)
self.assertIn("Runtime health", response.text)
def test_route_stubs_render(self):
for path in ("/runtime", "/audit"):
for path in ("/audit",):
with self.subTest(path=path):
response = self.client.get(path)
self.assertEqual(response.status_code, 200)