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
|
import pytest
from django.db import models
from psqlextra.partitioning import (
PostgresPartitioningError,
PostgresPartitioningManager,
partition_by_current_time,
)
from .fake_model import define_fake_partitioned_model, get_fake_model
def test_partitioning_manager_duplicate_model():
"""Tests whether it is not possible to have more than one partitioning
config per model."""
model = define_fake_partitioned_model(
{"timestamp": models.DateTimeField()}, {"key": ["timestamp"]}
)
with pytest.raises(PostgresPartitioningError):
PostgresPartitioningManager(
[
partition_by_current_time(model, years=1, count=3),
partition_by_current_time(model, years=1, count=3),
]
)
def test_partitioning_manager_find_config_for_model():
"""Tests that finding a partitioning config by the model works as
expected."""
model1 = define_fake_partitioned_model(
{"timestamp": models.DateTimeField()}, {"key": ["timestamp"]}
)
config1 = partition_by_current_time(model1, years=1, count=3)
model2 = define_fake_partitioned_model(
{"timestamp": models.DateTimeField()}, {"key": ["timestamp"]}
)
config2 = partition_by_current_time(model2, months=1, count=2)
manager = PostgresPartitioningManager([config1, config2])
assert manager.find_config_for_model(model1) == config1
assert manager.find_config_for_model(model2) == config2
def test_partitioning_manager_plan_not_partitioned_model():
"""Tests that the auto partitioner does not try to auto partition for non-
partitioned models/tables."""
model = get_fake_model({"timestamp": models.DateTimeField()})
with pytest.raises(PostgresPartitioningError):
manager = PostgresPartitioningManager(
[partition_by_current_time(model, months=1, count=2)]
)
manager.plan()
def test_partitioning_manager_plan_non_existent_model():
"""Tests that the auto partitioner does not try to partition for non-
existent partitioned tables."""
model = define_fake_partitioned_model(
{"timestamp": models.DateTimeField()}, {"key": ["timestamp"]}
)
with pytest.raises(PostgresPartitioningError):
manager = PostgresPartitioningManager(
[partition_by_current_time(model, months=1, count=2)]
)
manager.plan()
|
