mirror of
https://github.com/netbox-community/netbox.git
synced 2025-08-13 19:18:16 -06:00
Changed ConditionSet in conditions.py to allow conditions without logical operator. Added some tests.
This commit is contained in:
parent
ebf6ce1b01
commit
cbde1bebee
@ -132,20 +132,26 @@ class ConditionSet:
|
|||||||
def __init__(self, ruleset):
|
def __init__(self, ruleset):
|
||||||
if type(ruleset) is not dict:
|
if type(ruleset) is not dict:
|
||||||
raise ValueError(f"Ruleset must be a dictionary, not {type(ruleset)}.")
|
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)})")
|
raise ValueError(f"Ruleset must have exactly one logical operator (found {len(ruleset)})")
|
||||||
|
|
||||||
# Determine the logic type
|
self.logic = None
|
||||||
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
|
# If logic type use it, else return the ruleset
|
||||||
self.conditions = [
|
if len(ruleset) == 1:
|
||||||
ConditionSet(rule) if is_ruleset(rule) else Condition(**rule)
|
logic = list(ruleset.keys())[0]
|
||||||
for rule in ruleset[self.logic]
|
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):
|
def eval(self, data):
|
||||||
"""
|
"""
|
||||||
|
@ -376,3 +376,69 @@ class EventRuleTest(APITestCase):
|
|||||||
# Patch the Session object with our dummy_send() method, then process the webhook for sending
|
# 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:
|
with patch.object(Session, 'send', dummy_send) as mock_send:
|
||||||
send_webhook(**job.kwargs)
|
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))
|
||||||
|
Loading…
Reference in New Issue
Block a user