updated validator to restrict cohort_label for R, X and Y rules (#466)

* updated validator to restrict cohort_label for R, X and Y rules

* lint fix

Author: Karthikeyannhs

src/eligibility_signposting_api/services/processors/rule_processor.py

@@ -148,10 +148,6 @@ def evaluate_rules_priority_group(
 
         return best_status, exclusion_reasons, is_rule_stop
 
-    @staticmethod
-    def get_exclusion_rules(cohort: IterationCohort, rules: Iterable[IterationRule]) -> Iterator[IterationRule]:
-        return (ir for ir in rules if not ir.parsed_cohort_labels or cohort.cohort_label in ir.parsed_cohort_labels)
-
     def get_cohort_group_results(
         self, person: Person, active_iteration: Iteration
     ) -> dict[CohortLabel, CohortGroupResult]:

src/rules_validation_api/validators/iteration_rules_validator.py

@@ -2,7 +2,12 @@
 
 from pydantic import model_validator
 
-from eligibility_signposting_api.model.campaign_config import IterationRule, RuleAttributeLevel, RuleAttributeName
+from eligibility_signposting_api.model.campaign_config import (
+    IterationRule,
+    RuleAttributeLevel,
+    RuleAttributeName,
+    RuleType,
+)
 
 
 class IterationRuleValidation(IterationRule):
@@ -16,3 +21,14 @@ def check_cohort_attribute_name(self) -> Self:
             msg = "When attribute_level is COHORT, attribute_name must be COHORT_LABEL or None (default:COHORT_LABEL)"
             raise ValueError(msg)
         return self
+
+    @model_validator(mode="after")
+    def check_cohort_label_for_non_f_and_s_types(self) -> Self:
+        allowed_types = {RuleType("F"), RuleType("S")}
+        if self.cohort_label is not None and self.type not in allowed_types:
+            msg = (
+                f"CohortLabel is only allowed for rule types F and S. "
+                f"Found type: {self.type} with cohort_label: {self.cohort_label}"
+            )
+            raise ValueError(msg)
+        return self

tests/unit/services/processors/test_rule_processor.py

@@ -25,93 +25,6 @@ def rule_processor(mock_person_data_reader):
 MOCK_PERSON_DATA = Person([{"ATTRIBUTE_TYPE": "PERSON", "AGE": "30"}])
 
 
-def test_get_exclusion_rules_no_rules():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    rules_to_filter = []
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([]))
-
-
-def test_get_exclusion_rules_general_rule():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    no_cohort_label_rule = rule_builder.IterationRuleFactory.build(cohort_label=None)
-    rules_to_filter = [no_cohort_label_rule]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([no_cohort_label_rule]))
-
-
-def test_get_exclusion_rules_matching_cohort_label():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_A")
-    rules_to_filter = [matching_rule]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([matching_rule]))
-
-
-def test_get_exclusion_rules_matching_cohort_label_when_it_contains_multiple_cohort_labels():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_A,COHORT_B")
-    rules_to_filter = [matching_rule]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([matching_rule]))
-
-
-def test_get_exclusion_rules_non_matching_cohort_label():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    non_matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_B")
-    rules_to_filter = [non_matching_rule]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([]))
-
-
-def test_get_exclusion_rules_non_matching_cohort_label_when_it_contains_multiple_cohort_labels():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    non_matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_B,COHORT_C")
-    rules_to_filter = [non_matching_rule]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([]))
-
-
-def test_get_exclusion_rules_matching_from_list_cohort_label():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    rule1 = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_A")
-    rule2 = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_B")
-    rules_to_filter = [rule1, rule2]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([rule1]))
-
-
-def test_get_exclusion_rules_matching_from_list_cohort_label_when_it_contains_multiple_cohort_labels():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    rule1 = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_A")
-    rule2 = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_B,COHORT_C")
-    rules_to_filter = [rule1, rule2]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that(result, is_([rule1]))
-
-
-def test_get_exclusion_rules_mixed_rules():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    no_cohort_label_rule = rule_builder.IterationRuleFactory.build(cohort_label=None, name="General")
-    matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_A", name="Matching")
-    non_matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_B", name="NonMatching")
-
-    rules_to_filter = [no_cohort_label_rule, matching_rule, non_matching_rule]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that({r.name for r in result}, is_({"General", "Matching"}))
-
-
-def test_get_exclusion_rules_mixed_rules_when_it_contains_multiple_cohort_labels():
-    cohort = rule_builder.IterationCohortFactory.build(cohort_label="COHORT_A")
-    no_cohort_label_rule = rule_builder.IterationRuleFactory.build(cohort_label=None, name="General")
-    matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_A,COHORT_C", name="Matching")
-    non_matching_rule = rule_builder.IterationRuleFactory.build(cohort_label="COHORT_B,COHORT_C", name="NonMatching")
-
-    rules_to_filter = [no_cohort_label_rule, matching_rule, non_matching_rule]
-    result = list(RuleProcessor.get_exclusion_rules(cohort, rules_to_filter))
-    assert_that({r.name for r in result}, is_({"General", "Matching"}))
-
-
 @patch("eligibility_signposting_api.services.processors.rule_processor.RuleCalculator")
 def test_evaluate_rules_priority_group_all_actionable(mock_rule_calculator_class, rule_processor):
     mock_rule_calculator_class.return_value.evaluate_exclusion.return_value = (

tests/unit/validation/conftest.py

@@ -45,7 +45,6 @@ def valid_iteration_rule_with_only_mandatory_fields():
         "AttributeTarget": "RSV",
         "AttributeLevel": "TARGET",
         "AttributeName": "LAST_SUCCESSFUL_DATE",
-        "CohortLabel": "elid_all_people",
         "Priority": 100,
     }
 

