No Description
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

test_datasource.py 6.9KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. import copy
  2. import datetime
  3. import unittest
  4. from unittest import mock
  5. from unittest.mock import patch
  6. import tests.loader_utils
  7. from tests.leapi.query.utils import dyncode_module as dyncode
  8. from lodel.leapi.query import LeDeleteQuery, LeUpdateQuery, LeGetQuery, \
  9. LeInsertQuery
  10. from lodel.leapi.exceptions import *
  11. class LeQueryDatasourceTestCase(unittest.TestCase):
  12. """ Testing LeQuery objects connection with datasource """
  13. mockread = mock.MagicMock()
  14. mockwrite = mock.MagicMock()
  15. dyncode = None
  16. @classmethod
  17. def setUpClass(cls):
  18. """ Mocking read & write datasource of dyncode """
  19. cls.dyncode = dict()
  20. for dyncls in dyncode.dynclasses:
  21. dyncls._ro_datasource = cls.mockread
  22. dyncls._rw_datasource = cls.mockwrite
  23. cls.dyncode[dyncls.__name__] = dyncls
  24. def setUp(self):
  25. """ Reseting all mock calls before test """
  26. self.mockread.reset_mock()
  27. self.mockwrite.reset_mock()
  28. def check_nocall(self, read = True, exclude = None):
  29. """ Utility function to check if a datasource mock method has been
  30. called during test """
  31. exclude = [] if exclude is None else exclude
  32. if read:
  33. mockds = self.mockread
  34. else:
  35. mockds = self.mockwrite
  36. if 'select' not in exclude:
  37. self.assertFalse(
  38. mockds.select.called,
  39. "select method was not expected to be called")
  40. if 'delete' not in exclude:
  41. self.assertFalse(
  42. mockds.delete.called,
  43. "delete method was not expected to be called")
  44. if 'update' not in exclude:
  45. self.assertFalse(
  46. mockds.update.called,
  47. "update method was not expected to be called")
  48. if 'insert' not in exclude:
  49. self.assertFalse(
  50. mockds.insert.called,
  51. "insert method was not expected to be called")
  52. def test_delete_simple_filter(self):
  53. """ Testing delete query mocking datasource using simple query
  54. filters """
  55. cls = self.dyncode['Person']
  56. query = LeDeleteQuery(
  57. target_class = cls,
  58. query_filter = ['lodel_id = 1', 'alias.lodel_id = 2'])
  59. query.execute()
  60. # Cannot check with assert_called_once_with because of the array
  61. # that is not in a deterministic order
  62. call_args = self.mockwrite.delete.call_args[0]
  63. self.assertEqual(call_args[0], cls)
  64. self.assertEqual(
  65. sorted(call_args[1]),
  66. sorted([('lodel_id', '=', 1), ('alias', '=', '2')]))
  67. self.assertEqual(call_args[2], [])
  68. self.check_nocall(read = False, exclude = ['delete'])
  69. self.check_nocall(read = True)
  70. def test_delete_simple_rel_filters(self):
  71. """ Testing delete query mocking datasource using simple filters
  72. and relationnal filters"""
  73. cls = self.dyncode['Person']
  74. query = LeDeleteQuery(
  75. target_class = cls,
  76. query_filter = ['lodel_id = 1', 'alias.firstname = foo'])
  77. query.execute()
  78. self.mockwrite.delete.assert_called_once_with(
  79. cls,
  80. [('lodel_id', '=', 1)],
  81. [(('alias', {cls: 'firstname'}), '=', 'foo')])
  82. self.check_nocall(read = False, exclude = ['delete'])
  83. self.check_nocall(read = True)
  84. def test_delete_rel_filters(self):
  85. """ Testing delete query mocking datasource """
  86. cls = self.dyncode['Person']
  87. query = LeDeleteQuery(
  88. target_class = cls,
  89. query_filter = ['alias.firstname = foo'])
  90. query.execute()
  91. self.mockwrite.delete.assert_called_once_with(
  92. cls,
  93. [],
  94. [(('alias', {cls: 'firstname'}), '=', 'foo')])
  95. self.check_nocall(read = False, exclude = ['delete'])
  96. self.check_nocall(read = True)
  97. def test_insert(self):
  98. """ Testing LeInsertQuery mocking datasource """
  99. cls = self.dyncode['Person']
  100. query = LeInsertQuery(
  101. target_class = cls)
  102. self.mockwrite.insert.return_value = 1
  103. datas = {
  104. 'firstname': 'foo',
  105. 'lastname': 'bar',
  106. 'alias': None}
  107. query.execute(datas)
  108. self.assertEqual(self.mockwrite.insert.call_count, 1)
  109. cargs , _ = self.mockwrite.insert.call_args
  110. pdatas = cls.prepare_datas(datas, True, False)
  111. self.assertEqual(cargs[0], cls)
  112. cargs = cargs[1]
  113. self.assertEqual(set(pdatas.keys()), set(cargs.keys()))
  114. for dname in pdatas:
  115. if isinstance(pdatas[dname], datetime.datetime):
  116. d1 = pdatas[dname]
  117. d2 = cargs[dname]
  118. for vname in ('year', 'month', 'day', 'hour', 'minute'):
  119. self.assertEqual(
  120. getattr(d1, vname), getattr(d2, vname))
  121. pass
  122. else:
  123. self.assertEqual(pdatas[dname], cargs[dname])
  124. self.check_nocall(read = False, exclude = ['insert'])
  125. self.check_nocall(read = True)
  126. def test_update_instance(self):
  127. """ Testing LeUpdateQuery with an instance mocking datasource """
  128. cls = self.dyncode['Person']
  129. inst = cls(lodel_id = 1, firstname = 'foo', lastname = 'bar',
  130. alias = None, linked_texts = None)
  131. query = LeUpdateQuery(inst)
  132. with self.assertRaises(LeApiQueryError):
  133. # Bad call, giving datas while an instance was given to __init__
  134. query.execute(datas = {'firstname': 'ooba'})
  135. query.execute()
  136. self.mockwrite.update.assert_called_once_with(
  137. cls,
  138. [('lodel_id', '=', '1')],
  139. [],
  140. inst.datas(True))
  141. self.check_nocall(read=False, exclude = ['update'])
  142. self.check_nocall(read=True)
  143. def test_update_filter(self):
  144. """ Testing LeUpdateQuery with filter mocking datasource """
  145. cls = self.dyncode['Person']
  146. fake_db_datas = [{
  147. 'lodel_id': 1,
  148. 'firstname': 'barfoo',
  149. 'lastname': 'foobar',
  150. 'fullname': 'barfoo foobar',
  151. 'alias': None,
  152. 'linked_texts': None,
  153. 'help_text': None,
  154. 'classname': 'Person',
  155. 'date_create': None,
  156. 'date_update': None}]
  157. q_datas = {'firstname': 'foobar', 'lastname': 'barfoo'}
  158. expt_datas = copy.copy(fake_db_datas[0])
  159. expt_datas.update(q_datas)
  160. expt_datas = cls.prepare_datas(expt_datas, True, True)
  161. query = LeUpdateQuery(cls, [('lodel_id', '=', 1)])
  162. with self.assertRaises(LeApiQueryError):
  163. # Bad call, no datas given while a class and a filters were given
  164. # to __init__
  165. query.execute()
  166. self.mockread.select.return_value = fake_db_datas
  167. query.execute(q_datas)
  168. self.mockwrite.update.asser_called_once_with(
  169. cls,
  170. [('lodel_id', '=', '1')],
  171. [],
  172. expt_datas)