diff --git a/tests/conftest.py b/tests/conftest.py index 7a9291ee..a45e6e32 100644 --- a/tests/conftest.py +++ b/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) diff --git a/tests/test_rsa13.py b/tests/test_rsa13.py index f9bcf684..e42cee37 100644 --- a/tests/test_rsa13.py +++ b/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() diff --git a/tests/test_toulouse_smart.py b/tests/test_toulouse_smart.py index 1649cd85..eb2889ab 100644 --- a/tests/test_toulouse_smart.py +++ b/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')