diff --git a/netbox/extras/conditions.py b/netbox/extras/conditions.py index db054149e..6e764209b 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() + self.logic = None - # Compile the set of Conditions - self.conditions = [ - ConditionSet(rule) if is_ruleset(rule) else Condition(**rule) - for rule in ruleset[self.logic] - ] + # 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 549c33478..0b8672d8a 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -376,3 +376,69 @@ class EventRuleTest(APITestCase): # 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))