sysadminandClaude Opus 4.8
b2f6e9a6dc
Evolve the MVP project registry (#427) into a versioned, fail-closed project registry API for the console (Phase 1, read-only). - Add schema version 2 with project `status`, per-step onboarding `state`/`required`, optional redacted `last_seen_health`, and `remote_name`. Version 1 files stay loadable and are normalized with explicit defaults. - Serve `/api/v1/projects` and `/api/v1/projects/{project_id}` with API provenance (`api_version`, `schema_version`, `source`). `/api/projects` is retained as an unversioned Phase 1 alias. - Replace bare `ValueError` with `RegistryError`, carrying an operator `remediation` and `field_path`; invalid registries fail closed as a 500 JSON payload or a dedicated HTML error page instead of a traceback. - Reject credential-shaped keys before any DTO is built, reusing `registry_safety.is_forbidden_key` as the single source of truth shared with the worker registry (#798). - Render HTML views from `project_to_dict`, so the console and the JSON API cannot disagree about status or onboarding progress. - Document the contract in docs/webui-project-registry-api.md. Tests: registry load/validate (valid, missing project, schema validation, credential rejection) and API route coverage. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
417 lines
17 KiB
Python
417 lines
17 KiB
Python
"""Tests for web UI project registry (#427) and its API evolution (#635)."""
|
|
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.project_registry import (
|
|
CURRENT_SCHEMA_VERSION,
|
|
REGISTRY_API_VERSION,
|
|
SUPPORTED_SCHEMA_VERSIONS,
|
|
RegistryError,
|
|
default_registry_path,
|
|
load_registry,
|
|
onboarding_summary,
|
|
project_to_dict,
|
|
)
|
|
from webui.registry_safety import is_forbidden_key
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
_API_DOC = _REPO_ROOT / "docs" / "webui-project-registry-api.md"
|
|
|
|
|
|
def _valid_project(**overrides):
|
|
project = {
|
|
"id": "example",
|
|
"repo_name": "Example",
|
|
"gitea_owner": "Org",
|
|
"remote_host": "https://gitea.example.invalid",
|
|
"default_branch": "main",
|
|
"local_checkout_path": ".",
|
|
"profiles": {"author": "a", "reviewer": "r", "reconciler": "c"},
|
|
"workflow_paths": {"skill": "skills/x.md"},
|
|
}
|
|
project.update(overrides)
|
|
return project
|
|
|
|
|
|
def _write_registry(payload) -> Path:
|
|
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
|
|
json.dump(payload, handle)
|
|
return Path(handle.name)
|
|
|
|
|
|
class RegistryFileCase(unittest.TestCase):
|
|
"""Base class that cleans up temporary registry files."""
|
|
|
|
def setUp(self):
|
|
self._temp_paths: list[Path] = []
|
|
|
|
def tearDown(self):
|
|
for path in self._temp_paths:
|
|
path.unlink(missing_ok=True)
|
|
|
|
def write_registry(self, payload) -> Path:
|
|
path = _write_registry(payload)
|
|
self._temp_paths.append(path)
|
|
return path
|
|
|
|
|
|
class TestProjectRegistryLoader(RegistryFileCase):
|
|
def test_default_registry_loads_gitea_tools(self):
|
|
registry = load_registry()
|
|
self.assertEqual(registry.version, CURRENT_SCHEMA_VERSION)
|
|
self.assertEqual(registry.schema_version, CURRENT_SCHEMA_VERSION)
|
|
self.assertEqual(registry.api_version, REGISTRY_API_VERSION)
|
|
self.assertEqual(len(registry.projects), 1)
|
|
project = registry.projects[0]
|
|
self.assertEqual(project.id, "gitea-tools")
|
|
self.assertEqual(project.repo_name, "Gitea-Tools")
|
|
self.assertEqual(project.gitea_owner, "Scaled-Tech-Consulting")
|
|
self.assertEqual(project.repo_full_name, "Scaled-Tech-Consulting/Gitea-Tools")
|
|
self.assertEqual(project.remote_host, "https://gitea.prgs.cc")
|
|
self.assertEqual(project.remote_name, "prgs")
|
|
self.assertEqual(project.status, "active")
|
|
self.assertEqual(project.profiles["author"], "prgs-author")
|
|
self.assertEqual(project.profiles["reviewer"], "prgs-reviewer")
|
|
self.assertEqual(project.profiles["reconciler"], "prgs-reconciler")
|
|
self.assertIn("skill", project.workflow_paths)
|
|
self.assertGreaterEqual(len(project.onboarding_checklist), 4)
|
|
|
|
def test_default_registry_onboarding_summary_is_complete(self):
|
|
summary = onboarding_summary(load_registry().projects[0])
|
|
self.assertEqual(summary.total, summary.complete)
|
|
self.assertEqual(summary.required_outstanding, 0)
|
|
self.assertTrue(summary.onboarding_complete)
|
|
|
|
def test_version_1_registry_still_loads_with_defaults(self):
|
|
path = self.write_registry({
|
|
"version": 1,
|
|
"projects": [
|
|
_valid_project(
|
|
onboarding_checklist=[
|
|
{"id": "step", "title": "Step", "description": "Do it"}
|
|
]
|
|
)
|
|
],
|
|
})
|
|
registry = load_registry(path)
|
|
self.assertEqual(registry.schema_version, 1)
|
|
self.assertIn(1, SUPPORTED_SCHEMA_VERSIONS)
|
|
project = registry.projects[0]
|
|
self.assertEqual(project.status, "active")
|
|
self.assertIsNone(project.remote_name)
|
|
self.assertIsNone(project.last_seen_health)
|
|
step = project.onboarding_checklist[0]
|
|
self.assertEqual(step.state, "pending")
|
|
self.assertTrue(step.required)
|
|
self.assertFalse(onboarding_summary(project).onboarding_complete)
|
|
|
|
def test_onboarding_summary_counts_states(self):
|
|
path = self.write_registry({
|
|
"version": 2,
|
|
"projects": [
|
|
_valid_project(
|
|
onboarding_checklist=[
|
|
{"id": "a", "title": "A", "description": "d", "state": "complete"},
|
|
{"id": "b", "title": "B", "description": "d", "state": "blocked"},
|
|
{
|
|
"id": "c",
|
|
"title": "C",
|
|
"description": "d",
|
|
"state": "pending",
|
|
"required": False,
|
|
},
|
|
{
|
|
"id": "d",
|
|
"title": "D",
|
|
"description": "d",
|
|
"state": "not_applicable",
|
|
},
|
|
]
|
|
)
|
|
],
|
|
})
|
|
summary = onboarding_summary(load_registry(path).projects[0])
|
|
self.assertEqual(summary.total, 4)
|
|
self.assertEqual(summary.complete, 1)
|
|
self.assertEqual(summary.blocked, 1)
|
|
self.assertEqual(summary.pending, 1)
|
|
self.assertEqual(summary.not_applicable, 1)
|
|
# Only the blocked step is both required and outstanding.
|
|
self.assertEqual(summary.required_outstanding, 1)
|
|
self.assertFalse(summary.onboarding_complete)
|
|
|
|
def test_last_seen_health_is_parsed_when_present(self):
|
|
path = self.write_registry({
|
|
"version": 2,
|
|
"projects": [
|
|
_valid_project(
|
|
last_seen_health={
|
|
"status": "degraded",
|
|
"checked_at": "2026-01-01T00:00:00Z",
|
|
"detail": "daemon restart pending",
|
|
}
|
|
)
|
|
],
|
|
})
|
|
health = load_registry(path).projects[0].last_seen_health
|
|
self.assertIsNotNone(health)
|
|
self.assertEqual(health.status, "degraded")
|
|
self.assertEqual(health.checked_at, "2026-01-01T00:00:00Z")
|
|
|
|
def test_registry_rejects_credential_keys(self):
|
|
path = self.write_registry({
|
|
"version": 1,
|
|
"projects": [_valid_project(id="bad", api_token="redacted-placeholder")],
|
|
})
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertIn("credential", ctx.exception.remediation.lower())
|
|
self.assertEqual(ctx.exception.field_path, "projects[0].api_token")
|
|
|
|
def test_unsupported_version_fails_closed_with_remediation(self):
|
|
path = self.write_registry({"version": 99, "projects": [_valid_project()]})
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertIn("unsupported registry version", ctx.exception.message)
|
|
self.assertIn(str(CURRENT_SCHEMA_VERSION), ctx.exception.remediation)
|
|
self.assertEqual(ctx.exception.field_path, "version")
|
|
|
|
def test_missing_required_field_fails_closed(self):
|
|
broken = _valid_project()
|
|
del broken["default_branch"]
|
|
path = self.write_registry({"version": 2, "projects": [broken]})
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertIn("default_branch", ctx.exception.message)
|
|
self.assertEqual(ctx.exception.field_path, "projects[0]")
|
|
|
|
def test_unknown_status_fails_closed(self):
|
|
path = self.write_registry({
|
|
"version": 2,
|
|
"projects": [_valid_project(status="mystery")],
|
|
})
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertEqual(ctx.exception.field_path, "projects[0].status")
|
|
self.assertIn("active", ctx.exception.remediation)
|
|
|
|
def test_unknown_onboarding_state_fails_closed(self):
|
|
path = self.write_registry({
|
|
"version": 2,
|
|
"projects": [
|
|
_valid_project(
|
|
onboarding_checklist=[
|
|
{"id": "a", "title": "A", "description": "d", "state": "almost"}
|
|
]
|
|
)
|
|
],
|
|
})
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertEqual(
|
|
ctx.exception.field_path,
|
|
"projects[0].onboarding_checklist[0].state",
|
|
)
|
|
|
|
def test_missing_profile_role_fails_closed(self):
|
|
path = self.write_registry({
|
|
"version": 2,
|
|
"projects": [_valid_project(profiles={"author": "a", "reviewer": "r"})],
|
|
})
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertEqual(ctx.exception.field_path, "projects[0].profiles.reconciler")
|
|
|
|
def test_empty_projects_fails_closed(self):
|
|
path = self.write_registry({"version": 2, "projects": []})
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertEqual(ctx.exception.field_path, "projects")
|
|
|
|
def test_invalid_json_fails_closed_with_location(self):
|
|
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
|
|
handle.write("{not json")
|
|
path = Path(handle.name)
|
|
self._temp_paths.append(path)
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(path)
|
|
self.assertIn("not valid JSON", ctx.exception.message)
|
|
self.assertIn("line", ctx.exception.remediation)
|
|
|
|
def test_missing_file_fails_closed(self):
|
|
missing = Path(tempfile.gettempdir()) / "webui-registry-does-not-exist.json"
|
|
with self.assertRaises(RegistryError) as ctx:
|
|
load_registry(missing)
|
|
self.assertIn("could not be read", ctx.exception.message)
|
|
|
|
def test_default_registry_path_points_at_packaged_data(self):
|
|
path = default_registry_path()
|
|
self.assertTrue(path.name == "projects.registry.json")
|
|
self.assertTrue(path.parent.name == "data")
|
|
|
|
|
|
class TestProjectRegistryRoutes(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = TestClient(create_app())
|
|
|
|
def test_projects_page_lists_gitea_tools(self):
|
|
response = self.client.get("/projects")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("Gitea-Tools", response.text)
|
|
self.assertIn("Scaled-Tech-Consulting", response.text)
|
|
self.assertIn("prgs-author", response.text)
|
|
self.assertNotIn("child issue", response.text.lower())
|
|
|
|
def test_projects_page_shows_status_and_progress(self):
|
|
response = self.client.get("/projects")
|
|
self.assertIn("Status", response.text)
|
|
self.assertIn("Onboarding", response.text)
|
|
self.assertIn("4/4 complete", response.text)
|
|
|
|
def test_project_detail_renders_checklist(self):
|
|
response = self.client.get("/projects/gitea-tools")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("Onboarding checklist", response.text)
|
|
self.assertIn("Configure execution profiles", response.text)
|
|
self.assertIn("branches/", response.text)
|
|
self.assertIn("Complete", response.text)
|
|
self.assertIn("required outstanding 0", response.text)
|
|
|
|
def test_project_detail_404(self):
|
|
response = self.client.get("/projects/unknown-repo")
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_api_projects_alias_stays_compatible(self):
|
|
response = self.client.get("/api/projects")
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
# #427 consumers keep these keys.
|
|
self.assertEqual(data["version"], CURRENT_SCHEMA_VERSION)
|
|
self.assertIn("source_path", data)
|
|
self.assertEqual(len(data["projects"]), 1)
|
|
self.assertEqual(data["projects"][0]["id"], "gitea-tools")
|
|
self.assertIn("onboarding_checklist", data["projects"][0])
|
|
|
|
def test_api_v1_projects_payload(self):
|
|
response = self.client.get("/api/v1/projects")
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
|
|
self.assertEqual(data["schema_version"], CURRENT_SCHEMA_VERSION)
|
|
self.assertEqual(data["project_count"], 1)
|
|
self.assertEqual(data["source"]["kind"], "file")
|
|
self.assertTrue(data["source"]["inventory_complete"])
|
|
project = data["projects"][0]
|
|
self.assertEqual(project["status"], "active")
|
|
self.assertEqual(project["remote_name"], "prgs")
|
|
self.assertEqual(
|
|
project["repo_full_name"], "Scaled-Tech-Consulting/Gitea-Tools"
|
|
)
|
|
self.assertTrue(project["onboarding_summary"]["onboarding_complete"])
|
|
self.assertEqual(project["onboarding_checklist"][0]["state"], "complete")
|
|
self.assertIsNone(project["last_seen_health"])
|
|
|
|
def test_api_v1_project_detail(self):
|
|
response = self.client.get("/api/v1/projects/gitea-tools")
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
|
|
self.assertEqual(data["project"]["id"], "gitea-tools")
|
|
self.assertEqual(data["source"]["kind"], "file")
|
|
|
|
def test_api_v1_project_detail_missing_fails_closed(self):
|
|
response = self.client.get("/api/v1/projects/not-registered")
|
|
self.assertEqual(response.status_code, 404)
|
|
data = response.json()
|
|
self.assertEqual(data["error"], "project_not_found")
|
|
self.assertEqual(data["project_id"], "not-registered")
|
|
self.assertIn("gitea-tools", data["known_project_ids"])
|
|
self.assertIn("remediation", data)
|
|
|
|
def test_api_v1_projects_is_read_only(self):
|
|
response = self.client.post("/api/v1/projects", json={})
|
|
self.assertEqual(response.status_code, 405)
|
|
self.assertEqual(response.json()["error"], "read-only-mvp")
|
|
|
|
def test_project_to_dict_is_json_safe(self):
|
|
registry = load_registry()
|
|
dto = project_to_dict(registry.projects[0])
|
|
encoded = json.dumps(dto)
|
|
self.assertIn("gitea-tools", encoded)
|
|
# Prose may mention tokens; no serialized *key* may look like a secret.
|
|
for key in dto:
|
|
with self.subTest(key=key):
|
|
self.assertFalse(is_forbidden_key(key))
|
|
|
|
|
|
class TestInvalidRegistryFailsClosedOverHttp(RegistryFileCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.path = self.write_registry({"version": 42, "projects": []})
|
|
self.client = TestClient(create_app())
|
|
|
|
def _with_bad_registry(self, url: str):
|
|
import os
|
|
from unittest import mock
|
|
|
|
with mock.patch.dict(
|
|
os.environ, {"WEBUI_PROJECT_REGISTRY": str(self.path)}, clear=False
|
|
):
|
|
return self.client.get(url)
|
|
|
|
def test_api_v1_reports_actionable_error(self):
|
|
response = self._with_bad_registry("/api/v1/projects")
|
|
self.assertEqual(response.status_code, 500)
|
|
data = response.json()
|
|
self.assertEqual(data["error"], "registry_invalid")
|
|
self.assertIn("unsupported registry version", data["detail"])
|
|
self.assertTrue(data["remediation"])
|
|
self.assertEqual(data["field_path"], "version")
|
|
|
|
def test_unversioned_alias_reports_actionable_error(self):
|
|
response = self._with_bad_registry("/api/projects")
|
|
self.assertEqual(response.status_code, 500)
|
|
self.assertEqual(response.json()["error"], "registry_invalid")
|
|
|
|
def test_html_page_reports_actionable_error(self):
|
|
response = self._with_bad_registry("/projects")
|
|
self.assertEqual(response.status_code, 500)
|
|
self.assertIn("Project registry unavailable", response.text)
|
|
self.assertIn("Remediation", response.text)
|
|
|
|
|
|
class TestProjectRegistryApiDocs(unittest.TestCase):
|
|
def test_api_contract_is_documented(self):
|
|
self.assertTrue(_API_DOC.is_file(), f"missing {_API_DOC}")
|
|
text = _API_DOC.read_text(encoding="utf-8")
|
|
for token in (
|
|
"/api/v1/projects",
|
|
"/api/v1/projects/{project_id}",
|
|
"/api/projects",
|
|
"onboarding_summary",
|
|
"last_seen_health",
|
|
"registry_invalid",
|
|
"#635",
|
|
):
|
|
with self.subTest(token=token):
|
|
self.assertIn(token, text)
|
|
|
|
def test_route_table_lists_versioned_routes(self):
|
|
local_dev = (_REPO_ROOT / "docs" / "webui-local-dev.md").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("/api/v1/projects", local_dev)
|
|
self.assertIn("webui-project-registry-api.md", local_dev)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|