business-rules extension updates

Author: jerryvig

business_rules/engine.py

@@ -25,6 +25,9 @@ def run(rule, defined_variables, defined_actions):
 
 def check_conditions_recursively(conditions, defined_variables):
     keys = list(conditions.keys())
+    if keys == ['not']:
+        return not check_conditions_recursively(conditions['not'], defined_variables)
+
     if keys == ['all']:
         assert len(conditions['all']) >= 1
         for condition in conditions['all']:
@@ -40,9 +43,10 @@ def check_conditions_recursively(conditions, defined_variables):
         return False
 
     else:
-        # help prevent errors - any and all can only be in the condition dict
+        # help prevent errors - not, any, and all can only be in the condition dict
         # if they're the only item
-        assert not ('any' in keys or 'all' in keys)
+        if ('not' in keys or 'any' in keys or 'all' in keys):
+            raise ValueError("Only one of 'not', 'any', or 'all' can be at the same level in the conditions dict")
         return check_condition(conditions, defined_variables)
 
 def check_condition(condition, defined_variables):

tests/test_engine_logic.py

@@ -1,30 +1,32 @@
+from unittest import TestCase
+from unittest.mock import patch, MagicMock
+
+import pytest
+
 from business_rules import engine
 from business_rules.variables import BaseVariables
 from business_rules.operators import StringType
 from business_rules.actions import BaseActions
 
-from mock import patch, MagicMock
-from unittest import TestCase
-
 
 class EngineTests(TestCase):
-
     ###
     ### Run
     ###
 
-    @patch.object(engine, 'run')
+    @patch.object(engine, "run")
     def test_run_all_some_rule_triggered(self, *args):
