205 lines
8.3 KiB
Python
205 lines
8.3 KiB
Python
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"
|