63 lines
2.8 KiB
Python
63 lines
2.8 KiB
Python
|
"""Exercise the real forward/reverse migration, including a refused legacy DB."""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from django.db import IntegrityError, connection, transaction
|
||
|
|
from django.db.migrations.executor import MigrationExecutor
|
||
|
|
|
||
|
|
from coulomb_social.apps.spaces.models import Space, SpaceMembership
|
||
|
|
from tests.test_space_permissions import make_member
|
||
|
|
|
||
|
|
OLD = ("spaces", "0002_space_visual")
|
||
|
|
NEW = ("spaces", "0003_space_slug_global_uniq")
|
||
|
|
|
||
|
|
|
||
|
|
def migrate(target):
|
||
|
|
executor = MigrationExecutor(connection)
|
||
|
|
executor.migrate([target])
|
||
|
|
return executor
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.django_db(transaction=True)
|
||
|
|
def test_forward_refusal_rollback_and_reapply_preserve_rows():
|
||
|
|
owner = make_member("migration-owner")
|
||
|
|
space = Space.objects.create(
|
||
|
|
tenant_id=owner.tenant_id, slug="retained", title="Retained", created_by=owner,
|
||
|
|
)
|
||
|
|
SpaceMembership.objects.create(space=space, member=owner, role="owner")
|
||
|
|
before_spaces = list(Space.objects.order_by("pk").values())
|
||
|
|
before_grants = list(SpaceMembership.objects.order_by("pk").values())
|
||
|
|
duplicate = None
|
||
|
|
try:
|
||
|
|
migrate(OLD)
|
||
|
|
assert list(Space.objects.order_by("pk").values()) == before_spaces
|
||
|
|
assert list(SpaceMembership.objects.order_by("pk").values()) == before_grants
|
||
|
|
# The tenant constraint must survive rollback of the global constraint.
|
||
|
|
with pytest.raises(IntegrityError), transaction.atomic():
|
||
|
|
Space.objects.create(tenant_id=space.tenant_id, slug=space.slug, title="Denied")
|
||
|
|
duplicate = Space.objects.create(
|
||
|
|
tenant_id="tenant:legacy", slug=space.slug, title="Legacy collision", is_active=False,
|
||
|
|
)
|
||
|
|
collided = list(Space.objects.order_by("pk").values())
|
||
|
|
with pytest.raises(RuntimeError, match="does not rename, merge, or delete"):
|
||
|
|
migrate(NEW)
|
||
|
|
assert list(Space.objects.order_by("pk").values()) == collided
|
||
|
|
assert list(SpaceMembership.objects.order_by("pk").values()) == before_grants
|
||
|
|
assert NEW not in MigrationExecutor(connection).loader.applied_migrations
|
||
|
|
# Reconciliation here removes only the test's extra synthetic row.
|
||
|
|
duplicate.delete()
|
||
|
|
duplicate = None
|
||
|
|
migrate(NEW)
|
||
|
|
with pytest.raises(IntegrityError), transaction.atomic():
|
||
|
|
Space.objects.create(tenant_id="tenant:other", slug=space.slug, title="Denied")
|
||
|
|
assert list(Space.objects.order_by("pk").values()) == before_spaces
|
||
|
|
assert list(SpaceMembership.objects.order_by("pk").values()) == before_grants
|
||
|
|
assert NEW in MigrationExecutor(connection).loader.applied_migrations
|
||
|
|
migrate(OLD)
|
||
|
|
migrate(NEW)
|
||
|
|
assert list(Space.objects.order_by("pk").values()) == before_spaces
|
||
|
|
assert list(SpaceMembership.objects.order_by("pk").values()) == before_grants
|
||
|
|
finally:
|
||
|
|
if duplicate is not None:
|
||
|
|
duplicate.delete()
|
||
|
|
migrate(NEW)
|