"""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("", 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()