1
|
|
|
import pytest |
2
|
|
|
from sqlalchemy import Column, String, Integer |
3
|
|
|
|
4
|
|
|
import groundwork |
5
|
|
|
from groundwork_validation.patterns import GwDbValidatorsPattern |
6
|
|
|
from groundwork_validation.patterns.gw_db_validators_pattern.gw_db_validators_pattern import ValidationError |
7
|
|
|
|
8
|
|
|
|
9
|
|
|
def test_db_validator_init(): |
10
|
|
|
""" |
11
|
|
|
.. test:: GbDbValidation initialisation and validation tests |
12
|
|
|
:tags: gwdbvalidator_pattern; |
13
|
|
|
:links: S_5917A |
14
|
|
|
|
15
|
|
|
Tests the initialisation of :ref:`gwdbvalidators` and performs some IO actions on a for validation |
16
|
|
|
activated database table. |
17
|
|
|
""" |
18
|
|
|
|
19
|
|
|
class My_Plugin(GwDbValidatorsPattern): |
20
|
|
|
def __init__(self, app, **kwargs): |
21
|
|
|
self.name = "My_Plugin" |
22
|
|
|
super(My_Plugin, self).__init__(app, **kwargs) |
23
|
|
|
self.db = None |
24
|
|
|
self.Test = None |
25
|
|
|
|
26
|
|
|
def activate(self): |
27
|
|
|
self.db = self.app.databases.register("test_db", |
28
|
|
|
"sqlite://", |
29
|
|
|
"database for test values") |
30
|
|
|
|
31
|
|
|
class Test(self.db.Base): |
32
|
|
|
__tablename__ = "test" |
33
|
|
|
id = Column(Integer, primary_key=True) |
34
|
|
|
name = Column(String(512), nullable=False, unique=True) |
35
|
|
|
|
36
|
|
|
self.Test = self.db.classes.register(Test) |
37
|
|
|
self.db.create_all() |
38
|
|
|
|
39
|
|
|
self.validators.db.register("db_test_validator", "my db test validator", self.Test) |
40
|
|
|
|
41
|
|
|
my_test = self.Test(name="blub") |
42
|
|
|
self.db.add(my_test) |
43
|
|
|
self.db.commit() |
44
|
|
|
self.db.query(self.Test).all() |
45
|
|
|
|
46
|
|
|
my_test.name = "Boohaaaa" |
47
|
|
|
self.db.add(my_test) |
48
|
|
|
self.db.commit() |
49
|
|
|
self.db.query(self.Test).all() |
50
|
|
|
|
51
|
|
|
# Execute sql-statement, which does not trigger the sqlalchemy events. So no hash gets updated. |
52
|
|
|
self.db.engine.execute("UPDATE test SET name='not_working' WHERE id=1") |
53
|
|
|
|
54
|
|
|
# Add another entry, so the query are executed on database and not on session |
55
|
|
|
# ( self.db.session.remove() does not do the trick, y?? ) |
56
|
|
|
my_test_2 = self.Test(name="blub") |
57
|
|
|
self.db.add(my_test_2) |
58
|
|
|
self.db.commit() |
59
|
|
|
|
60
|
|
|
with pytest.raises(ValidationError): |
61
|
|
|
self.db.query(self.Test).filter_by(id=1).first() |
62
|
|
|
|
63
|
|
|
def deactivate(self): |
64
|
|
|
pass |
65
|
|
|
|
66
|
|
|
app = groundwork.App() |
67
|
|
|
plugin = My_Plugin(app) |
68
|
|
|
plugin.activate() |
69
|
|
|
|