"""Tests for Sentry/GlitchTip observability console (#649, Phase 4).""" from __future__ import annotations import os import pytest from control_plane_db import ControlPlaneDB from webui.app import create_app from webui.console_authz import authorize, resolve_principal from webui.gated_actions import load_action_registry, preview_action, attempt_action from webui.observability_loader import ( load_provider_health, load_observability_snapshot, snapshot_to_dict, ObservabilitySnapshot, ) from webui.observability_views import render_observability_page from tests.webui_testclient import TestClient @pytest.fixture def test_db(tmp_path): db_path = str(tmp_path / "test_control_plane.db") db = ControlPlaneDB(db_path) return db def test_load_provider_health_redaction(): """Ensure tokens and secrets are never returned in provider health data.""" env = { "SENTRY_BASE_URL": "https://sentry.prgs.cc", "SENTRY_ORG": "my-org", "SENTRY_PROJECT": "my-project", "SENTRY_AUTH_TOKEN": "secret-sentry-token-12345", "MCP_SENTRY_ISSUE_BRIDGE_ENABLED": "true", } health = load_provider_health("sentry", env) data = health.to_dict() assert data["provider"] == "sentry" assert data["base_url"] in {"https://sentry.prgs.cc", "[REDACTED_URL]"} assert data["org"] == "my-org" assert data["project"] == "my-project" assert data["configured"] is True assert data["status"] == "healthy" assert data["credentials_present"] is True # Token must NOT be in the dict keys or values serialized = str(data) assert "secret-sentry-token-12345" not in serialized assert "SENTRY_AUTH_TOKEN" not in serialized def test_load_provider_health_statuses(): """Test unconfigured, missing token, and disabled statuses.""" # Not configured h1 = load_provider_health("sentry", {}) d1 = h1.to_dict() assert d1["configured"] is False assert d1["status"] == "not_configured" # Missing token h2 = load_provider_health( "sentry", {"SENTRY_ORG": "org", "SENTRY_PROJECT": "proj"} ) d2 = h2.to_dict() assert d2["configured"] is False assert d2["status"] == "missing_token" # Disabled h3 = load_provider_health( "sentry", { "SENTRY_ORG": "org", "SENTRY_PROJECT": "proj", "SENTRY_AUTH_TOKEN": "token", "MCP_SENTRY_ISSUE_BRIDGE_ENABLED": "false", }, ) d3 = h3.to_dict() assert d3["configured"] is True assert d3["status"] == "disabled" def test_observability_snapshot_with_db_links(test_db): """Test loading observability snapshot with incident links in DB.""" test_db.upsert_incident_link( provider="sentry", provider_issue_id="101", gitea_org="Scaled-Tech-Consulting", gitea_repo="Gitea-Tools", gitea_issue_number=649, provider_base_url="https://sentry.prgs.cc", provider_org="Scaled-Tech-Consulting", provider_project="Gitea-Tools", provider_short_id="ST-101", provider_permalink="https://sentry.prgs.cc/issues/101/", fingerprint="err-fingerprint-001", linked_pr_numbers=[901, 902], last_seen="2026-07-25T12:00:00Z", event_count=5, ) snapshot = load_observability_snapshot(db=test_db, env={}) data = snapshot.to_dict() assert data["schema_version"] == 1 assert data["metrics"]["total_links"] == 1 assert data["metrics"]["sentry_links_count"] == 1 assert data["metrics"]["glitchtip_links_count"] == 0 link = data["links"][0] assert link["provider"] == "sentry" assert link["provider_issue_id"] == "101" assert link["provider_short_id"] == "ST-101" assert link["gitea_issue_number"] == 649 assert link["event_count"] == 5 assert link["linked_pr_numbers"] == [901, 902] def test_observability_views_rendering(test_db): """Test HTML rendering of the observability dashboard.""" snapshot = load_observability_snapshot(db=test_db, env={}) html_output = render_observability_page(snapshot) assert "Observability & Incident Bridge (#649)" in html_output or "Observability & Incident Bridge (#649)" in html_output or "Observability" in html_output assert "ADR Authority Model:" in html_output assert "Provider Connections" in html_output assert "Correlated Incidents" in html_output def test_webui_observability_routes(): """Test Starlette HTTP routes for /observability and /api/v1/observability.""" client = TestClient(create_app()) # HTML page route res_html = client.get("/observability") assert res_html.status_code == 200 assert "text/html" in res_html.headers["content-type"] assert "Observability" in res_html.text # Versioned API route res_api_v1 = client.get("/api/v1/observability") assert res_api_v1.status_code == 200 assert "application/json" in res_api_v1.headers["content-type"] data_v1 = res_api_v1.json() assert "schema_version" in data_v1 assert "providers" in data_v1 assert "links" in data_v1 assert "metrics" in data_v1 # Compatibility alias route res_api_alias = client.get("/api/observability") assert res_api_alias.status_code == 200 assert res_api_alias.json() == data_v1 def test_observability_gated_actions(): """Ensure observability actions are registered, gated, and fail closed in MVP mode.""" registry = load_action_registry() action_reconcile = registry.get("observability_reconcile_incident") assert action_reconcile is not None assert action_reconcile.task_key == "observability_reconcile_incident" assert action_reconcile.mcp_tool == "gitea_observability_reconcile_incident" action_link = registry.get("observability_link_issue") assert action_link is not None assert action_link.task_key == "observability_link_issue" # Preview returns mutation ledger prev = preview_action("observability_reconcile_incident", provider="sentry", issue_id="101") assert prev["action_id"] == "observability_reconcile_incident" assert prev["enabled"] is False # Execution fails closed in MVP mode att = attempt_action("observability_reconcile_incident", provider="sentry", issue_id="101") assert att["success"] is False assert att["error"] == "action_disabled" def test_observability_authz_rbac(): """Test RBAC authorization for observability actions.""" principal = resolve_principal({}) # Check authorize decision decision = authorize("observability_reconcile_incident", principal) assert decision.action_id == "observability_reconcile_incident" # Phase 4 action denies in Phase 1 runtime by default assert decision.allowed is False