2
0

test_conditions.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315
  1. from django.contrib.contenttypes.models import ContentType
  2. from django.test import TestCase
  3. from dcim.choices import SiteStatusChoices
  4. from dcim.models import Site
  5. from extras.conditions import Condition, ConditionSet
  6. from extras.events import serialize_for_event
  7. from extras.forms import EventRuleForm
  8. from extras.models import EventRule, Webhook
  9. class ConditionTestCase(TestCase):
  10. def test_dotted_path_access(self):
  11. c = Condition('a.b.c', 1, 'eq')
  12. self.assertTrue(c.eval({'a': {'b': {'c': 1}}}))
  13. self.assertFalse(c.eval({'a': {'b': {'c': 2}}}))
  14. self.assertFalse(c.eval({'a': {'b': {'x': 1}}}))
  15. def test_undefined_attr(self):
  16. c = Condition('x', 1, 'eq')
  17. self.assertFalse(c.eval({}))
  18. self.assertTrue(c.eval({'x': 1}))
  19. #
  20. # Validation tests
  21. #
  22. def test_invalid_op(self):
  23. with self.assertRaises(ValueError):
  24. # 'blah' is not a valid operator
  25. Condition('x', 1, 'blah')
  26. def test_invalid_type(self):
  27. with self.assertRaises(ValueError):
  28. # dict type is unsupported
  29. Condition('x', 1, dict())
  30. def test_invalid_op_type(self):
  31. with self.assertRaises(ValueError):
  32. # 'gt' supports only numeric values
  33. Condition('x', 'foo', 'gt')
  34. #
  35. # Nested attrs tests
  36. #
  37. def test_nested(self):
  38. c = Condition('x.y.z', 1)
  39. self.assertTrue(c.eval({'x': {'y': {'z': 1}}}))
  40. self.assertFalse(c.eval({'x': {'y': {'z': 2}}}))
  41. self.assertFalse(c.eval({'a': {'b': {'c': 1}}}))
  42. #
  43. # Operator tests
  44. #
  45. def test_default_operator(self):
  46. c = Condition('x', 1)
  47. self.assertEqual(c.eval_func, c.eval_eq)
  48. def test_eq(self):
  49. c = Condition('x', 1, 'eq')
  50. self.assertTrue(c.eval({'x': 1}))
  51. self.assertFalse(c.eval({'x': 2}))
  52. def test_eq_negated(self):
  53. c = Condition('x', 1, 'eq', negate=True)
  54. self.assertFalse(c.eval({'x': 1}))
  55. self.assertTrue(c.eval({'x': 2}))
  56. def test_gt(self):
  57. c = Condition('x', 1, 'gt')
  58. self.assertTrue(c.eval({'x': 2}))
  59. self.assertFalse(c.eval({'x': 1}))
  60. def test_gte(self):
  61. c = Condition('x', 1, 'gte')
  62. self.assertTrue(c.eval({'x': 2}))
  63. self.assertTrue(c.eval({'x': 1}))
  64. self.assertFalse(c.eval({'x': 0}))
  65. def test_lt(self):
  66. c = Condition('x', 2, 'lt')
  67. self.assertTrue(c.eval({'x': 1}))
  68. self.assertFalse(c.eval({'x': 2}))
  69. def test_lte(self):
  70. c = Condition('x', 2, 'lte')
  71. self.assertTrue(c.eval({'x': 1}))
  72. self.assertTrue(c.eval({'x': 2}))
  73. self.assertFalse(c.eval({'x': 3}))
  74. def test_in(self):
  75. c = Condition('x', [1, 2, 3], 'in')
  76. self.assertTrue(c.eval({'x': 1}))
  77. self.assertFalse(c.eval({'x': 9}))
  78. def test_in_negated(self):
  79. c = Condition('x', [1, 2, 3], 'in', negate=True)
  80. self.assertFalse(c.eval({'x': 1}))
  81. self.assertTrue(c.eval({'x': 9}))
  82. def test_contains(self):
  83. c = Condition('x', 1, 'contains')
  84. self.assertTrue(c.eval({'x': [1, 2, 3]}))
  85. self.assertFalse(c.eval({'x': [2, 3, 4]}))
  86. def test_contains_negated(self):
  87. c = Condition('x', 1, 'contains', negate=True)
  88. self.assertFalse(c.eval({'x': [1, 2, 3]}))
  89. self.assertTrue(c.eval({'x': [2, 3, 4]}))
  90. def test_regex(self):
  91. c = Condition('x', '[a-z]+', 'regex')
  92. self.assertTrue(c.eval({'x': 'abc'}))
  93. self.assertFalse(c.eval({'x': '123'}))
  94. def test_regex_negated(self):
  95. c = Condition('x', '[a-z]+', 'regex', negate=True)
  96. self.assertFalse(c.eval({'x': 'abc'}))
  97. self.assertTrue(c.eval({'x': '123'}))
  98. class ConditionSetTest(TestCase):
  99. def test_empty(self):
  100. with self.assertRaises(ValueError):
  101. ConditionSet({})
  102. def test_invalid_logic(self):
  103. with self.assertRaises(ValueError):
  104. ConditionSet({'foo': []})
  105. def test_null_value(self):
  106. cs = ConditionSet({
  107. 'and': [
  108. {'attr': 'a', 'value': None, 'op': 'eq', 'negate': True},
  109. ]
  110. })
  111. self.assertFalse(cs.eval({'a': None}))
  112. self.assertTrue(cs.eval({'a': "string"}))
  113. self.assertTrue(cs.eval({'a': {"key": "value"}}))
  114. def test_and_single_depth(self):
  115. cs = ConditionSet({
  116. 'and': [
  117. {'attr': 'a', 'value': 1, 'op': 'eq'},
  118. {'attr': 'b', 'value': 1, 'op': 'eq', 'negate': True},
  119. ]
  120. })
  121. self.assertTrue(cs.eval({'a': 1, 'b': 2}))
  122. self.assertFalse(cs.eval({'a': 1, 'b': 1}))
  123. def test_or_single_depth(self):
  124. cs = ConditionSet({
  125. 'or': [
  126. {'attr': 'a', 'value': 1, 'op': 'eq'},
  127. {'attr': 'b', 'value': 1, 'op': 'eq'},
  128. ]
  129. })
  130. self.assertTrue(cs.eval({'a': 1, 'b': 2}))
  131. self.assertTrue(cs.eval({'a': 2, 'b': 1}))
  132. self.assertFalse(cs.eval({'a': 2, 'b': 2}))
  133. def test_and_multi_depth(self):
  134. cs = ConditionSet({
  135. 'and': [
  136. {'attr': 'a', 'value': 1, 'op': 'eq'},
  137. {'and': [
  138. {'attr': 'b', 'value': 2, 'op': 'eq'},
  139. {'attr': 'c', 'value': 3, 'op': 'eq'},
  140. ]}
  141. ]
  142. })
  143. self.assertTrue(cs.eval({'a': 1, 'b': 2, 'c': 3}))
  144. self.assertFalse(cs.eval({'a': 9, 'b': 2, 'c': 3}))
  145. self.assertFalse(cs.eval({'a': 1, 'b': 9, 'c': 3}))
  146. self.assertFalse(cs.eval({'a': 1, 'b': 2, 'c': 9}))
  147. def test_or_multi_depth(self):
  148. cs = ConditionSet({
  149. 'or': [
  150. {'attr': 'a', 'value': 1, 'op': 'eq'},
  151. {'or': [
  152. {'attr': 'b', 'value': 2, 'op': 'eq'},
  153. {'attr': 'c', 'value': 3, 'op': 'eq'},
  154. ]}
  155. ]
  156. })
  157. self.assertTrue(cs.eval({'a': 1, 'b': 9, 'c': 9}))
  158. self.assertTrue(cs.eval({'a': 9, 'b': 2, 'c': 9}))
  159. self.assertTrue(cs.eval({'a': 9, 'b': 9, 'c': 3}))
  160. self.assertFalse(cs.eval({'a': 9, 'b': 9, 'c': 9}))
  161. def test_mixed_and(self):
  162. cs = ConditionSet({
  163. 'and': [
  164. {'attr': 'a', 'value': 1, 'op': 'eq'},
  165. {'or': [
  166. {'attr': 'b', 'value': 2, 'op': 'eq'},
  167. {'attr': 'c', 'value': 3, 'op': 'eq'},
  168. ]}
  169. ]
  170. })
  171. self.assertTrue(cs.eval({'a': 1, 'b': 2, 'c': 9}))
  172. self.assertTrue(cs.eval({'a': 1, 'b': 9, 'c': 3}))
  173. self.assertFalse(cs.eval({'a': 1, 'b': 9, 'c': 9}))
  174. self.assertFalse(cs.eval({'a': 9, 'b': 2, 'c': 3}))
  175. def test_mixed_or(self):
  176. cs = ConditionSet({
  177. 'or': [
  178. {'attr': 'a', 'value': 1, 'op': 'eq'},
  179. {'and': [
  180. {'attr': 'b', 'value': 2, 'op': 'eq'},
  181. {'attr': 'c', 'value': 3, 'op': 'eq'},
  182. ]}
  183. ]
  184. })
  185. self.assertTrue(cs.eval({'a': 1, 'b': 9, 'c': 9}))
  186. self.assertTrue(cs.eval({'a': 9, 'b': 2, 'c': 3}))
  187. self.assertTrue(cs.eval({'a': 1, 'b': 2, 'c': 9}))
  188. self.assertFalse(cs.eval({'a': 9, 'b': 2, 'c': 9}))
  189. self.assertFalse(cs.eval({'a': 9, 'b': 9, 'c': 3}))
  190. def test_event_rule_conditions_without_logic_operator(self):
  191. """
  192. Test evaluation of EventRule conditions without logic operator.
  193. """
  194. event_rule = EventRule(
  195. name='Event Rule 1',
  196. type_create=True,
  197. type_update=True,
  198. conditions={
  199. 'attr': 'status.value',
  200. 'value': 'active',
  201. }
  202. )
  203. # Create a Site to evaluate - Status = active
  204. site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE)
  205. data = serialize_for_event(site)
  206. # Evaluate the conditions (status='active')
  207. self.assertTrue(event_rule.eval_conditions(data))
  208. def test_event_rule_conditions_with_logical_operation(self):
  209. """
  210. Test evaluation of EventRule conditions without logic operator, but with logical operation (in).
  211. """
  212. event_rule = EventRule(
  213. name='Event Rule 1',
  214. type_create=True,
  215. type_update=True,
  216. conditions={
  217. "attr": "status.value",
  218. "value": ["planned", "staging"],
  219. "op": "in",
  220. }
  221. )
  222. # Create a Site to evaluate - Status = active
  223. site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE)
  224. data = serialize_for_event(site)
  225. # Evaluate the conditions (status in ['planned, 'staging'])
  226. self.assertFalse(event_rule.eval_conditions(data))
  227. def test_event_rule_conditions_with_logical_operation_and_negate(self):
  228. """
  229. Test evaluation of EventRule with logical operation (in) and negate.
  230. """
  231. event_rule = EventRule(
  232. name='Event Rule 1',
  233. type_create=True,
  234. type_update=True,
  235. conditions={
  236. "attr": "status.value",
  237. "value": ["planned", "staging"],
  238. "op": "in",
  239. "negate": True,
  240. }
  241. )
  242. # Create a Site to evaluate - Status = active
  243. site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE)
  244. data = serialize_for_event(site)
  245. # Evaluate the conditions (status NOT in ['planned, 'staging'])
  246. self.assertTrue(event_rule.eval_conditions(data))
  247. def test_event_rule_conditions_with_incorrect_key_must_return_false(self):
  248. """
  249. Test Event Rule with incorrect condition (key "foo" is wrong). Must return false.
  250. """
  251. ct = ContentType.objects.get(app_label='extras', model='webhook')
  252. site_ct = ContentType.objects.get_for_model(Site)
  253. webhook = Webhook.objects.create(name='Webhook 100', payload_url='http://example.com/?1', http_method='POST')
  254. form = EventRuleForm({
  255. "name": "Event Rule 1",
  256. "type_create": True,
  257. "type_update": True,
  258. "action_object_type": ct.pk,
  259. "action_type": "webhook",
  260. "action_choice": webhook.pk,
  261. "content_types": [site_ct.pk],
  262. "conditions": {
  263. "foo": "status.value",
  264. "value": "active"
  265. }
  266. })
  267. self.assertFalse(form.is_valid())