from __future__ import annotations import tempfile import unittest from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path from unittest.mock import MagicMock, patch from backend.sql_catalog import server class SQLCatalogDomainTest(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() self.previous_db_path = server.DB_PATH server.DB_PATH = Path(self.temp_dir.name) / "test.sqlite3" server.init_db() def tearDown(self) -> None: server.DB_PATH = self.previous_db_path self.temp_dir.cleanup() def finding_payload(self, event_id: str, digest: str = "digest-test") -> dict[str, object]: return { "source": "test", "source_type": "integration", "source_event_id": event_id, "org_id": "org-test", "cluster_id": "cluster-test", "sql_digest": digest, "normalized_sql": "select * from orders where id = ?", "time_window_start": "2026-07-01T00:00:00Z", "time_window_end": "2026-07-01T01:00:00Z", "observed_at": "2026-07-01T01:00:00Z", "detector_code": "test.slow_query", "metric_schema_version": "test.slow_query/v1", "metrics": {"slow_count": 12, "max_latency_ms": 9200}, "symptoms": [ { "code": "high_total_latency", "severity": "high", "signal_codes": ["slowlog.max_latency"], "metric_keys": ["max_latency_ms"], } ], } def create_item(self, event_id: str = "event-1") -> dict[str, object]: status, body = server.create_finding(self.finding_payload(event_id)) self.assertEqual(201, status) return body["catalog_item"] def treatment_payload(self, title: str = "Improve SQL behavior") -> dict[str, object]: return { "title": title, "objective": "Reduce the operational impact of the linked SQL.", "success_criteria": "A comparable observation confirms material improvement without regression.", } def test_treatment_number_accepts_tidb_decimal_aggregate(self) -> None: conn = MagicMock() conn.execute.return_value.fetchone.return_value = [Decimal("9")] self.assertEqual("TRT-0010", server.next_treatment_no(conn)) def create_ready_analysis(self, treatment_id: str) -> dict[str, object]: status, body = server.create_record( "treatment", treatment_id, { "record_type": "analysis_record", "title": "SQL diagnosis", "review_status": "ready", "structured_fields": { "conclusion_type": "index_and_scan_review", "confidence": "high", "risk_level": "medium", }, "body_markdown": "The evidence supports preparing a concrete optimization plan.", }, ) self.assertEqual(201, status) return body["record"] def move_treatment_to_execution(self, treatment_id: str) -> None: self.create_ready_analysis(treatment_id) status, _ = server.transition_treatment(treatment_id, {"stage": "planning"}) self.assertEqual(200, status) status, _ = server.transition_treatment(treatment_id, {"stage": "execution"}) self.assertEqual(200, status) def create_reviewed_observation(self, treatment_id: str) -> dict[str, object]: status, body = server.create_record( "treatment", treatment_id, { "record_type": "observation_record", "title": "Validated improvement", "review_status": "reviewed", "structured_fields": { "baseline_window": "2026-07-01T00:00:00Z/2026-07-01T01:00:00Z", "observation_window": "2026-07-02T00:00:00Z/2026-07-02T01:00:00Z", "baseline_metrics": {"max_latency_ms": 9200}, "observation_metrics": {"max_latency_ms": 850}, "traffic_comparable": True, "conclusion_type": "fixed", "confidence": "high", }, "body_markdown": "Latency and scan volume both dropped.", }, ) self.assertEqual(201, status) return body["record"] def test_finding_ingestion_is_idempotent_and_creates_catalog_item(self) -> None: status, first = server.create_finding(self.finding_payload("same-event")) self.assertEqual(201, status) self.assertTrue(first["created"]) self.assertEqual("registered", first["catalog_item"]["current_posture"]["state"]) self.assertIsNone(first["catalog_item"]["current_posture"]["risk"]) self.assertFalse( first["catalog_item"]["current_posture"]["review_admission"]["required"] ) status, second = server.create_finding(self.finding_payload("same-event")) self.assertEqual(200, status) self.assertFalse(second["created"]) self.assertEqual(first["finding"]["id"], second["finding"]["id"]) self.assertEqual("high", first["finding"]["severity"]) self.assertEqual("high_total_latency", first["finding"]["symptoms"][0]["code"]) server.init_db() _, detail = server.finding_detail(first["finding"]["id"]) self.assertEqual(1, len(detail["timeline"])) def test_catalog_item_can_be_registered_before_a_finding(self) -> None: status, body = server.create_catalog_item( { "org_id": "org-test", "cluster_id": "cluster-test", "sql_digest": "manual-digest", "normalized_sql": "select * from orders where customer_id = ?", "registration_reason": "customer_ticket", "source_reference": "SUPPORT-4821", "source_url": "https://example.com/tickets/SUPPORT-4821", "note": "Customer asked us to review this high-volume query.", "initial_action": "register_only", "owner": "demo-user", "created_by": "demo-user", } ) self.assertEqual(201, status) item = body["catalog_item"] self.assertEqual("registered", item["current_posture"]["state"]) self.assertIsNone(item["current_posture"]["risk"]) self.assertEqual("active", item["lifecycle"]) self.assertEqual(0, item["finding_count"]) self.assertIsNone(item["first_seen_at"]) self.assertIsNone(item["latest_finding_at"]) self.assertEqual("manual", item["registration"]["origin"]) self.assertEqual("customer_ticket", item["registration"]["reason"]) detail_status, detail = server.catalog_item_detail(item["id"]) self.assertEqual(200, detail_status) self.assertEqual(0, len(detail["findings"])) registration = next( entry for entry in detail["timeline"] if entry["event_type"] == "catalog_item_created" ) self.assertEqual("SQL manually registered in Catalog", registration["title"]) self.assertTrue( any( field["label"] == "Source reference" and field["value"] == "SUPPORT-4821" for field in registration["details"] ) ) def test_catalog_finding_and_symptom_projections_support_chinese(self) -> None: status, created = server.create_finding(self.finding_payload("zh-projection")) self.assertEqual(201, status) _, catalog_detail = server.catalog_item_detail( created["catalog_item"]["id"], "zh-CN" ) registration = next( entry for entry in catalog_detail["timeline"] if entry["event_type"] == "catalog_item_created" ) self.assertIn("SQL 目录", registration["title"]) self.assertTrue( any(field["label"] == "来源" for field in registration["details"]) ) _, finding_detail = server.finding_detail(created["finding"]["id"], "zh-CN") self.assertIn("异常发现", finding_detail["timeline"][0]["summary"]) definitions = server.list_symptom_definitions({}, "zh-CN")["items"] symptom = next( item for item in definitions if item["code"] == "high_total_latency" ) self.assertEqual("高延迟贡献", symptom["name"]) self.assertEqual("延迟", symptom["category_name"]) def test_duplicate_manual_catalog_registration_returns_existing_item(self) -> None: payload = { "org_id": "org-test", "cluster_id": "cluster-test", "sql_digest": "duplicate-manual-digest", "registration_reason": "high_frequency", } first_status, first = server.create_catalog_item(payload) second_status, second = server.create_catalog_item(payload) self.assertEqual(201, first_status) self.assertEqual(409, second_status) self.assertEqual("catalog_item_exists", second["error"]) self.assertEqual(first["catalog_item"]["id"], second["catalog_item"]["id"]) self.assertEqual( "registered", first["catalog_item"]["current_posture"]["state"] ) self.assertEqual( "register_only", first["catalog_item"]["registration"]["initial_action"] ) self.assertEqual(1, server.list_catalog_items({})["total"]) def test_first_finding_reuses_a_manually_registered_catalog_item(self) -> None: status, registered = server.create_catalog_item( { "org_id": "org-test", "cluster_id": "cluster-test", "sql_digest": "manual-then-finding", "registration_reason": "proactive_watch", "initial_action": "watch", } ) self.assertEqual(201, status) status, finding_body = server.create_finding( self.finding_payload("manual-then-finding-event", "manual-then-finding") ) self.assertEqual(201, status) self.assertEqual( registered["catalog_item"]["id"], finding_body["catalog_item"]["id"] ) self.assertFalse(finding_body["finding"]["is_new_digest"]) self.assertTrue(finding_body["finding"]["is_first_observation"]) self.assertEqual( "watching", finding_body["catalog_item"]["current_posture"]["state"] ) _, detail = server.catalog_item_detail(registered["catalog_item"]["id"]) self.assertEqual(1, len(detail["findings"])) self.assertEqual("manual", detail["item"]["registration"]["origin"]) self.assertEqual("2026-07-01T00:00:00Z", detail["item"]["first_seen_at"]) self.assertFalse(detail["item"]["is_reappeared"]) self.assertFalse( any( entry["event_type"] == "catalog_item_reappeared" for entry in detail["timeline"] ) ) _, preview = server.finding_deletion_preview( {"selection": {"ids": [finding_body["finding"]["id"]]}} ) delete_status, _ = server.delete_findings( { "selection": preview["selection"], "expected_count": 1, "reason": "The first reported Finding was entered incorrectly.", } ) self.assertEqual(200, delete_status) self.assertEqual(1, server.list_catalog_items({})["total"]) _, after_delete = server.catalog_item_detail(registered["catalog_item"]["id"]) self.assertEqual("watching", after_delete["item"]["current_posture"]["state"]) self.assertIsNone(after_delete["item"]["current_posture"]["risk"]) self.assertIsNone(after_delete["item"]["current_posture"]["historical_peak_risk"]) self.assertIsNone(after_delete["item"]["first_seen_at"]) def test_catalog_registration_uses_dossier_creation_time(self) -> None: item = self.create_item("registration-chronology") with server.get_conn() as conn: conn.execute( "UPDATE catalog_items SET created_at = ? WHERE id = ?", ("2026-06-30T23:59:00Z", item["id"]), ) conn.execute( """ UPDATE timeline_events SET created_at = ? WHERE target_type = 'catalog_item' AND target_id = ? AND event_type = 'catalog_item_created' """, ("2026-07-14T06:18:41Z", item["id"]), ) _, detail = server.catalog_item_detail(item["id"]) registration = next( entry for entry in detail["timeline"] if entry["event_type"] == "catalog_item_created" ) self.assertEqual("2026-06-30T23:59:00Z", registration["occurred_at"]) self.assertEqual( "2026-06-30T23:59:00Z", detail["item"]["registration"]["registered_at"] ) self.assertEqual("catalog_item_created", detail["timeline"][-1]["event_type"]) def test_finding_requires_external_idempotency_key(self) -> None: payload = self.finding_payload("missing-event") payload.pop("source_event_id") status, body = server.create_finding(payload) self.assertEqual(400, status) self.assertEqual("invalid_finding", body["error"]) def test_catalog_timeline_projects_repeated_findings_as_one_signal_episode(self) -> None: status, first = server.create_finding(self.finding_payload("timeline-event-1")) self.assertEqual(201, status) second_payload = self.finding_payload("timeline-event-2") second_payload.update( { "time_window_start": "2026-07-02T00:00:00Z", "time_window_end": "2026-07-02T01:00:00Z", "observed_at": "2026-07-02T01:00:00Z", } ) status, second = server.create_finding(second_payload) self.assertEqual(201, status) status, detail = server.catalog_item_detail(first["catalog_item"]["id"]) self.assertEqual(200, status) signal_entries = [ entry for entry in detail["timeline"] if entry["event_type"] == "finding_signal_episode" ] self.assertEqual(1, len(signal_entries)) self.assertEqual(2, signal_entries[0]["count"]) self.assertEqual("2026-07-01T01:00:00Z", signal_entries[0]["started_at"]) self.assertEqual("2026-07-02T01:00:00Z", signal_entries[0]["occurred_at"]) self.assertEqual( [second["finding"]["id"], first["finding"]["id"]], [child["id"] for child in signal_entries[0]["children"]], ) self.assertNotIn( "finding_ingested", [entry["event_type"] for entry in detail["timeline"]], ) self.assertEqual(2, detail["timeline_meta"]["finding_count"]) self.assertEqual(1, detail["timeline_meta"]["signal_episode_count"]) def test_catalog_timeline_starts_a_new_episode_for_changed_severity(self) -> None: item = self.create_item("timeline-high") payload = self.finding_payload("timeline-critical") payload["observed_at"] = "2026-07-02T01:00:00Z" payload["symptoms"][0]["severity"] = "critical" status, _ = server.create_finding(payload) self.assertEqual(201, status) _, detail = server.catalog_item_detail(item["id"]) signal_entries = [ entry for entry in detail["timeline"] if entry["event_type"] == "finding_signal_episode" ] self.assertEqual(2, len(signal_entries)) self.assertEqual({"high", "critical"}, {entry["severity"] for entry in signal_entries}) def test_catalog_timeline_keeps_governance_milestones_without_status_sync_noise(self) -> None: item = self.create_item("timeline-governance") status, treatment_body = server.create_treatment(self.treatment_payload(), item["id"]) self.assertEqual(201, status) treatment_id = treatment_body["treatment"]["id"] self.create_ready_analysis(treatment_id) status, _ = server.transition_treatment(treatment_id, {"stage": "planning"}) self.assertEqual(200, status) _, detail = server.catalog_item_detail(item["id"]) journey = next( entry for entry in detail["timeline"] if entry["event_type"] == "treatment_journey" ) self.assertEqual("governance", journey["category"]) self.assertEqual( [ "treatment_created", "treatment_record_created", "treatment_milestone", ], [entry["event_type"] for entry in journey["children"]], ) self.assertFalse( any( entry["event_type"] == "catalog_item_status_changed" for entry in detail["timeline"] ) ) self.assertTrue( any( link.get("target_type") == "treatment" and link.get("target_id") == treatment_id for link in journey["links"] ) ) self.assertEqual(1, detail["timeline_meta"]["treatment_journey_count"]) self.assertFalse( any( entry["event_type"] == "treatment_milestone" for entry in detail["timeline"] ) ) self.assertEqual( detail["timeline_meta"]["highlight_count"], detail["timeline_meta"]["entry_count"], ) def test_catalog_metric_history_groups_activity_and_symptom_evidence(self) -> None: first_payload = self.finding_payload("metric-history-1") first_payload["metrics"] = { "slow_count": 12, "total_latency_s": 48.5, "total_keys": 900_000, "process_keys": 760_000, } first_payload["symptoms"] = [ { "code": "large_key_scan", "severity": "high", "metric_keys": ["total_keys", "process_keys"], } ] status, first = server.create_finding(first_payload) self.assertEqual(201, status) second_payload = self.finding_payload("metric-history-2") second_payload.update( { "observed_at": "2026-07-03T01:00:00Z", "time_window_start": "2026-07-03T00:00:00Z", "time_window_end": "2026-07-03T01:00:00Z", "metrics": { "total_latency_s": 31.25, "total_keys": 610_000, "process_keys": 520_000, }, "symptoms": [ { "code": "large_key_scan", "severity": "high", "metric_keys": ["total_keys", "process_keys"], } ], } ) status, second = server.create_finding(second_payload) self.assertEqual(201, status) _, detail = server.catalog_item_detail(first["catalog_item"]["id"]) history = detail["metric_history"] self.assertEqual(2, history["metric_observation_count"]) self.assertEqual( [first["finding"]["id"], second["finding"]["id"]], [point["finding_id"] for point in history["points"]], ) self.assertIn("slow_count", history["points"][0]["values"]) self.assertNotIn("slow_count", history["points"][1]["values"]) activity = next(group for group in history["groups"] if group["id"] == "activity") key_scan = next( group for group in history["groups"] if group["id"] == "symptom:large_key_scan" ) self.assertEqual(["slow_count", "total_latency_s"], activity["metric_keys"]) self.assertEqual(["total_keys", "process_keys"], key_scan["metric_keys"]) self.assertEqual("Large key scan", key_scan["title"]) _, finding_detail = server.finding_detail(second["finding"]["id"]) self.assertEqual(history, finding_detail["metric_history"]) def test_catalog_metric_windows_use_observation_time(self) -> None: findings = [ { "observed_at": "2026-07-09T12:00:00Z", "created_at": "2026-07-12T12:00:00Z", "metrics": {"slow_count": 10}, }, { "observed_at": "2026-07-11T12:00:00Z", "created_at": "2026-07-01T12:00:00Z", "metrics": {"slow_count": 5}, }, ] with patch.object(server, "since_iso", return_value="2026-07-10T00:00:00Z"): windows = server.catalog_metric_windows(findings) self.assertEqual(1, windows["24h"]["finding_count"]) self.assertEqual(5.0, windows["24h"]["slow_count"]) def test_catalog_metric_windows_respect_metric_semantics(self) -> None: findings = [ { "id": "metric-semantics-1", "observed_at": "2026-07-01T12:00:00Z", "metrics": { "slow_count": 10, "avg_topsql_share_percent": 40, "max_topsql_share_percent": 55, "affected_instances": 2, "custom_gauge": 7, }, }, { "id": "metric-semantics-2", "observed_at": "2026-07-02T12:00:00Z", "metrics": { "slow_count": 5, "avg_topsql_share_percent": 60, "max_topsql_share_percent": 72, "affected_instances": 3, "custom_gauge": 9, }, }, ] summary = server.summarize_metrics(findings) self.assertEqual(15, summary["slow_count"]) self.assertEqual(50, summary["avg_topsql_share_percent"]) self.assertEqual(72, summary["max_topsql_share_percent"]) self.assertEqual(3, summary["affected_instances"]) self.assertEqual(9, summary["custom_gauge"]) history = server.project_catalog_metric_history(findings) self.assertEqual( "average", history["definitions"]["avg_topsql_share_percent"]["aggregation"], ) self.assertEqual( "max", history["definitions"]["max_topsql_share_percent"]["aggregation"], ) def test_catalog_posture_separates_current_risk_from_unreviewed_peak(self) -> None: observed_at = datetime.now(timezone.utc) - timedelta(hours=2) critical = self.finding_payload("risk-peak", "risk-history-digest") critical["time_window_start"] = (observed_at - timedelta(minutes=30)).isoformat() critical["time_window_end"] = observed_at.isoformat() critical["observed_at"] = observed_at.isoformat() critical["reported_at"] = observed_at.isoformat() critical["symptoms"][0]["severity"] = "critical" status, first = server.create_finding(critical) self.assertEqual(201, status) latest = self.finding_payload("risk-latest", "risk-history-digest") latest_at = observed_at + timedelta(hours=1) latest["time_window_start"] = (latest_at - timedelta(minutes=30)).isoformat() latest["time_window_end"] = latest_at.isoformat() latest["observed_at"] = latest_at.isoformat() latest["reported_at"] = latest_at.isoformat() latest["symptoms"][0]["severity"] = "medium" status, latest_body = server.create_finding(latest) self.assertEqual(201, status) _, detail = server.catalog_item_detail(first["catalog_item"]["id"]) posture = detail["item"]["current_posture"] self.assertEqual("medium", posture["risk"]) self.assertEqual("critical", posture["unreviewed_peak_risk"]) self.assertEqual("critical", posture["historical_peak_risk"]) self.assertEqual("needs_review", posture["state"]) self.assertEqual( latest_body["finding"]["id"], posture["current_signal_finding_id"] ) self.assertIn("critical_peak_unreviewed", posture["reasons"]) self.assertNotIn("critical_without_treatment", posture["reasons"]) current_critical = server.list_catalog_items( {"queue": ["current_critical"]} ) critical_history = server.list_catalog_items( {"queue": ["critical_history"]} ) self.assertEqual(0, current_critical["total"]) self.assertEqual(1, critical_history["total"]) def test_catalog_triage_has_no_intermediate_start_review_action(self) -> None: item = self.create_item("claim-on-review") self.assertIsNone(item["owner"]) status, body = server.triage_catalog_item( item["id"], {"action": "start_review", "actor": "demo-user"} ) self.assertEqual(400, status) self.assertEqual("invalid_triage_action", body["error"]) def test_archived_catalog_is_reactivated_by_a_new_finding(self) -> None: item = self.create_item("archive-before-recurrence") status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Archive obsolete review", "review_status": "ready", "structured_fields": { "conclusion_type": "obsolete", "confidence": "high", }, "body_markdown": "The original report is no longer actionable.", }, ) self.assertEqual(201, status) status, archived = server.triage_catalog_item( item["id"], {"action": "archive", "record_id": record_body["record"]["id"]}, ) self.assertEqual(200, status) self.assertEqual("archived", archived["item"]["lifecycle"]) self.assertEqual("archived", archived["item"]["current_posture"]["state"]) status, reactivated = server.create_finding( self.finding_payload("finding-after-archive") ) self.assertEqual(201, status) self.assertEqual("active", reactivated["catalog_item"]["lifecycle"]) self.assertEqual( "registered", reactivated["catalog_item"]["current_posture"]["state"], ) _, detail = server.catalog_item_detail(item["id"]) self.assertTrue( any( event["event_type"] == "catalog_item_reactivated" for event in detail["timeline"] ) ) def test_catalog_status_columns_migrate_to_posture_events_and_lifecycle(self) -> None: reviewing = self.create_item("legacy-reviewing") other_payload = self.finding_payload("legacy-archived", "legacy-archive-digest") _, archived = server.create_finding(other_payload) with server.get_conn() as conn: conn.execute("ALTER TABLE catalog_items ADD COLUMN status TEXT") conn.execute("ALTER TABLE catalog_items ADD COLUMN severity TEXT") conn.execute( "UPDATE catalog_items SET status = 'reviewing', severity = 'high' WHERE id = ?", (reviewing["id"],), ) conn.execute( "UPDATE catalog_items SET status = 'archived', severity = 'high' WHERE id = ?", (archived["catalog_item"]["id"],), ) server.migrate_catalog_item_posture_schema(conn) with server.get_conn() as conn: columns = server.table_columns(conn, "catalog_items") self.assertNotIn("status", columns) self.assertNotIn("severity", columns) _, reviewing_detail = server.catalog_item_detail(reviewing["id"]) self.assertEqual( "registered", reviewing_detail["item"]["current_posture"]["state"] ) _, archived_detail = server.catalog_item_detail(archived["catalog_item"]["id"]) self.assertEqual("archived", archived_detail["item"]["lifecycle"]) self.assertEqual( "archived", archived_detail["item"]["current_posture"]["state"] ) def test_workbench_projects_repeated_findings_as_one_review_backlog_item(self) -> None: observed_at = datetime.now(timezone.utc) - timedelta(hours=2) first_payload = self.finding_payload("workbench-first") first_payload["time_window_start"] = (observed_at - timedelta(minutes=30)).isoformat() first_payload["time_window_end"] = observed_at.isoformat() first_payload["observed_at"] = observed_at.isoformat() first_payload["reported_at"] = observed_at.isoformat() status, first = server.create_finding(first_payload) self.assertEqual(201, status) second_payload = self.finding_payload("workbench-latest") second_payload["symptoms"][0]["severity"] = "critical" latest_at = observed_at + timedelta(hours=1) second_payload["time_window_start"] = (latest_at - timedelta(minutes=30)).isoformat() second_payload["time_window_end"] = latest_at.isoformat() second_payload["observed_at"] = latest_at.isoformat() second_payload["reported_at"] = latest_at.isoformat() status, second = server.create_finding(second_payload) self.assertEqual(201, status) with server.get_conn() as conn: conn.execute( "UPDATE findings SET created_at = ? WHERE id = ?", (observed_at.isoformat(), first["finding"]["id"]), ) conn.execute( "UPDATE findings SET created_at = ? WHERE id = ?", (latest_at.isoformat(), second["finding"]["id"]), ) result = server.workbench({"cluster_id": ["cluster-test"]}) review = result["review_flow"]["summary"] self.assertEqual(1, review["review_inflow"]) self.assertEqual(1, review["backlog_end"]) self.assertEqual(1, review["critical_backlog_end"]) self.assertEqual("expanding", result["assessment"]["state"]) self.assertEqual( 1, next( item["count"] for item in result["bottlenecks"] if item["code"] == "critical_without_treatment" ), ) status, _ = server.create_treatment( self.treatment_payload("Handle repeated Finding"), first["catalog_item"]["id"], ) self.assertEqual(201, status) handled = server.workbench({"cluster_id": ["cluster-test"]}) self.assertEqual(1, handled["review_flow"]["summary"]["review_handled"]) self.assertEqual(0, handled["review_flow"]["summary"]["backlog_end"]) self.assertEqual(1, handled["treatment_flow"]["summary"]["new_treatments"]) self.assertTrue(handled["treatment_flow"]["time_series"]) self.assertEqual( 1, next( item["count"] for item in handled["treatment_flow"]["stage_health"] if item["stage"] == "assessment" ), ) def test_workbench_loads_cluster_health_through_a_separate_projection(self) -> None: self.create_item("separate-cluster-health") with patch.object( server, "workbench_cluster_health", side_effect=AssertionError("main Workbench must not wait for cluster trends"), ): payload = server.workbench({"cluster_id": ["cluster-test"]}) self.assertNotIn("cluster_health", payload) attention = server.workbench_attention( {"cluster_id": ["cluster-test"]}, 0 ) trends = server.workbench_cluster_health_trends( {"cluster_id": ["cluster-test"]}, 0 ) health = server.workbench_cluster_health( {"cluster_id": ["cluster-test"]}, 0 ) self.assertEqual( attention["catalog_queue"], payload["catalog_queue"] ) self.assertEqual( attention["cluster_health"]["summary"]["needs_attention"], health["summary"]["needs_attention"], ) self.assertEqual( trends["summary"]["worsening"], health["summary"]["worsening"] ) self.assertEqual(1, health["summary"]["total_clusters"]) self.assertEqual("cluster-test", health["items"][0]["cluster_id"]) period_params = { "cluster_id": ["cluster-test"], "from": [payload["period"]["from"]], "to": [payload["period"]["to"]], } outcomes = server.workbench_period_outcomes(period_params) self.assertEqual(payload["period"], outcomes["period"]) self.assertEqual( payload["treatment_flow"]["time_series"], outcomes["treatment_flow"]["time_series"], ) def test_workbench_drilldown_queues_filter_canonical_lists(self) -> None: review_payload = self.finding_payload("queue-review") observed_at = datetime.now(timezone.utc) - timedelta(hours=1) review_payload["time_window_start"] = (observed_at - timedelta(minutes=30)).isoformat() review_payload["time_window_end"] = observed_at.isoformat() review_payload["observed_at"] = observed_at.isoformat() review_payload["reported_at"] = observed_at.isoformat() review_payload["symptoms"][0]["severity"] = "critical" status, review = server.create_finding(review_payload) self.assertEqual(201, status) review_item = review["catalog_item"] treated_payload = self.finding_payload("queue-treated", "digest-treated") status, treated = server.create_finding(treated_payload) self.assertEqual(201, status) status, _ = server.create_treatment( self.treatment_payload("Treat queue item"), treated["catalog_item"]["id"], ) self.assertEqual(201, status) catalog_queue = server.list_catalog_items({"queue": ["review"]}) self.assertIn(review_item["id"], {item["id"] for item in catalog_queue["items"]}) self.assertNotIn( treated["catalog_item"]["id"], {item["id"] for item in catalog_queue["items"]}, ) treatment_ids: list[str] = [] for title in ["Alignment", "Observation due", "Assessment", "Closed"]: status, body = server.create_treatment( self.treatment_payload(title), review_item["id"] ) self.assertEqual(201, status) treatment_ids.append(body["treatment"]["id"]) with server.get_conn() as conn: conn.execute( "UPDATE treatments SET stage = 'alignment' WHERE id = ?", (treatment_ids[0],), ) conn.execute( "UPDATE treatments SET stage = 'observation', observation_due_at = ? WHERE id = ?", ("2026-07-01T00:00:00Z", treatment_ids[1]), ) conn.execute( "UPDATE treatments SET state = 'completed', stage = 'completed', " "outcome = 'fixed', completed_at = ? WHERE id = ?", ("2026-07-02T00:00:00Z", treatment_ids[3]), ) follow_up = server.list_treatments({"queue": ["follow_up"]}) self.assertEqual( {treatment_ids[0], treatment_ids[1]}, {item["id"] for item in follow_up["items"]}, ) outcomes = server.list_treatments({"queue": ["recent_outcomes"]}) self.assertEqual( {treatment_ids[3]}, {item["id"] for item in outcomes["items"]}, ) def test_finding_rejects_invalid_optional_identity_fields(self) -> None: payload = self.finding_payload("invalid-optional-field") payload["created_by"] = {"unexpected": "object"} status, body = server.create_finding(payload) self.assertEqual(400, status) self.assertEqual("invalid_finding", body["error"]) def test_finding_requires_structured_symptoms(self) -> None: payload = self.finding_payload("missing-symptoms") payload.pop("symptoms") status, body = server.create_finding(payload) self.assertEqual(400, status) self.assertIn("symptoms must be a non-empty array", body["message"]) def test_finding_rejects_removed_legacy_fields(self) -> None: payload = self.finding_payload("legacy-fields") payload["reason_codes"] = ["high_latency"] status, body = server.create_finding(payload) self.assertEqual(400, status) self.assertIn("unknown fields: reason_codes", body["message"]) def test_finding_derives_event_severity_from_symptoms(self) -> None: payload = self.finding_payload("multi-symptom") payload["symptoms"] = [ {"code": "slow_query_frequency", "severity": "medium"}, { "code": "large_key_scan", "severity": "critical", "signal_codes": ["slowlog.total_keys"], "metric_keys": ["slow_count"], "confidence": 0.92, }, ] status, body = server.create_finding(payload) self.assertEqual(201, status) self.assertEqual("critical", body["finding"]["severity"]) self.assertEqual(2, body["finding"]["symptom_count"]) self.assertEqual( ["large_key_scan", "slow_query_frequency"], [entry["code"] for entry in body["finding"]["symptoms"]], ) def test_finding_requires_versioned_metrics_and_valid_metric_bindings(self) -> None: unversioned = self.finding_payload("unversioned-metrics") unversioned.pop("metric_schema_version") status, body = server.create_finding(unversioned) self.assertEqual(400, status) self.assertIn("metric_schema_version is required", body["message"]) unknown_binding = self.finding_payload("unknown-metric-binding") unknown_binding["symptoms"][0]["metric_keys"] = ["missing_metric"] status, body = server.create_finding(unknown_binding) self.assertEqual(400, status) self.assertIn("reference unknown metrics: missing_metric", body["message"]) def test_observed_time_and_ingested_time_are_distinct(self) -> None: status, body = server.create_finding(self.finding_payload("observed-event")) self.assertEqual(201, status) self.assertEqual("2026-07-01T01:00:00Z", body["finding"]["observed_at"]) self.assertNotEqual(body["finding"]["observed_at"], body["finding"]["ingested_at"]) observed = server.finding_stats( { "time_basis": ["observed"], "from": ["2026-07-01T00:00:00Z"], "to": ["2026-07-02T00:00:00Z"], } ) ingested = server.finding_stats( { "time_basis": ["ingested"], "from": ["2026-07-01T00:00:00Z"], "to": ["2026-07-02T00:00:00Z"], } ) self.assertEqual(1, observed["total_findings"]) self.assertEqual(0, ingested["total_findings"]) def test_created_time_is_the_default_finding_feed_basis(self) -> None: status, body = server.create_finding(self.finding_payload("created-feed-event")) self.assertEqual(201, status) created_at = datetime.fromisoformat( body["finding"]["created_at"].replace("Z", "+00:00") ) start = server.iso_datetime(created_at - timedelta(seconds=1)) end = server.iso_datetime(created_at + timedelta(seconds=1)) stats = server.finding_stats({"from": [start], "to": [end], "compare": ["none"]}) feed = server.list_findings({"from": [start], "to": [end]}) self.assertEqual("created", stats["time_basis"]) self.assertEqual(1, stats["total_findings"]) self.assertEqual(body["finding"]["id"], feed["items"][0]["id"]) def test_finding_stats_bucket_respects_timezone_offset(self) -> None: payload = self.finding_payload("timezone-bucket") payload["time_window_start"] = "2026-07-01T22:30:00Z" payload["time_window_end"] = "2026-07-01T23:30:00Z" payload["observed_at"] = "2026-07-01T23:30:00Z" server.create_finding(payload) stats = server.finding_stats( { "time_basis": ["observed"], "from": ["2026-07-01T00:00:00Z"], "to": ["2026-07-03T00:00:00Z"], "compare": ["none"], "timezone_offset_minutes": ["480"], } ) buckets = {entry["bucket"]: entry["count"] for entry in stats["time_series"]} self.assertEqual(480, stats["timezone_offset_minutes"]) self.assertEqual(1, buckets["2026-07-01T16:00:00Z"]) self.assertEqual(0, buckets["2026-06-30T16:00:00Z"]) def test_finding_stats_uses_daily_buckets_for_short_ranges(self) -> None: server.create_finding(self.finding_payload("daily-bucket")) stats = server.finding_stats( { "time_basis": ["observed"], "from": ["2026-07-01T00:00:00Z"], "to": ["2026-07-01T02:00:00Z"], "compare": ["none"], } ) self.assertEqual( [{"bucket": "2026-07-01T00:00:00Z", "count": 1}], [ {"bucket": entry["bucket"], "count": entry["count"]} for entry in stats["time_series"] ], ) def test_cluster_scope_library_starts_empty(self) -> None: library = server.cluster_scope_library() self.assertIsNone(library["recommended_scope_id"]) self.assertEqual([], library["my_scopes"]) self.assertEqual([], library["shared_scopes"]) def test_dynamic_cluster_scope_matches_cluster_metadata_and_can_be_default(self) -> None: payload = self.finding_payload("dynamic-scope-source") payload["dimensions"] = { "cluster": { "name": "payments-production", "org_name": "Payments Organization", "provider": "gcp", "region": "asia-northeast1", } } status, _ = server.create_finding(payload) self.assertEqual(201, status) status, body = server.create_cluster_scope( { "name": "GCP Japan clusters", "visibility": "private", "membership_mode": "dynamic", "include_deleted": True, "definition": { "cluster_ids": [], "conditions": [ { "org_ids": ["org-test"], "providers": ["gcp"], "regions": ["asia-northeast1"], "customer_regions": [], } ], "excluded_cluster_ids": [], }, } ) self.assertEqual(201, status) scope = body["cluster_scope"] self.assertEqual("dynamic", scope["membership_mode"]) self.assertEqual(["cluster-test"], [item["cluster_id"] for item in scope["members"]]) status, preference = server.set_default_cluster_scope({"scope_id": scope["id"]}) self.assertEqual(200, status) self.assertEqual(scope["id"], preference["default_scope_id"]) refreshed = server.cluster_scope_library() selected = next(item for item in refreshed["my_scopes"] if item["id"] == scope["id"]) self.assertTrue(selected["is_default"]) def test_cluster_scope_library_separates_owned_and_shared_public_scopes(self) -> None: status, _ = server.create_finding(self.finding_payload("scope-visibility-source")) self.assertEqual(201, status) base_payload = { "membership_mode": "snapshot", "include_deleted": True, "member_ids": ["cluster-test"], } status, public_body = server.create_cluster_scope( {**base_payload, "name": "Team scope", "visibility": "public"} ) self.assertEqual(201, status) status, private_body = server.create_cluster_scope( {**base_payload, "name": "Private scope", "visibility": "private"} ) self.assertEqual(201, status) owner_library = server.cluster_scope_library() self.assertEqual( {public_body["cluster_scope"]["id"], private_body["cluster_scope"]["id"]}, {scope["id"] for scope in owner_library["my_scopes"]}, ) self.assertEqual([], owner_library["shared_scopes"]) request_token = server.REQUEST_USER.set({"id": "another-user"}) try: viewer_library = server.cluster_scope_library() finally: server.REQUEST_USER.reset(request_token) self.assertEqual([], viewer_library["my_scopes"]) self.assertEqual( [public_body["cluster_scope"]["id"]], [scope["id"] for scope in viewer_library["shared_scopes"]], ) def test_catalog_review_admission_uses_critical_or_five_findings_in_fourteen_days(self) -> None: item_id = None for index in range(5): payload = self.finding_payload( f"review-frequency-{index}", "review-frequency-digest" ) payload["reported_at"] = f"2026-07-{index + 1:02d}T01:00:00Z" status, body = server.create_finding(payload) self.assertEqual(201, status) item_id = body["catalog_item"]["id"] self.assertEqual( "registered", body["catalog_item"]["current_posture"]["state"] ) _, detail = server.catalog_item_detail(item_id) admission = detail["item"]["current_posture"]["review_admission"] self.assertTrue(admission["required"]) self.assertEqual(["finding_frequency"], admission["matched_rules"]) self.assertEqual(5, admission["max_findings_in_window"]) queue = server.workbench({"cluster_id": ["cluster-test"]})["catalog_queue"] self.assertEqual(0, queue["total"]) self.assertEqual({"p0": 0, "p1": 0}, queue["priority_counts"]) critical = self.finding_payload("review-critical", "review-critical-digest") critical["symptoms"][0]["severity"] = "critical" status, body = server.create_finding(critical) self.assertEqual(201, status) self.assertEqual("registered", body["catalog_item"]["current_posture"]["state"]) self.assertEqual( ["critical_finding"], body["catalog_item"]["current_posture"]["review_admission"]["matched_rules"], ) self.assertIsNone( body["catalog_item"]["current_posture"]["review_priority"]["tier"] ) self.assertIsNone( body["catalog_item"]["current_posture"]["review_priority"]["focus"]["kind"] ) def test_ranking_observations_do_not_create_review_work(self) -> None: item_id = None for index in range(6): payload = self.finding_payload( f"ranking-observation-{index}", "ranking-only-digest" ) payload["source"] = "clinic_health_check" payload["source_type"] = "inspection" payload["detector_code"] = "clinic.health_check.sql_digest_report" payload["reported_at"] = f"2026-07-{index + 1:02d}T01:00:00Z" payload["dimensions"] = { "check_names": ["slowlog_top_digest"], "top_digest_ranks": {"top_by_total_latency": 1}, } status, body = server.create_finding(payload) self.assertEqual(201, status) item_id = body["catalog_item"]["id"] _, detail = server.catalog_item_detail(item_id) posture = detail["item"]["current_posture"] self.assertEqual("registered", posture["state"]) self.assertEqual(0, posture["review_admission"]["eligible_finding_count"]) self.assertEqual(6, posture["review_admission"]["ranking_observation_count"]) self.assertIsNone(posture["current_signal_finding_id"]) self.assertIsNone(posture["review_priority"]["tier"]) def test_non_tunable_statement_classification_is_conservative(self) -> None: excluded = { "/* daily maintenance */ ANALYZE TABLE orders": "analyze", "-- inspect schema\nSHOW CREATE TABLE orders": "show", "# inspect plan\nEXPLAIN ANALYZE SELECT * FROM orders": "explain", "ALTER TABLE orders ADD INDEX idx_customer(customer_id)": "alter", "COMMIT": "commit", } for sql, statement_type in excluded.items(): with self.subTest(sql=sql): relevance = server.catalog_governance_relevance( {"normalized_sql": sql} ) self.assertFalse(relevance["review_eligible"]) self.assertEqual(statement_type, relevance["statement_type"]) self.assertEqual("non_tunable_statement", relevance["reason"]) for sql in ( "SELECT * FROM orders WHERE id = ?", "WITH recent AS (SELECT * FROM orders) SELECT * FROM recent", "UPDATE orders SET status = ? WHERE id = ?", "", ): with self.subTest(sql=sql): self.assertTrue( server.catalog_governance_relevance( {"normalized_sql": sql} )["review_eligible"] ) def test_non_tunable_findings_remain_evidence_without_creating_review_work(self) -> None: analyze = self.finding_payload( "non-tunable-analyze", "non-tunable-analyze-digest" ) analyze["normalized_sql"] = "ANALYZE TABLE orders" analyze["symptoms"][0]["severity"] = "critical" status, created = server.create_finding(analyze) self.assertEqual(201, status) self.assertEqual("critical", created["finding"]["severity"]) analyze_item = created["catalog_item"] posture = analyze_item["current_posture"] self.assertEqual("registered", posture["state"]) self.assertIsNone(posture["risk"]) self.assertIsNone(posture["historical_peak_risk"]) self.assertFalse(posture["governance_relevance"]["review_eligible"]) self.assertEqual("analyze", posture["governance_relevance"]["statement_type"]) self.assertFalse(posture["review_admission"]["required"]) self.assertEqual(0, posture["review_admission"]["eligible_finding_count"]) self.assertEqual(1, posture["excluded_non_tunable_finding_count"]) self.assertIsNone(posture["review_priority"]["tier"]) self.assertEqual(0, posture["review_priority"]["anomaly_finding_count"]) show_item_id = None for index in range(5): show = self.finding_payload( f"non-tunable-show-{index}", "non-tunable-show-digest" ) show["normalized_sql"] = "SHOW TABLE STATUS LIKE ?" show["reported_at"] = f"2026-07-{index + 1:02d}T01:00:00Z" status, body = server.create_finding(show) self.assertEqual(201, status) show_item_id = body["catalog_item"]["id"] self.assertEqual( "registered", body["catalog_item"]["current_posture"]["state"] ) _, show_detail = server.catalog_item_detail(show_item_id) show_posture = show_detail["item"]["current_posture"] self.assertFalse(show_posture["review_admission"]["required"]) self.assertEqual(5, show_posture["excluded_non_tunable_finding_count"]) queue = server.workbench({"cluster_id": ["cluster-test"]})[ "catalog_queue" ] self.assertEqual(0, queue["total"]) catalog = server.list_catalog_items({"cluster_id": ["cluster-test"]}) self.assertEqual( {analyze_item["id"], show_item_id}, {item["id"] for item in catalog["items"]}, ) stats = server.catalog_stats({"cluster_id": ["cluster-test"]}) self.assertEqual(2, stats["non_tunable_statement_count"]) self.assertEqual(0, stats["decision_required_count"]) status, body = server.triage_catalog_item( analyze_item["id"], {"action": "start_review"} ) self.assertEqual(400, status) self.assertEqual("invalid_triage_action", body["error"]) status, body = server.create_treatment( self.treatment_payload(), analyze_item["id"] ) self.assertEqual(409, status) self.assertEqual("catalog_item_not_review_eligible", body["error"]) def test_non_tunable_catalog_cannot_be_registered_into_watch(self) -> None: status, body = server.create_catalog_item( { "org_id": "org-test", "cluster_id": "cluster-test", "sql_digest": "manual-analyze-digest", "normalized_sql": "ANALYZE TABLE orders", "registration_reason": "proactive_watch", "initial_action": "watch", } ) self.assertEqual(409, status) self.assertEqual("catalog_item_not_review_eligible", body["error"]) self.assertEqual(0, server.catalog_stats({})["total_items"]) def test_historical_persistent_high_is_not_in_the_weekly_focus(self) -> None: item_id = None for index in range(5): payload = self.finding_payload( f"p1-impact-{index}", "p1-impact-digest" ) payload["reported_at"] = f"2026-07-{index + 1:02d}T01:00:00Z" payload["dimensions"] = { "top_digest_ranks": {"top_by_total_latency": 2} } status, body = server.create_finding(payload) self.assertEqual(201, status) item_id = body["catalog_item"]["id"] _, detail = server.catalog_item_detail(item_id) priority = detail["item"]["current_posture"]["review_priority"] self.assertIsNone(priority["tier"]) self.assertEqual(["not_in_weekly_focus"], priority["reasons"]) self.assertIsNone(priority["focus"]["kind"]) queue = server.workbench({"cluster_id": ["cluster-test"]})["catalog_queue"] self.assertEqual(0, queue["total"]) self.assertEqual({"p0": 0, "p1": 0}, queue["priority_counts"]) def test_recent_first_critical_finding_is_in_the_weekly_focus(self) -> None: payload = self.finding_payload("recent-critical", "recent-critical-digest") observed_at = datetime.now(timezone.utc) payload["time_window_start"] = (observed_at - timedelta(hours=1)).isoformat() payload["time_window_end"] = observed_at.isoformat() payload["observed_at"] = observed_at.isoformat() payload["reported_at"] = observed_at.isoformat() payload["symptoms"][0]["severity"] = "critical" status, body = server.create_finding(payload) self.assertEqual(201, status) priority = body["catalog_item"]["current_posture"]["review_priority"] self.assertEqual("p1", priority["tier"]) self.assertEqual(["new_this_week"], priority["reasons"]) self.assertEqual("new", priority["focus"]["kind"]) def test_persistent_week_over_week_severity_increase_is_p0(self) -> None: now = datetime.now(timezone.utc) digest = "weekly-worsening-digest" events = [ ("previous-high-1", now - timedelta(days=10), "high"), ("previous-high-2", now - timedelta(days=8), "high"), ("current-critical-1", now - timedelta(days=3), "critical"), ("current-critical-2", now - timedelta(hours=2), "critical"), ] item_id = None for event_id, occurred_at, severity in events: payload = self.finding_payload(event_id, digest) payload["time_window_start"] = (occurred_at - timedelta(hours=1)).isoformat() payload["time_window_end"] = occurred_at.isoformat() payload["observed_at"] = occurred_at.isoformat() payload["reported_at"] = occurred_at.isoformat() payload["symptoms"][0]["severity"] = severity status, body = server.create_finding(payload) self.assertEqual(201, status) item_id = body["catalog_item"]["id"] _, detail = server.catalog_item_detail(item_id) priority = detail["item"]["current_posture"]["review_priority"] self.assertEqual("p0", priority["tier"]) self.assertEqual(["weekly_severity_worsened"], priority["reasons"]) self.assertEqual("worsened", priority["focus"]["kind"]) self.assertEqual("high", priority["focus"]["previous_risk"]) self.assertEqual("critical", priority["focus"]["current_risk"]) self.assertEqual(2, priority["focus"]["current_peak_count"]) def test_catalog_decision_is_recorded_and_applied_in_one_action(self) -> None: item = self.create_item("simple-decision") status, body = server.decide_catalog_item( item["id"], { "action": "no_action", "conclusion_type": "expected_workload", "rationale": "Expected nightly batch. https://example.com/context", "actor": "frontline-user", }, ) self.assertEqual(200, status) self.assertEqual("ready", body["record"]["review_status"]) self.assertEqual("no_intervention", body["item"]["current_posture"]["state"]) self.assertEqual( "expected_workload", body["item"]["current_posture"]["governance_memory"]["conclusion_type"], ) def test_catalog_decision_turns_source_url_into_markdown_link(self) -> None: item = self.create_item("decision-link") status, body = server.decide_catalog_item( item["id"], { "action": "watch", "conclusion_type": "watch", "rationale": "Observe the next Health Check run.", "source_url": "https://example.com/review/42", }, ) self.assertEqual(200, status) self.assertIn( "[Related source](https://example.com/review/42)", body["record"]["body_markdown"], ) def test_cluster_review_policy_recomputes_catalog_posture_and_can_reset(self) -> None: item = self.create_item("cluster-policy-high") status, initial = server.get_cluster_configuration("cluster-test") self.assertEqual(200, status) self.assertEqual("system_default", initial["sql_review_policy"]["source"]) self.assertEqual(1, initial["review_impact"]["registered_count"]) status, updated = server.update_cluster_configuration( "cluster-test", { "sql_review_policy": { "critical_finding_triggers_review": False, "finding_count_threshold": 1, "finding_window_days": 7, } }, ) self.assertEqual(200, status) self.assertEqual("cluster", updated["sql_review_policy"]["source"]) self.assertEqual(0, updated["review_impact"]["decision_required_count"]) _, detail = server.catalog_item_detail(item["id"]) self.assertEqual("registered", detail["item"]["current_posture"]["state"]) status, reset = server.reset_cluster_configuration("cluster-test") self.assertEqual(200, status) self.assertEqual("system_default", reset["sql_review_policy"]["source"]) self.assertEqual(1, reset["review_impact"]["registered_count"]) def test_cluster_scope_configuration_is_copied_only_to_selected_clusters(self) -> None: self.create_item("scope-config-first") other = self.finding_payload("scope-config-other", "scope-config-other-digest") other["org_id"] = "org-other" other["cluster_id"] = "cluster-other" status, _ = server.create_finding(other) self.assertEqual(201, status) status, created = server.create_cluster_scope( { "name": "Configuration selection", "member_ids": ["cluster-test", "cluster-other"], } ) self.assertEqual(201, status) scope_id = created["cluster_scope"]["id"] status, result = server.apply_cluster_scope_configurations( scope_id, { "cluster_ids": ["cluster-other"], "sql_review_policy": { "critical_finding_triggers_review": False, "finding_count_threshold": 1, "finding_window_days": 30, }, }, ) self.assertEqual(200, status) self.assertEqual(1, result["applied_count"]) self.assertEqual( "system_default", server.get_cluster_configuration("cluster-test")[1]["sql_review_policy"]["source"], ) self.assertEqual( "cluster", server.get_cluster_configuration("cluster-other")[1]["sql_review_policy"]["source"], ) status, scope_configuration = server.cluster_scope_configurations(scope_id) self.assertEqual(200, status) sources = { member["cluster_id"]: member["sql_review_policy"]["source"] for member in scope_configuration["cluster_scope"]["members"] } self.assertEqual( {"cluster-test": "system_default", "cluster-other": "cluster"}, sources, ) def test_cluster_health_deduplicates_slow_count_by_report_and_digest(self) -> None: report_time = "2026-07-20T08:00:00Z" status, _ = server.create_finding_run( { "source": "test", "source_type": "inspection", "source_run_id": "health-report-1", "org_id": "org-test", "cluster_id": "cluster-test", "time_window_start": "2026-07-19T08:00:00Z", "time_window_end": report_time, "status": "success", "expected_detector_count": 2, "completed_detector_count": 2, }, received_at=report_time, ) self.assertEqual(201, status) for index in range(2): payload = self.finding_payload(f"health-report-finding-{index}") payload["source_run_id"] = "health-report-1" payload["detector_code"] = f"test.detector_{index}" status, _ = server.create_finding(payload, received_at=report_time) self.assertEqual(201, status) health = server.workbench_cluster_health({"cluster_id": ["cluster-test"]}, 0) item = health["items"][0] point = next( value for value in item["series"] if value["bucket"].startswith("2026-07-20") ) self.assertEqual(2, point["finding_count"]) self.assertEqual(12, point["slow_log_count"]) self.assertEqual(1, point["new_digest_count"]) workbench = server.workbench({"cluster_id": ["cluster-test"]}) self.assertEqual(0, workbench["catalog_queue"]["total"]) catalog_item = server.list_catalog_items({"cluster_id": ["cluster-test"]})["items"][0] self.assertEqual("registered", catalog_item["current_posture"]["state"]) def test_cluster_health_separates_observation_day_from_backfill_intake_day(self) -> None: observed_at = "2026-07-20T08:00:00Z" reported_at = "2026-07-28T08:00:00Z" status, _ = server.create_finding_run( { "source": "test", "source_type": "inspection", "source_run_id": "backfilled-health-report", "org_id": "org-test", "cluster_id": "cluster-test", "time_window_start": "2026-07-19T08:00:00Z", "time_window_end": observed_at, "status": "success", "expected_detector_count": 1, "completed_detector_count": 1, }, received_at=reported_at, ) self.assertEqual(201, status) payload = self.finding_payload("backfilled-health-finding") payload.update( { "source_run_id": "backfilled-health-report", "reported_at": reported_at, "observed_at": observed_at, "time_window_start": "2026-07-19T08:00:00Z", "time_window_end": observed_at, } ) status, _ = server.create_finding(payload, received_at=reported_at) self.assertEqual(201, status) health = server.workbench_cluster_health({"cluster_id": ["cluster-test"]}, 0) item = health["items"][0] observation_point = next( point for point in item["series"] if point["bucket"].startswith("2026-07-20") ) intake_point = next( point for point in item["series"] if point["bucket"].startswith("2026-07-28") ) self.assertEqual(1, item["coverage_days"]) self.assertTrue(observation_point["covered"]) self.assertFalse(observation_point["intake_covered"]) self.assertEqual(12, observation_point["slow_log_count"]) self.assertIsNone(observation_point["finding_count"]) self.assertFalse(intake_point["covered"]) self.assertTrue(intake_point["intake_covered"]) self.assertIsNone(intake_point["slow_log_count"]) self.assertEqual(1, intake_point["finding_count"]) self.assertEqual(1, intake_point["new_digest_count"]) def test_cluster_health_does_not_add_blank_day_for_next_day_report_intake(self) -> None: first_end = datetime(2026, 7, 29, 1, tzinfo=timezone.utc) delayed_intake = datetime(2026, 8, 11, 17, tzinfo=timezone.utc) for index in range(14): report_end = first_end + timedelta(days=index) received_at = delayed_intake if index == 5 else report_end status, _ = server.create_finding_run( { "source": "clinic_health_check", "source_type": "inspection", "source_run_id": f"health-report-{index}", "org_id": "org-test", "cluster_id": "cluster-test", "time_window_start": server.iso_datetime( report_end - timedelta(days=1) ), "time_window_end": server.iso_datetime(report_end), "status": "success", "expected_detector_count": 1, "completed_detector_count": 1, }, received_at=server.iso_datetime(received_at), ) self.assertEqual(201, status) health = server.workbench_cluster_health( {"cluster_id": ["cluster-test"]}, 480 ) item = health["items"][0] self.assertEqual("2026-08-11T16:00:00Z", health["period"]["to"]) self.assertEqual(14, item["coverage_days"]) self.assertTrue(all(point["covered"] for point in item["series"])) def test_cluster_health_uses_each_clusters_latest_report_day(self) -> None: first_ends = { "cluster-early": datetime(2026, 7, 28, 23, tzinfo=timezone.utc), "cluster-late": datetime(2026, 7, 29, 1, tzinfo=timezone.utc), } for cluster_id, first_end in first_ends.items(): for index in range(14): report_end = first_end + timedelta(days=index) status, _ = server.create_finding_run( { "source": "clinic_health_check", "source_type": "inspection", "source_run_id": f"{cluster_id}-report-{index}", "org_id": "org-test", "cluster_id": cluster_id, "time_window_start": server.iso_datetime( report_end - timedelta(days=1) ), "time_window_end": server.iso_datetime(report_end), "status": "success", "expected_detector_count": 1, "completed_detector_count": 1, }, received_at=server.iso_datetime(report_end), ) self.assertEqual(201, status) health = server.workbench_cluster_health( {"cluster_id": ["cluster-early,cluster-late"]}, 0 ) items = {item["cluster_id"]: item for item in health["items"]} self.assertEqual(14, items["cluster-early"]["coverage_days"]) self.assertEqual(14, items["cluster-late"]["coverage_days"]) self.assertNotEqual( items["cluster-early"]["period"], items["cluster-late"]["period"] ) def test_cluster_sql_health_supports_multiple_owners(self) -> None: self.create_item("cluster-health-owner") status, payload = server.update_cluster_sql_health_owners( "cluster-test", {"owner_ids": ["alice", "bob", "alice"]} ) self.assertEqual(200, status) self.assertEqual(["alice", "bob"], payload["owners"]) health = server.workbench_cluster_health( {"cluster_id": ["cluster-test"]}, 0 ) self.assertEqual(["alice", "bob"], health["items"][0]["owners"]) status, payload = server.update_cluster_sql_health_owners( "cluster-test", {"owner_ids": ["bob"]} ) self.assertEqual(200, status) self.assertEqual(["bob"], payload["owners"]) status, payload = server.update_cluster_sql_health_owners( "cluster-test", {"owner_ids": []} ) self.assertEqual(200, status) self.assertEqual([], payload["owners"]) def test_cluster_sql_health_level_priority_is_explicit(self) -> None: def item( state: str, risk: str | None, priority: str | None = None, active_treatments: int = 0, review_eligible: bool = True, ) -> dict[str, object]: return { "current_posture": { "state": state, "risk": risk, "active_treatment_count": active_treatments, "review_priority": {"tier": priority}, "governance_relevance": { "review_eligible": review_eligible, }, } } urgent = server.cluster_health_risk( [item("needs_review", "critical", "p0")], 14, 0 ) high = server.cluster_health_risk( [item("needs_review", "high", "p1")], 14, 0 ) attention = server.cluster_health_risk( [item("in_treatment", "medium", None, 1)], 14, 1 ) active = server.cluster_health_risk( [item("in_treatment", "critical", None, 1)], 14, 1 ) clear = server.cluster_health_risk([item("watching", None)], 4, 0) insufficient = server.cluster_health_risk( [item("watching", None)], 3, 0 ) excluded = server.cluster_health_risk( [item("needs_review", "critical", "p0", review_eligible=False)], 14, 0, ) self.assertEqual("urgent", urgent["level"]) self.assertEqual("high", high["level"]) self.assertEqual("attention", attention["level"]) self.assertEqual("attention", active["level"]) self.assertEqual("clear", clear["level"]) self.assertEqual("insufficient_data", insufficient["level"]) self.assertEqual("clear", excluded["level"]) self.assertEqual(1, excluded["excluded_non_tunable_count"]) def test_scope_library_bootstraps_from_real_finding_identities(self) -> None: self.create_item("scope-library") library = server.scope_library() self.assertEqual("demo-user", library["owner_id"]) self.assertEqual( [("org-test", "org-test")], [(item["resource_id"], item["name"]) for item in library["organizations"]], ) self.assertEqual( [("cluster-test", "org-test")], [(item["resource_id"], item["org_id"]) for item in library["clusters"]], ) def test_scope_library_does_not_restore_scopes_the_user_removed(self) -> None: self.create_item("scope-library-removal") library = server.scope_library() for item in [*library["organizations"], *library["clusters"]]: status, _ = server.delete_followed_scope(item["id"]) self.assertEqual(200, status) refreshed = server.scope_library() self.assertEqual([], refreshed["organizations"]) self.assertEqual([], refreshed["clusters"]) def test_clinic_scope_resolution_and_follow_use_canonical_org_identity(self) -> None: clinic_item = { "clusterId": "cluster-clinic", "name": "production-east", "orgId": "tenant-org-test", "deployTypeV2": "dedicated", "status": "active", "provider": "aws", "region": "us-east-1", "diagnosticContext": {"projectId": "project-test"}, } with ( patch.object(server, "CLINIC_API_KEY", "test-key"), patch.object( server, "clinic_get_cluster_by_id", return_value=(200, clinic_item) ), ): status, resolved = server.clinic_scope_resolve( {"type": ["cluster"], "id": ["cluster-clinic"]} ) self.assertEqual(200, status) self.assertEqual("production-east", resolved["scope"]["name"]) self.assertEqual( "tenant-org-test", resolved["scope"]["org_id"] ) self.assertEqual("clinic", resolved["name_source"]) status, followed = server.create_followed_scope( { "type": "cluster", "resource_id": "cluster-clinic", } ) self.assertEqual(201, status) self.assertEqual("production-east", followed["scope"]["name"]) self.assertEqual( "tenant-org-test", followed["scope"]["org_id"] ) def test_clinic_scope_resolution_preserves_portal_org_identity_and_name(self) -> None: server.upsert_cluster_metadata( { "cluster_id": "cluster-clinic", "clinic_org_id": "portal-org-test", "tenant_id": "tenant-org-test", "org_name": "Example Organization", "cluster_name": "old-cluster-name", "deploy_type": "dedicated", "metadata": {}, "fetched_at": server.now_iso(), } ) clinic_item = { "clusterId": "cluster-clinic", "name": "production-east", "orgId": "tenant-org-test", "deployTypeV2": "dedicated", "status": "active", "diagnosticContext": {}, } with ( patch.object(server, "CLINIC_API_KEY", "test-key"), patch.object( server, "clinic_get_cluster_by_id", return_value=(200, clinic_item) ), ): status, resolved = server.clinic_scope_resolve( {"type": ["cluster"], "id": ["cluster-clinic"]} ) self.assertEqual(200, status) self.assertEqual("production-east", resolved["scope"]["name"]) self.assertEqual("portal-org-test", resolved["scope"]["org_id"]) self.assertEqual("tenant-org-test", resolved["scope"]["tenant_id"]) self.assertEqual("Example Organization", resolved["scope"]["org_name"]) def test_org_scope_resolution_uses_mapped_dbaas_id(self) -> None: server.upsert_cluster_metadata( { "cluster_id": "cluster-clinic", "clinic_org_id": "portal-org-test", "tenant_id": "tenant-org-test", "org_name": "Example Organization", "cluster_name": "production-east", "deploy_type": "dedicated", "metadata": {}, "fetched_at": server.now_iso(), } ) with ( patch.object(server, "CLINIC_API_KEY", "test-key"), patch.object( server, "clinic_list_org_clusters", return_value=(200, {"items": [], "total": 0}), ) as mocked_list, ): status, resolved = server.clinic_scope_resolve( {"type": ["organization"], "id": ["portal-org-test"]} ) self.assertEqual(200, status) mocked_list.assert_called_once_with( "tenant-org-test", limit=1, page=1 ) self.assertEqual("portal-org-test", resolved["scope"]["id"]) self.assertEqual("tenant-org-test", resolved["scope"]["tenant_id"]) self.assertEqual("Example Organization", resolved["scope"]["name"]) def test_clinic_organization_clusters_lists_and_caches_canonical_metadata(self) -> None: clinic_item = { "clusterId": "cluster-clinic", "name": "production-east", "orgId": "tenant-org-test", "orgName": "Example Organization", "deployTypeV2": "dedicated", "status": "active", "provider": "aws", "region": "us-east-1", "diagnosticContext": {"projectId": "project-test"}, } with ( patch.object(server, "CLINIC_API_KEY", "test-key"), patch.object( server, "clinic_list_org_clusters", return_value=( 200, { "items": [clinic_item], "total": 1, "page": 1, "page_size": 50, }, ), ) as mocked_list, ): status, response = server.clinic_organization_clusters( "portal-org-test", {"dbaas_org_id": ["tenant-org-test"]}, ) self.assertEqual(200, status) mocked_list.assert_called_once_with( "tenant-org-test", query=None, limit=50, page=1 ) self.assertEqual("cluster-clinic", response["items"][0]["id"]) self.assertEqual("portal-org-test", response["items"][0]["org_id"]) self.assertEqual("tenant-org-test", response["items"][0]["tenant_id"]) self.assertEqual("Example Organization", response["organization"]["name"]) with server.get_conn() as conn: cached = conn.execute( "SELECT * FROM cluster_metadata WHERE cluster_id = 'cluster-clinic'" ).fetchone() self.assertEqual("production-east", cached["cluster_name"]) self.assertEqual("portal-org-test", cached["clinic_org_id"]) def test_clinic_exact_metadata_uses_canonical_path_and_timeout(self) -> None: response = MagicMock() response.__enter__.return_value.read.return_value = ( b'{"clusterId":"cluster-clinic","orgId":"123","name":"production-east"}' ) with ( patch.object(server, "CLINIC_API_KEY", "test-key"), patch.object(server, "CLINIC_API_TIMEOUT_SECONDS", 30.0), patch.object(server, "urlopen", return_value=response) as mocked_urlopen, ): status, result = server.clinic_get_cluster_by_id("cluster-clinic") self.assertEqual(200, status) self.assertEqual("production-east", result["name"]) self.assertEqual(30.0, mocked_urlopen.call_args.kwargs["timeout"]) request = mocked_urlopen.call_args.args[0] self.assertEqual( "https://clinic.pingcap.com/data-proxy/api/v1/clusters/cluster-clinic", request.full_url, ) self.assertEqual("Bearer test-key", request.headers["Authorization"]) def test_clinic_cluster_can_be_added_to_group_before_any_finding(self) -> None: clinic_item = { "clusterId": "cluster-clinic", "name": "production-east", "orgId": "tenant-org-test", "deployType": "dedicated", "diagnosticContext": {}, } with ( patch.object(server, "CLINIC_API_KEY", "test-key"), patch.object( server, "clinic_get_cluster_by_id", return_value=(200, clinic_item), ), ): status, created = server.create_cluster_group( { "name": "Clinic-only group", "member_ids": ["cluster-clinic"], } ) self.assertEqual(201, status) self.assertEqual( ["cluster-clinic"], [member["cluster_id"] for member in created["cluster_group"]["members"]], ) def test_cluster_group_reports_single_and_cross_org_membership(self) -> None: self.create_item("group-member-a") other = self.finding_payload("group-member-b", "digest-other") other["org_id"] = "org-other" other["cluster_id"] = "cluster-other" server.create_finding(other) status, created = server.create_cluster_group( { "name": "Shared service", "description": "Clusters serving one workload", "member_ids": ["cluster-test"], } ) self.assertEqual(201, status) group_id = created["cluster_group"]["id"] self.assertFalse(created["cluster_group"]["is_cross_org"]) self.assertEqual("org-test", created["cluster_group"]["org_id"]) status, updated = server.update_cluster_group( group_id, {"member_ids": ["cluster-test", "cluster-other"]}, ) self.assertEqual(200, status) self.assertTrue(updated["cluster_group"]["is_cross_org"]) self.assertEqual(2, updated["cluster_group"]["organization_count"]) self.assertIsNone(updated["cluster_group"]["org_id"]) self.assertEqual( {"cluster-test", "cluster-other"}, {member["cluster_id"] for member in updated["cluster_group"]["members"]}, ) def test_scope_filters_catalog_treatments_and_workbench_consistently(self) -> None: first_payload = self.finding_payload("scope-first") first_payload["symptoms"][0]["severity"] = "critical" status, first = server.create_finding(first_payload) self.assertEqual(201, status) first_item = first["catalog_item"] other = self.finding_payload("scope-second", "digest-other") other["org_id"] = "org-other" other["cluster_id"] = "cluster-other" other["symptoms"][0]["severity"] = "critical" status, second = server.create_finding(other) self.assertEqual(201, status) second_item = second["catalog_item"] scoped_workbench = server.workbench({"cluster_id": ["cluster-test"]}) self.assertEqual(1, scoped_workbench["review_flow"]["summary"]["backlog_end"]) all_scope_workbench = server.workbench( {"cluster_id": ["cluster-test,cluster-other"]} ) self.assertEqual(2, all_scope_workbench["review_flow"]["summary"]["backlog_end"]) _, first_treatment = server.create_treatment(self.treatment_payload("First scope"), first_item["id"]) _, second_treatment = server.create_treatment(self.treatment_payload("Second scope"), second_item["id"]) catalog = server.list_catalog_items( {"cluster_id": ["cluster-test,cluster-other"]} ) self.assertEqual(2, catalog["total"]) treatments = server.list_treatments( {"org_id": ["org-other"], "cluster_id": ["cluster-other"]} ) self.assertEqual(1, treatments["total"]) self.assertEqual(second_treatment["treatment"]["id"], treatments["items"][0]["id"]) self.assertEqual(["org-other"], treatments["items"][0]["org_ids"]) self.assertEqual(["cluster-other"], treatments["items"][0]["cluster_ids"]) self.assertNotEqual(first_treatment["treatment"]["id"], treatments["items"][0]["id"]) def test_clinic_finding_builds_source_url_from_report_provenance(self) -> None: payload = self.finding_payload("clinic-report") payload["source"] = "clinic_health_check" payload["source_type"] = "inspection" payload["detector_code"] = "clinic.slowlog_top_digest" payload["source_run_id"] = "report-run" payload["raw_payload"] = { "report": { "org_id": "org-test", "portal_org_id": "portal-org-test", "cluster_id": "cluster-test", "result_uuid": "00000000-0000-4000-8000-000000000001", "strategy_uuid": "strategy-that-must-not-be-used-as-the-report-id", "time_range_a_start": "2026-07-09T07:00:15.636Z", "time_range_a_end": "2026-07-10T07:00:15.636Z", } } status, body = server.create_finding(payload) self.assertEqual(201, status) self.assertEqual( "https://clinic.pingcap.com/portal/#/orgs/portal-org-test/clusters/cluster-test/health-check/" "00000000-0000-4000-8000-000000000001?" "from=2026-07-09T07%3A00%3A15.636Z&to=2026-07-10T07%3A00%3A15.636Z", body["finding"]["source_url"], ) canonical_only = self.finding_payload("clinic-report-without-portal-org") canonical_only["source"] = "clinic_health_check" canonical_only["source_type"] = "inspection" canonical_only["detector_code"] = "clinic.slowlog_top_digest" canonical_only["source_run_id"] = "report-run-without-portal-org" canonical_only["raw_payload"] = { "report": { "org_id": "org-test", "cluster_id": "cluster-test", "result_uuid": "00000000-0000-4000-8000-000000000002", "time_range_a_start": "2026-07-09T07:00:15.636Z", "time_range_a_end": "2026-07-10T07:00:15.636Z", } } status, body = server.create_finding(canonical_only) self.assertEqual(201, status) self.assertIsNone(body["finding"]["source_url"]) manual = self.finding_payload("manual-finding") manual["source"] = "manual" manual["source_type"] = "manual" manual["detector_code"] = "manual.user_report" status, body = server.create_finding(manual) self.assertEqual(201, status) self.assertIsNone(body["finding"]["source_url"]) def test_reported_at_drives_created_time_basis(self) -> None: earlier = self.finding_payload("reported-earlier", "digest-earlier") earlier["reported_at"] = "2026-07-19T08:00:00Z" earlier["observed_at"] = "2026-07-19T08:00:00Z" status, earlier_body = server.create_finding(earlier) self.assertEqual(201, status) self.assertEqual( "2026-07-19T08:00:00Z", earlier_body["finding"]["reported_at"], ) later = self.finding_payload("reported-later", "digest-later") later["reported_at"] = "2026-07-20T08:00:00Z" later["observed_at"] = "2026-07-20T08:00:00Z" status, _ = server.create_finding(later) self.assertEqual(201, status) listed = server.list_findings( { "time_basis": ["created"], "from": ["2026-07-19T00:00:00Z"], "to": ["2026-07-20T00:00:00Z"], } ) self.assertEqual(1, listed["total"]) self.assertEqual( earlier_body["finding"]["id"], listed["items"][0]["id"], ) def test_historical_replay_controls_all_system_owned_intake_times(self) -> None: first_replay_at = str( server.parse_iso( "2026-07-20T00:00:16.556Z", "first_replay_at", required=True, ) ) run_status, run_body = server.create_finding_run( { "source": "test", "source_type": "integration", "source_run_id": "historical-run-1", "org_id": "org-test", "cluster_id": "cluster-test", "time_window_start": "2026-07-19T00:00:16.556Z", "time_window_end": first_replay_at, "status": "success", "started_at": "2026-07-19T00:00:16.556Z", "completed_at": first_replay_at, }, received_at=first_replay_at, ) self.assertEqual(201, run_status) self.assertEqual(first_replay_at, run_body["run"]["ingested_at"]) first_payload = self.finding_payload( "historical-event-1", "historical-digest", ) first_payload.update( { "source_run_id": "historical-run-1", "reported_at": first_replay_at, "observed_at": first_replay_at, "time_window_start": "2026-07-19T00:00:16.556Z", "time_window_end": first_replay_at, } ) finding_status, finding_body = server.create_finding( first_payload, received_at=first_replay_at, ) self.assertEqual(201, finding_status) self.assertEqual(first_replay_at, finding_body["finding"]["ingested_at"]) self.assertEqual(first_replay_at, finding_body["finding"]["created_at"]) self.assertEqual(first_replay_at, finding_body["catalog_item"]["created_at"]) self.assertEqual(first_replay_at, finding_body["catalog_item"]["updated_at"]) second_replay_at = str( server.parse_iso( "2026-07-21T00:00:16.612Z", "second_replay_at", required=True, ) ) second_payload = self.finding_payload( "historical-event-2", "historical-digest", ) second_payload.update( { "reported_at": second_replay_at, "observed_at": second_replay_at, "time_window_start": first_replay_at, "time_window_end": second_replay_at, } ) batch_defaults = { key: value for key, value in second_payload.items() if key not in {"source", "source_type", "source_event_id", "sql_digest"} } batch_status, batch_body = server.create_finding_batch( { "source": "test", "source_type": "integration", "source_batch_id": "historical-batch-2", "defaults": batch_defaults, "findings": [ { "source_event_id": "historical-event-2", "sql_digest": "historical-digest", } ], }, received_at=second_replay_at, ) self.assertEqual(201, batch_status) self.assertEqual(1, batch_body["summary"]["created"]) catalog_item_id = finding_body["catalog_item"]["id"] with server.get_conn() as conn: item = conn.execute( "SELECT * FROM catalog_items WHERE id = ?", (catalog_item_id,), ).fetchone() findings = conn.execute( """ SELECT source_event_id, ingested_at, created_at FROM findings WHERE catalog_item_id = ? ORDER BY reported_at """, (catalog_item_id,), ).fetchall() relations = conn.execute( """ SELECT created_at FROM relations WHERE to_type = 'catalog_item' AND to_id = ? ORDER BY created_at """, (catalog_item_id,), ).fetchall() timeline = conn.execute( """ SELECT event_type, created_at FROM timeline_events WHERE (target_type = 'catalog_item' AND target_id = ?) OR ( target_type = 'finding' AND target_id IN ( SELECT id FROM findings WHERE catalog_item_id = ? ) ) ORDER BY created_at, event_type """, (catalog_item_id, catalog_item_id), ).fetchall() self.assertEqual(first_replay_at, item["created_at"]) self.assertEqual(second_replay_at, item["updated_at"]) self.assertEqual( [ ("historical-event-1", first_replay_at, first_replay_at), ("historical-event-2", second_replay_at, second_replay_at), ], [ (row["source_event_id"], row["ingested_at"], row["created_at"]) for row in findings ], ) self.assertEqual( [first_replay_at, second_replay_at], [row["created_at"] for row in relations], ) self.assertEqual( {first_replay_at, second_replay_at}, {row["created_at"] for row in timeline}, ) def test_historical_replay_header_is_gated(self) -> None: replay_at = "2026-07-20T00:00:16.556Z" self.assertEqual( server.parse_iso(replay_at, "replay_at", required=True), server.historical_replay_at(replay_at), ) with patch.object(server, "ALLOW_HISTORICAL_REPLAY", False): with self.assertRaises(PermissionError): server.historical_replay_at(replay_at) def test_catalog_latest_finding_uses_reported_time_across_backfill(self) -> None: later = self.finding_payload("catalog-later", "catalog-time-digest") later["reported_at"] = "2026-07-20T08:00:00Z" later["observed_at"] = "2026-07-20T08:00:00Z" later["time_window_start"] = "2026-07-20T07:00:00Z" later["time_window_end"] = "2026-07-20T08:00:00Z" status, later_body = server.create_finding(later) self.assertEqual(201, status) earlier = self.finding_payload("catalog-earlier", "catalog-time-digest") earlier["reported_at"] = "2026-07-19T08:00:00Z" earlier["observed_at"] = "2026-07-19T08:00:00Z" earlier["time_window_start"] = "2026-07-19T07:00:00Z" earlier["time_window_end"] = "2026-07-19T08:00:00Z" status, _ = server.create_finding(earlier) self.assertEqual(201, status) status, detail = server.catalog_item_detail( later_body["catalog_item"]["id"] ) self.assertEqual(200, status) self.assertEqual( "2026-07-20T08:00:00Z", detail["item"]["latest_finding_at"], ) self.assertEqual( ["catalog-later", "catalog-earlier"], [finding["source_event_id"] for finding in detail["findings"]], ) def test_cluster_metadata_uses_cached_user_facing_names(self) -> None: server.upsert_cluster_metadata( { "cluster_id": "cluster-test", "clinic_org_id": "org-test", "tenant_id": "tenant-test", "org_name": "Test Organization", "cluster_name": "production-east", "deploy_type": "dedicated", "metadata": {}, "fetched_at": server.now_iso(), } ) result = server.cluster_metadata({"cluster_id": ["cluster-test"]}) self.assertEqual("Test Organization", result["items"][0]["org_name"]) self.assertEqual("production-east", result["items"][0]["cluster_name"]) self.assertEqual("org-test", result["items"][0]["org_id"]) self.assertEqual("tenant-test", result["items"][0]["dbaas_org_id"]) self.assertTrue(result["items"][0]["resolved"]) def test_finding_cluster_identity_is_cached_as_metadata_fallback(self) -> None: payload = self.finding_payload("source-reported-cluster-metadata") payload["dimensions"] = { "cluster": { "name": "main", "org_name": "mk-jp-tidb-gcp", "dbaas_org_id": "dbaas-org-1", "provider": "gcp", "region": "asia-northeast1", } } status, _ = server.create_finding(payload) self.assertEqual(201, status) metadata = server.cluster_metadata({"cluster_id": ["cluster-test"]})["items"][0] self.assertEqual("main", metadata["cluster_name"]) self.assertEqual("mk-jp-tidb-gcp", metadata["org_name"]) self.assertEqual("org-test", metadata["org_id"]) self.assertEqual("dbaas-org-1", metadata["dbaas_org_id"]) self.assertEqual("finding_dimensions", metadata["metadata"]["identity_source"]) library = server.scope_library() self.assertEqual("mk-jp-tidb-gcp", library["organizations"][0]["name"]) self.assertEqual("main", library["clusters"][0]["name"]) server.upsert_cluster_metadata( { "cluster_id": "cluster-test", "clinic_org_id": "org-test", "tenant_id": "dbaas-org-1", "org_name": "Authoritative Organization", "cluster_name": "authoritative-main", "deploy_type": "dedicated", "metadata": {"identity_source": "clinic_data_proxy"}, "fetched_at": server.now_iso(), } ) updated_library = server.scope_library() self.assertEqual( "Authoritative Organization", updated_library["organizations"][0]["name"], ) self.assertEqual("authoritative-main", updated_library["clusters"][0]["name"]) def test_finding_upgrades_dbaas_org_placeholder_to_portal_org_identity(self) -> None: server.upsert_cluster_metadata( { "cluster_id": "cluster-test", "clinic_org_id": None, "tenant_id": "dbaas-org-1", "org_name": None, "cluster_name": "main", "deploy_type": "dedicated", "metadata": {"identity_source": "clinic_data_proxy"}, "fetched_at": server.now_iso(), } ) payload = self.finding_payload("portal-org-mapping") payload["dimensions"] = { "cluster": { "name": "main", "org_name": "Test Organization", "dbaas_org_id": "dbaas-org-1", } } status, _ = server.create_finding(payload) self.assertEqual(201, status) metadata = server.cluster_metadata({"cluster_id": ["cluster-test"]})["items"][0] self.assertEqual("org-test", metadata["org_id"]) self.assertEqual("dbaas-org-1", metadata["dbaas_org_id"]) self.assertEqual("Test Organization", metadata["org_name"]) def test_late_backfill_recomputes_first_observation(self) -> None: later = self.finding_payload("later") later["time_window_start"] = "2026-07-02T00:00:00Z" later["time_window_end"] = "2026-07-02T01:00:00Z" later["observed_at"] = "2026-07-02T01:00:00Z" _, later_body = server.create_finding(later) earlier = self.finding_payload("earlier") _, earlier_body = server.create_finding(earlier) with server.get_conn() as conn: later_row = server.row_to_finding( conn.execute( "SELECT * FROM findings WHERE id = ?", (later_body["finding"]["id"],) ).fetchone() ) earlier_row = server.row_to_finding( conn.execute( "SELECT * FROM findings WHERE id = ?", (earlier_body["finding"]["id"],) ).fetchone() ) self.assertFalse(later_row["is_first_observation"]) self.assertTrue(earlier_row["is_first_observation"]) def test_run_coverage_counts_successful_zero_finding_run(self) -> None: for day in (1, 2): status, _ = server.create_finding_run( { "source": "clinic_health_check", "source_type": "inspection", "source_run_id": f"run-{day}", "org_id": "org-test", "cluster_id": "cluster-test", "time_window_start": f"2026-07-0{day}T00:00:00Z", "time_window_end": f"2026-07-0{day}T23:00:00Z", "status": "success", "expected_detector_count": 3, "completed_detector_count": 3, } ) self.assertEqual(201, status) payload = self.finding_payload("run-finding") payload["source"] = "clinic_health_check" payload["source_type"] = "inspection" payload["source_run_id"] = "run-1" payload["symptoms"] = [{"code": "high_total_latency", "severity": "high"}] server.create_finding(payload) stats = server.finding_stats( { "time_basis": ["observed"], "from": ["2026-07-01T00:00:00Z"], "to": ["2026-07-03T00:00:00Z"], "cluster_id": ["cluster-test"], } ) self.assertEqual(100, stats["coverage"]["coverage_percent"]) self.assertEqual(1, stats["coverage"]["zero_finding_runs"]) self.assertEqual(1, stats["symptom_counts"]["high_total_latency"]) def test_run_coverage_detects_a_cluster_missing_from_the_current_period(self) -> None: for day, cluster_ids in ((1, ("cluster-a", "cluster-b")), (2, ("cluster-a",))): for cluster_id in cluster_ids: status, _ = server.create_finding_run( { "source": "clinic_health_check", "source_type": "inspection", "source_run_id": f"run-{day}-{cluster_id}", "org_id": "org-test", "cluster_id": cluster_id, "time_window_start": f"2026-07-0{day}T00:00:00Z", "time_window_end": f"2026-07-0{day}T23:00:00Z", "status": "success", "expected_detector_count": 3, "completed_detector_count": 3, } ) self.assertEqual(201, status) stats = server.finding_stats( { "time_basis": ["observed"], "from": ["2026-07-02T00:00:00Z"], "to": ["2026-07-03T00:00:00Z"], } ) self.assertEqual(2, stats["coverage"]["expected_runs"]) self.assertEqual(1, stats["coverage"]["missing_runs"]) self.assertEqual(50, stats["coverage"]["coverage_percent"]) self.assertFalse(stats["comparison_quality"]["comparable"]) def test_multi_scope_filter_and_previous_period_comparison(self) -> None: first = self.finding_payload("cluster-a", "digest-a") first["cluster_id"] = "cluster-a" first["symptoms"] = [{"code": "large_key_scan", "severity": "high"}] server.create_finding(first) second = self.finding_payload("cluster-b", "digest-b") second["cluster_id"] = "cluster-b" second["time_window_start"] = "2026-07-02T00:00:00Z" second["time_window_end"] = "2026-07-02T01:00:00Z" second["observed_at"] = "2026-07-02T01:00:00Z" second["symptoms"] = [{"code": "large_key_scan", "severity": "high"}] server.create_finding(second) stats = server.finding_stats( { "time_basis": ["observed"], "from": ["2026-07-02T00:00:00Z"], "to": ["2026-07-03T00:00:00Z"], "cluster_id": ["cluster-a,cluster-b"], "symptom_code": ["large_key_scan"], } ) self.assertEqual(1, stats["summary"]["total_findings"]) self.assertEqual(1, stats["comparison"]["summary"]["total_findings"]) self.assertEqual(0, stats["deltas"]["total_findings"]["absolute"]) self.assertEqual("previous", stats["comparison_mode"]) self.assertEqual("2026-07-01T00:00:00Z", stats["comparison_period"]["from"]) self.assertEqual("2026-07-02T00:00:00Z", stats["comparison_period"]["to"]) def test_no_action_requires_related_record(self) -> None: item = self.create_item() status, _ = server.triage_catalog_item(item["id"], {"action": "no_action"}) self.assertEqual(409, status) status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Accepted background workload", "review_status": "ready", "structured_fields": { "conclusion_type": "accepted_risk", "confidence": "high", "reopen_policy": "meaningful_change", }, "body_markdown": "Continue monitoring at the current threshold.", }, ) self.assertEqual(201, status) status, body = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": record_body["record"]["id"]}, ) self.assertEqual(200, status) self.assertEqual("no_intervention", body["item"]["current_posture"]["state"]) self.assertIsNone(body["item"]["current_posture"]["risk"]) self.assertEqual("high", body["item"]["current_posture"]["last_known_risk"]) def test_catalog_decision_requires_ready_record_and_reopening_policy(self) -> None: item = self.create_item("decision-validation") status, draft_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Draft decision", "structured_fields": { "conclusion_type": "known_stable", "confidence": "high", "reopen_policy": "meaningful_change", }, "body_markdown": "The decision is not ready yet.", }, ) self.assertEqual(201, status) status, body = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": draft_body["record"]["id"]}, ) self.assertEqual(409, status) self.assertEqual("triage_record_not_ready", body["error"]) status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Incomplete reopening policy", "review_status": "ready", "structured_fields": { "conclusion_type": "known_stable", "confidence": "high", }, "body_markdown": "The decision needs an explicit reopening policy.", }, ) self.assertEqual(201, status) status, body = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": record_body["record"]["id"]}, ) self.assertEqual(409, status) self.assertEqual("invalid_governance_memory", body["error"]) def test_governance_memory_covers_stable_repeated_findings(self) -> None: item = self.create_item("stable-baseline") status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Known stable business workload", "review_status": "ready", "structured_fields": { "conclusion_type": "known_stable", "confidence": "high", "reopen_policy": "meaningful_change", }, "body_markdown": "The symptom and workload pattern are understood and stable.", }, ) self.assertEqual(201, status) status, _ = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": record_body["record"]["id"]}, ) self.assertEqual(200, status) repeated = self.finding_payload("stable-repeat") repeated["reported_at"] = ( datetime.now(timezone.utc) + timedelta(minutes=1) ).isoformat() repeated["metrics"] = {"slow_count": 13, "max_latency_ms": 9400} status, created = server.create_finding(repeated) self.assertEqual(201, status) posture = created["catalog_item"]["current_posture"] self.assertEqual("no_intervention", posture["state"]) self.assertEqual(1, posture["new_anomaly_count"]) self.assertEqual(0, posture["unreviewed_anomaly_count"]) self.assertEqual("active", posture["governance_memory"]["status"]) self.assertEqual(1, posture["governance_memory"]["covered_finding_count"]) self.assertTrue( posture["review_admission"]["suppressed_by_governance_memory"] ) self.assertIsNone(posture["review_priority"]["tier"]) self.assertFalse(created["catalog_item"]["is_reappeared"]) def test_historical_decision_only_covers_evidence_through_its_cutoff(self) -> None: baseline = self.finding_payload("historical-baseline") baseline["reported_at"] = "2026-07-01T01:00:00Z" status, baseline_body = server.create_finding(baseline) self.assertEqual(201, status) item = baseline_body["catalog_item"] later = self.finding_payload("historical-later") later["reported_at"] = "2026-07-03T01:00:00Z" status, _ = server.create_finding(later) self.assertEqual(201, status) status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Imported weekly review", "review_status": "reviewed", "structured_fields": { "conclusion_type": "known_stable", "confidence": "high", "reopen_policy": "meaningful_change", }, "body_markdown": "This historical review covered evidence through July 1 only.", }, ) self.assertEqual(201, status) status, body = server.triage_catalog_item( item["id"], { "action": "no_action", "record_id": record_body["record"]["id"], "reviewed_through_at": "2026-07-01T23:59:59Z", }, ) self.assertEqual(200, status) posture = body["item"]["current_posture"] self.assertEqual("active", posture["governance_memory"]["status"]) self.assertEqual(1, posture["governance_memory"]["new_evidence_count"]) self.assertEqual(1, posture["governance_memory"]["covered_finding_count"]) self.assertEqual(1, posture["new_anomaly_count"]) self.assertEqual(0, posture["unreviewed_anomaly_count"]) def test_governance_memory_reopens_for_a_new_symptom(self) -> None: item = self.create_item("new-symptom-baseline") status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Expected latency pattern", "review_status": "ready", "structured_fields": { "conclusion_type": "expected_workload", "confidence": "high", "reopen_policy": "meaningful_change", }, "body_markdown": "Reassess only when the symptom profile changes.", }, ) self.assertEqual(201, status) status, _ = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": record_body["record"]["id"]}, ) self.assertEqual(200, status) changed = self.finding_payload("new-symptom-change") changed["reported_at"] = ( datetime.now(timezone.utc) + timedelta(minutes=1) ).isoformat() changed["symptoms"].append( { "code": "large_key_scan", "severity": "high", "metric_keys": [], } ) status, created = server.create_finding(changed) self.assertEqual(201, status) posture = created["catalog_item"]["current_posture"] self.assertEqual("needs_review", posture["state"]) self.assertEqual("p1", posture["review_priority"]["tier"]) self.assertEqual("change_detected", posture["governance_memory"]["status"]) self.assertIn("new_symptom", posture["governance_memory"]["trigger_reasons"]) self.assertTrue(created["catalog_item"]["is_reappeared"]) def test_governance_memory_reopens_critical_severity_increase_as_p0(self) -> None: item = self.create_item("severity-baseline") status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Accepted high-severity baseline", "review_status": "ready", "structured_fields": { "conclusion_type": "accepted_risk", "confidence": "medium", "reopen_policy": "meaningful_change", }, "body_markdown": "Reassess if source severity increases.", }, ) self.assertEqual(201, status) status, _ = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": record_body["record"]["id"]}, ) self.assertEqual(200, status) critical = self.finding_payload("severity-increase") critical["reported_at"] = ( datetime.now(timezone.utc) + timedelta(minutes=1) ).isoformat() critical["symptoms"][0]["severity"] = "critical" status, created = server.create_finding(critical) self.assertEqual(201, status) posture = created["catalog_item"]["current_posture"] self.assertEqual("needs_review", posture["state"]) self.assertEqual("critical", posture["risk"]) self.assertEqual("p0", posture["review_priority"]["tier"]) self.assertIn( "severity_increase", posture["governance_memory"]["trigger_reasons"] ) def test_governance_memory_reopens_when_metric_guardrail_is_breached(self) -> None: item = self.create_item("metric-baseline") status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Stable latency baseline", "review_status": "ready", "structured_fields": { "conclusion_type": "known_stable", "confidence": "high", "reopen_policy": "meaningful_change", "metric_guardrails": { "max_latency_ms": {"increase_percent": 20} }, }, "body_markdown": "Reassess when maximum latency increases materially.", }, ) self.assertEqual(201, status) status, _ = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": record_body["record"]["id"]}, ) self.assertEqual(200, status) changed = self.finding_payload("metric-change") changed["reported_at"] = ( datetime.now(timezone.utc) + timedelta(minutes=1) ).isoformat() changed["metrics"] = {"slow_count": 12, "max_latency_ms": 12000} status, created = server.create_finding(changed) self.assertEqual(201, status) memory = created["catalog_item"]["current_posture"]["governance_memory"] self.assertEqual("change_detected", memory["status"]) self.assertIn("metric_guardrail_breached", memory["trigger_reasons"]) self.assertEqual("max_latency_ms", memory["metric_changes"][0]["metric"]) self.assertEqual(9200.0, memory["metric_changes"][0]["baseline"]) self.assertEqual(12000.0, memory["metric_changes"][0]["current"]) def test_ranking_observation_does_not_reopen_governance_memory(self) -> None: item = self.create_item("ranking-after-decision-baseline") status, record_body = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": "Known workload", "review_status": "ready", "structured_fields": { "conclusion_type": "expected_workload", "confidence": "high", "reopen_policy": "any_anomaly", }, "body_markdown": "Ranking observations alone are not anomaly evidence.", }, ) self.assertEqual(201, status) status, _ = server.triage_catalog_item( item["id"], {"action": "no_action", "record_id": record_body["record"]["id"]}, ) self.assertEqual(200, status) ranking = self.finding_payload("ranking-after-decision") ranking["source"] = "clinic_health_check" ranking["source_type"] = "inspection" ranking["detector_code"] = "clinic.health_check.sql_digest_report" ranking["reported_at"] = ( datetime.now(timezone.utc) + timedelta(minutes=1) ).isoformat() ranking["dimensions"] = { "check_names": ["slowlog_top_digest"], "top_digest_ranks": {"top_by_total_latency": 1}, } status, created = server.create_finding(ranking) self.assertEqual(201, status) posture = created["catalog_item"]["current_posture"] self.assertEqual("no_intervention", posture["state"]) self.assertEqual("active", posture["governance_memory"]["status"]) self.assertEqual(0, posture["new_anomaly_count"]) self.assertFalse(created["catalog_item"]["is_reappeared"]) def test_ready_records_require_substantive_output(self) -> None: item = self.create_item("record-quality") status, draft = server.create_record( "catalog_item", item["id"], { "record_type": "triage_record", "title": "Triage decision", "body_markdown": "## Finding summary\n\n## Decision\n", }, ) self.assertEqual(201, status) status, body = server.update_record( draft["record"]["id"], {"review_status": "ready"}, ) self.assertEqual(400, status) self.assertEqual("record_content_required", body["error"]) self.assertEqual( {"conclusion_type", "confidence", "body_markdown"}, set(body["details"]["missing_fields"]), ) status, _ = server.update_record( draft["record"]["id"], { "structured_fields": { "conclusion_type": "create_treatment", "confidence": "high", }, "body_markdown": "The repeated latency signal requires intervention.", "review_status": "ready", }, ) self.assertEqual(200, status) def test_fixed_requires_reviewed_observation(self) -> None: item = self.create_item() status, treatment_body = server.create_treatment(self.treatment_payload(), item["id"]) self.assertEqual(201, status) treatment_id = treatment_body["treatment"]["id"] status, body = server.complete_treatment(treatment_id, {"outcome": "fixed"}) self.assertEqual(409, status) self.assertEqual("reviewed_observation_required", body["error"]) record = self.create_reviewed_observation(treatment_id) self.move_treatment_to_execution(treatment_id) due_at = ( datetime.now(timezone.utc) + timedelta(days=1) ).replace(microsecond=0).isoformat().replace("+00:00", "Z") status, _ = server.start_observation( treatment_id, {"record_id": record["id"], "observation_due_at": due_at}, ) self.assertEqual(200, status) status, body = server.complete_treatment(treatment_id, {"outcome": "fixed"}) self.assertEqual(200, status) self.assertEqual("completed", body["treatment"]["state"]) self.assertEqual("fixed", body["treatment"]["outcome"]) _, detail = server.catalog_item_detail(item["id"]) self.assertEqual("watching", detail["item"]["current_posture"]["state"]) self.assertIsNone(detail["item"]["current_posture"]["risk"]) self.assertEqual("high", detail["item"]["current_posture"]["last_known_risk"]) self.assertIsNotNone(detail["item"]["latest_record"]) self.assertIn( record["id"], [entry["id"] for entry in detail["records"]], "Catalog dossier must aggregate Records from its related Treatments.", ) self.assertEqual( {"analysis_record", "observation_record"}, {entry["record_type"] for entry in detail["records"]}, ) def test_treatment_requires_objective_and_success_criteria(self) -> None: item = self.create_item() status, body = server.create_treatment({"title": "Incomplete"}, item["id"]) self.assertEqual(400, status) self.assertEqual("invalid_treatment", body["error"]) self.assertIn("objective is required", body["message"]) status, body = server.create_treatment( {"title": "Incomplete", "objective": "Reduce latency."}, item["id"] ) self.assertEqual(400, status) self.assertIn("success_criteria is required", body["message"]) def test_treatment_requires_ready_analysis_before_planning(self) -> None: item = self.create_item("analysis-gate") _, created = server.create_treatment(self.treatment_payload(), item["id"]) treatment_id = created["treatment"]["id"] status, body = server.transition_treatment( treatment_id, {"stage": "planning"} ) self.assertEqual(409, status) self.assertEqual("analysis_record_required", body["error"]) self.create_ready_analysis(treatment_id) status, body = server.transition_treatment( treatment_id, {"stage": "planning"} ) self.assertEqual(200, status) self.assertEqual("planning", body["treatment"]["stage"]) def test_cancelled_treatment_requires_outcome_record_and_can_be_reopened(self) -> None: item = self.create_item() _, created = server.create_treatment(self.treatment_payload(), item["id"]) treatment_id = created["treatment"]["id"] status, body = server.complete_treatment(treatment_id, {"outcome": "cancelled"}) self.assertEqual(409, status) self.assertEqual("outcome_record_required", body["error"]) status, record_body = server.create_record( "treatment", treatment_id, { "record_type": "outcome_record", "title": "Treatment cancelled", "review_status": "ready", "structured_fields": {"outcome": "cancelled"}, "body_markdown": "The submitted remediation target was incorrect.", }, ) self.assertEqual(201, status) status, body = server.complete_treatment( treatment_id, { "outcome": "cancelled", "outcome_record_id": record_body["record"]["id"], }, ) self.assertEqual(200, status) self.assertEqual("cancelled", body["treatment"]["state"]) self.assertEqual("cancelled", body["treatment"]["outcome"]) _, catalog_detail = server.catalog_item_detail(item["id"]) self.assertEqual( "needs_review", catalog_detail["item"]["current_posture"]["state"] ) self.assertEqual( "review", catalog_detail["item"]["current_posture"]["next_action"] ) status, body = server.reopen_treatment(treatment_id, {}) self.assertEqual(200, status) self.assertEqual("active", body["treatment"]["state"]) self.assertEqual("assessment", body["treatment"]["stage"]) self.assertIsNone(body["treatment"]["outcome"]) _, catalog_detail = server.catalog_item_detail(item["id"]) self.assertEqual( "in_treatment", catalog_detail["item"]["current_posture"]["state"] ) self.assertEqual( "continue_assessment", catalog_detail["item"]["current_posture"]["next_action"], ) stats = server.treatment_stats({}) self.assertEqual(1, stats["summary"]["new_treatments"]) self.assertEqual(1, stats["summary"]["reopened_treatments"]) self.assertEqual(1, stats["summary"]["closed_treatments"]) self.assertEqual(1, stats["summary"]["net_backlog_change"]) def test_superseded_treatment_keeps_shared_catalog_in_treatment(self) -> None: item = self.create_item() _, first = server.create_treatment(self.treatment_payload("First approach"), item["id"]) _, successor = server.create_treatment(self.treatment_payload("Replacement approach"), item["id"]) status, body = server.complete_treatment( first["treatment"]["id"], { "outcome": "superseded", "superseded_by_treatment_id": successor["treatment"]["id"], }, ) self.assertEqual(200, status) self.assertEqual("superseded", body["treatment"]["outcome"]) self.assertEqual(successor["treatment"]["id"], body["treatment"]["superseded_by_treatment_id"]) _, detail = server.catalog_item_detail(item["id"]) self.assertEqual("in_treatment", detail["item"]["current_posture"]["state"]) self.assertEqual(1, detail["item"]["active_treatment_count"]) def test_completing_one_of_two_active_treatments_keeps_catalog_in_treatment(self) -> None: item = self.create_item() _, first = server.create_treatment(self.treatment_payload("First"), item["id"]) _, second = server.create_treatment(self.treatment_payload("Second"), item["id"]) self.create_reviewed_observation(first["treatment"]["id"]) status, _ = server.complete_treatment(first["treatment"]["id"], {"outcome": "fixed"}) self.assertEqual(200, status) _, detail = server.catalog_item_detail(item["id"]) self.assertEqual("in_treatment", detail["item"]["current_posture"]["state"]) self.assertEqual(1, detail["item"]["active_treatment_count"]) self.assertEqual("active", server.treatment_detail(second["treatment"]["id"])[1]["treatment"]["state"]) def test_latest_same_second_treatment_outcome_determines_catalog_posture(self) -> None: item = self.create_item("same-second-outcomes") _, first = server.create_treatment(self.treatment_payload("First"), item["id"]) _, second = server.create_treatment(self.treatment_payload("Second"), item["id"]) self.create_reviewed_observation(first["treatment"]["id"]) status, _ = server.complete_treatment(first["treatment"]["id"], {"outcome": "fixed"}) self.assertEqual(200, status) status, record_body = server.create_record( "treatment", second["treatment"]["id"], { "record_type": "outcome_record", "title": "Risk accepted", "review_status": "ready", "body_markdown": "The remaining impact is accepted by the owner.", }, ) self.assertEqual(201, status) status, _ = server.complete_treatment( second["treatment"]["id"], { "outcome": "accepted_risk", "outcome_record_id": record_body["record"]["id"], }, ) self.assertEqual(200, status) _, detail = server.catalog_item_detail(item["id"]) self.assertEqual("no_intervention", detail["item"]["current_posture"]["state"]) self.assertEqual( "accepted_risk", detail["item"]["current_posture"]["latest_treatment_outcome"], ) def test_new_finding_after_completed_treatment_marks_item_reappeared(self) -> None: item = self.create_item() _, treatment_body = server.create_treatment(self.treatment_payload(), item["id"]) self.create_reviewed_observation(treatment_body["treatment"]["id"]) status, _ = server.complete_treatment(treatment_body["treatment"]["id"], {"outcome": "fixed"}) self.assertEqual(200, status) status, _ = server.create_finding(self.finding_payload("event-2")) self.assertEqual(201, status) _, detail = server.catalog_item_detail(item["id"]) self.assertTrue(detail["item"]["is_reappeared"]) self.assertEqual("needs_review", detail["item"]["current_posture"]["state"]) self.assertEqual("high", detail["item"]["current_posture"]["risk"]) self.assertEqual("attention", detail["item"]["current_posture"]["attention"]) self.assertEqual("p1", detail["item"]["current_posture"]["review_priority"]["tier"]) self.assertIn("reappeared", detail["item"]["current_posture"]["reasons"]) def test_generic_patch_cannot_change_workflow_fields(self) -> None: item = self.create_item() status, body = server.update_catalog_item(item["id"], {"status": "archived"}) self.assertEqual(400, status) self.assertEqual("no_fields_to_update", body["error"]) _, treatment_body = server.create_treatment(self.treatment_payload(), item["id"]) status, _ = server.update_treatment(treatment_body["treatment"]["id"], {"stage": "completed"}) self.assertEqual(409, status) def test_symptom_registry_rejects_unknown_codes_and_preserves_source_assessment(self) -> None: definitions = server.list_symptom_definitions({}) self.assertIn( "high_total_latency", {item["code"] for item in definitions["items"]} ) unknown = self.finding_payload("unknown-symptom") unknown["symptoms"][0]["code"] = "producer_invented_code" status, body = server.create_finding(unknown) self.assertEqual(400, status) self.assertIn("unregistered codes", body["message"]) payload = self.finding_payload("assessment") payload["symptoms"][0]["assessment"] = { "method": "rule", "summary": "The producer applied the cluster-specific latency policy.", "rule_code": "test.slow_query", "rule_version": "v3", "policy_id": "latency-sensitive", "policy_version": "7", "parameters": {"high_ms": 5000}, } status, body = server.create_finding(payload) self.assertEqual(201, status) symptom = body["finding"]["symptoms"][0] self.assertEqual("high", symptom["severity"]) self.assertEqual("latency-sensitive", symptom["assessment"]["policy_id"]) self.assertEqual(5000, symptom["assessment"]["parameters"]["high_ms"]) def test_unclassified_symptom_is_manual_and_requires_assessment(self) -> None: payload = self.finding_payload("unclassified") payload["symptoms"] = [ {"code": "unclassified_sql_anomaly", "severity": "medium"} ] status, _ = server.create_finding(payload) self.assertEqual(400, status) payload["source"] = "manual" payload["source_type"] = "manual" payload["symptoms"][0]["assessment"] = { "method": "human", "summary": "Customer-visible latency regression needs classification.", } status, _ = server.create_finding(payload) self.assertEqual(201, status) def test_batch_intake_reuses_single_finding_validation_and_idempotency(self) -> None: base = self.finding_payload("unused") defaults = { key: value for key, value in base.items() if key not in {"source", "source_type", "source_event_id", "sql_digest"} } invalid_symptoms = [ {"code": "not_registered", "severity": "medium"} ] status, body = server.create_finding_batch( { "source": "test", "source_type": "integration", "source_batch_id": "batch-42", "defaults": defaults, "findings": [ {"source_event_id": "batch-event-1", "sql_digest": "digest-batch-1"}, { "source_event_id": "batch-event-2", "sql_digest": "digest-batch-2", "symptoms": invalid_symptoms, }, ], } ) self.assertEqual(207, status) self.assertEqual( {"submitted": 2, "created": 1, "duplicate": 0, "rejected": 1}, body["summary"], ) self.assertEqual("batch-42", body["source_batch_id"]) status, retry = server.create_finding_batch( { "source": "test", "source_type": "integration", "source_batch_id": "batch-42", "defaults": defaults, "findings": [ {"source_event_id": "batch-event-1", "sql_digest": "digest-batch-1"} ], } ) self.assertEqual(200, status) self.assertEqual(1, retry["summary"]["duplicate"]) listed = server.list_findings({"source_batch_id": ["batch-42"]}) self.assertEqual(1, listed["total"]) def test_finding_deletion_recomputes_catalog_and_can_be_restored(self) -> None: _, first = server.create_finding(self.finding_payload("delete-first")) second_payload = self.finding_payload("delete-second") second_payload["observed_at"] = "2026-07-02T01:00:00Z" second_payload["time_window_start"] = "2026-07-02T00:00:00Z" second_payload["time_window_end"] = "2026-07-02T01:00:00Z" second_payload["symptoms"][0]["severity"] = "medium" _, second = server.create_finding(second_payload) item_id = first["catalog_item"]["id"] _, preview = server.finding_deletion_preview( {"selection": {"ids": [first["finding"]["id"]]}} ) status, deleted = server.delete_findings( { "selection": preview["selection"], "expected_count": preview["matched_count"], "reason": "The manually submitted values were incorrect.", "deleted_by": "test-user", } ) self.assertEqual(200, status) self.assertEqual(1, deleted["deleted_count"]) _, detail = server.catalog_item_detail(item_id) self.assertIsNone(detail["item"]["current_posture"]["risk"]) self.assertEqual([second["finding"]["id"]], [item["id"] for item in detail["findings"]]) _, preview = server.finding_deletion_preview( {"selection": {"ids": [second["finding"]["id"]]}} ) server.delete_findings( { "selection": preview["selection"], "expected_count": 1, "reason": "The source batch was invalid.", } ) self.assertEqual(0, server.list_catalog_items({})["total"]) self.assertEqual(1, server.list_catalog_items({"include_orphaned": ["true"]})["total"]) _, deleted_detail = server.finding_detail(first["finding"]["id"]) self.assertTrue(deleted_detail["finding"]["is_deleted"]) status, restored = server.restore_findings( {"ids": [first["finding"]["id"]], "reason": "Deletion was selected in error."} ) self.assertEqual(200, status) self.assertEqual(1, restored["restored_count"]) self.assertEqual(1, server.list_catalog_items({})["total"]) _, restored_item = server.catalog_item_detail(item_id) self.assertIsNone(restored_item["item"]["current_posture"]["risk"]) def test_query_selection_supports_cross_page_exclusions_and_count_conflict(self) -> None: ids = [] for index in range(3): payload = self.finding_payload( f"bulk-delete-{index}", digest=f"digest-bulk-delete-{index}" ) payload["source_batch_id"] = "bad-source-batch" _, body = server.create_finding(payload) ids.append(body["finding"]["id"]) selection = { "filter": {"source_batch_id": "bad-source-batch"}, "exclude_ids": [ids[2]], } _, preview = server.finding_deletion_preview({"selection": selection}) self.assertEqual(2, preview["matched_count"]) status, body = server.delete_findings( { "selection": preview["selection"], "expected_count": 2, "reason": "The automated source batch used the wrong report.", } ) self.assertEqual(200, status) self.assertEqual(2, body["deleted_count"]) self.assertEqual( 1, server.list_findings({"source_batch_id": ["bad-source-batch"]})["total"] ) _, stale_preview = server.finding_deletion_preview( {"selection": {"filter": {"source": "test"}}} ) server.create_finding( self.finding_payload("arrived-after-preview", "digest-after-preview") ) status, conflict = server.delete_findings( { "selection": stale_preview["selection"], "expected_count": stale_preview["matched_count"], "reason": "This must detect selection drift.", } ) self.assertEqual(409, status) self.assertEqual("finding_selection_changed", conflict["error"]) def test_deleting_last_finding_keeps_catalog_with_finding_record(self) -> None: _, created = server.create_finding(self.finding_payload("recorded-finding")) finding_id = created["finding"]["id"] item_id = created["catalog_item"]["id"] status, _ = server.create_record( "finding", finding_id, { "record_type": "triage_record", "title": "Human review already exists", "body_markdown": "The input was wrong, but the review remains auditable.", }, ) self.assertEqual(201, status) _, preview = server.finding_deletion_preview( {"selection": {"ids": [finding_id]}} ) status, _ = server.delete_findings( { "selection": preview["selection"], "expected_count": 1, "reason": "Incorrect source event.", } ) self.assertEqual(200, status) self.assertEqual(1, server.list_catalog_items({})["total"]) _, detail = server.catalog_item_detail(item_id) self.assertFalse(detail["item"]["is_orphaned"]) self.assertIsNone(detail["item"]["current_posture"]["risk"]) self.assertIsNone(detail["item"]["current_posture"]["historical_peak_risk"]) def test_catalog_and_treatment_management_signals_are_aggregated(self) -> None: item = self.create_item("management-first") status, _ = server.create_finding(self.finding_payload("management-second")) self.assertEqual(201, status) listed_item = server.list_catalog_items({})["items"][0] self.assertEqual(2, listed_item["finding_count"]) self.assertEqual(2, listed_item["finding_count_7d"]) self.assertEqual(["high_total_latency"], listed_item["symptom_codes"]) self.assertEqual(0, listed_item["treatment_count"]) self.assertEqual(0, listed_item["record_count"]) catalog_stats = server.catalog_stats({}) self.assertEqual(0, catalog_stats["decision_required_count"]) self.assertEqual(1, catalog_stats["recurring_7d_count"]) self.assertEqual(0, catalog_stats["unowned_count"]) self.assertEqual(2, catalog_stats["finding_count_7d"]) self.assertEqual(0, catalog_stats["risk_counts"]["high"]) status, treatment_body = server.create_treatment(self.treatment_payload(), item["id"]) self.assertEqual(201, status) treatment_id = treatment_body["treatment"]["id"] status, _ = server.create_record( "treatment", treatment_id, { "record_type": "analysis_record", "title": "Initial diagnosis", "body_markdown": "The query plan requires review.", }, ) self.assertEqual(201, status) treatment = server.treatment_detail(treatment_id)[1]["treatment"] self.assertEqual(1, treatment["record_count"]) self.assertEqual(2, treatment["linked_finding_count"]) self.assertIsNotNone(treatment["cycle_time_hours"]) treatment_stats = server.treatment_stats({}) self.assertEqual(1, treatment_stats["current_snapshot"]["active_treatments"]) self.assertEqual(1, treatment_stats["current_snapshot"]["unowned_active_treatments"]) assessment_health = next( item for item in treatment_stats["stage_health"] if item["stage"] == "assessment" ) self.assertEqual( 1, assessment_health["count"], ) self.assertEqual(1, assessment_health["entered_in_period_count"]) self.assertEqual(0, assessment_health["exited_in_period_count"]) self.assertEqual(1, treatment_stats["summary"]["new_treatments"]) self.assertEqual(1, treatment_stats["summary"]["net_backlog_change"]) self.assertEqual( 1, sum(item["count"] for item in treatment_stats["distributions"]["active_age"]), ) self.assertEqual( 1, next( item["count"] for item in treatment_stats["distributions"]["started_cohort"] if item["key"] == "assessment" ), ) def test_catalog_list_and_stats_use_bulk_projection(self) -> None: self.create_item("bulk-catalog-first") status, _ = server.create_finding( self.finding_payload("bulk-catalog-second", "bulk-catalog-second-digest") ) self.assertEqual(201, status) with patch.object( server, "enrich_catalog_item", side_effect=AssertionError("Catalog collections must not enrich items one by one"), ): listed = server.list_catalog_items({}) stats = server.catalog_stats({}) self.assertEqual(2, listed["total"]) self.assertEqual(2, stats["total_items"]) self.assertTrue(all("current_posture" in item for item in listed["items"])) self.assertTrue(all("record_count" in item for item in listed["items"])) combined = server.list_catalog_items( {"queue": ["recommended"], "include_stats": ["true"]} ) self.assertEqual(0, combined["total"]) self.assertEqual(2, combined["stats"]["total_items"]) def test_treatment_stats_supports_calendar_and_custom_comparison_periods(self) -> None: calendar_stats = server.treatment_stats( { "from": ["2026-06-01T00:00:00Z"], "to": ["2026-07-01T00:00:00Z"], "timezone_offset_minutes": ["0"], } ) self.assertEqual("previous_calendar", calendar_stats["comparison_mode"]) self.assertEqual("2026-05-01T00:00:00Z", calendar_stats["comparison_period"]["from"]) self.assertEqual("2026-06-01T00:00:00Z", calendar_stats["comparison_period"]["to"]) custom_stats = server.treatment_stats( { "from": ["2026-06-01T00:00:00Z"], "to": ["2026-07-01T00:00:00Z"], "compare_from": ["2026-05-10T00:00:00Z"], "compare_to": ["2026-05-17T00:00:00Z"], "timezone_offset_minutes": ["0"], } ) self.assertEqual("custom", custom_stats["comparison_mode"]) self.assertEqual("2026-05-10T00:00:00Z", custom_stats["comparison_period"]["from"]) self.assertEqual("2026-05-17T00:00:00Z", custom_stats["comparison_period"]["to"]) def test_customer_visible_reviewed_proposal_records_one_delivery_milestone(self) -> None: item = self.create_item("recommendation-finding") _, created = server.create_treatment(self.treatment_payload(), item["id"]) treatment_id = created["treatment"]["id"] status, record_body = server.create_record( "treatment", treatment_id, { "record_type": "proposal_record", "title": "Add a selective index", "review_status": "ready", "visibility": "internal", "structured_fields": { "action_type": "add_index", "risk_level": "low", }, "body_markdown": "Validate the index before rollout.", }, ) self.assertEqual(201, status) record_id = record_body["record"]["id"] self.assertEqual(0, server.treatment_stats({})["summary"]["recommendations_shared"]) status, _ = server.update_record( record_id, {"review_status": "reviewed", "visibility": "customer_visible"}, ) self.assertEqual(200, status) status, _ = server.update_record(record_id, {"title": "Add a selective index safely"}) self.assertEqual(200, status) _, detail = server.treatment_detail(treatment_id) deliveries = [ event for event in detail["timeline"] if event["event_type"] == "recommendation_shared" ] self.assertEqual(1, len(deliveries)) self.assertEqual(record_id, deliveries[0]["payload"]["record_id"]) self.assertEqual("alignment", detail["treatment"]["stage"]) alignment_events = [ event for event in detail["timeline"] if event["event_type"] == "treatment_stage_changed" and event["payload"].get("to") == "alignment" ] self.assertEqual(1, len(alignment_events)) self.assertEqual("recommendation_shared", alignment_events[0]["payload"]["trigger"]) self.assertEqual(1, server.treatment_stats({})["summary"]["recommendations_shared"]) def test_stage_entry_time_comes_from_timestamped_transition_event(self) -> None: item = self.create_item("stage-time-finding") _, created = server.create_treatment(self.treatment_payload(), item["id"]) treatment_id = created["treatment"]["id"] self.create_ready_analysis(treatment_id) status, body = server.transition_treatment(treatment_id, {"stage": "planning"}) self.assertEqual(200, status) _, detail = server.treatment_detail(treatment_id) transition = next( event for event in detail["timeline"] if event["event_type"] == "treatment_stage_changed" ) self.assertEqual(body["treatment"]["updated_at"], transition["created_at"]) self.assertEqual(transition["created_at"], detail["treatment"]["stage_entered_at"]) self.assertEqual("planning", transition["payload"]["to"]) self.assertEqual(item["id"], detail["treatment"]["catalog_items"][0]["id"]) def test_treatment_list_prioritizes_active_work_over_completed_history(self) -> None: first_item = self.create_item("completed-treatment-finding") _, completed_body = server.create_treatment(self.treatment_payload("Completed history"), first_item["id"]) self.create_reviewed_observation(completed_body["treatment"]["id"]) status, _ = server.complete_treatment(completed_body["treatment"]["id"], {"outcome": "fixed"}) self.assertEqual(200, status) _, second_finding = server.create_finding( self.finding_payload("open-treatment-finding", "digest-open-treatment") ) _, active_body = server.create_treatment( self.treatment_payload("Active treatment"), second_finding["catalog_item"]["id"] ) treatments = server.list_treatments({})["items"] self.assertEqual(active_body["treatment"]["id"], treatments[0]["id"]) self.assertEqual("active", treatments[0]["state"]) self.assertEqual("completed", treatments[1]["state"]) stats = server.treatment_stats({}) self.assertEqual(1, stats["summary"]["completed_treatments"]) self.assertIsNotNone(stats["summary"]["median_completion_time_hours"]) self.assertEqual( 1, next( item["count"] for item in stats["distributions"]["closed_outcomes"] if item["key"] == "fixed" ), ) if __name__ == "__main__": unittest.main()