from __future__ import annotations import json from typing import TYPE_CHECKING from unittest.mock import MagicMock from unittest.mock import patch import pytest from django.test import override_settings from django.urls import reverse from control_plane.models import Deployment from control_plane.models import DeploymentStatus from control_plane.models import HostedSite from control_plane.models import RuntimeService from control_plane.models import RuntimeServiceStatus from control_plane.models import Tenant from control_plane.observability import build_test_deployment_diagnostics_snapshot_path from control_plane.observability import capture_test_deployment_diagnostics from control_plane.observability import load_test_deployment_diagnostics from control_plane.observability import probe_test_deployment_health if TYPE_CHECKING: from pathlib import Path from django.test import Client def _create_deployment(*, status: str = DeploymentStatus.QUEUED.value, last_error: str = "") -> Deployment: tenant = Tenant.objects.create(slug=f"tenant-{status}", display_name=f"Tenant {status}") hosted_site = HostedSite.objects.create( tenant=tenant, slug=f"site-{status}", display_name=f"Site {status}", wsgi_module="tenant_site.wsgi:application", ) return Deployment.objects.create( hosted_site=hosted_site, idempotency_key=f"deploy-{status}", source_sha256=status[0] * 64, status=status, last_error=last_error, ) @pytest.mark.django_db def test_deployment_dashboard_lists_recent_deployments(client: Client) -> None: """Dashboard list view should surface deployment identity and runtime counts.""" deployment = _create_deployment(status=DeploymentStatus.RUNNING.value) deployment.ensure_test_runtime_services() RuntimeService.objects.filter(deployment=deployment).update(status=RuntimeServiceStatus.READY.value) response = client.get(reverse("control_plane:deployment-dashboard")) assert response.status_code == 200 content = response.content.decode() assert deployment.hosted_site.tenant.slug in content assert deployment.hosted_site.slug in content assert "Inspect deployment" in content assert "Open sentinel" in content @pytest.mark.django_db def test_deployment_detail_renders_persisted_diagnostics(tmp_path: Path, client: Client) -> None: """Detail view should render captured pod state and container logs.""" deployment = _create_deployment(status=DeploymentStatus.RUNNING.value) deployment.ensure_test_runtime_services() RuntimeService.objects.filter(deployment=deployment).update(status=RuntimeServiceStatus.READY.value) runtime_services = tuple(RuntimeService.objects.filter(deployment=deployment).order_by("kind")) with ( override_settings(DATA_DIR=tmp_path), patch( "control_plane.views.probe_test_deployment_health", return_value={ "status": "healthy", "label": "Healthy", "checked_at": "now", "ok": True, "payload": {"status": "ok"}, "error": "", }, ), ): snapshot_path = build_test_deployment_diagnostics_snapshot_path(deployment) snapshot_path.parent.mkdir(parents=True, exist_ok=True) snapshot_path.write_text( json.dumps( { "captured_at": "2026-04-27T18:10:00+00:00", "pod": {"name": runtime_services[0].network_name, "status": "Running", "error": ""}, "django": { "container_name": "django-test", "container_status": "running", "logs": "gunicorn boot ok", "inspect_error": "", "log_error": "", }, "runtime_services": [ { "label": runtime_services[0].kind, "container_name": runtime_services[0].container_name, "container_status": "healthy", "logs": "postgres boot ok", "inspect_error": "", "log_error": "", }, { "label": runtime_services[1].kind, "container_name": runtime_services[1].container_name, "container_status": "healthy", "logs": "redis ready", "inspect_error": "", "log_error": "", }, ], }, ), encoding="utf-8", ) response = client.get(reverse("control_plane:deployment-detail", args=[deployment.id])) assert response.status_code == 200 content = response.content.decode() assert "gunicorn boot ok" in content assert "postgres boot ok" in content assert "redis ready" in content assert deployment.hosted_site.slug in content @pytest.mark.django_db def test_deployment_health_endpoint_returns_probe_payload(client: Client) -> None: """Health endpoint should proxy structured sentinel probe state as JSON.""" deployment = _create_deployment(status=DeploymentStatus.RUNNING.value) with patch( "control_plane.views.probe_test_deployment_health", return_value={ "checked_at": "2026-04-27T18:12:00+00:00", "deployment_id": str(deployment.id), "deployment_status": deployment.status, "sentinel_url": "http://127.0.0.1:9999/sentinel/", "ok": True, "status": "healthy", "label": "Healthy", "payload": {"status": "ok", "postgres": 1}, "error": "", "http_status": 200, }, ): response = client.get(reverse("control_plane:deployment-health", args=[deployment.id])) assert response.status_code == 200 assert response.json()["status"] == "healthy" assert response.json()["payload"]["postgres"] == 1 @pytest.mark.django_db def test_capture_test_deployment_diagnostics_writes_snapshot(tmp_path: Path) -> None: """Diagnostics capture should persist pod status plus log tails for each container.""" deployment = _create_deployment(status=DeploymentStatus.RUNNING.value) deployment.ensure_test_runtime_services() RuntimeService.objects.filter(deployment=deployment).update(status=RuntimeServiceStatus.READY.value) runtime_services = tuple(RuntimeService.objects.filter(deployment=deployment).order_by("kind")) with ( override_settings(DATA_DIR=tmp_path), patch( "control_plane.observability.run_host_command", side_effect=[ MagicMock(stdout="Running\n", stderr=""), MagicMock(stdout="running\n", stderr=""), MagicMock(stdout="gunicorn started\n", stderr=""), MagicMock(stdout="healthy\n", stderr=""), MagicMock(stdout="postgres boot ok\n", stderr=""), MagicMock(stdout="healthy\n", stderr=""), MagicMock(stdout="redis ready\n", stderr=""), ], ), ): capture_test_deployment_diagnostics(str(deployment.id)) snapshot = load_test_deployment_diagnostics(deployment) assert snapshot is not None assert snapshot["pod"]["name"] == runtime_services[0].network_name assert snapshot["django"]["logs"] == "gunicorn started" assert snapshot["runtime_services"][0]["logs"] == "postgres boot ok" assert snapshot["runtime_services"][1]["logs"] == "redis ready" @pytest.mark.django_db def test_probe_test_deployment_health_reports_success() -> None: """Sentinel probe helper should classify healthy payloads as healthy.""" deployment = _create_deployment(status=DeploymentStatus.RUNNING.value) response = MagicMock() response.__enter__.return_value = response response.read.return_value = b'{"status": "ok", "postgres": 1, "redis": "site-running"}' response.status = 200 with patch("control_plane.observability.urlopen", return_value=response): payload = probe_test_deployment_health(deployment) assert payload["ok"] is True assert payload["status"] == "healthy" assert payload["payload"]["status"] == "ok"