tests/unit/validation/test_iteration_rules_validator.py

@@ -259,3 +259,35 @@ def test_invalid_when_attribute_level_is_cohort_but_attribute_name_is_neither_no
             IterationRuleValidation(**data)
         msg = "When attribute_level is COHORT, attribute_name must be COHORT_LABEL or None (default:COHORT_LABEL)"
         assert msg in str(error.value)
+
+    @pytest.mark.parametrize("rule_type", ["F", "S"])
+    def test_valid_when_cohort_label_present_for_type_f_or_s(
+        self, rule_type, valid_iteration_rule_with_only_mandatory_fields
+    ):
+        data = valid_iteration_rule_with_only_mandatory_fields.copy()
+        data["Type"] = rule_type
+        data["CohortLabel"] = "Test Cohort"
+        result = IterationRuleValidation(**data)
+        assert result.cohort_label == "Test Cohort"
+
+    @pytest.mark.parametrize("rule_type", ["R", "X", "Y"])
+    def test_invalid_when_cohort_label_present_for_non_f_s_types(
+        self, rule_type, valid_iteration_rule_with_only_mandatory_fields
+    ):
+        data = valid_iteration_rule_with_only_mandatory_fields.copy()
+        data["Type"] = rule_type
+        data["CohortLabel"] = "Invalid Cohort"
+        with pytest.raises(ValidationError) as error:
+            IterationRuleValidation(**data)
+        msg = "CohortLabel is only allowed for rule types F and S."
+        assert msg in str(error.value)
+
+    @pytest.mark.parametrize("rule_type", ["R", "X", "Y"])
+    def test_valid_when_cohort_label_absent_for_non_f_s_types(
+        self, rule_type, valid_iteration_rule_with_only_mandatory_fields
+    ):
+        data = valid_iteration_rule_with_only_mandatory_fields.copy()
+        data["Type"] = rule_type
+        data.pop("CohortLabel", None)
+        result = IterationRuleValidation(**data)
+        assert result.cohort_label is None

tests/unit/validation/test_iteration_validator.py

@@ -353,7 +353,6 @@ def test_valid_iteration_if_actions_mapper_exists_for_rule_routing(
             "AttributeTarget": "RSV",
             "AttributeLevel": "TARGET",
             "AttributeName": "LAST_SUCCESSFUL_DATE",
-            "CohortLabel": "elid_all_people",
             "Priority": 100,
             "CommsRouting": default_routing,
         }
@@ -389,7 +388,6 @@ def test_invalid_iteration_if_actions_mapper_exists_for_rule_routing(
             "AttributeTarget": "RSV",
             "AttributeLevel": "TARGET",
             "AttributeName": "LAST_SUCCESSFUL_DATE",
-            "CohortLabel": "elid_all_people",
             "Priority": 100,
             "CommsRouting": default_routing,
         }