72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from io import StringIO
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
from django.test import override_settings
|
|
|
|
from control_plane.models import DeploymentStatus
|
|
from control_plane.models import HostedSite
|
|
from control_plane.models import Tenant
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_test_deployment_command_creates_randomized_records() -> None:
|
|
"""Management command should create fresh tenant and site slugs for each invocation."""
|
|
with (
|
|
patch(
|
|
"control_plane.management.commands.create_test_deployment.provision_test_deployment",
|
|
return_value=SimpleNamespace(status=DeploymentStatus.RUNNING.value),
|
|
) as mock_provision,
|
|
):
|
|
first_output = StringIO()
|
|
second_output = StringIO()
|
|
call_command("create_test_deployment", stdout=first_output)
|
|
call_command("create_test_deployment", stdout=second_output)
|
|
|
|
tenants = tuple(Tenant.objects.order_by("created_at"))
|
|
hosted_sites = tuple(HostedSite.objects.order_by("created_at"))
|
|
|
|
assert len(tenants) == 2
|
|
assert len(hosted_sites) == 2
|
|
assert tenants[0].slug != tenants[1].slug
|
|
assert hosted_sites[0].slug != hosted_sites[1].slug
|
|
assert mock_provision.call_count == 2
|
|
assert "execution_mode=inline" in first_output.getvalue()
|
|
assert "execution_mode=inline" in second_output.getvalue()
|
|
assert "status=running" in first_output.getvalue()
|
|
assert "status=running" in second_output.getvalue()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_test_deployment_command_can_return_without_waiting() -> None:
|
|
"""Management command should support queue-only local workflows."""
|
|
output = StringIO()
|
|
|
|
with (
|
|
patch(
|
|
"control_plane.management.commands.create_test_deployment.queue_test_deployment_provisioning",
|
|
return_value="task-1",
|
|
) as mock_queue,
|
|
patch(
|
|
"control_plane.management.commands.create_test_deployment.provision_test_deployment",
|
|
) as mock_provision,
|
|
):
|
|
call_command("create_test_deployment", no_wait=True, stdout=output)
|
|
|
|
assert mock_queue.call_count == 1
|
|
mock_provision.assert_not_called()
|
|
assert "status=queued" in output.getvalue()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_test_deployment_command_rejects_queue_only_mode_without_real_broker() -> None:
|
|
"""Queue-only mode should fail fast when Celery has no usable cross-process broker."""
|
|
output = StringIO()
|
|
|
|
with override_settings(CELERY_BROKER_URL="memory://"), pytest.raises(CommandError, match="cannot use memory://"):
|
|
call_command("create_test_deployment", no_wait=True, stdout=output)
|