Merge pull request #153 from NHSDigital/fix/magic_cohorts

fix/magic cohorts

Author: robbailiff2

src/eligibility_signposting_api/model/rules.py

@@ -95,6 +95,10 @@ class IterationCohort(BaseModel):
 
     model_config = {"populate_by_name": True, "extra": "ignore"}
 
+    @cached_property
+    def is_magic_cohort(self) -> bool:
+        return self.cohort_label.upper() == MAGIC_COHORT_LABEL.upper()
+
 
 class IterationRule(BaseModel):
     type: RuleType = Field(..., alias="Type")
@@ -144,17 +148,6 @@ def parse_dates(cls, v: str | date) -> date:
     def serialize_dates(v: date, _info: SerializationInfo) -> str:
         return v.strftime("%Y%m%d")
 
-    @cached_property
-    def has_magic_cohort(self) -> bool:
-        return next(
-            (
-                True
-                for cc in self.iteration_cohorts
-                if cc.cohort_label and cc.cohort_label.upper() == MAGIC_COHORT_LABEL.upper()
-            ),
-            False,
-        )
-
 
 class CampaignConfig(BaseModel):
     id: CampaignID = Field(..., alias="ID")

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -116,26 +116,7 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
             iteration_results: dict[str, IterationResult] = {}
 
             for active_iteration in [cc.current_iteration for cc in campaign_group]:
-                cohort_results: dict[str, CohortGroupResult] = {}
-
-                filter_rules, suppression_rules = self.get_rules_by_type(active_iteration)
-
-                for cohort in sorted(active_iteration.iteration_cohorts, key=attrgetter("priority")):
-                    # Base Eligibility - check
-                    if cohort.cohort_label in self.person_cohorts or active_iteration.has_magic_cohort:
-                        # Eligibility - check
-                        if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
-                            # Actionability - evaluation
-                            self.evaluate_suppression_rules(cohort, cohort_results, suppression_rules)
-
-                    # Not base eligible
-                    elif cohort.cohort_label is not None:
-                        cohort_results[cohort.cohort_label] = CohortGroupResult(
-                            (cohort.cohort_group),
-                            Status.not_eligible,
-                            [],
-                            cohort.negative_description,
-                        )
+                cohort_results: dict[str, CohortGroupResult] = self.get_cohort_results(active_iteration)
 
                 # Determine Result between cohorts - get the best
                 status, best_cohorts = self.get_the_best_cohort_memberships(cohort_results)
@@ -153,6 +134,27 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
         final_result = self.build_condition_results(condition_results)
         return eligibility.EligibilityStatus(conditions=final_result)
 
+    def get_cohort_results(self, active_iteration: rules.Iteration) -> dict[str, CohortGroupResult]:
+        cohort_results: dict[str, CohortGroupResult] = {}
+        filter_rules, suppression_rules = self.get_rules_by_type(active_iteration)
+        for cohort in sorted(active_iteration.iteration_cohorts, key=attrgetter("priority")):
+            # Base Eligibility - check
+            if cohort.cohort_label in self.person_cohorts or cohort.is_magic_cohort:
+                # Eligibility - check
+                if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
+                    # Actionability - evaluation
+                    self.evaluate_suppression_rules(cohort, cohort_results, suppression_rules)
+
+            # Not base eligible
+            elif cohort.cohort_label is not None:
+                cohort_results[cohort.cohort_label] = CohortGroupResult(
+                    (cohort.cohort_group),
+                    Status.not_eligible,
+                    [],
+                    cohort.negative_description,
+                )
+        return cohort_results
+
     @staticmethod
     def build_condition_results(
         condition_results: dict[ConditionName, IterationResult],

tests/unit/services/calculators/test_eligibility_calculator.py

@@ -1042,10 +1042,19 @@ def test_eligibility_results_when_multiple_cohorts(
                 ("magic cohort group", "magic negative description"),
                 ("rsv_age_range", "rsv_age_range negative description"),
             ],
-            "all the cohorts are not-eligible",
+            "rsv_75_rolling is not base-eligible & magic cohort group not eligible by F rules ",
         ),
         (
-            person_rows_builder(nhs_number="123", cohorts=[], postcode="SW19", de=False, icb="QE1"),
+            person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling"], postcode="AC01", de=True, icb="QE1"),
+            Status.not_eligible,
+            [
+                ("magic cohort group", "magic negative description"),
+                ("rsv_age_range", "rsv_age_range negative description"),
+            ],
+            "all the cohorts are not-eligible by F rules",
+        ),
+        (
+            person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling"], postcode="SW19", de=False, icb="QE1"),
             Status.not_actionable,
             [
                 ("magic cohort group", "magic positive description"),
@@ -1054,7 +1063,7 @@ def test_eligibility_results_when_multiple_cohorts(
             "all the cohorts are not-actionable",
         ),
         (
-            person_rows_builder(nhs_number="123", cohorts=[], postcode="AC01", de=False, icb="QE1"),
+            person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling"], postcode="AC01", de=False, icb="QE1"),
             Status.actionable,
             [
                 ("magic cohort group", "magic positive description"),
@@ -1063,20 +1072,20 @@ def test_eligibility_results_when_multiple_cohorts(
             "all the cohorts are actionable",
         ),
         (
-            person_rows_builder(nhs_number="123", cohorts=[], postcode="AC01", de=False, icb="NOT_QE1"),
+            person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling"], postcode="AC01", de=False, icb="NOT_QE1"),
             Status.actionable,
             [("magic cohort group", "magic positive description")],
             "magic_cohort is actionable, but not others",
         ),
         (
-            person_rows_builder(nhs_number="123", cohorts=[], postcode="SW19", de=False, icb="NOT_QE1"),
+            person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling"], postcode="SW19", de=False, icb="NOT_QE1"),
             Status.not_actionable,
             [("magic cohort group", "magic positive description")],
             "magic_cohort is not-actionable, but others are not eligible",
         ),
     ],
 )
-def test_cohort_groups_and_their_descriptions_when_magic_cohort_is_having_the_best_status(
+def test_cohort_groups_and_their_descriptions_when_magic_cohort_is_present(
     person_rows: list[dict[str, Any]],
     expected_status: str,
     expected_cohort_group_and_description: list[tuple[str, str]],
@@ -1089,19 +1098,20 @@ def test_cohort_groups_and_their_descriptions_when_magic_cohort_is_having_the_be
             iterations=[
                 rule_builder.IterationFactory.build(
                     iteration_cohorts=[
-                        rule_builder.MagicCohortFactory.build(),
                         rule_builder.Rsv75RollingCohortFactory.build(),
+                        rule_builder.MagicCohortFactory.build(),
                     ],
                     iteration_rules=[
-                        # common rules
+                        # F common rule
                         rule_builder.DetainedEstateSuppressionRuleFactory.build(type=rules.RuleType.filter),
+                        # F rules for rsv_75_rolling
+                        rule_builder.ICBFilterRuleFactory.build(
+                            type=rules.RuleType.filter, cohort_label=rules.CohortLabel("rsv_75_rolling")
+                        ),
+                        # S common rule
                         rule_builder.PostcodeSuppressionRuleFactory.build(
                             comparator=rules.RuleComparator("SW19"),
                         ),
-                        # rules for specific cohorts
-                        rule_builder.ICBFilterRuleFactory.build(
-                            cohort_label=rules.CohortLabel("rsv_75_rolling"),
-                        ),
                     ],
                 )
             ],