|
@@ -0,0 +1,104 @@
|
|
1
|
+import unittest
|
|
2
|
+from unittest import mock
|
|
3
|
+from unittest.mock import patch
|
|
4
|
+
|
|
5
|
+import tests.loader_utils
|
|
6
|
+from tests.leapi.query.utils import dyncode_module as dyncode
|
|
7
|
+from lodel.leapi.query import LeDeleteQuery, LeUpdateQuery, LeGetQuery, \
|
|
8
|
+ LeInsertQuery
|
|
9
|
+
|
|
10
|
+class LeQueryDatasourceTestCase(unittest.TestCase):
|
|
11
|
+ """ Testing LeQuery objects connection with datasource """
|
|
12
|
+
|
|
13
|
+ mockread = mock.MagicMock()
|
|
14
|
+ mockwrite = mock.MagicMock()
|
|
15
|
+ dyncode = None
|
|
16
|
+
|
|
17
|
+ @classmethod
|
|
18
|
+ def setUpClass(cls):
|
|
19
|
+ """ Mocking read & write datasource of dyncode """
|
|
20
|
+ cls.dyncode = dict()
|
|
21
|
+ for dyncls in dyncode.dynclasses:
|
|
22
|
+ dyncls._ro_datasource = cls.mockread
|
|
23
|
+ dyncls._rw_datasource = cls.mockwrite
|
|
24
|
+ cls.dyncode[dyncls.__name__] = dyncls
|
|
25
|
+
|
|
26
|
+ def setUp(self):
|
|
27
|
+ """ Reseting all mock calls before test """
|
|
28
|
+ self.mockread.reset_mock()
|
|
29
|
+ self.mockwrite.reset_mock()
|
|
30
|
+
|
|
31
|
+ def check_nocall(self, read = True, exclude = None):
|
|
32
|
+ """ Utility function to check if a datasource mock method has been
|
|
33
|
+ called during test """
|
|
34
|
+ exclude = [] if exclude is None else exclude
|
|
35
|
+ if read:
|
|
36
|
+ mockds = self.mockread
|
|
37
|
+ else:
|
|
38
|
+ mockds = self.mockwrite
|
|
39
|
+
|
|
40
|
+ if 'select' not in exclude:
|
|
41
|
+ self.assertFalse(
|
|
42
|
+ mockds.select.called,
|
|
43
|
+ "select method was not expected to be called")
|
|
44
|
+ if 'delete' not in exclude:
|
|
45
|
+ self.assertFalse(
|
|
46
|
+ mockds.delete.called,
|
|
47
|
+ "delete method was not expected to be called")
|
|
48
|
+ if 'update' not in exclude:
|
|
49
|
+ self.assertFalse(
|
|
50
|
+ mockds.update.called,
|
|
51
|
+ "update method was not expected to be called")
|
|
52
|
+ if 'insert' not in exclude:
|
|
53
|
+ self.assertFalse(
|
|
54
|
+ mockds.insert.called,
|
|
55
|
+ "insert method was not expected to be called")
|
|
56
|
+
|
|
57
|
+ def test_delete_simple_filter(self):
|
|
58
|
+ """ Testing delete query mocking datasource using simple query
|
|
59
|
+ filters """
|
|
60
|
+ cls = self.dyncode['Person']
|
|
61
|
+ query = LeDeleteQuery(
|
|
62
|
+ target_class = cls,
|
|
63
|
+ query_filter = ['lodel_id = 1', 'alias.lodel_id = 2'])
|
|
64
|
+ query.execute()
|
|
65
|
+ # Cannot check with assert_called_once_with because of the array
|
|
66
|
+ # that is not in a deterministic order
|
|
67
|
+ call_args = self.mockwrite.delete.call_args[0]
|
|
68
|
+ self.assertEqual(call_args[0], cls)
|
|
69
|
+ self.assertEqual(
|
|
70
|
+ sorted(call_args[1]),
|
|
71
|
+ sorted([('lodel_id', '=', '1'), ('alias', '=', '2')]))
|
|
72
|
+ self.assertEqual(call_args[2], [])
|
|
73
|
+ self.check_nocall(read = False, exclude = ['delete'])
|
|
74
|
+ self.check_nocall(read = True)
|
|
75
|
+
|
|
76
|
+
|
|
77
|
+ def test_delete_simple_rel_filters(self):
|
|
78
|
+ """ Testing delete query mocking datasource using simple filters
|
|
79
|
+ and relationnal filters"""
|
|
80
|
+ cls = self.dyncode['Person']
|
|
81
|
+ query = LeDeleteQuery(
|
|
82
|
+ target_class = cls,
|
|
83
|
+ query_filter = ['lodel_id = 1', 'alias.firstname = foo'])
|
|
84
|
+ query.execute()
|
|
85
|
+ self.mockwrite.delete.assert_called_once_with(
|
|
86
|
+ cls,
|
|
87
|
+ [('lodel_id', '=', '1')],
|
|
88
|
+ [(('alias', {cls: 'firstname'}), '=', 'foo')])
|
|
89
|
+ self.check_nocall(read = False, exclude = ['delete'])
|
|
90
|
+ self.check_nocall(read = True)
|
|
91
|
+
|
|
92
|
+ def test_delete_rel_filters(self):
|
|
93
|
+ """ Testing delete query mocking datasource """
|
|
94
|
+ cls = self.dyncode['Person']
|
|
95
|
+ query = LeDeleteQuery(
|
|
96
|
+ target_class = cls,
|
|
97
|
+ query_filter = ['alias.firstname = foo'])
|
|
98
|
+ query.execute()
|
|
99
|
+ self.mockwrite.delete.assert_called_once_with(
|
|
100
|
+ cls,
|
|
101
|
+ [],
|
|
102
|
+ [(('alias', {cls: 'firstname'}), '=', 'foo')])
|
|
103
|
+ self.check_nocall(read = False, exclude = ['delete'])
|
|
104
|
+ self.check_nocall(read = True)
|