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
|
"""Tests multi-database support"""
import contextlib
import ddf
import pytest
from django.contrib.auth.models import User
from django.core.management import call_command
from django.db import transaction
import pgtrigger
from pgtrigger import core
from pgtrigger.tests import models, utils
class ToLogRouter:
"""
Route the "ToLog" model to the "other" database
"""
route_app_labels = {"auth", "contenttypes"}
def db_for_write(self, model, **hints):
if model == models.ToLogModel:
return "other"
return None
@pytest.fixture(autouse=True)
def routed_db(settings):
settings.DATABASE_ROUTERS = [
"pgtrigger.tests.test_multi_db.ToLogRouter",
"pgtrigger.tests.models.Router",
]
@pytest.mark.django_db(databases=["default", "sqlite", "other"], transaction=True)
def test_multi_db_ignore():
"""Tests ignoring triggers across multiple databases"""
trigger = pgtrigger.Protect(operation=pgtrigger.Delete, name="protect_deletes")
with contextlib.ExitStack() as stack:
stack.enter_context(trigger.register(models.ToLogModel))
stack.enter_context(trigger.register(User))
stack.enter_context(trigger.install(models.ToLogModel, database="other"))
stack.enter_context(trigger.install(User))
with utils.raises_trigger_error(match="Cannot delete", database="other"):
log = ddf.G(models.ToLogModel)
log.delete()
with utils.raises_trigger_error(match="Cannot delete"):
user = ddf.G(User)
user.delete()
with transaction.atomic():
with pgtrigger.ignore("tests.ToLogModel:protect_deletes", "auth.User:protect_deletes"):
log = models.ToLogModel.objects.create()
log.delete()
user = ddf.G(User)
user.delete()
with utils.raises_trigger_error(match="Cannot delete"):
user = User.objects.create(username="hi")
user.delete()
with utils.raises_trigger_error(match="Cannot delete", database="other"):
log = models.ToLogModel.objects.create()
log.delete()
@pytest.mark.django_db(databases=["default", "sqlite", "other"])
def test_full_ls(capsys):
call_command("pgtrigger", "ls")
captured = capsys.readouterr()
lines = [line for line in captured.out.split("\n") if line]
for line in lines:
assert "\x1b[92mINSTALLED\x1b[0m" in line
call_command("pgtrigger", "ls", "-d", "other")
captured = capsys.readouterr()
lines = [line for line in captured.out.split("\n") if line]
for line in lines:
# The router ignores partition models for the default DB
if "tests.PartitionModel:protect_delete" in line:
assert "\x1b[94mUNALLOWED\x1b[0m" in line
else:
assert "\x1b[92mINSTALLED\x1b[0m" in line
call_command("pgtrigger", "ls", "-d", "sqlite")
captured = capsys.readouterr()
lines = [line for line in captured.out.split("\n") if line]
for line in lines:
assert "\x1b[94mUNALLOWED\x1b[0m" in line
@pytest.mark.django_db(databases=["other"])
def test_disable_enable(capsys):
call_command("pgtrigger", "disable", "-d", "other")
for model, trigger in pgtrigger.registered():
expected_status = None if model == models.PartitionModel else False
assert trigger.get_installation_status(model, database="other")[1] is expected_status
call_command("pgtrigger", "enable", "--database", "other")
for model, trigger in pgtrigger.registered():
expected_status = None if model == models.PartitionModel else True
assert trigger.get_installation_status(model, database="other")[1] is expected_status
@pytest.mark.django_db(databases=["sqlite"])
def test_ignore_non_postgres_dbs():
call_command("pgtrigger", "uninstall", "-d", "sqlite")
call_command("pgtrigger", "install", "-d", "sqlite")
call_command("pgtrigger", "install", "-d", "sqlite")
call_command("pgtrigger", "prune", "-d", "sqlite")
@pytest.mark.django_db(databases=["other", "default", "sqlite"])
def test_uninstall_install():
for model, trigger in pgtrigger.registered():
expected_status = core.UNALLOWED if model == models.PartitionModel else core.INSTALLED
assert trigger.get_installation_status(model, database="other")[0] == expected_status
call_command("pgtrigger", "uninstall", "-d", "other")
call_command("pgtrigger", "uninstall", "-d", "default")
for model, trigger in pgtrigger.registered():
expected_status = core.UNALLOWED if model == models.PartitionModel else core.UNINSTALLED
assert trigger.get_installation_status(model, database="other")[0] == expected_status
call_command("pgtrigger", "install", "--database", "other")
for model, trigger in pgtrigger.registered():
expecetd_status = core.UNALLOWED if model == models.PartitionModel else core.INSTALLED
assert trigger.get_installation_status(model, database="other")[0] == expecetd_status
for model, trigger in pgtrigger.registered():
assert trigger.get_installation_status(model, database="default")[0] == core.UNINSTALLED
|
