Merge branch 'master' into feat/issue-543-mcp-namespace-health-check
This commit is contained in:
@@ -0,0 +1,187 @@
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import branch_cleanup_guard as guard # noqa: E402
|
||||
import mcp_server # noqa: E402
|
||||
import task_capability_map # noqa: E402
|
||||
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||
from mcp_server import gitea_cleanup_merged_pr_branch # noqa: E402
|
||||
|
||||
FAKE_AUTH = "token fake"
|
||||
|
||||
|
||||
class TestRawBranchDeleteGuard(unittest.TestCase):
|
||||
def test_detects_local_and_remote_raw_git_delete_commands(self):
|
||||
text = """
|
||||
Cleanup mutations:
|
||||
- git branch -d feat/issue-485-lease-comments-non-list-guard
|
||||
- git push prgs --delete feat/issue-485-lease-comments-non-list-guard
|
||||
"""
|
||||
commands = guard.raw_branch_delete_commands(text)
|
||||
self.assertEqual(len(commands), 2)
|
||||
self.assertIn("git branch -d", commands[0])
|
||||
self.assertIn("git push prgs --delete", commands[1])
|
||||
|
||||
def test_final_report_blocks_raw_delete_as_cleanup_proof(self):
|
||||
report = """
|
||||
## Controller Handoff
|
||||
- Task: review_pr
|
||||
- Cleanup mutations: git push prgs --delete feat/branch
|
||||
"""
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any("shared.raw_branch_delete_bypass" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_cleanup_task_requires_branch_delete_capability(self):
|
||||
self.assertEqual(
|
||||
task_capability_map.required_permission("cleanup_merged_pr_branch"),
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
self.assertEqual(
|
||||
task_capability_map.required_role("cleanup_merged_pr_branch"),
|
||||
"reconciler",
|
||||
)
|
||||
|
||||
|
||||
class TestMergedPrBranchCleanupTool(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(
|
||||
mcp_server.REMOTES,
|
||||
{
|
||||
"prgs": {
|
||||
"host": "gitea.example.com",
|
||||
"org": "Example-Org",
|
||||
"repo": "Example-Repo",
|
||||
}
|
||||
},
|
||||
)
|
||||
self._remotes.start()
|
||||
patch("gitea_audit.audit_enabled", return_value=False).start()
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start()
|
||||
patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
|
||||
return_value=True,
|
||||
).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
|
||||
def test_reviewer_without_delete_authority_fails_closed(self):
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "reviewer",
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.review"],
|
||||
"forbidden_operations": ["gitea.branch.delete"],
|
||||
},
|
||||
).start()
|
||||
res = gitea_cleanup_merged_pr_branch(
|
||||
pr_number=487,
|
||||
confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
|
||||
branch="feat/branch",
|
||||
remote="prgs",
|
||||
worktree_path="/tmp/repo/branches/cleanup",
|
||||
)
|
||||
self.assertFalse(res["performed"])
|
||||
self.assertEqual(res["required_permission"], "gitea.branch.delete")
|
||||
self.mock_api.assert_not_called()
|
||||
|
||||
def test_root_checkout_cleanup_fails_closed(self):
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "branch-cleanup",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
res = gitea_cleanup_merged_pr_branch(
|
||||
pr_number=487,
|
||||
confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
|
||||
branch="feat/branch",
|
||||
remote="prgs",
|
||||
worktree_path="/tmp/repo/Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(res["performed"])
|
||||
self.assertIn("root checkout", " ".join(res["reasons"]))
|
||||
self.mock_api.assert_not_called()
|
||||
|
||||
def test_authorized_cleanup_deletes_through_api_path(self):
|
||||
branch = "feat/issue-485-lease-comments-non-list-guard"
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "branch-cleanup",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
self.mock_api.side_effect = [
|
||||
{
|
||||
"number": 487,
|
||||
"merged": True,
|
||||
"merged_at": "2026-07-08T01:00:00Z",
|
||||
"head": {"ref": branch, "sha": "a" * 40},
|
||||
"base": {"ref": "master"},
|
||||
},
|
||||
{},
|
||||
{},
|
||||
]
|
||||
res = gitea_cleanup_merged_pr_branch(
|
||||
pr_number=487,
|
||||
confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
|
||||
branch=branch,
|
||||
remote="prgs",
|
||||
worktree_path="/tmp/repo/branches/cleanup",
|
||||
)
|
||||
self.assertTrue(res["performed"])
|
||||
delete_calls = [
|
||||
call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
|
||||
]
|
||||
self.assertEqual(len(delete_calls), 1)
|
||||
self.assertIn("branches/feat%2Fissue-485", delete_calls[0].args[1])
|
||||
|
||||
def test_confirmation_required_before_delete(self):
|
||||
branch = "feat/issue-485-lease-comments-non-list-guard"
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "branch-cleanup",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
self.mock_api.side_effect = [
|
||||
{
|
||||
"number": 487,
|
||||
"merged": True,
|
||||
"merged_at": "2026-07-08T01:00:00Z",
|
||||
"head": {"ref": branch, "sha": "a" * 40},
|
||||
"base": {"ref": "master"},
|
||||
},
|
||||
{},
|
||||
]
|
||||
res = gitea_cleanup_merged_pr_branch(
|
||||
pr_number=487,
|
||||
confirmation="wrong",
|
||||
branch=branch,
|
||||
remote="prgs",
|
||||
worktree_path="/tmp/repo/branches/cleanup",
|
||||
)
|
||||
self.assertFalse(res["performed"])
|
||||
self.assertIn("confirmation", " ".join(res["reasons"]))
|
||||
delete_calls = [
|
||||
call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
|
||||
]
|
||||
self.assertFalse(delete_calls)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -4288,3 +4288,236 @@ class TestIssue546Deadlock(unittest.TestCase):
|
||||
self.assertIsNone(reviewer_pr_lease.get_session_lease())
|
||||
# verify comment phase is released
|
||||
self.assertIn("phase: released", posted_comments[0]["body"])
|
||||
|
||||
def test_reviewer_pr_lease_gate_resolves_identity_with_host(self):
|
||||
"""Lease mutation gate must resolve identity using host, not remote alias."""
|
||||
import mcp_server
|
||||
|
||||
resolved_hosts = []
|
||||
|
||||
def tracking_username(host):
|
||||
resolved_hosts.append(host)
|
||||
return None if host in {"prgs", "dadeschools"} else "reviewer-bot"
|
||||
|
||||
with _install_owned_reviewer_lease(
|
||||
550,
|
||||
session_id=_DEFAULT_LEASE_SESSION,
|
||||
head_sha="abc123",
|
||||
), patch(
|
||||
"mcp_server._resolve",
|
||||
return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
|
||||
), patch(
|
||||
"mcp_server._authenticated_username",
|
||||
side_effect=tracking_username,
|
||||
):
|
||||
reasons = mcp_server._reviewer_pr_lease_gate(
|
||||
pr_number=550,
|
||||
remote="prgs",
|
||||
host=None,
|
||||
org=None,
|
||||
repo=None,
|
||||
mutation="approve",
|
||||
live_head_sha="abc123",
|
||||
pinned_head_sha="abc123",
|
||||
)
|
||||
|
||||
self.assertEqual(reasons, [])
|
||||
self.assertIn("gitea.prgs.cc", resolved_hosts)
|
||||
self.assertNotIn("prgs", resolved_hosts)
|
||||
|
||||
def test_acquire_reviewer_pr_lease_resolves_identity_with_host(self):
|
||||
"""Acquire path must not pass a remote alias into identity lookup."""
|
||||
import mcp_server
|
||||
|
||||
resolved_hosts = []
|
||||
posted = []
|
||||
|
||||
def tracking_username(host):
|
||||
resolved_hosts.append(host)
|
||||
return None if host in {"prgs", "dadeschools"} else "reviewer-bot"
|
||||
|
||||
def mock_api(method, url, auth, data=None):
|
||||
if "/pulls/" in url:
|
||||
return {
|
||||
"user": {"login": "author-user"},
|
||||
"state": "open",
|
||||
"head": {"sha": "abc123"},
|
||||
"mergeable": True,
|
||||
}
|
||||
if "/comments" in url:
|
||||
if method == "POST":
|
||||
posted.append(data["body"])
|
||||
return {"id": 1003}
|
||||
return []
|
||||
return {}
|
||||
|
||||
with patch("mcp_server.api_request", side_effect=mock_api), patch(
|
||||
"mcp_server._authenticated_username",
|
||||
side_effect=tracking_username,
|
||||
), patch(
|
||||
"mcp_server._resolve",
|
||||
return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
|
||||
):
|
||||
res = mcp_server.gitea_acquire_reviewer_pr_lease(
|
||||
pr_number=550,
|
||||
worktree="/workspace",
|
||||
candidate_head="abc123",
|
||||
remote="prgs",
|
||||
)
|
||||
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertTrue(res.get("acquired"), res)
|
||||
self.assertEqual(res.get("comment_id"), 1003)
|
||||
self.assertTrue(posted)
|
||||
self.assertIn("gitea.prgs.cc", resolved_hosts)
|
||||
self.assertNotIn("prgs", resolved_hosts)
|
||||
|
||||
def test_adopt_merger_pr_lease_resolves_identity_with_host(self):
|
||||
"""Adopt path must not pass a remote alias into identity lookup."""
|
||||
import mcp_server
|
||||
import reviewer_pr_lease
|
||||
|
||||
reviewer_pr_lease.clear_session_lease()
|
||||
head = "a" * 40
|
||||
resolved_hosts = []
|
||||
posted = []
|
||||
reviewer_comment = _reviewer_lease_comment(
|
||||
550,
|
||||
session_id="reviewer-session",
|
||||
head_sha=head,
|
||||
reviewer="reviewer-bot",
|
||||
)
|
||||
|
||||
def tracking_username(host):
|
||||
resolved_hosts.append(host)
|
||||
return None if host in {"prgs", "dadeschools"} else "merger-bot"
|
||||
|
||||
def mock_api(method, url, auth, data=None):
|
||||
if "/pulls/" in url and "/comments" not in url:
|
||||
return {
|
||||
"number": 550,
|
||||
"user": {"login": "author-user"},
|
||||
"state": "open",
|
||||
"head": {"sha": head},
|
||||
"mergeable": True,
|
||||
"merged": False,
|
||||
}
|
||||
if "/comments" in url:
|
||||
if method == "POST":
|
||||
posted.append(data["body"])
|
||||
return {"id": 1004}
|
||||
return [reviewer_comment]
|
||||
return {}
|
||||
|
||||
merger_profile = {
|
||||
"profile_name": "prgs-merger",
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"],
|
||||
"forbidden_operations": [],
|
||||
}
|
||||
with patch("mcp_server.get_profile", return_value=merger_profile), patch(
|
||||
"mcp_server.api_request",
|
||||
side_effect=mock_api,
|
||||
), patch(
|
||||
"mcp_server._authenticated_username",
|
||||
side_effect=tracking_username,
|
||||
), patch(
|
||||
"mcp_server._resolve",
|
||||
return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
|
||||
), patch(
|
||||
"mcp_server.gitea_get_pr_review_feedback",
|
||||
return_value={
|
||||
"approval_at_current_head": True,
|
||||
"latest_approved_head_sha": head,
|
||||
},
|
||||
):
|
||||
res = mcp_server.gitea_adopt_merger_pr_lease(
|
||||
pr_number=550,
|
||||
worktree="/workspace",
|
||||
expected_head_sha=head,
|
||||
remote="prgs",
|
||||
)
|
||||
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertTrue(res.get("adopted"), res)
|
||||
self.assertEqual(res.get("adoption_comment_id"), 1004)
|
||||
self.assertTrue(posted)
|
||||
self.assertIn("gitea.prgs.cc", resolved_hosts)
|
||||
self.assertNotIn("prgs", resolved_hosts)
|
||||
|
||||
def test_release_reviewer_pr_lease_same_identity_cross_session(self):
|
||||
"""Same reviewer identity may release a lease claimed by another session.
|
||||
|
||||
Regression: release used to call ``_authenticated_username(remote)``
|
||||
with the remote alias (e.g. ``prgs``) instead of the resolved host,
|
||||
so identity resolved empty and same-identity release fail-closed even
|
||||
when the authenticated user owned the lease.
|
||||
"""
|
||||
import mcp_server
|
||||
import reviewer_pr_lease
|
||||
|
||||
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
reviewer_pr_lease.clear_session_lease()
|
||||
|
||||
foreign_session = "46820-dd78cdc4aacc"
|
||||
comments = [
|
||||
_reviewer_lease_comment(
|
||||
457,
|
||||
session_id=foreign_session,
|
||||
head_sha="ef287eaf5841d9e542a4e888ce0ae7d679dbd685",
|
||||
reviewer="sysadmin",
|
||||
)
|
||||
]
|
||||
posted = []
|
||||
resolved_hosts = []
|
||||
|
||||
def mock_api(method, url, auth, data=None):
|
||||
if "/api/v1/user" in url:
|
||||
return {"login": "sysadmin"}
|
||||
if "/comments" in url:
|
||||
if method == "POST":
|
||||
new_c = {
|
||||
"id": 8071,
|
||||
"body": data["body"],
|
||||
"user": {"login": "sysadmin"},
|
||||
}
|
||||
comments.append(new_c)
|
||||
posted.append(new_c)
|
||||
return {"id": 8071}
|
||||
return comments
|
||||
if "/pulls/" in url:
|
||||
return {
|
||||
"user": {"login": "jcwalker3"},
|
||||
"state": "open",
|
||||
"head": {"sha": "ef287eaf5841d9e542a4e888ce0ae7d679dbd685"},
|
||||
"mergeable": False,
|
||||
}
|
||||
return {}
|
||||
|
||||
def tracking_username(host):
|
||||
resolved_hosts.append(host)
|
||||
# Empty identity when passed a remote alias (the pre-fix bug path).
|
||||
if host in {"prgs", "dadeschools"}:
|
||||
return None
|
||||
return "sysadmin"
|
||||
|
||||
with patch("mcp_server.api_request", side_effect=mock_api), patch(
|
||||
"mcp_server._authenticated_username", side_effect=tracking_username
|
||||
), patch(
|
||||
"mcp_server._resolve",
|
||||
return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
|
||||
), patch("mcp_server._auth", return_value={"Authorization": "token x"}), patch(
|
||||
"mcp_server._verify_role_mutation_workspace", return_value=None
|
||||
):
|
||||
res = mcp_server.gitea_release_reviewer_pr_lease(
|
||||
pr_number=457, remote="prgs"
|
||||
)
|
||||
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertTrue(res.get("released"), res)
|
||||
self.assertEqual(res.get("comment_id"), 8071)
|
||||
self.assertTrue(posted)
|
||||
self.assertIn("phase: released", posted[0]["body"])
|
||||
# Must resolve identity with host, never the remote alias alone.
|
||||
self.assertIn("gitea.prgs.cc", resolved_hosts)
|
||||
self.assertNotIn("prgs", resolved_hosts)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
|
||||
import sys
|
||||
import unittest
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
@@ -26,6 +27,90 @@ class TestShellHealthCircuitBreaker(unittest.TestCase):
|
||||
self.assertEqual(status["consecutive_spawn_failures"], 0)
|
||||
|
||||
|
||||
class TestTerminalDiagnostics(unittest.TestCase):
|
||||
def test_diagnose_missing_cwd(self):
|
||||
res = nmp.diagnose_terminal_failure("/nonexistent_directory_for_test", ["git", "status"])
|
||||
self.assertFalse(res["cwd_exists"])
|
||||
self.assertEqual(res["error_type"], "missing cwd")
|
||||
self.assertIn("does not exist", res["error_msg"])
|
||||
|
||||
def test_diagnose_cwd_is_not_dir(self):
|
||||
import tempfile
|
||||
import os
|
||||
with tempfile.NamedTemporaryFile() as tmp:
|
||||
res = nmp.diagnose_terminal_failure(tmp.name, ["git", "status"])
|
||||
self.assertFalse(res["cwd_is_dir"])
|
||||
self.assertEqual(res["error_type"], "cwd is not a directory")
|
||||
|
||||
def test_diagnose_missing_executable(self):
|
||||
res = nmp.diagnose_terminal_failure(None, ["nonexistent_exec_for_test"])
|
||||
self.assertFalse(res["executable_exists"])
|
||||
self.assertEqual(res["error_type"], "missing executable")
|
||||
|
||||
def test_diagnose_missing_runtime_wrapper(self):
|
||||
res = nmp.diagnose_terminal_failure("/tmp", ["venv/bin/pytest"])
|
||||
self.assertFalse(res["executable_exists"])
|
||||
self.assertEqual(res["error_type"], "missing runtime wrapper")
|
||||
|
||||
def test_diagnose_relative_executable_from_cwd(self):
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
script = Path(tmp) / "tool"
|
||||
script.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
||||
os.chmod(script, 0o755)
|
||||
|
||||
res = nmp.diagnose_terminal_failure(tmp, ["./tool"])
|
||||
self.assertTrue(res["executable_exists"])
|
||||
self.assertEqual(res["resolved_executable"], str(script))
|
||||
|
||||
def test_probe_terminal_spawn_success(self):
|
||||
res = nmp.probe_terminal_spawn()
|
||||
self.assertTrue(res["healthy"])
|
||||
self.assertIn("command", res)
|
||||
self.assertIn("diagnostics", res)
|
||||
|
||||
def test_probe_terminal_spawn_missing_cwd(self):
|
||||
res = nmp.probe_terminal_spawn("/nonexistent_directory_for_test")
|
||||
self.assertFalse(res["healthy"])
|
||||
self.assertEqual(res["error_type"], "missing cwd")
|
||||
|
||||
def test_probe_terminal_spawn_honors_custom_command(self):
|
||||
res = nmp.probe_terminal_spawn(
|
||||
command=[sys.executable, "-c", "raise SystemExit(3)"]
|
||||
)
|
||||
self.assertTrue(res["healthy"])
|
||||
self.assertEqual(res["command"][0], sys.executable)
|
||||
self.assertEqual(res["exit_code"], 3)
|
||||
|
||||
@unittest.skipUnless(shutil.which("git"), "git is required for git -C probe")
|
||||
def test_probe_terminal_spawn_supports_explicit_git_c_worktree(self):
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
subprocess.run(
|
||||
["git", "init", tmp],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
res = nmp.probe_terminal_spawn(
|
||||
cwd="/",
|
||||
command=["git", "-C", tmp, "status", "--short"],
|
||||
)
|
||||
|
||||
self.assertTrue(res["healthy"])
|
||||
self.assertEqual(res["command"][1], "-C")
|
||||
|
||||
def test_probe_terminal_spawn_blocks_missing_custom_command(self):
|
||||
res = nmp.probe_terminal_spawn(command=["nonexistent_exec_for_test"])
|
||||
self.assertFalse(res["healthy"])
|
||||
self.assertEqual(res["error_type"], "missing executable")
|
||||
self.assertEqual(res["command"], ["nonexistent_exec_for_test"])
|
||||
|
||||
|
||||
class TestNativeMcpPreferenceGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
nmp.clear_shell_health_for_tests()
|
||||
@@ -118,4 +203,4 @@ class TestNativeMcpPreferenceGate(unittest.TestCase):
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
||||
@@ -96,6 +96,14 @@ class TestResolveTaskCapability(unittest.TestCase):
|
||||
"GITEA_TOKEN_MERGER": "merger-pass",
|
||||
}
|
||||
|
||||
def test_diagnose_terminal_success_has_no_error(self):
|
||||
res = mcp_server.gitea_diagnose_terminal(
|
||||
command=[sys.executable, "-c", "raise SystemExit(0)"]
|
||||
)
|
||||
self.assertTrue(res["healthy"])
|
||||
self.assertIsNone(res["error_type"])
|
||||
self.assertEqual(res["error_msg"], "")
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolve_author_capable_task(self, _auth, _api):
|
||||
@@ -110,6 +118,39 @@ class TestResolveTaskCapability(unittest.TestCase):
|
||||
self.assertFalse(res["different_mcp_namespace_required"])
|
||||
self.assertIn("ready for operations", res["exact_safe_next_action"])
|
||||
|
||||
@patch("mcp_server.native_mcp_preference.probe_terminal_spawn")
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolve_mutation_task_blocks_when_terminal_launcher_unhealthy(
|
||||
self,
|
||||
_auth,
|
||||
_api,
|
||||
mock_probe,
|
||||
):
|
||||
mock_probe.return_value = {
|
||||
"healthy": False,
|
||||
"error_type": "missing cwd",
|
||||
"error_msg": "Current working directory does not exist: /missing",
|
||||
"cwd": "/missing",
|
||||
"command": ["git", "--version"],
|
||||
"diagnostics": {"cwd_exists": False},
|
||||
}
|
||||
env = self._env("author-profile")
|
||||
# Probe is skipped under pytest unless forced (mirrors production path).
|
||||
env["GITEA_FORCE_TERMINAL_PROBE"] = "1"
|
||||
with patch.dict(os.environ, env):
|
||||
res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
|
||||
|
||||
self.assertFalse(res["allowed_in_current_session"])
|
||||
self.assertTrue(res["stop_required"])
|
||||
self.assertTrue(res["infra_stop"])
|
||||
self.assertTrue(res["terminal_launcher_unhealthy"])
|
||||
self.assertTrue(res["blocked"])
|
||||
self.assertEqual(res["blocked_reason"], "BLOCKED + DIAGNOSE")
|
||||
self.assertEqual(res["terminal_launcher_diagnostics"]["error_type"], "missing cwd")
|
||||
self.assertIn("terminal-launcher-failure", res["exact_safe_next_action"])
|
||||
self.assertIn("BLOCKED + DIAGNOSE", res["exact_safe_next_action"])
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolve_reviewer_capable_task_blocked(self, _auth, _api):
|
||||
|
||||
@@ -98,6 +98,45 @@ class TestReviewerLeaseFreshness(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestReviewerLeaseNewestWins(unittest.TestCase):
|
||||
"""#577: terminal release must supersede older claimed markers."""
|
||||
|
||||
def test_released_after_claimed_has_no_active_lease(self):
|
||||
claim = _lease_comment(516, "session-a", phase="claimed")
|
||||
claim["id"] = 1
|
||||
release = _lease_comment(516, "session-a", phase="released")
|
||||
release["id"] = 2
|
||||
active = leases.find_active_reviewer_lease(
|
||||
[claim, release], pr_number=516
|
||||
)
|
||||
self.assertIsNone(active)
|
||||
|
||||
def test_new_claim_after_release_is_active(self):
|
||||
claim = _lease_comment(516, "session-a", phase="claimed")
|
||||
claim["id"] = 1
|
||||
release = _lease_comment(516, "session-a", phase="released")
|
||||
release["id"] = 2
|
||||
reclaim = _lease_comment(516, "session-b", phase="claimed")
|
||||
reclaim["id"] = 3
|
||||
active = leases.find_active_reviewer_lease(
|
||||
[claim, release, reclaim], pr_number=516
|
||||
)
|
||||
self.assertIsNotNone(active)
|
||||
self.assertEqual(active.get("session_id"), "session-b")
|
||||
self.assertEqual(active.get("phase"), "claimed")
|
||||
self.assertEqual(active.get("comment_id"), 3)
|
||||
|
||||
def test_done_after_claimed_has_no_active_lease(self):
|
||||
claim = _lease_comment(516, "session-a", phase="claimed")
|
||||
claim["id"] = 10
|
||||
done = _lease_comment(516, "session-a", phase="done")
|
||||
done["id"] = 11
|
||||
active = leases.find_active_reviewer_lease(
|
||||
[claim, done], pr_number=516
|
||||
)
|
||||
self.assertIsNone(active)
|
||||
|
||||
|
||||
class TestReviewerLeaseMutationGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
leases.clear_session_lease()
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Tests for web UI CI path-filter gate (#436)."""
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from webui.ci_paths import should_run_webui_ci, touches_webui_surface
|
||||
from webui.lease_loader import load_lease_snapshot
|
||||
from webui.queue_loader import load_queue_snapshot
|
||||
from webui.runtime_health import load_runtime_snapshot
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
class TestCiPathMatching(unittest.TestCase):
|
||||
def test_positive_paths(self):
|
||||
for path in (
|
||||
"webui/app.py",
|
||||
"tests/test_webui_skeleton.py",
|
||||
"docs/webui-local-dev.md",
|
||||
"scripts/test-webui",
|
||||
):
|
||||
with self.subTest(path=path):
|
||||
self.assertTrue(touches_webui_surface(path))
|
||||
|
||||
def test_negative_paths(self):
|
||||
for path in ("mcp_server.py", "tests/test_mcp_server.py", "README.md"):
|
||||
with self.subTest(path=path):
|
||||
self.assertFalse(touches_webui_surface(path))
|
||||
|
||||
def test_should_run_aggregate(self):
|
||||
self.assertTrue(should_run_webui_ci(["README.md", "webui/layout.py"]))
|
||||
self.assertFalse(should_run_webui_ci(["mcp_server.py", "gitea_auth.py"]))
|
||||
|
||||
|
||||
class TestCiRunnerScript(unittest.TestCase):
|
||||
def test_test_webui_smoke(self):
|
||||
script = _REPO_ROOT / "scripts" / "test-webui"
|
||||
result = subprocess.run(
|
||||
["bash", str(script), "-q"],
|
||||
cwd=_REPO_ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
env={
|
||||
**os.environ,
|
||||
"WEBUI_TEST_PATTERN": "test_webui_skeleton.py",
|
||||
},
|
||||
)
|
||||
self.assertEqual(result.returncode, 0, result.stderr or result.stdout)
|
||||
|
||||
|
||||
class TestOfflineWebuiCiMode(unittest.TestCase):
|
||||
def test_offline_mode_avoids_live_queue_auth(self):
|
||||
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
|
||||
with mock.patch(
|
||||
"webui.queue_loader.get_auth_header",
|
||||
side_effect=AssertionError("live auth should not run"),
|
||||
):
|
||||
snapshot = load_queue_snapshot()
|
||||
self.assertIsNone(snapshot.fetch_error)
|
||||
self.assertEqual(snapshot.prs, ())
|
||||
self.assertEqual(snapshot.issues, ())
|
||||
|
||||
def test_offline_mode_avoids_live_lease_auth(self):
|
||||
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
|
||||
with mock.patch(
|
||||
"webui.lease_loader.get_auth_header",
|
||||
side_effect=AssertionError("live auth should not run"),
|
||||
):
|
||||
snapshot = load_lease_snapshot()
|
||||
self.assertIsNone(snapshot.fetch_error)
|
||||
self.assertEqual(snapshot.reviewer_leases, ())
|
||||
|
||||
def test_offline_mode_avoids_live_runtime_identity(self):
|
||||
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
|
||||
with mock.patch(
|
||||
"webui.runtime_health.get_auth_header",
|
||||
side_effect=AssertionError("live auth should not run"),
|
||||
):
|
||||
snapshot = load_runtime_snapshot()
|
||||
self.assertEqual(snapshot.identity_error, "offline web UI test mode")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Tests for web UI deployment and auth boundary (#435)."""
|
||||
import os
|
||||
import subprocess
|
||||
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.deployment_boundary import (
|
||||
assess_bind_host,
|
||||
runtime_assumptions,
|
||||
scan_text_for_client_secrets,
|
||||
)
|
||||
|
||||
|
||||
class TestBindAssessment(unittest.TestCase):
|
||||
def test_loopback_is_safe(self):
|
||||
for host in ("127.0.0.1", "localhost", "::1"):
|
||||
with self.subTest(host=host):
|
||||
result = assess_bind_host(host)
|
||||
self.assertEqual(result.disposition, "safe")
|
||||
|
||||
def test_all_interfaces_refused_by_default(self):
|
||||
for host in ("0.0.0.0", "::", "*"):
|
||||
with self.subTest(host=host):
|
||||
result = assess_bind_host(host)
|
||||
self.assertEqual(result.disposition, "refuse")
|
||||
|
||||
def test_all_interfaces_allowed_with_override(self):
|
||||
prior = os.environ.pop("WEBUI_ALLOW_PUBLIC_BIND", None)
|
||||
try:
|
||||
os.environ["WEBUI_ALLOW_PUBLIC_BIND"] = "1"
|
||||
result = assess_bind_host("0.0.0.0")
|
||||
self.assertEqual(result.disposition, "warn")
|
||||
finally:
|
||||
os.environ.pop("WEBUI_ALLOW_PUBLIC_BIND", None)
|
||||
if prior is not None:
|
||||
os.environ["WEBUI_ALLOW_PUBLIC_BIND"] = prior
|
||||
|
||||
def test_non_loopback_warns_without_override(self):
|
||||
prior = os.environ.pop("WEBUI_ALLOW_REMOTE_BIND", None)
|
||||
try:
|
||||
os.environ.pop("WEBUI_ALLOW_REMOTE_BIND", None)
|
||||
result = assess_bind_host("192.168.1.10")
|
||||
self.assertEqual(result.disposition, "warn")
|
||||
finally:
|
||||
if prior is not None:
|
||||
os.environ["WEBUI_ALLOW_REMOTE_BIND"] = prior
|
||||
|
||||
def test_runtime_assumptions_exclude_secrets(self):
|
||||
assumptions = runtime_assumptions()
|
||||
blob = " ".join(str(v) for v in assumptions.values())
|
||||
self.assertNotIn("GITEA_TOKEN", blob)
|
||||
self.assertIn("internal-operator-console", assumptions["deployment_mode"])
|
||||
|
||||
|
||||
class TestDeploymentRoutes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app(bind_host="127.0.0.1"))
|
||||
|
||||
def test_health_includes_deployment_metadata(self):
|
||||
response = self.client.get("/health")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertIn("deployment", data)
|
||||
deployment = data["deployment"]
|
||||
self.assertEqual(deployment["mode"], "internal-operator-console")
|
||||
self.assertEqual(deployment["mvp_auth"], "none")
|
||||
self.assertEqual(deployment["bind"]["disposition"], "safe")
|
||||
self.assertIn("runtime_assumptions", deployment)
|
||||
|
||||
def test_rendered_pages_contain_no_embedded_secrets(self):
|
||||
for path in ("/", "/projects", "/prompts", "/queue"):
|
||||
with self.subTest(path=path):
|
||||
text = self.client.get(path).text
|
||||
self.assertEqual(scan_text_for_client_secrets(text), [])
|
||||
|
||||
|
||||
class TestStartupRefusal(unittest.TestCase):
|
||||
def test_refuses_all_interface_bind(self):
|
||||
env = {**os.environ, "WEBUI_HOST": "0.0.0.0"}
|
||||
env.pop("WEBUI_ALLOW_PUBLIC_BIND", None)
|
||||
repo_root = Path(__file__).resolve().parent.parent
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "webui"],
|
||||
cwd=repo_root,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=3,
|
||||
)
|
||||
self.assertEqual(result.returncode, 1)
|
||||
|
||||
|
||||
class TestClientSecretScanner(unittest.TestCase):
|
||||
def test_detects_token_like_content(self):
|
||||
findings = scan_text_for_client_secrets("var x = GITEA_TOKEN_PRGS")
|
||||
self.assertTrue(findings)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,128 @@
|
||||
"""Tests for web UI gated action framework (#434)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from task_capability_map import TASK_CAPABILITY_MAP
|
||||
from webui.app import create_app
|
||||
from webui.gated_actions import (
|
||||
attempt_action,
|
||||
build_action_registry,
|
||||
load_action_registry,
|
||||
preview_action,
|
||||
)
|
||||
|
||||
|
||||
class TestGatedActionRegistry(unittest.TestCase):
|
||||
def test_all_actions_disabled_by_default(self):
|
||||
registry = build_action_registry()
|
||||
self.assertGreater(len(registry.actions), 0)
|
||||
for action in registry.actions:
|
||||
with self.subTest(action=action.action_id):
|
||||
self.assertFalse(action.enabled)
|
||||
|
||||
def test_actions_align_with_task_capability_map(self):
|
||||
registry = build_action_registry()
|
||||
for action in registry.actions:
|
||||
with self.subTest(task=action.task_key):
|
||||
self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
|
||||
self.assertEqual(
|
||||
action.required_permission,
|
||||
TASK_CAPABILITY_MAP[action.task_key]["permission"],
|
||||
)
|
||||
self.assertEqual(
|
||||
action.required_role,
|
||||
TASK_CAPABILITY_MAP[action.task_key]["role"],
|
||||
)
|
||||
|
||||
def test_preview_includes_mutation_ledger(self):
|
||||
preview = preview_action("merge_pr", pr_number=42)
|
||||
self.assertNotIn("error", preview)
|
||||
ledger = preview["mutation_ledger"]
|
||||
self.assertEqual(len(ledger), 1)
|
||||
self.assertEqual(ledger[0]["mcp_tool"], "gitea_merge_pr")
|
||||
self.assertEqual(ledger[0]["permission"], "gitea.pr.merge")
|
||||
self.assertIn("42", ledger[0]["target"])
|
||||
|
||||
def test_attempt_fails_closed_without_mcp_calls(self):
|
||||
registry = build_action_registry()
|
||||
with patch("webui.gated_actions._MVP_ACTIONS_ENABLED", False):
|
||||
for action in registry.actions:
|
||||
with self.subTest(action=action.action_id):
|
||||
result = attempt_action(action.action_id, issue_number=1)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["error"], "action_disabled")
|
||||
self.assertIn("preview", result)
|
||||
|
||||
def test_attempt_module_has_no_mcp_imports(self):
|
||||
"""Ungated execution path must not import MCP server tools."""
|
||||
import webui.gated_actions as ga
|
||||
|
||||
source_path = Path(ga.__file__).resolve()
|
||||
source = source_path.read_text(encoding="utf-8")
|
||||
self.assertNotIn("mcp_server", source)
|
||||
result = attempt_action("merge_pr", pr_number=99)
|
||||
self.assertFalse(result["success"])
|
||||
|
||||
def test_unknown_action_fails_closed(self):
|
||||
result = attempt_action("nonexistent_action")
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["error"], "unknown_action")
|
||||
|
||||
|
||||
class TestGatedActionRoutes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_actions_page_renders_disabled_buttons(self):
|
||||
response = self.client.get("/actions")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Gated actions", response.text)
|
||||
self.assertIn("disabled", response.text.lower())
|
||||
self.assertIn("mutation ledger", response.text.lower())
|
||||
self.assertIn("aria-disabled", response.text)
|
||||
self.assertIn(" disabled", response.text)
|
||||
|
||||
def test_api_actions_registry(self):
|
||||
response = self.client.get("/api/actions")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertFalse(data["mvp_actions_enabled"])
|
||||
action_ids = {item["action_id"] for item in data["actions"]}
|
||||
self.assertIn("merge_pr", action_ids)
|
||||
self.assertIn("claim_issue", action_ids)
|
||||
for item in data["actions"]:
|
||||
self.assertFalse(item["enabled"])
|
||||
|
||||
def test_api_action_preview_json(self):
|
||||
response = self.client.get(
|
||||
"/api/actions/review_pr/preview?pr_number=12"
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertEqual(data["task_key"], "review_pr")
|
||||
self.assertEqual(data["required_role"], "reviewer")
|
||||
self.assertEqual(len(data["mutation_ledger"]), 1)
|
||||
|
||||
def test_post_attempt_fails_closed(self):
|
||||
response = self.client.post(
|
||||
"/api/actions/merge_pr/attempt",
|
||||
json={"pr_number": 1},
|
||||
)
|
||||
self.assertEqual(response.status_code, 403)
|
||||
data = response.json()
|
||||
self.assertFalse(data["success"])
|
||||
self.assertEqual(data["error"], "action_disabled")
|
||||
|
||||
def test_nav_includes_actions_link(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn('href="/actions"', text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Consolidated MVP coverage for the internal web UI (#436)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
from starlette.routing import Route
|
||||
|
||||
from webui.app import create_app
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
# HTML pages and JSON exports registered on master MVP.
|
||||
MVP_GET_ROUTES: tuple[tuple[str, str | tuple[str, ...]], ...] = (
|
||||
("/", "Operator console"),
|
||||
("/health", "mcp-control-plane-webui"),
|
||||
("/queue", "Live queue"),
|
||||
("/api/queue", ("prs", "issues")),
|
||||
("/projects", "Gitea-Tools"),
|
||||
("/api/projects", "projects"),
|
||||
("/prompts", "Prompt library"),
|
||||
("/api/prompts", "prompts"),
|
||||
("/runtime", "Runtime"),
|
||||
("/audit", "Audit"),
|
||||
("/worktrees", "Worktrees"),
|
||||
("/leases", "Leases"),
|
||||
)
|
||||
|
||||
MUTATION_METHODS = ("POST", "PUT", "PATCH", "DELETE")
|
||||
|
||||
|
||||
def _import_optional(module_name: str):
|
||||
try:
|
||||
return importlib.import_module(module_name)
|
||||
except ImportError:
|
||||
return None
|
||||
|
||||
|
||||
class TestWebuiServerStartup(unittest.TestCase):
|
||||
def test_create_app_imports_cleanly(self):
|
||||
app = create_app()
|
||||
self.assertIsNotNone(app)
|
||||
|
||||
def test_main_module_imports(self):
|
||||
import webui.__main__ as main_mod
|
||||
|
||||
self.assertTrue(callable(main_mod.main))
|
||||
|
||||
|
||||
class TestWebuiRouteRendering(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_all_mvp_get_routes_render(self):
|
||||
for path, needle in MVP_GET_ROUTES:
|
||||
with self.subTest(path=path):
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 200, path)
|
||||
if path == "/health":
|
||||
assert isinstance(needle, str)
|
||||
self.assertEqual(response.json()["service"], needle)
|
||||
elif path.startswith("/api/"):
|
||||
data = response.json()
|
||||
if isinstance(needle, tuple):
|
||||
for key in needle:
|
||||
self.assertIn(key, data)
|
||||
else:
|
||||
self.assertIn(needle, data)
|
||||
else:
|
||||
assert isinstance(needle, str)
|
||||
self.assertIn(needle, response.text)
|
||||
|
||||
def test_registered_routes_include_mvp_paths(self):
|
||||
app = create_app()
|
||||
get_paths = {
|
||||
route.path
|
||||
for route in app.routes
|
||||
if isinstance(route, Route) and "GET" in route.methods
|
||||
}
|
||||
for path, _ in MVP_GET_ROUTES:
|
||||
with self.subTest(path=path):
|
||||
self.assertIn(path, get_paths)
|
||||
|
||||
|
||||
class TestWebuiReadOnlyGuards(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_mutation_methods_rejected_on_core_paths(self):
|
||||
paths = ("/health", "/queue", "/projects", "/prompts", "/api/queue")
|
||||
for path in paths:
|
||||
for method in MUTATION_METHODS:
|
||||
with self.subTest(path=path, method=method):
|
||||
response = self.client.request(method, path)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
|
||||
class TestWebuiOptionalModules(unittest.TestCase):
|
||||
"""Exercise child-issue modules when present on the branch under test."""
|
||||
|
||||
def test_audit_validator_basics_when_present(self):
|
||||
mod = _import_optional("webui.audit_validator")
|
||||
if mod is None:
|
||||
self.skipTest("webui.audit_validator not merged on this branch")
|
||||
report = "## Controller Handoff\n- Task: review PR #1\n"
|
||||
self.assertEqual(mod.infer_task_kind(report), "review_pr")
|
||||
result = mod.audit_report(report)
|
||||
self.assertIn("grade", result)
|
||||
|
||||
def test_gated_actions_fail_closed_when_present(self):
|
||||
mod = _import_optional("webui.gated_actions")
|
||||
if mod is None:
|
||||
self.skipTest("webui.gated_actions not merged on this branch")
|
||||
registry = mod.build_action_registry()
|
||||
for action in registry.actions:
|
||||
with self.subTest(action=action.action_id):
|
||||
self.assertFalse(action.enabled)
|
||||
result = mod.attempt_action("merge_pr", pr_number=1)
|
||||
self.assertFalse(result["success"])
|
||||
|
||||
def test_deployment_boundary_when_present(self):
|
||||
mod = _import_optional("webui.deployment_boundary")
|
||||
if mod is None:
|
||||
self.skipTest("webui.deployment_boundary not merged on this branch")
|
||||
assessment = mod.assess_bind_host("0.0.0.0")
|
||||
self.assertEqual(assessment.disposition, "refuse")
|
||||
|
||||
|
||||
class TestWebuiTestInventory(unittest.TestCase):
|
||||
def test_webui_test_modules_exist(self):
|
||||
modules = sorted(_REPO_ROOT.glob("tests/test_webui_*.py"))
|
||||
names = [path.name for path in modules]
|
||||
self.assertIn("test_webui_skeleton.py", names)
|
||||
self.assertIn("test_webui_project_registry.py", names)
|
||||
self.assertIn("test_webui_prompt_library.py", names)
|
||||
self.assertIn("test_webui_queue_dashboard.py", names)
|
||||
self.assertGreaterEqual(len(names), 5)
|
||||
|
||||
|
||||
class TestWebuiCiScripts(unittest.TestCase):
|
||||
def test_runner_scripts_exist(self):
|
||||
for name in ("test-webui", "ci-webui-check"):
|
||||
script = _REPO_ROOT / "scripts" / name
|
||||
self.assertTrue(script.is_file(), name)
|
||||
self.assertTrue(os.access(script, os.X_OK), name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -206,8 +206,9 @@ class TestQueueRoutes(unittest.TestCase):
|
||||
self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
|
||||
|
||||
def test_queue_fail_closed_without_credentials(self):
|
||||
with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
|
||||
snapshot = load_queue_snapshot()
|
||||
with mock.patch.dict("os.environ", {"WEBUI_TEST_OFFLINE": "0"}):
|
||||
with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
|
||||
snapshot = load_queue_snapshot()
|
||||
self.assertIsNotNone(snapshot.fetch_error)
|
||||
self.assertEqual(len(snapshot.prs), 0)
|
||||
|
||||
@@ -285,4 +286,4 @@ class TestQueueFailClosedUx(unittest.TestCase):
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
||||
@@ -22,6 +22,8 @@ class TestWebuiSkeleton(unittest.TestCase):
|
||||
self.assertEqual(data["service"], "mcp-control-plane-webui")
|
||||
self.assertEqual(data["mode"], "read-only-mvp")
|
||||
self.assertIn("timestamp", data)
|
||||
self.assertIn("deployment", data)
|
||||
self.assertEqual(data["deployment"]["mode"], "internal-operator-console")
|
||||
|
||||
def test_home_renders(self):
|
||||
response = self.client.get("/")
|
||||
@@ -61,6 +63,11 @@ class TestWebuiSkeleton(unittest.TestCase):
|
||||
self.assertIn("Collision warnings", response.text)
|
||||
|
||||
|
||||
def test_actions_is_implemented(self):
|
||||
response = self.client.get("/actions")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Gated actions", response.text)
|
||||
|
||||
def test_post_is_rejected(self):
|
||||
response = self.client.post("/health")
|
||||
self.assertEqual(response.status_code, 405)
|
||||
@@ -72,8 +79,8 @@ class TestWebuiSkeleton(unittest.TestCase):
|
||||
self.assertIn("Live queue", response.text)
|
||||
|
||||
def test_nav_links_on_all_pages(self):
|
||||
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
|
||||
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
|
||||
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
|
||||
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
|
||||
for path in paths:
|
||||
with self.subTest(path=path):
|
||||
text = self.client.get(path).text
|
||||
|
||||
Reference in New Issue
Block a user