seed script

tests/test_production_seed.py

@@ -0,0 +1,72 @@
+import gzip
+import json
+
+import pytest
+from django.contrib.auth.models import Permission
+from django.core.management import call_command
+from django.db import connection
+
+from apps.accounts.models import User
+from apps.core.models import ProductionSeedState
+from apps.notifications.models import NotificationJob
+from apps.staff.models import OrganizationalUnit, StaffProfile
+
+
+@pytest.mark.django_db(transaction=True)
+def test_seed_bundle_round_trip_is_complete_idempotent_and_excludes_outbox(tmp_path, settings):
+    media_root = tmp_path / "source-media"
+    settings.MEDIA_ROOT = media_root
+    (media_root / "branding" / "logo").mkdir(parents=True)
+    (media_root / "branding" / "logo" / "current.png").write_bytes(b"seed-media")
+
+    unit = OrganizationalUnit.objects.create(name="واحد Seed", slug="seed-unit")
+    user = User.objects.create_user(
+        username="seed-staff",
+        password="not-a-production-password",
+        account_type=User.AccountType.STAFF,
+    )
+    StaffProfile.objects.create(user=user, primary_unit=unit, display_name="کارشناس Seed")
+    permission = Permission.objects.order_by("pk").first()
+    permission_key = (permission.content_type.app_label, permission.codename)
+    unit.permissions.add(permission)
+    membership = user.unit_memberships.get(unit=unit)
+    membership.permissions.add(permission)
+    user.user_permissions.add(permission)
+    user.groups.get().permissions.add(permission)
+    NotificationJob.objects.create(
+        channel="BALE",
+        event_type="SHOULD_NOT_IMPORT",
+        dedupe_key="excluded-outbox-job",
+        next_attempt_at="2026-09-10T00:00:00Z",
+    )
+
+    bundle_dir = tmp_path / "bundle"
+    call_command("export_production_seed", output=str(bundle_dir))
+    with gzip.open(bundle_dir / "data.json.gz", "rt", encoding="utf-8") as stream:
+        serialized = json.load(stream)
+    assert any(row["model"] == "accounts.user" for row in serialized)
+    assert not any(row["model"] == "notifications.notificationjob" for row in serialized)
+
+    call_command("flush", interactive=False, verbosity=0)
+    target_media = tmp_path / "target-media"
+    settings.MEDIA_ROOT = target_media
+    original_vendor = connection.vendor
+    connection.vendor = "postgresql"
+    try:
+        call_command("import_production_seed", str(bundle_dir))
+        call_command("import_production_seed", str(bundle_dir))
+    finally:
+        connection.vendor = original_vendor
+
+    restored = User.objects.get(username="seed-staff")
+    assert restored.check_password("not-a-production-password")
+    assert restored.staff_profile.primary_unit.slug == "seed-unit"
+    restored_membership = restored.unit_memberships.get(unit__slug="seed-unit")
+    permission_filter = {"content_type__app_label": permission_key[0], "codename": permission_key[1]}
+    assert restored.user_permissions.filter(**permission_filter).exists()
+    assert restored.groups.get().permissions.filter(**permission_filter).exists()
+    assert restored_membership.permissions.filter(**permission_filter).exists()
+    assert restored_membership.unit.permissions.filter(**permission_filter).exists()
+    assert ProductionSeedState.objects.count() == 1
+    assert NotificationJob.objects.count() == 0
+    assert (target_media / "branding" / "logo" / "current.png").read_bytes() == b"seed-media"