from __future__ import annotations from typing import TYPE_CHECKING from unittest.mock import MagicMock from unittest.mock import patch import pytest from django.test import override_settings from control_plane import tasks as control_plane_tasks from control_plane.host_commands import HostCommandError from control_plane.host_commands import HostCommandResult 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 RuntimeServiceKind from control_plane.models import RuntimeServiceStatus from control_plane.models import Tenant from control_plane.tasks import mark_deployment_booting from control_plane.tasks import mark_deployment_provisioning from control_plane.tasks import provision_test_django_runtime from control_plane.tasks import provision_test_runtime_services if TYPE_CHECKING: from pathlib import Path def _create_deployment(*, status: str = DeploymentStatus.QUEUED.value, last_error: str = "") -> Deployment: tenant = Tenant.objects.create(slug="acme", display_name="Acme") hosted_site = HostedSite.objects.create( tenant=tenant, slug="portal", display_name="Portal", wsgi_module="portal.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_mark_deployment_provisioning_moves_queued_deployment_to_provisioning() -> None: """Provisioning task should persist the stored provisioning value and clear prior errors.""" deployment = _create_deployment(last_error="previous failure") result = mark_deployment_provisioning.run(str(deployment.id)) deployment.refresh_from_db() assert result == DeploymentStatus.PROVISIONING.value assert deployment.status == DeploymentStatus.PROVISIONING.value assert not deployment.last_error @pytest.mark.django_db def test_mark_deployment_provisioning_leaves_terminal_state_unchanged() -> None: """Provisioning task should not mutate deployments already in a terminal state.""" deployment = _create_deployment( status=DeploymentStatus.FAILED.value, last_error="still failed", ) result = mark_deployment_provisioning.run(str(deployment.id)) deployment.refresh_from_db() assert result == DeploymentStatus.FAILED.value assert deployment.status == DeploymentStatus.FAILED.value assert deployment.last_error == "still failed" @pytest.mark.django_db def test_mark_deployment_booting_moves_provisioning_deployment_to_booting() -> None: """Booting task should persist the stored booting value for active deployments.""" deployment = _create_deployment(status=DeploymentStatus.PROVISIONING.value) result = mark_deployment_booting.run(str(deployment.id)) deployment.refresh_from_db() assert result == DeploymentStatus.BOOTING.value assert deployment.status == DeploymentStatus.BOOTING.value @pytest.mark.django_db def test_provision_test_runtime_services_creates_and_launches_test_containers( tmp_path: Path, ) -> None: """Provisioning task should seed missing services and run Podman commands for them.""" deployment = _create_deployment() with ( override_settings(DATA_DIR=tmp_path), patch( "control_plane.tasks.run_host_command", side_effect=[ HostCommandError( "missing pod", args=("podman", "pod", "exists", "ignored"), returncode=1, stdout="", stderr="", ), HostCommandResult( args=("podman", "pod", "create", "ignored"), returncode=0, stdout="pod-id\n", stderr="", ), HostCommandResult( args=("podman", "run", "postgres"), returncode=0, stdout="postgres-id\n", stderr="", ), HostCommandResult( args=("podman", "inspect", "postgres"), returncode=0, stdout="healthy\n", stderr="", ), HostCommandResult( args=("podman", "pod", "exists", "ignored"), returncode=0, stdout="", stderr="", ), HostCommandResult( args=("podman", "run", "redis"), returncode=0, stdout="redis-id\n", stderr="", ), HostCommandResult( args=("podman", "inspect", "redis"), returncode=0, stdout="healthy\n", stderr="", ), ], ) as mock_run_host_command, ): result = provision_test_runtime_services.run(str(deployment.id)) runtime_services = tuple(RuntimeService.objects.filter(deployment=deployment).order_by("kind")) assert result == RuntimeServiceStatus.READY.value assert len(runtime_services) == 2 assert {runtime_service.status for runtime_service in runtime_services} == { RuntimeServiceStatus.READY.value, } for runtime_service in runtime_services: runtime_service_root = tmp_path / "runtime-services" / str(deployment.id) / runtime_service.kind assert (runtime_service_root / "data").is_dir() is True assert (runtime_service_root / "secrets" / "password").read_text(encoding="utf-8").strip() assert mock_run_host_command.call_count == 7 @pytest.mark.django_db def test_provision_test_runtime_services_skips_ready_runtime_services(tmp_path: Path) -> None: """Provisioning task should remain idempotent once all runtime services are already ready.""" deployment = _create_deployment() deployment.ensure_test_runtime_services() RuntimeService.objects.filter(deployment=deployment).update(status=RuntimeServiceStatus.READY.value) with override_settings(DATA_DIR=tmp_path), patch("control_plane.tasks.run_host_command") as mock_run_host_command: result = provision_test_runtime_services.run(str(deployment.id)) assert result == RuntimeServiceStatus.READY.value mock_run_host_command.assert_not_called() @pytest.mark.django_db def test_provision_test_django_runtime_builds_and_runs_generated_app(tmp_path: Path) -> None: """Django runtime task should build image, run migrations, and publish the app container.""" deployment = _create_deployment(status=DeploymentStatus.BOOTING.value) deployment.ensure_test_runtime_services() RuntimeService.objects.filter(deployment=deployment).update(status=RuntimeServiceStatus.READY.value) with ( override_settings(DATA_DIR=tmp_path), patch( "control_plane.tasks.run_host_command", side_effect=[ HostCommandError( "missing image", args=("podman", "image", "exists", "ignored"), returncode=1, stdout="", stderr="", ), HostCommandResult( args=("podman", "build", "ignored"), returncode=0, stdout="built\n", stderr="", ), HostCommandResult( args=("podman", "run", "migrate"), returncode=0, stdout="migrated\n", stderr="", ), HostCommandResult( args=("podman", "run", "django"), returncode=0, stdout="container-id\n", stderr="", ), ], ) as mock_run_host_command, patch( "control_plane.tasks._wait_for_http_ready", return_value={"status": "ok", "postgres": 1, "redis": deployment.hosted_site.slug}, ) as mock_wait_for_http_ready, ): result = provision_test_django_runtime.run(str(deployment.id)) deployment.refresh_from_db() project_root = tmp_path / "test-deployments" / str(deployment.id) / "django-app" assert result == DeploymentStatus.RUNNING.value assert deployment.status == DeploymentStatus.RUNNING.value assert deployment.started_at is not None assert (project_root / "manage.py").is_file() is True assert (project_root / "tenant_site" / "settings.py").is_file() is True assert mock_run_host_command.call_count == 4 assert mock_run_host_command.call_args_list[0].kwargs["command"][:3] == ( "podman", "image", "exists", ) assert "--rm" in mock_run_host_command.call_args_list[2].kwargs["command"] assert "--pod" in mock_run_host_command.call_args_list[3].kwargs["command"] assert "--publish" not in mock_run_host_command.call_args_list[3].kwargs["command"] mock_wait_for_http_ready.assert_called_once() @pytest.mark.django_db def test_provision_test_django_runtime_marks_failure_when_services_are_not_ready(tmp_path: Path) -> None: """Django runtime task should fail fast when backing services are not ready yet.""" deployment = _create_deployment(status=DeploymentStatus.BOOTING.value) deployment.ensure_test_runtime_services() RuntimeService.objects.filter( deployment=deployment, kind=RuntimeServiceKind.POSTGRESQL.value, ).update(status=RuntimeServiceStatus.FAILED.value) with override_settings(DATA_DIR=tmp_path), pytest.raises(ValueError, match="must be ready"): provision_test_django_runtime.run(str(deployment.id)) deployment.refresh_from_db() assert deployment.status == DeploymentStatus.FAILED.value assert "must be ready" in deployment.last_error def test_wait_for_http_ready_retries_connection_reset() -> None: """Sentinel polling should retry transient socket resets while Gunicorn finishes booting.""" response = MagicMock() response.__enter__.return_value = response response.read.return_value = b'{"status": "ok", "postgres": 1, "redis": "portal"}' with ( patch( "control_plane.tasks.urlopen", side_effect=[ConnectionResetError(104, "Connection reset by peer"), response], ), patch("control_plane.tasks.time.sleep") as mock_sleep, ): payload = control_plane_tasks._wait_for_http_ready( "http://127.0.0.1:18000/sentinel/", timeout_seconds=2.0, ) assert payload == {"status": "ok", "postgres": 1, "redis": "portal"} mock_sleep.assert_called_once_with(1.0)