18 |
18 |
|
19 |
19 |
import django_webtest
|
20 |
20 |
import pytest
|
|
21 |
from django.core.management import call_command
|
|
22 |
from django.db import connection
|
|
23 |
from django.db.migrations.executor import MigrationExecutor
|
21 |
24 |
|
22 |
25 |
|
23 |
26 |
@pytest.fixture(autouse=True)
|
... | ... | |
82 |
85 |
with metadata_path.open('w') as fd:
|
83 |
86 |
fd.write(metadata)
|
84 |
87 |
yield str(metadata_path)
|
|
88 |
|
|
89 |
|
|
90 |
@pytest.fixture()
|
|
91 |
def migration(request, transactional_db):
|
|
92 |
# see https://gist.github.com/asfaltboy/b3e6f9b5d95af8ba2cc46f2ba6eae5e2
|
|
93 |
"""
|
|
94 |
This fixture returns a helper object to test Django data migrations.
|
|
95 |
The fixture returns an object with two methods;
|
|
96 |
- `before` to initialize db to the state before the migration under test
|
|
97 |
- `after` to execute the migration and bring db to the state after the migration
|
|
98 |
The methods return `old_apps` and `new_apps` respectively; these can
|
|
99 |
be used to initiate the ORM models as in the migrations themselves.
|
|
100 |
For example:
|
|
101 |
def test_foo_set_to_bar(migration):
|
|
102 |
old_apps = migration.before([('my_app', '0001_inital')])
|
|
103 |
Foo = old_apps.get_model('my_app', 'foo')
|
|
104 |
Foo.objects.create(bar=False)
|
|
105 |
assert Foo.objects.count() == 1
|
|
106 |
assert Foo.objects.filter(bar=False).count() == Foo.objects.count()
|
|
107 |
# executing migration
|
|
108 |
new_apps = migration.apply([('my_app', '0002_set_foo_bar')])
|
|
109 |
Foo = new_apps.get_model('my_app', 'foo')
|
|
110 |
|
|
111 |
assert Foo.objects.filter(bar=False).count() == 0
|
|
112 |
assert Foo.objects.filter(bar=True).count() == Foo.objects.count()
|
|
113 |
Based on: https://gist.github.com/blueyed/4fb0a807104551f103e6
|
|
114 |
"""
|
|
115 |
|
|
116 |
class Migrator:
|
|
117 |
def before(self, targets, at_end=True):
|
|
118 |
"""Specify app and starting migration names as in:
|
|
119 |
before([('app', '0001_before')]) => app/migrations/0001_before.py
|
|
120 |
"""
|
|
121 |
executor = MigrationExecutor(connection)
|
|
122 |
executor.migrate(targets)
|
|
123 |
executor.loader.build_graph()
|
|
124 |
return executor._create_project_state(with_applied_migrations=True).apps
|
|
125 |
|
|
126 |
def apply(self, targets):
|
|
127 |
"""Migrate forwards to the "targets" migration"""
|
|
128 |
executor = MigrationExecutor(connection)
|
|
129 |
executor.migrate(targets)
|
|
130 |
executor.loader.build_graph()
|
|
131 |
return executor._create_project_state(with_applied_migrations=True).apps
|
|
132 |
|
|
133 |
yield Migrator()
|
|
134 |
|
|
135 |
call_command('migrate', verbosity=0)
|
85 |
|
-
|