-        """ By default, does not stop on first triggered rule. Returns True if
+        """By default, does not stop on first triggered rule. Returns True if
         any rule was triggered, otherwise False
         """
-        rule1 = {'conditions': 'condition1', 'actions': 'action name 1'}
-        rule2 = {'conditions': 'condition2', 'actions': 'action name 2'}
+        rule1 = {"conditions": "condition1", "actions": "action name 1"}
+        rule2 = {"conditions": "condition2", "actions": "action name 2"}
         variables = BaseVariables()
         actions = BaseActions()
 
         def return_action1(rule, *args, **kwargs):
-            return rule['actions'] == 'action name 1'
+            return rule["actions"] == "action name 1"
+
         engine.run.side_effect = return_action1
 
         result = engine.run_all([rule1, rule2], variables, actions)
@@ -38,158 +40,198 @@ def return_action1(rule, *args, **kwargs):
         self.assertTrue(result)
         self.assertEqual(engine.run.call_count, 2)
 
-    @patch.object(engine, 'run', return_value=True)
+    @patch.object(engine, "run", return_value=True)
     def test_run_all_stop_on_first(self, *args):
-        rule1 = {'conditions': 'condition1', 'actions': 'action name 1'}
-        rule2 = {'conditions': 'condition2', 'actions': 'action name 2'}
+        rule1 = {"conditions": "condition1", "actions": "action name 1"}
+        rule2 = {"conditions": "condition2", "actions": "action name 2"}
         variables = BaseVariables()
         actions = BaseActions()
 
-        result = engine.run_all([rule1, rule2], variables, actions,
-                stop_on_first_trigger=True)
+        result = engine.run_all(
+            [rule1, rule2], variables, actions, stop_on_first_trigger=True
+        )
         self.assertEqual(result, True)
         self.assertEqual(engine.run.call_count, 1)
         engine.run.assert_called_once_with(rule1, variables, actions)
 
-    @patch.object(engine, 'check_conditions_recursively', return_value=True)
-    @patch.object(engine, 'do_actions')
+    @patch.object(engine, "check_conditions_recursively", return_value=True)
+    @patch.object(engine, "do_actions")
     def test_run_that_triggers_rule(self, *args):
-        rule = {'conditions': 'blah', 'actions': 'blah2'}
+        rule = {"conditions": "blah", "actions": "blah2"}
         variables = BaseVariables()
         actions = BaseActions()
 
         result = engine.run(rule, variables, actions)
         self.assertEqual(result, True)
         engine.check_conditions_recursively.assert_called_once_with(
-                rule['conditions'], variables)
-        engine.do_actions.assert_called_once_with(rule['actions'], actions)
+            rule["conditions"], variables
+        )
+        engine.do_actions.assert_called_once_with(rule["actions"], actions)
 
-
-    @patch.object(engine, 'check_conditions_recursively', return_value=False)
-    @patch.object(engine, 'do_actions')
+    @patch.object(engine, "check_conditions_recursively", return_value=False)
+    @patch.object(engine, "do_actions")
     def test_run_that_doesnt_trigger_rule(self, *args):
-        rule = {'conditions': 'blah', 'actions': 'blah2'}
+        rule = {"conditions": "blah", "actions": "blah2"}
         variables = BaseVariables()
         actions = BaseActions()
 
         result = engine.run(rule, variables, actions)
         self.assertEqual(result, False)
         engine.check_conditions_recursively.assert_called_once_with(
-                rule['conditions'], variables)
+            rule["conditions"], variables
+        )
         self.assertEqual(engine.do_actions.call_count, 0)
 
-
-    @patch.object(engine, 'check_condition', return_value=True)
-    def test_check_all_conditions_with_all_true(self, *args):
-        conditions = {'all': [{'thing1': ''}, {'thing2': ''}]}
-        variables = BaseVariables()
-
-        result = engine.check_conditions_recursively(conditions, variables)
-        self.assertEqual(result, True)
-        # assert call count and most recent call are as expected
-        self.assertEqual(engine.check_condition.call_count, 2)
-        engine.check_condition.assert_called_with({'thing2': ''}, variables)
-
-
-    ###
-    ### Check conditions
-    ###
-    @patch.object(engine, 'check_condition', return_value=False)
-    def test_check_all_conditions_with_all_false(self, *args):
-        conditions = {'all': [{'thing1': ''}, {'thing2': ''}]}
-        variables = BaseVariables()
-
-        result = engine.check_conditions_recursively(conditions, variables)
-        self.assertEqual(result, False)
-        engine.check_condition.assert_called_once_with({'thing1': ''}, variables)
-
-
-    def test_check_all_condition_with_no_items_fails(self):
-        with self.assertRaises(AssertionError):
-            engine.check_conditions_recursively({'all': []}, BaseVariables())
-
-
-    @patch.object(engine, 'check_condition', return_value=True)
-    def test_check_any_conditions_with_all_true(self, *args):
-        conditions = {'any': [{'thing1': ''}, {'thing2': ''}]}
-        variables = BaseVariables()
-
-        result = engine.check_conditions_recursively(conditions, variables)
-        self.assertEqual(result, True)
-        engine.check_condition.assert_called_once_with({'thing1': ''}, variables)
-
-
-    @patch.object(engine, 'check_condition', return_value=False)
-    def test_check_any_conditions_with_all_false(self, *args):
-        conditions = {'any': [{'thing1': ''}, {'thing2': ''}]}
-        variables = BaseVariables()
-
-        result = engine.check_conditions_recursively(conditions, variables)
-        self.assertEqual(result, False)
-        # assert call count and most recent call are as expected
-        self.assertEqual(engine.check_condition.call_count, 2)
-        engine.check_condition.assert_called_with({'thing2': ''}, variables)
-
-
-    def test_check_any_condition_with_no_items_fails(self):
-        with self.assertRaises(AssertionError):
-            engine.check_conditions_recursively({'any': []}, BaseVariables())
-
-
-    def test_check_all_and_any_together(self):
-        conditions = {'any': [], 'all': []}
-        variables = BaseVariables()
-        with self.assertRaises(AssertionError):
-            engine.check_conditions_recursively(conditions, variables)
-
-    @patch.object(engine, 'check_condition')
-    def test_nested_all_and_any(self, *args):
-        conditions = {'all': [
-            {'any': [{'name': 1}, {'name': 2}]},
-            {'name': 3}]}
-        bv = BaseVariables()
-
-        def side_effect(condition, _):
-            return condition['name'] in [2,3]
-        engine.check_condition.side_effect = side_effect
-
-        engine.check_conditions_recursively(conditions, bv)
-        self.assertEqual(engine.check_condition.call_count, 3)
-        engine.check_condition.assert_any_call({'name': 1}, bv)
-        engine.check_condition.assert_any_call({'name': 2}, bv)
-        engine.check_condition.assert_any_call({'name': 3}, bv)
-
-
     ###
     ### Operator comparisons
     ###
     def test_check_operator_comparison(self):
-        string_type = StringType('yo yo')
-        with patch.object(string_type, 'contains', return_value=True):
+        string_type = StringType("yo yo")
+        with patch.object(string_type, "contains", return_value=True):
             result = engine._do_operator_comparison(
-                    string_type, 'contains', 'its mocked')
+                string_type, "contains", "its mocked"
+            )
             self.assertTrue(result)
-            string_type.contains.assert_called_once_with('its mocked')
-
+            string_type.contains.assert_called_once_with("its mocked")
 
     ###
     ### Actions
     ###
     def test_do_actions(self):
-        actions = [ {'name': 'action1'},
-                    {'name': 'action2',
-                     'params': {'param1': 'foo', 'param2': 10}}]
+        actions = [
+            {"name": "action1"},
+            {"name": "action2", "params": {"param1": "foo", "param2": 10}},
+        ]
         defined_actions = BaseActions()
         defined_actions.action1 = MagicMock()
         defined_actions.action2 = MagicMock()
 
         engine.do_actions(actions, defined_actions)
 
         defined_actions.action1.assert_called_once_with()
-        defined_actions.action2.assert_called_once_with(param1='foo', param2=10)
+        defined_actions.action2.assert_called_once_with(param1="foo", param2=10)
 
     def test_do_with_invalid_action(self):
-        actions = [{'name': 'fakeone'}]
+        actions = [{"name": "fakeone"}]
         err_string = "Action fakeone is not defined in class BaseActions"
         with self.assertRaisesRegex(AssertionError, err_string):
             engine.do_actions(actions, BaseActions())
+
+
+@patch.object(engine, "check_condition", return_value=True)
+def test_check_all_conditions_with_all_true(_mock_check_condition: MagicMock) -> None:
+    # set up
+    conditions = {"all": [{"thing1": ""}, {"thing2": ""}]}
+    variables = BaseVariables()
+    # test
+    result = engine.check_conditions_recursively(conditions, variables)
+    assert result is True
+    # assert call count and most recent call are as expected
+    assert engine.check_condition.call_count == 2
+    engine.check_condition.assert_called_with({"thing2": ""}, variables)
+
+
+@patch.object(engine, "check_condition", return_value=False)
+def test_check_all_conditions_with_all_false(_mock_check_condition: MagicMock) -> None:
+    # set up
+    conditions = {"all": [{"thing1": ""}, {"thing2": ""}]}
+    variables = BaseVariables()
+    # test
+    result = engine.check_conditions_recursively(conditions, variables)
+    # self.assertEqual(result, False)
+    assert result is False
+    engine.check_condition.assert_called_once_with({"thing1": ""}, variables)
+
+
+def test_check_all_and_any_together() -> None:
+    conditions = {"any": [], "all": []}
+    variables = BaseVariables()
+    with pytest.raises(
+        ValueError,
+        match="Only one of 'not', 'any', or 'all' can be at the same level in the conditions dict",
+    ):
+        engine.check_conditions_recursively(conditions, variables)
+
+
+def test_check_all_conditions_with_no_items_fails() -> None:
+    with pytest.raises(AssertionError):
+        engine.check_conditions_recursively({"all": []}, BaseVariables())
+
+
+@patch.object(engine, "check_condition", return_value=True)
+def test_check_any_conditions_with_all_true(_mock_check_condition: MagicMock) -> None:
+    conditions = {"any": [{"thing1": ""}, {"thing2": ""}]}
+    variables = BaseVariables()
+
+    result = engine.check_conditions_recursively(conditions, variables)
+    assert result is True
+    engine.check_condition.assert_called_once_with({"thing1": ""}, variables)
+
+
+@patch.object(engine, "check_condition", return_value=False)
+def test_check_any_conditions_with_all_false(_mock_check_condition: MagicMock) -> None:
+    conditions = {"any": [{"thing1": ""}, {"thing2": ""}]}
+    variables = BaseVariables()
+
+    result = engine.check_conditions_recursively(conditions, variables)
+    assert result is False
+    # assert call count and most recent call are as expected
+    assert engine.check_condition.call_count == 2
+    engine.check_condition.assert_called_with({"thing2": ""}, variables)
+
+
+def test_check_any_condition_with_no_items_fails() -> None:
+    with pytest.raises(AssertionError):
+        engine.check_conditions_recursively({"any": []}, BaseVariables())
+
+
+@patch.object(engine, "check_condition")
+def test_nested_all_and_any(_mock_check_condition: MagicMock) -> None:
+    conditions = {"all": [{"any": [{"name": 1}, {"name": 2}]}, {"name": 3}]}
+    bv = BaseVariables()
+
+    def side_effect(condition, _):
+        return condition["name"] in [2, 3]
+
+    engine.check_condition.side_effect = side_effect
+
+    engine.check_conditions_recursively(conditions, bv)
+    # self.assertEqual(engine.check_condition.call_count, 3)
+    assert engine.check_condition.call_count == 3
+    engine.check_condition.assert_any_call({"name": 1}, bv)
+    engine.check_condition.assert_any_call({"name": 2}, bv)
+    engine.check_condition.assert_any_call({"name": 3}, bv)
+
+
+@pytest.mark.parametrize("other_key", ["all", "any"])
+def test_check_not_with_all_any(other_key: str) -> None:
+    """
+    DOCUMENT ME.
+    """
+    conditions = {"not": [], other_key: []}
+    variables = BaseVariables()
+    with pytest.raises(
+        ValueError,
+        match="Only one of 'not', 'any', or 'all' can be at the same level in the conditions dict",
+    ):
+        engine.check_conditions_recursively(conditions, variables)
+
+
+@pytest.mark.parametrize(
+    "check_condition_result, expected_result", [(True, False), (False, True)]
+)
+def test_check_not_negates_result(
+    check_condition_result: bool, expected_result: bool
+) -> None:
+    """
+    DOCUMENT ME.
+    """
+    conditions = {"not": {"thing1": ""}}
+    variables = BaseVariables()
+
+    with patch.object(
+        engine, "check_condition", return_value=check_condition_result
+    ) as mock_check_condition:
+        result = engine.check_conditions_recursively(conditions, variables)
+    assert result is expected_result
+    mock_check_condition.assert_called_once_with({"thing1": ""}, variables)