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
162
163
164
165
166
167
168
169
170
171
172
173
174
|
"""Tests multi-database support"""
# flake8: noqa
import contextlib
import ddf
from django.core.management import call_command
import pytest
import pgtrigger
from pgtrigger.tests import models, utils
class SchemaRouter:
"""
A router to control tables that should be migrated to different schemas
"""
def db_for_read(self, model, **hints):
if model == models.OrderSchema:
return "order"
elif model == models.ReceiptSchema: # pragma: no branch
return "receipt"
def db_for_write(self, model, **hints):
return self.db_for_read(model, **hints)
def allow_migrate(self, db, app_label, model_name=None, **hints):
if model_name == "orderschema":
return db == "order"
elif model_name == "receiptschema":
return db == "receipt"
@pytest.fixture(autouse=True)
def routed_db(settings):
settings.DATABASE_ROUTERS = [
"pgtrigger.tests.test_multi_schema.SchemaRouter",
"pgtrigger.tests.models.Router",
]
@pytest.fixture(autouse=True)
def schema_triggers():
protect_deletes = pgtrigger.Protect(name="protect_deletes", operation=pgtrigger.Delete)
protect_updates = pgtrigger.Protect(name="protect_updates", operation=pgtrigger.Update)
with contextlib.ExitStack() as stack:
stack.enter_context(protect_deletes.register(models.OrderSchema))
stack.enter_context(protect_updates.register(models.ReceiptSchema))
yield
@pytest.mark.django_db(databases=["order", "receipt"], transaction=True)
def test_multi_schema_triggers_work():
"""Verify the triggers in the schema_triggers fixture work"""
call_command("pgtrigger", "install", "-d", "order")
call_command("pgtrigger", "install", "-d", "receipt")
order = ddf.G("tests.OrderSchema")
receipt = ddf.G("tests.ReceiptSchema")
with utils.raises_trigger_error(match="Cannot delete", database="order"):
order.delete()
with utils.raises_trigger_error(match="Cannot update", database="receipt"):
receipt.char_field = "hello"
receipt.save()
receipt.delete()
order = ddf.G("tests.OrderSchema")
with pgtrigger.ignore("tests.OrderSchema:protect_deletes"):
order.delete()
@pytest.mark.django_db(databases=["order", "receipt", "default", "other"], transaction=True)
def test_commands(capsys):
"""Verify commands work"""
call_command("pgtrigger", "ls")
lines = capsys.readouterr().out.split("\n")
expected_lines = [
"",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.CustomSoftDelete:soft_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.CustomTableName:protect_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.FSM:fsm",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.SearchModel:add_body_title_to_vector",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.SearchModel:add_body_to_vector",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.SoftDelete:soft_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.TestDefaultThrough:protect_it",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.TestTrigger:protect_misc_insert",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.TestTriggerProxy:protect_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.ToLogModel:after_update_row_test",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.ToLogModel:after_update_statement_test",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.ToLogModel:update_of_statement_test",
"\x1b[94mUNALLOWED\x1b[0m \x1b[94mN/A\x1b[0m tests.OrderSchema:protect_deletes",
"\x1b[94mUNALLOWED\x1b[0m \x1b[94mN/A\x1b[0m tests.ReceiptSchema:protect_updates",
]
assert set(expected_lines).issubset(set(lines))
call_command("pgtrigger", "ls", "-d", "receipt")
lines = capsys.readouterr().out.split("\n")
expected_lines = [
"",
"\x1b[91mUNINSTALLED\x1b[0m \x1b[94mN/A\x1b[0m tests.ReceiptSchema:protect_updates",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.CustomSoftDelete:soft_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.CustomTableName:protect_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.FSM:fsm",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.SearchModel:add_body_title_to_vector",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.SearchModel:add_body_to_vector",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.SoftDelete:soft_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.TestDefaultThrough:protect_it",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.TestTrigger:protect_misc_insert",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.TestTriggerProxy:protect_delete",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.ToLogModel:after_update_row_test",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.ToLogModel:after_update_statement_test",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.ToLogModel:update_of_statement_test",
"\x1b[94mUNALLOWED\x1b[0m \x1b[94mN/A\x1b[0m tests.OrderSchema:protect_deletes",
]
assert set(expected_lines).issubset(set(lines))
call_command("pgtrigger", "install", "-d", "receipt")
call_command("pgtrigger", "ls", "-d", "receipt")
lines = capsys.readouterr().out.split("\n")
expected_lines = [
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.ReceiptSchema:protect_updates",
]
assert set(expected_lines).issubset(set(lines))
# Installed a trigger to be pruned.
protect_inserts = pgtrigger.Protect(name="protect_inserts", operation=pgtrigger.Insert)
protect_inserts.install(models.OrderSchema, database="order")
call_command("pgtrigger", "ls", "-d", "order")
lines = capsys.readouterr().out.split("\n")
expected_lines = [
"\x1b[91mUNINSTALLED\x1b[0m \x1b[94mN/A\x1b[0m tests.OrderSchema:protect_deletes",
"\x1b[94mUNALLOWED\x1b[0m \x1b[94mN/A\x1b[0m tests.ReceiptSchema:protect_updates",
"\x1b[96mPRUNE\x1b[0m \x1b[92mENABLED\x1b[0m tests_orderschema:pgtrigger_protect_inserts_a0767",
]
assert set(expected_lines).issubset(set(lines))
call_command("pgtrigger", "prune", "-d", "order")
call_command("pgtrigger", "install", "-d", "order")
call_command("pgtrigger", "ls", "-d", "order")
lines = capsys.readouterr().out.split("\n")
for line in lines:
assert "PRUNE" not in line
expected_lines = [
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.OrderSchema:protect_deletes",
"\x1b[94mUNALLOWED\x1b[0m \x1b[94mN/A\x1b[0m tests.ReceiptSchema:protect_updates",
]
assert set(expected_lines).issubset(set(lines))
# Set the search path to a schema and check results
call_command("pgtrigger", "uninstall", "-s", "receipt")
call_command("pgtrigger", "ls", "tests.CustomSoftDelete:soft_delete", "-s", "receipt")
lines = capsys.readouterr().out.split("\n")
expected_lines = [
"",
"\x1b[91mUNINSTALLED\x1b[0m \x1b[94mN/A\x1b[0m tests.CustomSoftDelete:soft_delete",
]
assert set(expected_lines) == set(lines)
call_command("pgtrigger", "install", "tests.CustomSoftDelete:soft_delete", "-s", "receipt")
call_command("pgtrigger", "ls", "tests.CustomSoftDelete:soft_delete", "-s", "receipt")
lines = capsys.readouterr().out.split("\n")
expected_lines = [
"",
"\x1b[92mINSTALLED\x1b[0m \x1b[92mENABLED\x1b[0m tests.CustomSoftDelete:soft_delete",
]
assert set(expected_lines) == set(lines)
|
