1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
from django.apps import apps
from django.db import DEFAULT_DB_ALIAS, connections
from django.db.migrations import Migration
from django.db.migrations.graph import Node
from django.db.migrations.loader import MigrationLoader
from django_test_migrations.exceptions import MigrationNotInPlan
from django_test_migrations.types import MigrationPlan, MigrationTarget
def all_migrations(
database: str = DEFAULT_DB_ALIAS,
app_names: list[str] | None = None,
) -> list[Node]:
"""
Returns the sorted list of migrations nodes.
The order is the same as when migrations are applied.
When you might need this function?
When you are testing the migration order.
For example, imagine that you have a direct dependency:
``main_app.0002_migration`` and ``other_app.0001_initial``
where ``other_app.0001_initial`` relies on the model or field
introduced in ``main_app.0002_migration``.
You can use ``dependencies`` field
to ensure that everything works correctly.
But, sometimes migrations are squashed,
sometimes they are renamed, refactored, and moved.
It would be better to have a test that will ensure
that ``other_app.0001_initial`` comes after ``main_app.0002_migration``.
And everything works as expected.
"""
loader = MigrationLoader(connections[database])
if app_names:
_validate_app_names(app_names)
targets = [
key for key in loader.graph.leaf_nodes() if key[0] in app_names
]
else:
targets = loader.graph.leaf_nodes()
return _generate_plan(targets, loader)
def nodes_to_tuples(nodes: list[Node]) -> list[tuple[str, str]]:
"""Utility function to transform nodes to tuples."""
return [(node[0], node[1]) for node in nodes]
def _validate_app_names(app_names: list[str]) -> None:
"""
Validates the provided app names.
Raises ```LookupError`` when incorrect app names are provided.
"""
for app_name in app_names:
apps.get_app_config(app_name)
def _generate_plan(
targets: list[Node],
loader: MigrationLoader,
) -> list[Node]:
plan = []
seen: set[Node] = set()
# Generate the plan
for target in targets:
for migration in loader.graph.forwards_plan(target):
if migration not in seen:
node = loader.graph.node_map[migration]
plan.append(node)
seen.add(migration)
return plan
def truncate_plan(
targets: list[MigrationTarget],
plan: MigrationPlan,
) -> MigrationPlan:
"""Truncate migrations ``plan`` up to ``targets``.
This method is used mainly to truncate full/clean migrations plan
to get as broad plan as possible.
By "broad" plan we mean plan with all targets migrations included
as well as all older migrations not related with targets.
"Broad" plan is needed mostly to make ``Migrator`` API developers'
friendly, just to not force developers to include migrations targets
for all other models they want to use in test (e.g. to setup some
model instances) in ``migrate_from``.
Such plan will also produce database state really similar to state
from our production environment just before new migrations are applied.
Migrations plan for targets generated by Django's
``MigrationExecutor.migration_plan`` is minimum plan needed to apply
targets migrations, it includes only migrations targets with all its
dependencies, so it doesn't fit to our approach, that's why following
function is needed.
"""
if not targets or not plan:
return plan
target_max_index = max(_get_index(target, plan) for target in targets)
return plan[: (target_max_index + 1)]
def _get_index(target: MigrationTarget, plan: MigrationPlan) -> int:
try:
index = next(
index
for index, (migration, _) in enumerate(plan)
if _filter_predicate(target, migration)
)
except StopIteration:
raise MigrationNotInPlan(target) from None
return index - (target[1] is None)
def _filter_predicate(target: MigrationTarget, migration: Migration) -> bool:
# when ``None`` passed as migration name then initial migration from
# target's app will be chosen and handled properly in ``_get_index``
# so in final all target app migrations will be excluded from plan
index = 2 - (target[1] is None)
return (migration.app_label, migration.name)[:index] == target[:index]
|
