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
|
import uuid
import pytest
from django.apps import apps
from django.db import models
from psqlextra.backend.migrations.state import (
PostgresPartitionedModelState,
PostgresPartitionState,
)
from psqlextra.manager import PostgresManager
from psqlextra.models import PostgresPartitionedModel
from psqlextra.types import PostgresPartitioningMethod
from .fake_model import define_fake_partitioned_model
@pytest.fixture
def model():
fields = {"name": models.TextField(), "category": models.TextField()}
partitioning_options = {
"method": PostgresPartitioningMethod.LIST,
"key": ["category"],
}
model = define_fake_partitioned_model(fields, partitioning_options)
return model
def test_partitioned_model_state_copies():
"""Tests whether cloning the model state properly copies all the options.
If it does not copy them, bad things can happen as the state is
mutated to build up migration state.
"""
options = dict(method=PostgresPartitioningMethod.RANGE, key=["timestamp"])
state = PostgresPartitionedModelState(
app_label="tests",
name=str(uuid.uuid4()),
fields=[],
options=None,
partitioning_options=options,
bases=(PostgresPartitionedModel,),
)
assert options is not state.partitioning_options
def test_partitioned_model_state_from_model(model):
"""Tests whether creating state from an existing model works as
expected."""
state = PostgresPartitionedModelState.from_model(model)
assert state.partitions == {}
assert (
state.partitioning_options["method"] == model._partitioning_meta.method
)
assert state.partitioning_options["key"] == model._partitioning_meta.key
def test_partitioned_model_clone(model):
"""Tests whether cloning the state actually clones the partitioning
options.
If its not a copy, but a reference instead, bad things can happen as
the options are mutated to build up migration state.
"""
state = PostgresPartitionedModelState.from_model(model)
state.partitions = {
"pt1": PostgresPartitionState(
app_label="tests", model_name="tests", name="pt1"
)
}
state_copy = state.clone()
assert state.partitions is not state_copy.partitions
assert state.partitioning_options is not state_copy.partitioning_options
def test_partitioned_model_render(model):
"""Tests whether the state can be rendered into a valid model class."""
options = dict(method=PostgresPartitioningMethod.RANGE, key=["timestamp"])
state = PostgresPartitionedModelState(
app_label="tests",
name=str(uuid.uuid4()),
fields=[("name", models.TextField())],
options=None,
partitioning_options=options,
bases=(PostgresPartitionedModel,),
managers=[("cookie", PostgresManager())],
)
rendered_model = state.render(apps)
assert issubclass(rendered_model, PostgresPartitionedModel)
assert rendered_model.name
assert isinstance(rendered_model.objects, PostgresManager)
assert isinstance(rendered_model.cookie, PostgresManager)
assert rendered_model.__name__ == state.name
assert rendered_model._meta.apps == apps
assert rendered_model._meta.app_label == "tests"
assert rendered_model._partitioning_meta.method == options["method"]
assert rendered_model._partitioning_meta.key == options["key"]
|
