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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
import argparse
from unittest.mock import MagicMock, create_autospec, patch
import pytest
from django.db import models
from django.test import override_settings
from psqlextra.backend.introspection import (
PostgresIntrospectedPartitionTable,
PostgresIntrospectedPartitonedTable,
)
from psqlextra.management.commands.pgpartition import Command
from psqlextra.partitioning import PostgresPartitioningManager
from psqlextra.partitioning.config import PostgresPartitioningConfig
from psqlextra.partitioning.partition import PostgresPartition
from psqlextra.partitioning.strategy import PostgresPartitioningStrategy
from .fake_model import define_fake_partitioned_model
@pytest.fixture
def fake_strategy():
strategy = create_autospec(PostgresPartitioningStrategy)
strategy.createable_partition = create_autospec(PostgresPartition)
strategy.createable_partition.name = MagicMock(return_value="tobecreated")
strategy.to_create = MagicMock(return_value=[strategy.createable_partition])
strategy.deleteable_partition = create_autospec(PostgresPartition)
strategy.deleteable_partition.name = MagicMock(return_value="tobedeleted")
strategy.to_delete = MagicMock(return_value=[strategy.deleteable_partition])
return strategy
@pytest.fixture
def fake_model(fake_strategy):
model = define_fake_partitioned_model(
{"timestamp": models.DateTimeField()}, {"key": ["timestamp"]}
)
# consistent model name so snapshot tests work
model.__name__ = "test"
# we have to trick the system into thinking the model/table
# actually exists with one partition (so we can simulate deletions)
deleteable_partition_name = fake_strategy.deleteable_partition.name()
mocked_partitioned_table = PostgresIntrospectedPartitonedTable(
name=model._meta.db_table,
method=model._partitioning_meta.method,
key=model._partitioning_meta.key,
partitions=[
PostgresIntrospectedPartitionTable(
name=deleteable_partition_name,
full_name=f"{model._meta.db_table}_{deleteable_partition_name}",
comment="psqlextra_auto_partitioned",
)
],
)
introspection_package = "psqlextra.backend.introspection"
introspection_class = f"{introspection_package}.PostgresIntrospection"
get_partitioned_table_path = f"{introspection_class}.get_partitioned_table"
with patch(get_partitioned_table_path) as mock:
mock.return_value = mocked_partitioned_table
yield model
@pytest.fixture
def fake_partitioning_manager(fake_model, fake_strategy):
manager = PostgresPartitioningManager(
[PostgresPartitioningConfig(fake_model, fake_strategy)]
)
with override_settings(PSQLEXTRA_PARTITIONING_MANAGER=manager):
yield manager
@pytest.fixture
def run(capsys):
def _run(*args):
parser = argparse.ArgumentParser()
command = Command()
command.add_arguments(parser)
command.handle(**vars(parser.parse_args(args)))
return capsys.readouterr()
return _run
@pytest.mark.parametrize("args", ["-d", "--dry"])
def test_management_command_partition_dry_run(
args, snapshot, run, fake_model, fake_partitioning_manager
):
"""Tests whether the --dry option actually makes it a dry run and does not
create/delete partitions."""
config = fake_partitioning_manager.find_config_for_model(fake_model)
snapshot.assert_match(run(args))
config.strategy.createable_partition.create.assert_not_called()
config.strategy.createable_partition.delete.assert_not_called()
config.strategy.deleteable_partition.create.assert_not_called()
config.strategy.deleteable_partition.delete.assert_not_called()
@pytest.mark.parametrize("args", ["-y", "--yes"])
def test_management_command_partition_auto_confirm(
args, snapshot, run, fake_model, fake_partitioning_manager
):
"""Tests whether the --yes option makes it not ask for confirmation before
creating/deleting partitions."""
config = fake_partitioning_manager.find_config_for_model(fake_model)
snapshot.assert_match(run(args))
config.strategy.createable_partition.create.assert_called_once()
config.strategy.createable_partition.delete.assert_not_called()
config.strategy.deleteable_partition.create.assert_not_called()
config.strategy.deleteable_partition.delete.assert_called_once()
@pytest.mark.parametrize("answer", ["y", "Y", "yes", "YES"])
def test_management_command_partition_confirm_yes(
answer, monkeypatch, snapshot, run, fake_model, fake_partitioning_manager
):
"""Tests whether the --yes option makes it not ask for confirmation before
creating/deleting partitions."""
config = fake_partitioning_manager.find_config_for_model(fake_model)
monkeypatch.setattr("builtins.input", lambda _: answer)
snapshot.assert_match(run())
config.strategy.createable_partition.create.assert_called_once()
config.strategy.createable_partition.delete.assert_not_called()
config.strategy.deleteable_partition.create.assert_not_called()
config.strategy.deleteable_partition.delete.assert_called_once()
@pytest.mark.parametrize("answer", ["n", "N", "no", "No", "NO"])
def test_management_command_partition_confirm_no(
answer, monkeypatch, snapshot, run, fake_model, fake_partitioning_manager
):
"""Tests whether the --yes option makes it not ask for confirmation before
creating/deleting partitions."""
config = fake_partitioning_manager.find_config_for_model(fake_model)
monkeypatch.setattr("builtins.input", lambda _: answer)
snapshot.assert_match(run())
config.strategy.createable_partition.create.assert_not_called()
config.strategy.createable_partition.delete.assert_not_called()
config.strategy.deleteable_partition.create.assert_not_called()
config.strategy.deleteable_partition.delete.assert_not_called()
|
