feat: add sanitized internal main deployment

tests/apps/core/test_sanitize_cloned_environment_command.py

@@ -0,0 +1,212 @@
+from __future__ import annotations
+
+import json
+from io import StringIO
+
+from apps.core.models import (
+    BackgroundJob,
+    JobStatus,
+    ReportUpload,
+    ReportUploadStatus,
+)
+from django.core.management import call_command
+from django.test import TestCase
+from django.utils import timezone
+from django_celery_beat.models import IntervalSchedule, PeriodicTask
+
+from tests.apps.user.factories import UserFactory
+
+
+class SanitizeClonedEnvironmentCommandTest(TestCase):
+    def setUp(self) -> None:
+        PeriodicTask.objects.all().delete()
+        interval = IntervalSchedule.objects.create(
+            every=1,
+            period=IntervalSchedule.HOURS,
+        )
+        self.enabled_schedule = PeriodicTask.objects.create(
+            name="enabled-schedule",
+            task="apps.core.tasks.enabled",
+            interval=interval,
+            enabled=True,
+        )
+        self.disabled_schedule = PeriodicTask.objects.create(
+            name="disabled-schedule",
+            task="apps.core.tasks.disabled",
+            interval=interval,
+            enabled=False,
+        )
+
+        self.user = UserFactory.create_user()
+        self.active_jobs = [
+            BackgroundJob.objects.create(
+                task_id=f"active-{status}",
+                task_name="apps.core.tasks.active",
+                status=status,
+                user_id=self.user.id,
+            )
+            for status in (JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY)
+        ]
+        self.finished_job = BackgroundJob.objects.create(
+            task_id="finished-success",
+            task_name="apps.core.tasks.finished",
+            status=JobStatus.SUCCESS,
+            progress=100,
+            result={"records": 42},
+            user_id=self.user.id,
+            completed_at=timezone.now(),
+        )
+
+        self.active_uploads = [
+            self._create_upload(status=status, suffix=str(index))
+            for index, status in enumerate(
+                (ReportUploadStatus.QUEUED, ReportUploadStatus.PROCESSING),
+                start=1,
+            )
+        ]
+        self.finished_upload = self._create_upload(
+            status=ReportUploadStatus.SUCCESS,
+            suffix="finished",
+            load_batch=42,
+        )
+
+    def _create_upload(
+        self,
+        *,
+        status: str,
+        suffix: str,
+        load_batch: int | None = None,
+    ) -> ReportUpload:
+        return ReportUpload.objects.create(
+            form="f1",
+            load_batch=load_batch,
+            original_file=f"report_uploads/f1/{suffix}/report.xlsx",
+            file_name=f"report-{suffix}.xlsx",
+            content_type=(
+                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+            ),
+            file_size=128,
+            file_hash=suffix.ljust(64, "0")[:64],
+            status=status,
+            uploaded_by=self.user,
+            completed_at=(
+                timezone.now()
+                if status
+                in {
+                    ReportUploadStatus.SUCCESS,
+                    ReportUploadStatus.PARTIAL,
+                    ReportUploadStatus.FAILED,
+                }
+                else None
+            ),
+        )
+
+    def test_disables_schedules_and_closes_only_in_flight_records(self) -> None:
+        result = json.loads(
+            call_command(
+                "sanitize_cloned_environment",
+                stdout=StringIO(),
+            )
+        )
+
+        self.assertEqual(
+            result,
+            {
+                "disabled_periodic_tasks": 1,
+                "failed_report_uploads": 2,
+                "revoked_background_jobs": 3,
+            },
+        )
+        self.enabled_schedule.refresh_from_db()
+        self.disabled_schedule.refresh_from_db()
+        self.assertFalse(self.enabled_schedule.enabled)
+        self.assertFalse(self.disabled_schedule.enabled)
+
+        for job in self.active_jobs:
+            job.refresh_from_db()
+            self.assertEqual(job.status, JobStatus.REVOKED)
+            self.assertIsNotNone(job.completed_at)
+
+        for upload in self.active_uploads:
+            upload.refresh_from_db()
+            self.assertEqual(upload.status, ReportUploadStatus.FAILED)
+            self.assertIsNotNone(upload.completed_at)
+
+        self.finished_job.refresh_from_db()
+        self.assertEqual(self.finished_job.status, JobStatus.SUCCESS)
+        self.assertEqual(self.finished_job.result, {"records": 42})
+        self.finished_upload.refresh_from_db()
+        self.assertEqual(self.finished_upload.status, ReportUploadStatus.SUCCESS)
+        self.assertEqual(self.finished_upload.load_batch, 42)
+        self.assertEqual(self.finished_upload.uploaded_by, self.user)
+        self.assertTrue(type(self.user).objects.filter(pk=self.user.pk).exists())
+
+    def test_is_idempotent(self) -> None:
+        call_command("sanitize_cloned_environment", stdout=StringIO())
+        first_completed_at = {
+            job.pk: BackgroundJob.objects.get(pk=job.pk).completed_at
+            for job in self.active_jobs
+        }
+        first_upload_completed_at = {
+            upload.pk: ReportUpload.objects.get(pk=upload.pk).completed_at
+            for upload in self.active_uploads
+        }
+
+        result = json.loads(
+            call_command(
+                "sanitize_cloned_environment",
+                stdout=StringIO(),
+            )
+        )
+
+        self.assertEqual(
+            result,
+            {
+                "disabled_periodic_tasks": 0,
+                "failed_report_uploads": 0,
+                "revoked_background_jobs": 0,
+            },
+        )
+        self.assertEqual(
+            first_completed_at,
+            {
+                job.pk: BackgroundJob.objects.get(pk=job.pk).completed_at
+                for job in self.active_jobs
+            },
+        )
+        self.assertEqual(
+            first_upload_completed_at,
+            {
+                upload.pk: ReportUpload.objects.get(pk=upload.pk).completed_at
+                for upload in self.active_uploads
+            },
+        )
+
+    def test_dry_run_rolls_back_all_changes(self) -> None:
+        result = json.loads(
+            call_command(
+                "sanitize_cloned_environment",
+                dry_run=True,
+                stdout=StringIO(),
+            )
+        )
+
+        self.assertEqual(result["disabled_periodic_tasks"], 1)
+        self.enabled_schedule.refresh_from_db()
+        self.assertTrue(self.enabled_schedule.enabled)
+        self.assertEqual(
+            set(
+                BackgroundJob.objects.filter(
+                    pk__in=[job.pk for job in self.active_jobs]
+                ).values_list("status", flat=True)
+            ),
+            {JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY},
+        )
+        self.assertEqual(
+            set(
+                ReportUpload.objects.filter(
+                    pk__in=[upload.pk for upload in self.active_uploads]
+                ).values_list("status", flat=True)
+            ),
+            {ReportUploadStatus.QUEUED, ReportUploadStatus.PROCESSING},
+        )

tests/apps/core/test_validate_cloned_migrations_command.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import json
+from io import StringIO
+
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.db import connection
+from django.db.migrations.recorder import MigrationRecorder
+from django.test import TestCase
+
+
+class ValidateClonedMigrationsCommandTest(TestCase):
+    def test_accepts_database_when_all_applied_migrations_exist_in_image(self) -> None:
+        result = json.loads(
+            call_command(
+                "validate_cloned_migrations",
+                stdout=StringIO(),
+            )
+        )
+
+        self.assertEqual(result["unknown_applied_migrations"], 0)
+        self.assertGreaterEqual(
+            result["disk_migrations"],
+            result["applied_migrations"],
+        )
+
+    def test_rejects_applied_migration_missing_from_image(self) -> None:
+        MigrationRecorder(connection).record_applied(
+            "removed_legacy_app",
+            "0001_removed",
+        )
+
+        with self.assertRaisesMessage(
+            CommandError,
+            "removed_legacy_app.0001_removed",
+        ):
+            call_command(
+                "validate_cloned_migrations",
+                stdout=StringIO(),
+            )