sysadminandClaude Opus 4.8
983e8ac2c7
Add Phase 4 advisory surfaces for declared AI-provider connection status (no secrets, no live probe claims) and operational insights derived only from durable traffic, health, provider, and analytics evidence. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
405 lines
14 KiB
Python
405 lines
14 KiB
Python
"""Tests for AI-provider connections and evidence-backed insights (#650).
|
|
|
|
Covers acceptance criteria:
|
|
|
|
1. Provider connection status is redacted and accurate (declared registry only).
|
|
2. At least three insight types with evidence citations.
|
|
3. Insights never claim actions completed without proof.
|
|
4. Evidence requirement is enforced (no evidence-less insights).
|
|
5. Interpretation limits appear in docs-facing payloads.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from tests.webui_testclient import TestClient
|
|
|
|
from webui.app import create_app
|
|
from webui.insights_loader import (
|
|
CONFIDENCE_HIGH,
|
|
CONNECTION_DECLARED_AVAILABLE,
|
|
CONNECTION_DECLARED_UNAVAILABLE,
|
|
INSIGHT_BLOCKED_QUEUE,
|
|
INSIGHT_CONTROLLER_ATTENTION,
|
|
INSIGHT_PROVIDER_WITHOUT_WORKERS,
|
|
INSIGHT_STALE_RUNTIME,
|
|
ProviderConnection,
|
|
build_provider_connection,
|
|
generate_insights,
|
|
insight_blocked_queue,
|
|
insight_controller_attention,
|
|
insight_providers_without_workers,
|
|
insight_stale_runtime,
|
|
load_insights_snapshot,
|
|
load_provider_snapshot,
|
|
snapshot_insights_to_dict,
|
|
snapshot_providers_to_dict,
|
|
)
|
|
from webui.insights_views import render_insights_page, render_providers_page
|
|
from webui.nav import STUB_PAGES, nav_hrefs
|
|
from webui.worker_registry import ProviderRecord, WorkerRecord, ScheduleSpec, SchedulerSpec
|
|
|
|
|
|
def _provider(
|
|
provider_id: str = "claude",
|
|
*,
|
|
available: bool = True,
|
|
models: tuple[str, ...] = ("claude-opus-4-8",),
|
|
notes: str = "",
|
|
) -> ProviderRecord:
|
|
return ProviderRecord(
|
|
id=provider_id,
|
|
display_name=provider_id.title(),
|
|
vendor="TestVendor",
|
|
executable=provider_id,
|
|
available=available,
|
|
models=models,
|
|
notes=notes,
|
|
)
|
|
|
|
|
|
def _worker(
|
|
worker_id: str = "claude-author",
|
|
*,
|
|
provider: str = "claude",
|
|
enabled: bool = True,
|
|
) -> WorkerRecord:
|
|
return WorkerRecord(
|
|
id=worker_id,
|
|
display_name=worker_id,
|
|
provider=provider,
|
|
model="m1",
|
|
project="gitea-tools",
|
|
role="author",
|
|
namespace="gitea-author",
|
|
profile="prgs-author",
|
|
workflow="skills/llm-project-workflow/workflows/work-issue.md",
|
|
schedule=ScheduleSpec(kind="manual", seconds=None, expression=None),
|
|
timeout_seconds=3600,
|
|
enabled=enabled,
|
|
scheduler=SchedulerSpec(kind="manual", label=None),
|
|
notes="",
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _TrafficItem:
|
|
kind: str
|
|
number: int
|
|
title: str = ""
|
|
traffic_state: str = "blocked"
|
|
expected_role: str = "author"
|
|
safe_for_roles: tuple[str, ...] = ()
|
|
badges: tuple[str, ...] = ()
|
|
block_reason: str | None = "dependency"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Traffic:
|
|
blocked: tuple = ()
|
|
needs_controller: tuple = ()
|
|
inventory_complete: bool = True
|
|
fetch_error: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Stale:
|
|
daemon_head: str | None
|
|
checkout_head: str | None
|
|
remote_head: str | None
|
|
stale: bool
|
|
determinable: bool
|
|
mutation_safe: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Health:
|
|
stale_runtime: _Stale | None
|
|
|
|
|
|
class TestProviderConnections(unittest.TestCase):
|
|
def test_available_provider_status(self):
|
|
conn = build_provider_connection(_provider(available=True), (_worker(),))
|
|
self.assertEqual(conn.connection_status, CONNECTION_DECLARED_AVAILABLE)
|
|
self.assertTrue(conn.available_declared)
|
|
self.assertEqual(conn.worker_count, 1)
|
|
self.assertEqual(conn.enabled_worker_count, 1)
|
|
self.assertFalse(conn.to_dict()["secrets_exposed"])
|
|
|
|
def test_unavailable_provider_status(self):
|
|
conn = build_provider_connection(_provider(available=False), ())
|
|
self.assertEqual(conn.connection_status, CONNECTION_DECLARED_UNAVAILABLE)
|
|
self.assertEqual(conn.worker_count, 0)
|
|
|
|
def test_notes_are_redacted(self):
|
|
conn = build_provider_connection(
|
|
_provider(notes="token=ghp_thisisnotarealsecretvalue0001"),
|
|
(),
|
|
)
|
|
self.assertNotIn("ghp_thisisnotarealsecretvalue0001", conn.notes)
|
|
self.assertNotIn(
|
|
"ghp_thisisnotarealsecretvalue0001",
|
|
json.dumps(conn.to_dict()),
|
|
)
|
|
|
|
def test_load_provider_snapshot_from_injected_registry(self):
|
|
from webui.worker_registry import WorkerRegistry
|
|
from pathlib import Path
|
|
|
|
registry = WorkerRegistry(
|
|
version=1,
|
|
revision=3,
|
|
updated_at="2026-07-25T00:00:00Z",
|
|
providers=(_provider("claude"), _provider("grok", available=False)),
|
|
workers=(_worker(),),
|
|
source_path=Path("/tmp/workers.registry.json"),
|
|
)
|
|
snapshot = load_provider_snapshot(registry=registry)
|
|
self.assertTrue(snapshot.ok)
|
|
self.assertEqual(snapshot.registry_revision, 3)
|
|
ids = {p.provider_id for p in snapshot.providers}
|
|
self.assertEqual(ids, {"claude", "grok"})
|
|
|
|
def test_registry_failure_is_fail_closed(self):
|
|
def _boom():
|
|
raise RuntimeError("disk gone")
|
|
|
|
snapshot = load_provider_snapshot(registry_loader=_boom)
|
|
self.assertFalse(snapshot.ok)
|
|
self.assertIn("unavailable", snapshot.fetch_error or "")
|
|
self.assertEqual(snapshot.providers, ())
|
|
|
|
|
|
class TestInsightGenerators(unittest.TestCase):
|
|
def test_blocked_queue_requires_evidence(self):
|
|
traffic = _Traffic(
|
|
blocked=(
|
|
_TrafficItem(kind="issue", number=647, block_reason="depends #646"),
|
|
_TrafficItem(kind="pr", number=902, block_reason="conflict"),
|
|
)
|
|
)
|
|
insight = insight_blocked_queue(traffic)
|
|
self.assertIsNotNone(insight)
|
|
self.assertEqual(insight.kind, INSIGHT_BLOCKED_QUEUE)
|
|
self.assertGreaterEqual(len(insight.evidence), 2)
|
|
self.assertTrue(insight.advisory_only)
|
|
self.assertFalse(insight.claims_action_completed)
|
|
refs = {e.ref for e in insight.evidence}
|
|
self.assertIn("#647", refs)
|
|
self.assertIn("#902", refs)
|
|
|
|
def test_empty_blocked_queue_yields_no_insight(self):
|
|
self.assertIsNone(insight_blocked_queue(_Traffic()))
|
|
|
|
def test_controller_attention_insight(self):
|
|
traffic = _Traffic(
|
|
needs_controller=(_TrafficItem(kind="issue", number=100, traffic_state="needs_controller"),)
|
|
)
|
|
insight = insight_controller_attention(traffic)
|
|
self.assertEqual(insight.kind, INSIGHT_CONTROLLER_ATTENTION)
|
|
self.assertEqual(insight.evidence[0].ref, "#100")
|
|
|
|
def test_stale_runtime_insight(self):
|
|
health = _Health(
|
|
stale_runtime=_Stale(
|
|
daemon_head="aaa",
|
|
checkout_head="bbb",
|
|
remote_head="ccc",
|
|
stale=True,
|
|
determinable=True,
|
|
mutation_safe=False,
|
|
)
|
|
)
|
|
insight = insight_stale_runtime(health)
|
|
self.assertEqual(insight.kind, INSIGHT_STALE_RUNTIME)
|
|
self.assertIn("stale", insight.evidence[0].detail)
|
|
self.assertFalse(insight.claims_action_completed)
|
|
|
|
def test_mutation_safe_runtime_yields_no_insight(self):
|
|
health = _Health(
|
|
stale_runtime=_Stale(
|
|
daemon_head="aaa",
|
|
checkout_head="aaa",
|
|
remote_head="aaa",
|
|
stale=False,
|
|
determinable=True,
|
|
mutation_safe=True,
|
|
)
|
|
)
|
|
self.assertIsNone(insight_stale_runtime(health))
|
|
|
|
def test_provider_without_workers(self):
|
|
providers = (
|
|
build_provider_connection(_provider("claude"), (_worker(),)),
|
|
build_provider_connection(_provider("grok"), ()),
|
|
)
|
|
insight = insight_providers_without_workers(providers)
|
|
self.assertEqual(insight.kind, INSIGHT_PROVIDER_WITHOUT_WORKERS)
|
|
self.assertEqual(insight.evidence[0].ref, "grok")
|
|
|
|
def test_generate_insights_composes_three_kinds(self):
|
|
from webui.worker_registry import WorkerRegistry
|
|
from pathlib import Path
|
|
|
|
registry = WorkerRegistry(
|
|
version=1,
|
|
revision=1,
|
|
updated_at="2026-07-25T00:00:00Z",
|
|
providers=(_provider("lonely"),),
|
|
workers=(),
|
|
source_path=Path("/tmp/w.json"),
|
|
)
|
|
provider_snapshot = load_provider_snapshot(registry=registry)
|
|
traffic = _Traffic(
|
|
blocked=(_TrafficItem(kind="issue", number=1),),
|
|
needs_controller=(_TrafficItem(kind="issue", number=2),),
|
|
)
|
|
health = _Health(
|
|
stale_runtime=_Stale("a", "b", "c", True, True, False)
|
|
)
|
|
insights, used, unavailable = generate_insights(
|
|
traffic=traffic,
|
|
health=health,
|
|
provider_snapshot=provider_snapshot,
|
|
analytics=None,
|
|
)
|
|
kinds = {i.kind for i in insights}
|
|
self.assertIn(INSIGHT_BLOCKED_QUEUE, kinds)
|
|
self.assertIn(INSIGHT_CONTROLLER_ATTENTION, kinds)
|
|
self.assertIn(INSIGHT_STALE_RUNTIME, kinds)
|
|
self.assertIn(INSIGHT_PROVIDER_WITHOUT_WORKERS, kinds)
|
|
self.assertGreaterEqual(len(kinds), 3)
|
|
self.assertIn("traffic", used)
|
|
self.assertIn("system_health", used)
|
|
self.assertIn("providers", used)
|
|
self.assertTrue(any(u["source"] == "analytics" for u in unavailable))
|
|
for insight in insights:
|
|
self.assertTrue(insight.advisory_only)
|
|
self.assertFalse(insight.claims_action_completed)
|
|
self.assertGreaterEqual(len(insight.evidence), 1)
|
|
|
|
def test_missing_source_is_reported_not_as_healthy_empty(self):
|
|
insights, used, unavailable = generate_insights(
|
|
traffic=None,
|
|
health=None,
|
|
provider_snapshot=None,
|
|
analytics=None,
|
|
)
|
|
self.assertEqual(insights, ())
|
|
self.assertEqual(used, ())
|
|
self.assertEqual(len(unavailable), 4)
|
|
|
|
|
|
class TestViewsAndRoutes(unittest.TestCase):
|
|
def test_providers_page_renders_connections(self):
|
|
from webui.worker_registry import WorkerRegistry
|
|
from pathlib import Path
|
|
|
|
registry = WorkerRegistry(
|
|
version=1,
|
|
revision=1,
|
|
updated_at="2026-07-25T00:00:00Z",
|
|
providers=(_provider("claude"),),
|
|
workers=(_worker(),),
|
|
source_path=Path("/tmp/w.json"),
|
|
)
|
|
snapshot = load_provider_snapshot(registry=registry)
|
|
html = render_providers_page(snapshot)
|
|
self.assertIn("AI provider connections", html)
|
|
self.assertIn("claude", html)
|
|
self.assertIn("declared_available", html)
|
|
self.assertIn("Interpretation limits", html)
|
|
self.assertNotIn("api_key", html.lower())
|
|
|
|
def test_failed_provider_snapshot_renders_no_table(self):
|
|
snapshot = load_provider_snapshot(registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("x")))
|
|
html = render_providers_page(snapshot)
|
|
self.assertIn("unavailable", html.lower())
|
|
self.assertNotIn("<tbody><tr><td><code>", html)
|
|
|
|
def test_insights_page_lists_evidence(self):
|
|
traffic = _Traffic(blocked=(_TrafficItem(kind="issue", number=42),))
|
|
snapshot = load_insights_snapshot(
|
|
traffic=traffic,
|
|
health=_Health(None),
|
|
provider_snapshot=load_provider_snapshot(
|
|
registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("skip"))
|
|
),
|
|
analytics=None,
|
|
load_live=False,
|
|
)
|
|
html = render_insights_page(snapshot)
|
|
self.assertIn("#42", html)
|
|
self.assertIn("advisory only", html.lower())
|
|
self.assertIn("Evidence", html)
|
|
|
|
def test_nav_exposes_live_insights_and_providers(self):
|
|
self.assertIn("/insights", nav_hrefs())
|
|
self.assertIn("/providers", nav_hrefs())
|
|
self.assertNotIn("/insights", STUB_PAGES)
|
|
|
|
def test_routes_are_read_only_and_export_json(self):
|
|
client = TestClient(create_app())
|
|
# Use live registry from package data — should be ok.
|
|
with mock.patch(
|
|
"webui.app.load_provider_snapshot",
|
|
return_value=load_provider_snapshot(
|
|
registry=__import__(
|
|
"webui.worker_registry", fromlist=["load_registry"]
|
|
).load_registry()
|
|
),
|
|
):
|
|
response = client.get("/providers")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("provider", response.text.lower())
|
|
api = client.get("/api/v1/providers")
|
|
self.assertEqual(api.status_code, 200)
|
|
payload = api.json()
|
|
self.assertTrue(payload["ok"])
|
|
self.assertIn("interpretation_limits", payload)
|
|
self.assertTrue(all(not p.get("secrets_exposed") for p in payload["providers"]))
|
|
|
|
with mock.patch(
|
|
"webui.app.load_insights_snapshot",
|
|
return_value=load_insights_snapshot(
|
|
traffic=_Traffic(blocked=(_TrafficItem(kind="issue", number=7),)),
|
|
health=_Health(None),
|
|
provider_snapshot=load_provider_snapshot(
|
|
registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("x"))
|
|
),
|
|
analytics=None,
|
|
load_live=False,
|
|
),
|
|
):
|
|
page = client.get("/insights")
|
|
self.assertEqual(page.status_code, 200)
|
|
self.assertIn("#7", page.text)
|
|
api = client.get("/api/v1/insights")
|
|
self.assertEqual(api.status_code, 200)
|
|
body = api.json()
|
|
self.assertTrue(body["ok"])
|
|
self.assertTrue(all(i["advisory_only"] for i in body["insights"]))
|
|
self.assertTrue(all(not i["claims_action_completed"] for i in body["insights"]))
|
|
self.assertTrue(all(i["evidence"] for i in body["insights"]))
|
|
|
|
for path in ("/providers", "/api/v1/providers", "/insights", "/api/v1/insights"):
|
|
with self.subTest(path=path):
|
|
self.assertEqual(client.post(path).status_code, 405)
|
|
|
|
def test_home_nav_links_providers_and_insights(self):
|
|
home = TestClient(create_app()).get("/").text
|
|
self.assertIn('href="/providers"', home)
|
|
self.assertIn('href="/insights"', home)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|