From cbde1bebeeadf4dda9ba9eb1b088adbc69d678f4 Mon Sep 17 00:00:00 2001 From: "julio.oliveira" Date: Fri, 19 Jan 2024 15:33:08 -0300 Subject: [PATCH 1/7] Changed ConditionSet in conditions.py to allow conditions without logical operator. Added some tests. --- netbox/extras/conditions.py | 30 ++++++----- netbox/extras/tests/test_event_rules.py | 66 +++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 12 deletions(-) diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index db054149efc..6e764209b4c 100644 --- a/netbox/extras/conditions.py +++ b/netbox/extras/conditions.py @@ -132,20 +132,26 @@ class ConditionSet: def __init__(self, ruleset): if type(ruleset) is not dict: raise ValueError(f"Ruleset must be a dictionary, not {type(ruleset)}.") - if len(ruleset) != 1: + if len(ruleset) < 1: raise ValueError(f"Ruleset must have exactly one logical operator (found {len(ruleset)})") - # Determine the logic type - logic = list(ruleset.keys())[0] - if type(logic) is not str or logic.lower() not in (AND, OR): - raise ValueError(f"Invalid logic type: {logic} (must be '{AND}' or '{OR}')") - self.logic = logic.lower() - - # Compile the set of Conditions - self.conditions = [ - ConditionSet(rule) if is_ruleset(rule) else Condition(**rule) - for rule in ruleset[self.logic] - ] + self.logic = None + + # If logic type use it, else return the ruleset + if len(ruleset) == 1: + logic = list(ruleset.keys())[0] + if logic.lower() in (AND, OR): + self.logic = logic.lower() + else: + raise ValueError(f"Invalid logic type: {logic} (must be '{AND}' or '{OR}')") + + # Compile the set of Conditions + self.conditions = [ + ConditionSet(rule) if is_ruleset(rule) else Condition(**rule) + for rule in ruleset[self.logic] + ] + else: + self.conditions = [Condition(**ruleset)] def eval(self, data): """ diff --git a/netbox/extras/tests/test_event_rules.py b/netbox/extras/tests/test_event_rules.py index 549c3347820..0b8672d8ae7 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -376,3 +376,69 @@ def dummy_send(_, request, **kwargs): # Patch the Session object with our dummy_send() method, then process the webhook for sending with patch.object(Session, 'send', dummy_send) as mock_send: send_webhook(**job.kwargs) + + def test_event_rule_conditions_without_logic_operator(self): + """ + Test evaluation of EventRule conditions without logic operator. + """ + event_rule = EventRule( + name='Event Rule 1', + type_create=True, + type_update=True, + conditions={ + 'attr': 'status.value', + 'value': 'active', + } + ) + + # Create a Site to evaluate - Status = active + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + data = serialize_for_event(site) + + # Evaluate the conditions (status='active') + self.assertTrue(event_rule.eval_conditions(data)) + + def test_event_rule_conditions_with_logical_operation(self): + """ + Test evaluation of EventRule conditions without logic operator, but with logical operation (in). + """ + event_rule = EventRule( + name='Event Rule 1', + type_create=True, + type_update=True, + conditions={ + "attr": "status.value", + "value": ["planned", "staging"], + "op": "in", + } + ) + + # Create a Site to evaluate - Status = active + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + data = serialize_for_event(site) + + # Evaluate the conditions (status in ['planned, 'staging']) + self.assertFalse(event_rule.eval_conditions(data)) + + def test_event_rule_conditions_with_logical_operation_and_negate(self): + """ + Test evaluation of EventRule with logical operation (in) and negate. + """ + event_rule = EventRule( + name='Event Rule 1', + type_create=True, + type_update=True, + conditions={ + "attr": "status.value", + "value": ["planned", "staging"], + "op": "in", + "negate": True, + } + ) + + # Create a Site to evaluate - Status = active + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + data = serialize_for_event(site) + + # Evaluate the conditions (status NOT in ['planned, 'staging']) + self.assertTrue(event_rule.eval_conditions(data)) From b14c447fcfb22c719cc5b4a28b3d6705b9fdbc39 Mon Sep 17 00:00:00 2001 From: "julio.oliveira" Date: Mon, 22 Jan 2024 15:19:37 -0300 Subject: [PATCH 2/7] Addressed comments in the PR. --- netbox/extras/conditions.py | 16 +++++++++------- netbox/extras/tests/test_event_rules.py | 25 +++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 7 deletions(-) diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index 6e764209b4c..11098fe08f5 100644 --- a/netbox/extras/conditions.py +++ b/netbox/extras/conditions.py @@ -130,20 +130,19 @@ class ConditionSet: :param ruleset: A dictionary mapping a logical operator to a list of conditional rules """ def __init__(self, ruleset): + if type(ruleset) is not dict: raise ValueError(f"Ruleset must be a dictionary, not {type(ruleset)}.") - if len(ruleset) < 1: - raise ValueError(f"Ruleset must have exactly one logical operator (found {len(ruleset)})") self.logic = None # If logic type use it, else return the ruleset if len(ruleset) == 1: logic = list(ruleset.keys())[0] - if logic.lower() in (AND, OR): - self.logic = logic.lower() - else: - raise ValueError(f"Invalid logic type: {logic} (must be '{AND}' or '{OR}')") + if logic not in (AND, OR): + raise ValueError( + f"Invalid logic type: {logic} (must be '{AND}' or '{OR}'). Please check documentation.") + self.logic = logic.lower() # Compile the set of Conditions self.conditions = [ @@ -151,7 +150,10 @@ def __init__(self, ruleset): for rule in ruleset[self.logic] ] else: - self.conditions = [Condition(**ruleset)] + try: + self.conditions = [Condition(**ruleset)] + except TypeError: + raise ValueError(f"Incorrect key(s) informed. Please check documentation.") def eval(self, data): """ diff --git a/netbox/extras/tests/test_event_rules.py b/netbox/extras/tests/test_event_rules.py index 0b8672d8ae7..74161e0a58c 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -10,6 +10,7 @@ from django.urls import reverse from extras.choices import EventRuleActionChoices, ObjectChangeActionChoices from extras.events import enqueue_object, flush_events, serialize_for_event +from extras.forms import SavedFilterForm, EventRuleForm from extras.models import EventRule, Tag, Webhook from extras.webhooks import generate_signature, send_webhook from requests import Session @@ -442,3 +443,27 @@ def test_event_rule_conditions_with_logical_operation_and_negate(self): # Evaluate the conditions (status NOT in ['planned, 'staging']) self.assertTrue(event_rule.eval_conditions(data)) + + def test_event_rule_conditions_with_incorrect_key_must_return_false(self): + """ + Test Event Rule with incorrect condition (key "foo" is wrong). Must return false. + """ + + ct = ContentType.objects.get(app_label='extras', model='webhook') + site_ct = ContentType.objects.get_for_model(Site) + webhook = Webhook.objects.create(name='Webhook 100', payload_url='http://example.com/?1', http_method='POST') + form = EventRuleForm({ + "name": "Event Rule 1", + "type_create": True, + "type_update": True, + "action_object_type": ct.pk, + "action_type": "webhook", + "action_choice": webhook.pk, + "content_types": [site_ct.pk], + "conditions": { + "foo": "status.value", + "value": "active" + } + }) + + self.assertFalse(form.is_valid()) From 8998a34b02f9013577b84bcc0e906f52668b6ce0 Mon Sep 17 00:00:00 2001 From: "julio.oliveira" Date: Fri, 26 Jan 2024 16:38:24 -0300 Subject: [PATCH 3/7] Addressed comments in the PR. --- netbox/extras/conditions.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index 11098fe08f5..e8ba2bed94c 100644 --- a/netbox/extras/conditions.py +++ b/netbox/extras/conditions.py @@ -134,15 +134,11 @@ def __init__(self, ruleset): if type(ruleset) is not dict: raise ValueError(f"Ruleset must be a dictionary, not {type(ruleset)}.") - self.logic = None - - # If logic type use it, else return the ruleset if len(ruleset) == 1: - logic = list(ruleset.keys())[0] - if logic not in (AND, OR): + self.logic = (list(ruleset.keys())[0]).lower() + if self.logic not in (AND, OR): raise ValueError( - f"Invalid logic type: {logic} (must be '{AND}' or '{OR}'). Please check documentation.") - self.logic = logic.lower() + f"Invalid logic type: {self.logic} (must be '{AND}' or '{OR}'). Please check documentation.") # Compile the set of Conditions self.conditions = [ @@ -151,6 +147,7 @@ def __init__(self, ruleset): ] else: try: + self.logic = None self.conditions = [Condition(**ruleset)] except TypeError: raise ValueError(f"Incorrect key(s) informed. Please check documentation.") From 5fd8dced0c494aec128f084c3f1d56bc2a35b32d Mon Sep 17 00:00:00 2001 From: Julio-Oliveira-Encora Date: Mon, 20 May 2024 14:52:18 -0300 Subject: [PATCH 4/7] Moved tests from test_event_rules.py to test_conditions.py. --- netbox/extras/conditions.py | 5 -- netbox/extras/tests/test_conditions.py | 96 +++++++++++++++++++++++++ netbox/extras/tests/test_event_rules.py | 90 ----------------------- 3 files changed, 96 insertions(+), 95 deletions(-) diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index 863397d7f86..22ab1541705 100644 --- a/netbox/extras/conditions.py +++ b/netbox/extras/conditions.py @@ -136,11 +136,6 @@ def __init__(self, ruleset): if type(ruleset) is not dict: raise ValueError(_("Ruleset must be a dictionary, not {ruleset}.").format(ruleset=type(ruleset))) - if len(ruleset) != 1: - raise ValueError(_("Ruleset must have exactly one logical operator (found {ruleset})").format( - ruleset=len(ruleset))) - raise ValueError(f"Ruleset must be a dictionary, not {type(ruleset)}.") - if len(ruleset) == 1: self.logic = (list(ruleset.keys())[0]).lower() if self.logic not in (AND, OR): diff --git a/netbox/extras/tests/test_conditions.py b/netbox/extras/tests/test_conditions.py index e7275482afd..dd528b918dc 100644 --- a/netbox/extras/tests/test_conditions.py +++ b/netbox/extras/tests/test_conditions.py @@ -1,6 +1,12 @@ +from django.contrib.contenttypes.models import ContentType from django.test import TestCase +from dcim.choices import SiteStatusChoices +from dcim.models import Site from extras.conditions import Condition, ConditionSet +from extras.events import serialize_for_event +from extras.forms import EventRuleForm +from extras.models import EventRule, Webhook class ConditionTestCase(TestCase): @@ -217,3 +223,93 @@ def test_mixed_or(self): self.assertTrue(cs.eval({'a': 1, 'b': 2, 'c': 9})) self.assertFalse(cs.eval({'a': 9, 'b': 2, 'c': 9})) self.assertFalse(cs.eval({'a': 9, 'b': 9, 'c': 3})) + + def test_event_rule_conditions_without_logic_operator(self): + """ + Test evaluation of EventRule conditions without logic operator. + """ + event_rule = EventRule( + name='Event Rule 1', + type_create=True, + type_update=True, + conditions={ + 'attr': 'status.value', + 'value': 'active', + } + ) + + # Create a Site to evaluate - Status = active + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + data = serialize_for_event(site) + + # Evaluate the conditions (status='active') + self.assertTrue(event_rule.eval_conditions(data)) + + def test_event_rule_conditions_with_logical_operation(self): + """ + Test evaluation of EventRule conditions without logic operator, but with logical operation (in). + """ + event_rule = EventRule( + name='Event Rule 1', + type_create=True, + type_update=True, + conditions={ + "attr": "status.value", + "value": ["planned", "staging"], + "op": "in", + } + ) + + # Create a Site to evaluate - Status = active + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + data = serialize_for_event(site) + + # Evaluate the conditions (status in ['planned, 'staging']) + self.assertFalse(event_rule.eval_conditions(data)) + + def test_event_rule_conditions_with_logical_operation_and_negate(self): + """ + Test evaluation of EventRule with logical operation (in) and negate. + """ + event_rule = EventRule( + name='Event Rule 1', + type_create=True, + type_update=True, + conditions={ + "attr": "status.value", + "value": ["planned", "staging"], + "op": "in", + "negate": True, + } + ) + + # Create a Site to evaluate - Status = active + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + data = serialize_for_event(site) + + # Evaluate the conditions (status NOT in ['planned, 'staging']) + self.assertTrue(event_rule.eval_conditions(data)) + + def test_event_rule_conditions_with_incorrect_key_must_return_false(self): + """ + Test Event Rule with incorrect condition (key "foo" is wrong). Must return false. + """ + + ct = ContentType.objects.get(app_label='extras', model='webhook') + site_ct = ContentType.objects.get_for_model(Site) + webhook = Webhook.objects.create(name='Webhook 100', payload_url='http://example.com/?1', http_method='POST') + form = EventRuleForm({ + "name": "Event Rule 1", + "type_create": True, + "type_update": True, + "action_object_type": ct.pk, + "action_type": "webhook", + "action_choice": webhook.pk, + "content_types": [site_ct.pk], + "conditions": { + "foo": "status.value", + "value": "active" + } + }) + + self.assertFalse(form.is_valid()) diff --git a/netbox/extras/tests/test_event_rules.py b/netbox/extras/tests/test_event_rules.py index 46809d38eb9..d1544a8fd1d 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -378,93 +378,3 @@ def dummy_send(_, request, **kwargs): # Patch the Session object with our dummy_send() method, then process the webhook for sending with patch.object(Session, 'send', dummy_send) as mock_send: send_webhook(**job.kwargs) - - def test_event_rule_conditions_without_logic_operator(self): - """ - Test evaluation of EventRule conditions without logic operator. - """ - event_rule = EventRule( - name='Event Rule 1', - type_create=True, - type_update=True, - conditions={ - 'attr': 'status.value', - 'value': 'active', - } - ) - - # Create a Site to evaluate - Status = active - site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) - data = serialize_for_event(site) - - # Evaluate the conditions (status='active') - self.assertTrue(event_rule.eval_conditions(data)) - - def test_event_rule_conditions_with_logical_operation(self): - """ - Test evaluation of EventRule conditions without logic operator, but with logical operation (in). - """ - event_rule = EventRule( - name='Event Rule 1', - type_create=True, - type_update=True, - conditions={ - "attr": "status.value", - "value": ["planned", "staging"], - "op": "in", - } - ) - - # Create a Site to evaluate - Status = active - site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) - data = serialize_for_event(site) - - # Evaluate the conditions (status in ['planned, 'staging']) - self.assertFalse(event_rule.eval_conditions(data)) - - def test_event_rule_conditions_with_logical_operation_and_negate(self): - """ - Test evaluation of EventRule with logical operation (in) and negate. - """ - event_rule = EventRule( - name='Event Rule 1', - type_create=True, - type_update=True, - conditions={ - "attr": "status.value", - "value": ["planned", "staging"], - "op": "in", - "negate": True, - } - ) - - # Create a Site to evaluate - Status = active - site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) - data = serialize_for_event(site) - - # Evaluate the conditions (status NOT in ['planned, 'staging']) - self.assertTrue(event_rule.eval_conditions(data)) - - def test_event_rule_conditions_with_incorrect_key_must_return_false(self): - """ - Test Event Rule with incorrect condition (key "foo" is wrong). Must return false. - """ - - ct = ContentType.objects.get(app_label='extras', model='webhook') - site_ct = ContentType.objects.get_for_model(Site) - webhook = Webhook.objects.create(name='Webhook 100', payload_url='http://example.com/?1', http_method='POST') - form = EventRuleForm({ - "name": "Event Rule 1", - "type_create": True, - "type_update": True, - "action_object_type": ct.pk, - "action_type": "webhook", - "action_choice": webhook.pk, - "content_types": [site_ct.pk], - "conditions": { - "foo": "status.value", - "value": "active" - } - }) - - self.assertFalse(form.is_valid()) From 270bd533014b7a3b59f1252eae0207c19b9fddbf Mon Sep 17 00:00:00 2001 From: Julio-Oliveira-Encora Date: Wed, 29 May 2024 14:49:07 -0300 Subject: [PATCH 5/7] Addressed PR comments. --- netbox/extras/conditions.py | 8 ++++---- netbox/extras/tests/test_event_rules.py | 1 - 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index 22ab1541705..15d198a1909 100644 --- a/netbox/extras/conditions.py +++ b/netbox/extras/conditions.py @@ -135,12 +135,12 @@ class ConditionSet: def __init__(self, ruleset): if type(ruleset) is not dict: raise ValueError(_("Ruleset must be a dictionary, not {ruleset}.").format(ruleset=type(ruleset))) - + print(len(ruleset)) if len(ruleset) == 1: self.logic = (list(ruleset.keys())[0]).lower() if self.logic not in (AND, OR): - raise ValueError( - f"Invalid logic type: {self.logic} (must be '{AND}' or '{OR}'). Please check documentation.") + raise ValueError(_( + f"Invalid logic type: {self.logic} (must be 'AND' or 'OR'). Please check documentation.")) # Compile the set of Conditions self.conditions = [ @@ -152,7 +152,7 @@ def __init__(self, ruleset): self.logic = None self.conditions = [Condition(**ruleset)] except TypeError: - raise ValueError(f"Incorrect key(s) informed. Please check documentation.") + raise ValueError(_("Incorrect key(s) informed. Please check documentation.")) def eval(self, data): """ diff --git a/netbox/extras/tests/test_event_rules.py b/netbox/extras/tests/test_event_rules.py index d1544a8fd1d..8cea2078a4c 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -13,7 +13,6 @@ from dcim.models import Site from extras.choices import EventRuleActionChoices, ObjectChangeActionChoices from extras.events import enqueue_object, flush_events, serialize_for_event -from extras.forms import SavedFilterForm, EventRuleForm from extras.models import EventRule, Tag, Webhook from extras.webhooks import generate_signature, send_webhook from utilities.testing import APITestCase From e94b2f39d080ac769997200d835b96a84a3d2268 Mon Sep 17 00:00:00 2001 From: Julio-Oliveira-Encora Date: Wed, 29 May 2024 14:50:14 -0300 Subject: [PATCH 6/7] Addressed PR comments. --- netbox/extras/conditions.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index 15d198a1909..b57ad1a35e1 100644 --- a/netbox/extras/conditions.py +++ b/netbox/extras/conditions.py @@ -135,7 +135,7 @@ class ConditionSet: def __init__(self, ruleset): if type(ruleset) is not dict: raise ValueError(_("Ruleset must be a dictionary, not {ruleset}.").format(ruleset=type(ruleset))) - print(len(ruleset)) + if len(ruleset) == 1: self.logic = (list(ruleset.keys())[0]).lower() if self.logic not in (AND, OR): From ccc6cf3d4fa5a80e0786eb83a1745ccdbcffa06e Mon Sep 17 00:00:00 2001 From: Julio-Oliveira-Encora Date: Mon, 3 Jun 2024 10:55:04 -0300 Subject: [PATCH 7/7] Addressed PR comment. --- netbox/extras/conditions.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index b57ad1a35e1..5680be4444c 100644 --- a/netbox/extras/conditions.py +++ b/netbox/extras/conditions.py @@ -139,8 +139,7 @@ def __init__(self, ruleset): if len(ruleset) == 1: self.logic = (list(ruleset.keys())[0]).lower() if self.logic not in (AND, OR): - raise ValueError(_( - f"Invalid logic type: {self.logic} (must be 'AND' or 'OR'). Please check documentation.")) + raise ValueError(_("Invalid logic type: must be 'AND' or 'OR'. Please check documentation.")) # Compile the set of Conditions self.conditions = [