test_conditions.py 11 KB

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