"""Unit tests for automation.evaluator — one test per operator + edge cases. Tests are pure unit tests — no database, no Django fixtures required. """ from __future__ import annotations import pytest from simorgh.apps.automation.evaluator import ( EvaluatorError, _resolve_path, evaluate_conditions, ) # --------------------------------------------------------------------------- # Fixtures / helpers # --------------------------------------------------------------------------- SIMPLE_PAYLOAD = { "lead": { "status": "new", "score": 75, "tags": ["vip", "priority"], "name": "Alice", "email": "alice@example.com", "closed_at": None, "assigned_to": 42, }, "source": "web", } def cond(field, op, value=None): """Build a simple condition dict.""" c = {"field": field, "op": op} if value is not None or op not in ("is_null", "is_not_null"): c["value"] = value return c # --------------------------------------------------------------------------- # Path resolution # --------------------------------------------------------------------------- class TestResolvePath: def test_simple_key(self): assert _resolve_path("source", {"source": "web"}) == "web" def test_nested_key(self): assert _resolve_path("lead.status", SIMPLE_PAYLOAD) == "new" def test_deeply_nested(self): payload = {"a": {"b": {"c": {"d": 99}}}} assert _resolve_path("a.b.c.d", payload) == 99 def test_payload_prefix_stripped(self): assert _resolve_path("payload.lead.score", SIMPLE_PAYLOAD) == 75 def test_missing_key_returns_none(self): assert _resolve_path("lead.nonexistent", SIMPLE_PAYLOAD) is None def test_missing_root_returns_none(self): assert _resolve_path("ghost", SIMPLE_PAYLOAD) is None def test_intermediate_none_returns_none(self): payload = {"lead": None} assert _resolve_path("lead.status", payload) is None def test_empty_path_raises(self): with pytest.raises(EvaluatorError): _resolve_path("", {}) # --------------------------------------------------------------------------- # Operator: eq # --------------------------------------------------------------------------- class TestEq: def test_string_match(self): assert evaluate_conditions([cond("lead.status", "eq", "new")], SIMPLE_PAYLOAD) def test_string_no_match(self): assert not evaluate_conditions([cond("lead.status", "eq", "closed")], SIMPLE_PAYLOAD) def test_int_match(self): assert evaluate_conditions([cond("lead.score", "eq", 75)], SIMPLE_PAYLOAD) def test_default_op_is_eq(self): assert evaluate_conditions([{"field": "lead.status", "value": "new"}], SIMPLE_PAYLOAD) def test_none_eq_none(self): assert evaluate_conditions([cond("lead.closed_at", "eq", None)], SIMPLE_PAYLOAD) # --------------------------------------------------------------------------- # Operator: neq # --------------------------------------------------------------------------- class TestNeq: def test_different_values(self): assert evaluate_conditions([cond("lead.status", "neq", "closed")], SIMPLE_PAYLOAD) def test_same_value(self): assert not evaluate_conditions([cond("lead.status", "neq", "new")], SIMPLE_PAYLOAD) # --------------------------------------------------------------------------- # Operator: gt / lt / gte / lte # --------------------------------------------------------------------------- class TestNumericComparisons: def test_gt_pass(self): assert evaluate_conditions([cond("lead.score", "gt", 50)], SIMPLE_PAYLOAD) def test_gt_fail(self): assert not evaluate_conditions([cond("lead.score", "gt", 75)], SIMPLE_PAYLOAD) def test_lt_pass(self): assert evaluate_conditions([cond("lead.score", "lt", 100)], SIMPLE_PAYLOAD) def test_lt_fail(self): assert not evaluate_conditions([cond("lead.score", "lt", 50)], SIMPLE_PAYLOAD) def test_gte_equal(self): assert evaluate_conditions([cond("lead.score", "gte", 75)], SIMPLE_PAYLOAD) def test_gte_greater(self): assert evaluate_conditions([cond("lead.score", "gte", 70)], SIMPLE_PAYLOAD) def test_gte_fail(self): assert not evaluate_conditions([cond("lead.score", "gte", 76)], SIMPLE_PAYLOAD) def test_lte_equal(self): assert evaluate_conditions([cond("lead.score", "lte", 75)], SIMPLE_PAYLOAD) def test_lte_less(self): assert evaluate_conditions([cond("lead.score", "lte", 80)], SIMPLE_PAYLOAD) def test_lte_fail(self): assert not evaluate_conditions([cond("lead.score", "lte", 74)], SIMPLE_PAYLOAD) def test_incompatible_types_returns_false(self): """gt/lt/gte/lte with incompatible types should return False, not raise.""" assert not evaluate_conditions([cond("lead.status", "gt", 100)], SIMPLE_PAYLOAD) # --------------------------------------------------------------------------- # Operator: contains # --------------------------------------------------------------------------- class TestContains: def test_string_contains(self): assert evaluate_conditions([cond("lead.email", "contains", "@example")], SIMPLE_PAYLOAD) def test_string_not_contains(self): assert not evaluate_conditions([cond("lead.email", "contains", "@other")], SIMPLE_PAYLOAD) def test_list_contains(self): assert evaluate_conditions([cond("lead.tags", "contains", "vip")], SIMPLE_PAYLOAD) def test_list_not_contains(self): assert not evaluate_conditions([cond("lead.tags", "contains", "ghost")], SIMPLE_PAYLOAD) # --------------------------------------------------------------------------- # Operator: startswith # --------------------------------------------------------------------------- class TestStartsWith: def test_match(self): assert evaluate_conditions([cond("lead.email", "startswith", "alice")], SIMPLE_PAYLOAD) def test_no_match(self): assert not evaluate_conditions([cond("lead.email", "startswith", "bob")], SIMPLE_PAYLOAD) def test_int_coercion(self): """startswith converts actual to str before comparing.""" payload = {"code": 404} assert evaluate_conditions([cond("code", "startswith", "4")], payload) # --------------------------------------------------------------------------- # Operator: in # --------------------------------------------------------------------------- class TestIn: def test_value_in_list(self): payload = {"status": "new"} assert evaluate_conditions([cond("status", "in", ["new", "open"])], payload) def test_value_not_in_list(self): payload = {"status": "closed"} assert not evaluate_conditions([cond("status", "in", ["new", "open"])], payload) def test_int_in_range(self): payload = {"score": 75} assert evaluate_conditions([cond("score", "in", [50, 75, 100])], payload) # --------------------------------------------------------------------------- # Operator: not_in # --------------------------------------------------------------------------- class TestNotIn: def test_not_in_list(self): payload = {"status": "closed"} assert evaluate_conditions([cond("status", "not_in", ["new", "open"])], payload) def test_in_list_fails(self): payload = {"status": "new"} assert not evaluate_conditions([cond("status", "not_in", ["new", "open"])], payload) # --------------------------------------------------------------------------- # Operator: is_null / is_not_null # --------------------------------------------------------------------------- class TestNullChecks: def test_is_null_on_none(self): assert evaluate_conditions([{"field": "lead.closed_at", "op": "is_null"}], SIMPLE_PAYLOAD) def test_is_null_on_value(self): assert not evaluate_conditions( [{"field": "lead.assigned_to", "op": "is_null"}], SIMPLE_PAYLOAD ) def test_is_not_null_on_value(self): assert evaluate_conditions( [{"field": "lead.assigned_to", "op": "is_not_null"}], SIMPLE_PAYLOAD ) def test_is_not_null_on_none(self): assert not evaluate_conditions( [{"field": "lead.closed_at", "op": "is_not_null"}], SIMPLE_PAYLOAD ) def test_missing_key_is_null(self): """Missing path resolves to None → is_null passes.""" assert evaluate_conditions( [{"field": "lead.nonexistent", "op": "is_null"}], SIMPLE_PAYLOAD ) # --------------------------------------------------------------------------- # Compound logical operators (all / any / not) # --------------------------------------------------------------------------- class TestLogicalCompound: def test_all_both_true(self): conditions = [ { "all": [ cond("lead.status", "eq", "new"), cond("lead.score", "gte", 50), ] } ] assert evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_all_one_false(self): conditions = [ { "all": [ cond("lead.status", "eq", "new"), cond("lead.score", "gte", 100), ] } ] assert not evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_any_one_true(self): conditions = [ { "any": [ cond("lead.status", "eq", "closed"), cond("lead.score", "gte", 50), ] } ] assert evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_any_all_false(self): conditions = [ { "any": [ cond("lead.status", "eq", "closed"), cond("lead.score", "gte", 200), ] } ] assert not evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_not_inverts(self): conditions = [{"not": cond("lead.status", "eq", "closed")}] assert evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_not_false_on_match(self): conditions = [{"not": cond("lead.status", "eq", "new")}] assert not evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_nested_all_any(self): conditions = [ { "all": [ cond("lead.status", "eq", "new"), { "any": [ cond("lead.score", "gte", 100), cond("source", "eq", "web"), ] }, ] } ] assert evaluate_conditions(conditions, SIMPLE_PAYLOAD) # --------------------------------------------------------------------------- # Edge cases for evaluate_conditions # --------------------------------------------------------------------------- class TestEvaluateConditionsEdgeCases: def test_empty_list_always_true(self): assert evaluate_conditions([], SIMPLE_PAYLOAD) def test_multiple_conditions_all_and(self): conditions = [ cond("lead.status", "eq", "new"), cond("lead.score", "gte", 50), cond("source", "eq", "web"), ] assert evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_multiple_conditions_one_fails(self): conditions = [ cond("lead.status", "eq", "new"), cond("lead.score", "gte", 100), # fails ] assert not evaluate_conditions(conditions, SIMPLE_PAYLOAD) def test_payload_prefix_equivalent_to_no_prefix(self): c1 = cond("lead.status", "eq", "new") c2 = cond("payload.lead.status", "eq", "new") assert evaluate_conditions([c1], SIMPLE_PAYLOAD) assert evaluate_conditions([c2], SIMPLE_PAYLOAD) def test_unknown_operator_raises(self): with pytest.raises(EvaluatorError, match="unknown operator"): evaluate_conditions([cond("lead.status", "regex", ".*")], SIMPLE_PAYLOAD) def test_non_dict_node_raises(self): with pytest.raises(EvaluatorError, match="must be a dict"): evaluate_conditions(["invalid"], SIMPLE_PAYLOAD) def test_missing_field_key_raises(self): with pytest.raises(EvaluatorError, match="'field' key"): evaluate_conditions([{"op": "eq", "value": "new"}], SIMPLE_PAYLOAD) def test_max_depth_exceeded_raises(self): # Build a deeply nested 'not' chain node: dict = cond("source", "eq", "web") for _ in range(12): node = {"not": node} with pytest.raises(EvaluatorError, match="nesting depth"): evaluate_conditions([node], SIMPLE_PAYLOAD)