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_misc.py

@@ -1,13 +1,9 @@
 import datetime
-from io import StringIO
 from unittest import mock
 from unittest.mock import patch
 
 import pytest
 from django.contrib.contenttypes.models import ContentType
-from django.core.files import File
-from django.db import connection
-from django.db.migrations.executor import MigrationExecutor
 from django.urls import reverse
 from django.utils import timezone
 from requests.exceptions import ReadTimeout
@@ -140,44 +136,6 @@ def test_trace_emails(app, settings, dummy_csv_datasource, email_handler, mailou
         assert mailoutbox[idx].to == ['john.doe@example.net']
 
 
-@pytest.mark.xfail
-def test_jsonb_migration(transactional_db):
-    app = 'csvdatasource'
-
-    migrate_from = [(app, '0017_auto_20200504_1402')]
-    migrate_to = [(app, '0018_text_to_jsonb')]
-    executor = MigrationExecutor(connection)
-    old_apps = executor.loader.project_state(migrate_from).apps
-    # state of the db is not important
-    executor.migrate(migrate_from, fake=True)
-
-    data = {'data': {'test': 1}}
-    CsvDataSource = old_apps.get_model(app, 'CsvDataSource')
-    connector = CsvDataSource.objects.create(csv_file=File(StringIO(''), 't.csv'), _dialect_options=data)
-    pk = connector.pk
-
-    field = CsvDataSource._meta.get_field('_dialect_options')
-    with connection.cursor() as cursor:
-        cursor.execute(
-            'ALTER TABLE {table} ALTER COLUMN {col} TYPE text USING {col}::text;'.format(
-                table=CsvDataSource._meta.db_table, col=field.get_attname_column()[1]
-            )
-        )
-    connector = CsvDataSource.objects.get(pk=pk)
-    # db is in a broken state
-    assert connector._dialect_options != data
-
-    # ensure migration fixes it
-    executor = MigrationExecutor(connection)
-    executor.migrate(migrate_to)
-    executor.loader.build_graph()
-
-    apps = executor.loader.project_state(migrate_to).apps
-    CsvDataSource = apps.get_model(app, 'CsvDataSource')
-    connector = CsvDataSource.objects.get(pk=pk)
-    assert connector._dialect_options == data
-
-
 def test_check_status_no_traceback_email(db, settings, email_handler, mailoutbox):
     settings.ADMINS = [('admin', 'admin@example.net')]
     connector = BaseAdresse.objects.create(slug='base-adresse')

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')