Add rule stop functionality (#106)

* Added rule_stop logic

* lint fixes

* added types to test func

* lint fixed

* fixed is_rule_stop

* added enum for rule stop

* RuleStop is False ("N") by default, if not set to "Y"

* RuleStop is set to bool

* added unit test for rules stop

* lint fixes

* intermittent unpack issue fixed

Author: Karthikeyannhs

src/eligibility_signposting_api/model/rules.py

@@ -29,6 +29,7 @@
 StartDate = NewType("StartDate", date)
 EndDate = NewType("EndDate", date)
 CohortLabel = NewType("CohortLabel", str)
+RuleStop = NewType("RuleStop", bool)
 
 
 class RuleType(StrEnum):
@@ -99,6 +100,13 @@ class IterationRule(BaseModel):
     operator: RuleOperator = Field(..., alias="Operator")
     comparator: RuleComparator = Field(..., alias="Comparator")
     attribute_target: RuleAttributeTarget | None = Field(None, alias="AttributeTarget")
+    rule_stop: RuleStop | None = Field(None, alias="RuleStop")
+
+    @field_validator("rule_stop", mode="before")
+    def parse_yn_to_bool(cls, v: str) -> bool:  # noqa: N805
+        if isinstance(v, str):
+            return v.upper() == "Y"
+        return False
 
     model_config = {"populate_by_name": True, "extra": "ignore"}
 

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -106,11 +106,15 @@ def evaluate_eligibility_by_iteration_rules(
             exclusion_reasons, actionable_reasons = [], []
             by_priority = sorted(iteration.iteration_rules, key=priority_getter)
             for _, rule_group in groupby(by_priority, key=priority_getter):
-                status, group_actionable, group_exclusions = self.evaluate_priority_group(rule_group, worst_status)
+                status, group_actionable, group_exclusions, is_rule_stop = self.evaluate_priority_group(
+                    rule_group, worst_status
+                )
                 # Merge results
                 worst_status = status
                 actionable_reasons.extend(group_actionable)
                 exclusion_reasons.extend(group_exclusions)
+                if is_rule_stop:
+                    break
             condition_status_entry = status_with_reasons.setdefault(worst_status, [])
             condition_status_entry.extend(
                 actionable_reasons if worst_status is eligibility.Status.actionable else exclusion_reasons
@@ -124,9 +128,9 @@ def evaluate_priority_group(
         self,
         iteration_rule_group: Iterator[rules.IterationRule],
         worst_status_so_far_for_condition: eligibility.Status,
-    ) -> tuple[eligibility.Status, list[eligibility.Reason], list[eligibility.Reason]]:
+    ) -> tuple[eligibility.Status, list[eligibility.Reason], list[eligibility.Reason], bool]:
+        is_rule_stop = False
         exclusion_reasons, actionable_reasons = [], []
-
         exclude_capable_rules = [
             ir
             for ir in iteration_rule_group
@@ -147,4 +151,6 @@ def evaluate_priority_group(
                 actionable_reasons.append(reason)
 
         worst_group_status = eligibility.Status.worst(best_status, worst_status_so_far_for_condition)
-        return worst_group_status, actionable_reasons, exclusion_reasons
+        if worst_group_status.is_exclusion:
+            is_rule_stop = any(rule.rule_stop for rule in exclude_capable_rules)
+        return worst_group_status, actionable_reasons, exclusion_reasons, is_rule_stop

tests/fixtures/builders/model/rule.py

@@ -22,6 +22,7 @@ class IterationCohortFactory(ModelFactory[rules.IterationCohort]): ...
 class IterationRuleFactory(ModelFactory[rules.IterationRule]):
     attribute_target = None
     cohort_label = None
+    rule_stop = False
 
 
 class IterationFactory(ModelFactory[rules.Iteration]):

tests/unit/services/calculators/test_eligibility_calculator.py

@@ -8,6 +8,7 @@
 from eligibility_signposting_api.model import rules
 from eligibility_signposting_api.model import rules as rules_model
 from eligibility_signposting_api.model.eligibility import ConditionName, DateOfBirth, NHSNumber, Postcode, Status
+from eligibility_signposting_api.model.rules import IterationRule
 from eligibility_signposting_api.services.calculators.eligibility_calculator import EligibilityCalculator
 from tests.fixtures.builders.model import rule as rule_builder
 from tests.fixtures.builders.repos.person import person_rows_builder
@@ -811,3 +812,65 @@ def test_status_if_iteration_rules_contains_cohort_label_field(
         ),
         test_comment,
     )
+
+
+@pytest.mark.parametrize(
+    ("rule_stop", "expected_status", "test_comment"),
+    [
+        ("Y", Status.not_actionable, "Stops at the first rule"),
+        ("N", Status.not_eligible, "Both the rules are executed"),
+        ("", Status.not_eligible, "Both the rules are executed"),
+        (None, Status.not_eligible, "Both the rules are executed"),
+    ],
+)
+def test_rules_stop_behavior(rule_stop: str | None, expected_status: Status, test_comment: str, faker: Faker) -> None:
+    # Given
+    nhs_number = NHSNumber(faker.nhs_number())
+    date_of_birth = DateOfBirth(faker.date_of_birth(minimum_age=18, maximum_age=74))
+    person_rows = person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1"])
+
+    # Base rule template
+    # Not using model factory to create Iteration rules since it sets boolean values for "Y"/"N"
+    simple_age_data = {
+        "Name": "Exclude too young less than 75",
+        "Description": "Exclude too young less than 75",
+        "AttributeLevel": "PERSON",
+        "AttributeName": "DATE_OF_BIRTH",
+        "Operator": "Y>",
+        "Comparator": "-75",
+    }
+
+    # Build rule variations
+    rule_variants = [
+        {"Type": "S", "Priority": 10, "RuleStop": rule_stop},
+        {"Type": "S", "Priority": 10},
+        {"Type": "F", "Priority": 15},
+    ]
+
+    iteration_rules = [IterationRule.model_validate({**simple_age_data, **variant}) for variant in rule_variants]
+
+    # Build campaign configuration
+    campaign_config = rule_builder.CampaignConfigFactory.build(
+        target="RSV",
+        iterations=[
+            rule_builder.IterationFactory.build(
+                iteration_rules=[],
+                iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")],
+            )
+        ],
+    )
+    campaign_config.iterations[0].iteration_rules.extend(iteration_rules)
+
+    calculator = EligibilityCalculator(person_rows, [campaign_config])
+
+    # When
+    actual = calculator.evaluate_eligibility()
+
+    # Then
+    assert_that(
+        actual,
+        is_eligibility_status().with_conditions(
+            has_item(is_condition().with_condition_name(ConditionName("RSV")).and_status(expected_status))
+        ),
+        test_comment,
+    )