1 | # -*- coding: utf-8 -*- |
||
2 | import unittest |
||
3 | from oe_utils.audit import audit, audit_with_request |
||
4 | from oe_utils.data.data_managers import AuditManager |
||
5 | from oe_utils.data.models import Wijziging |
||
6 | from oe_utils.data.db_audit_manager import includeme |
||
7 | from oe_utils.data.db_audit_manager import audit_manager |
||
8 | |||
9 | try: |
||
10 | from unittest.mock import Mock, MagicMock |
||
11 | except ImportError: |
||
12 | from mock import Mock, MagicMock, call # pragma: no cover |
||
13 | try: |
||
14 | import configparser |
||
15 | except ImportError: |
||
16 | import ConfigParser as configparser |
||
17 | import os |
||
18 | from sqlalchemy import engine_from_config |
||
19 | from sqlalchemy.orm import sessionmaker |
||
20 | from zope.sqlalchemy import ZopeTransactionExtension |
||
21 | from oe_utils.data.models import Base |
||
22 | from jsonpublish import register_adapter |
||
23 | |||
24 | |||
25 | def init_test_db(engine): |
||
26 | Base.metadata.drop_all(engine) |
||
27 | Base.metadata.create_all(engine) |
||
28 | session_maker = sessionmaker( |
||
29 | bind=engine |
||
30 | ) |
||
31 | session = session_maker() |
||
32 | |||
33 | session.commit() |
||
34 | session.close() |
||
35 | |||
36 | |||
37 | class Resource: |
||
38 | def __init__(self): |
||
39 | self.id = None, |
||
40 | self.naam = None |
||
41 | self.voornaam = None |
||
42 | |||
43 | |||
44 | @register_adapter(Resource) |
||
45 | def resource_adapter(obj, request=None): |
||
46 | return { |
||
47 | 'id': obj.id, |
||
48 | 'naam': obj.naam, |
||
49 | 'voornaam': obj.voornaam |
||
50 | } |
||
51 | |||
52 | |||
53 | class DummyParent(object): |
||
54 | |||
55 | def __init__(self): |
||
56 | self.request = MagicMock() |
||
57 | |||
58 | @audit() |
||
59 | def dummy(self): |
||
60 | resource = Resource() |
||
61 | resource.id = 123 |
||
62 | resource.naam = 'crash test' |
||
63 | resource.voornaam = 'dummy' |
||
64 | return resource |
||
65 | |||
66 | @audit() |
||
67 | def dummy_none(self): |
||
68 | return 'test' |
||
69 | |||
70 | @audit(result_id_key='naam') |
||
71 | def dummy_other_id(self): |
||
72 | resource = Resource() |
||
73 | resource.id = 123 |
||
74 | resource.naam = '456' |
||
75 | resource.voornaam = 'dummy' |
||
76 | return resource |
||
77 | |||
78 | @audit(actie='bijlage bewerken') |
||
79 | def dummy_bijlage(self): |
||
80 | return None |
||
81 | |||
82 | |||
83 | @audit_with_request(actie='bijlage blijven bewerken') |
||
84 | def dummy_bijlage_with_request(request): |
||
85 | return None |
||
86 | |||
87 | |||
88 | class AuditTests(unittest.TestCase): |
||
89 | |||
90 | View Code Duplication | @classmethod |
|
0 ignored issues
–
show
Duplication
introduced
by
![]() |
|||
91 | def setUpClass(cls): |
||
92 | config = configparser.ConfigParser() |
||
93 | config.read(os.path.join(os.path.dirname(__file__), 'test.ini')) |
||
94 | settings = config.items('app:oe_utils') |
||
95 | settings = dict((s[0], s[1]) for s in settings) |
||
96 | cls.engine = engine_from_config(settings, prefix='sqlalchemy.') |
||
97 | cls.session_maker = sessionmaker( |
||
98 | bind=cls.engine, |
||
99 | extension=ZopeTransactionExtension() |
||
100 | ) |
||
101 | |||
102 | def setUp(self): |
||
103 | init_test_db(self.engine) |
||
104 | self.session = self.session_maker() |
||
105 | self.dummy_parent = DummyParent() |
||
106 | self.audit_manager = AuditManager(self.session) |
||
107 | self.dummy_parent.request.audit_manager = self.audit_manager |
||
108 | |||
109 | def tearDown(self): |
||
110 | self.session.close() |
||
111 | |||
112 | View Code Duplication | def test_audit_get(self): |
|
0 ignored issues
–
show
|
|||
113 | self.dummy_parent.request.user = {'actor': {'uri': 'uri', 'omschrijving': 'omschrijving'}} |
||
114 | self.dummy_parent.request.method = 'GET' |
||
115 | self.dummy_parent.request.matchdict = {'id': 123} |
||
116 | self.dummy_parent.dummy() |
||
117 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
118 | self.assertEqual('opvragen', wijziging.actie) |
||
119 | self.assertEqual('omschrijving', wijziging.updated_by_omschrijving) |
||
120 | self.assertEqual('crash test', wijziging.resource_object_json['naam']) |
||
121 | |||
122 | def test_audit_put(self): |
||
123 | self.dummy_parent.request.method = 'PUT' |
||
124 | self.dummy_parent.request.matchdict = {'id': 123} |
||
125 | self.dummy_parent.dummy() |
||
126 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
127 | self.assertEqual('bewerken', wijziging.actie) |
||
128 | self.assertEqual('crash test', wijziging.resource_object_json['naam']) |
||
129 | self.assertEqual('publiek', wijziging.updated_by_omschrijving) |
||
130 | |||
131 | def test_audit_post(self): |
||
132 | self.dummy_parent.request.method = 'POST' |
||
133 | self.dummy_parent.dummy() |
||
134 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
135 | self.assertEqual('aanmaken', wijziging.actie) |
||
136 | self.assertEqual('crash test', wijziging.resource_object_json['naam']) |
||
137 | self.assertEqual('crash test', wijziging.resource_object_json['naam']) |
||
138 | |||
139 | View Code Duplication | def test_audit_delete(self): |
|
0 ignored issues
–
show
|
|||
140 | self.dummy_parent.request.method = 'DELETE' |
||
141 | self.dummy_parent.request.matchdict = {'id': 123} |
||
142 | self.dummy_parent.dummy_none() |
||
143 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
144 | self.assertEqual('verwijderen', wijziging.actie) |
||
145 | self.assertEqual(123, wijziging.resource_object_id) |
||
146 | self.assertIsNone(wijziging.resource_object_json) |
||
147 | |||
148 | View Code Duplication | def test_audit_other(self): |
|
0 ignored issues
–
show
|
|||
149 | self.dummy_parent.request.method = 'HEAD' |
||
150 | self.dummy_parent.request.matchdict = {'id': 123} |
||
151 | self.dummy_parent.dummy_none() |
||
152 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
153 | self.assertEqual('onbekend', wijziging.actie) |
||
154 | self.assertEqual(123, wijziging.resource_object_id) |
||
155 | self.assertIsNone(wijziging.resource_object_json) |
||
156 | |||
157 | View Code Duplication | def test_audit_bijlage(self): |
|
0 ignored issues
–
show
|
|||
158 | self.dummy_parent.request.method = 'PUT' |
||
159 | self.dummy_parent.request.matchdict = {'id': 123} |
||
160 | self.dummy_parent.dummy_bijlage() |
||
161 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
162 | self.assertEqual('bijlage bewerken', wijziging.actie) |
||
163 | self.assertEqual(123, wijziging.resource_object_id) |
||
164 | self.assertIsNone(wijziging.resource_object_json) |
||
165 | |||
166 | def test_db_audit_manager(self): |
||
167 | config = Mock() |
||
168 | includeme(config) |
||
169 | request = MagicMock() |
||
170 | request.db = self.session |
||
171 | manager = audit_manager(request) |
||
172 | self.assertIsInstance(manager, AuditManager) |
||
173 | |||
174 | def test_audit_bijlage_with_request(self): |
||
175 | request = MagicMock() |
||
176 | request.audit_manager = AuditManager(self.session) |
||
177 | request.method = 'PUT' |
||
178 | request.matchdict = {'id': 1234} |
||
179 | dummy_bijlage_with_request(request) |
||
180 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
181 | self.assertEqual('bijlage blijven bewerken', wijziging.actie) |
||
182 | self.assertEqual(1234, wijziging.resource_object_id) |
||
183 | self.assertIsNone(wijziging.resource_object_json) |
||
184 | |||
185 | View Code Duplication | def test_audit_other_id_key(self): |
|
0 ignored issues
–
show
|
|||
186 | self.dummy_parent.request.user = {'actor': {'uri': 'uri', 'omschrijving': 'omschrijving'}} |
||
187 | self.dummy_parent.request.method = 'GET' |
||
188 | self.dummy_parent.request.matchdict = {'id': None} |
||
189 | self.dummy_parent.dummy_other_id() |
||
190 | wijziging = self.session.query(Wijziging).order_by(Wijziging.updated_at.desc()).first() |
||
191 | self.assertEqual('opvragen', wijziging.actie) |
||
192 | self.assertEqual('omschrijving', wijziging.updated_by_omschrijving) |
||
193 | self.assertEqual('456', wijziging.resource_object_json['naam']) |
||
194 | self.assertEqual(456, wijziging.resource_object_id) |
||
195 |