tests: use migration fixture from authentic (#73760)

tests/conftest.py

@@ -6,6 +6,9 @@ import pytest
 from django.contrib.auth.models import User
 from django.core.cache import cache
 from django.core.files import File
+from django.core.management import call_command
+from django.db import connection
+from django.db.migrations.executor import MigrationExecutor
 from httmock import HTTMock, remember_called, response, urlmatch
 
 from tests.utils import make_resource
@@ -224,3 +227,51 @@ def httpbin():
     with HTTMock(httpbin_cookies_set) as fixture:
         fixture.url = 'http://httpbin.org'
         yield fixture
+
+
+@pytest.fixture()
+def migration(request, transactional_db):
+    # see https://gist.github.com/asfaltboy/b3e6f9b5d95af8ba2cc46f2ba6eae5e2
+    """
+    This fixture returns a helper object to test Django data migrations.
+    The fixture returns an object with two methods;
+     - `before` to initialize db to the state before the migration under test
+     - `after` to execute the migration and bring db to the state after the migration
+    The methods return `old_apps` and `new_apps` respectively; these can
+    be used to initiate the ORM models as in the migrations themselves.
+    For example:
+        def test_foo_set_to_bar(migration):
+            old_apps = migration.before([('my_app', '0001_inital')])
+            Foo = old_apps.get_model('my_app', 'foo')
+            Foo.objects.create(bar=False)
+            assert Foo.objects.count() == 1
+            assert Foo.objects.filter(bar=False).count() == Foo.objects.count()
+            # executing migration
+            new_apps = migration.apply([('my_app', '0002_set_foo_bar')])
+            Foo = new_apps.get_model('my_app', 'foo')
+
+            assert Foo.objects.filter(bar=False).count() == 0
+            assert Foo.objects.filter(bar=True).count() == Foo.objects.count()
+    Based on: https://gist.github.com/blueyed/4fb0a807104551f103e6
+    """
+
+    class Migrator:
+        def before(self, targets, at_end=True):
+            """Specify app and starting migration names as in:
+            before([('app', '0001_before')]) => app/migrations/0001_before.py
+            """
+            executor = MigrationExecutor(connection)
+            executor.migrate(targets)
+            executor.loader.build_graph()
+            return executor._create_project_state(with_applied_migrations=True).apps
+
+        def apply(self, targets):
+            """Migrate forwards to the "targets" migration"""
+            executor = MigrationExecutor(connection)
+            executor.migrate(targets)
+            executor.loader.build_graph()
+            return executor._create_project_state(with_applied_migrations=True).apps
+
+    yield Migrator()
+
+    call_command('migrate', verbosity=0)

tests/test_rsa13.py

@@ -22,8 +22,6 @@ from urllib.parse import parse_qs
 
 import httmock
 import pytest
-from django.db import connection
-from django.db.migrations.executor import MigrationExecutor
 
 import tests.utils
 from passerelle.contrib.rsa13.models import DEFAULTS, RSA13Resource, dump_csv_columns
@@ -1380,27 +1378,16 @@ def test_sous_action(app, rsa13, url):
     }
 
 
-def test_csv_columns_migration(transactional_db, settings):
-    migrate_from = [('rsa13', '0001_initial')]
-    migrate_to = [('rsa13', '0002_add_csv_columns_fields')]
-
-    executor = MigrationExecutor(connection)
-    old_apps = executor.loader.project_state(migrate_from).apps
-
-    # state of the db is not important
-    executor.migrate(migrate_from)
+def test_csv_columns_migration(migration, settings):
+    old_apps = migration.before([('rsa13', '0001_initial')])
 
     RSA13Resource = old_apps.get_model('rsa13', 'RSA13Resource')
     resource = RSA13Resource.objects.create()
-
     settings.RSA13_CSV_COLUMNS = [('A', '2'), 'B']
     settings.RSA13_FACTURATION_CSV_COLUMNS = ['C', 'D']
     settings.RSA13_BENEFICIAIRE_SORTI_CSV_COLUMNS = ['E', 'F']
 
-    executor = MigrationExecutor(connection)
-    executor.migrate(migrate_to)
-    executor.loader.build_graph()
-    apps = executor.loader.project_state(migrate_to).apps
+    apps = migration.apply([('rsa13', '0002_add_csv_columns_fields')])
 
     RSA13Resource = apps.get_model('rsa13', 'RSA13Resource')
     resource = RSA13Resource.objects.get()

tests/test_toulouse_smart.py

@@ -28,8 +28,6 @@ from unittest import mock
 import httmock
 import lxml.etree as ET
 import pytest
-from django.db import connection
-from django.db.migrations.executor import MigrationExecutor
 from django.utils.encoding import force_str
 from requests.exceptions import ReadTimeout
 
@@ -1167,15 +1165,10 @@ def test_create_intervention_multiple_step(mocked_uuid4, app, smart):
     assert smart.wcs_requests.get(uuid=NEW_UUID, wcs_form_step='reclamation-1')
 
 
-def test_pk_change_migration(transactional_db):
-    migrate_from = [('toulouse_smart', '0005_auto_20220105_1514'), ('base', '0029_auto_20210202_1627')]
-    migrate_to = [('toulouse_smart', '0012_migrate_jobs')]
-
-    executor = MigrationExecutor(connection)
-    old_apps = executor.loader.project_state(migrate_from).apps
-
-    # state of the db is not important
-    executor.migrate(migrate_from)
+def test_pk_change_migration(migration):
+    old_apps = migration.before(
+        [('toulouse_smart', '0005_auto_20220105_1514'), ('base', '0029_auto_20210202_1627')]
+    )
 
     Job = old_apps.get_model('base', 'Job')
     ContentType = old_apps.get_model('contenttypes', 'ContentType')
@@ -1200,10 +1193,7 @@ def test_pk_change_migration(transactional_db):
         },
     )
 
-    executor = MigrationExecutor(connection)
-    executor.migrate(migrate_to)
-    executor.loader.build_graph()
-    apps = executor.loader.project_state(migrate_to).apps
+    apps = migration.apply([('toulouse_smart', '0012_migrate_jobs')])
 
     Job = apps.get_model('base', 'Job')
     ContentType = apps.get_model('contenttypes', 'ContentType')