refer status of magic cohorts based on other cohorts info

Karthikeyannhs · Karthikeyannhs · commit 0a34d5340d95 · 2025-06-10T21:19:24.000+01:00
diff --git a/src/eligibility_signposting_api/app.py b/src/eligibility_signposting_api/app.py
@@ -8,7 +8,7 @@
 from mangum.types import LambdaContext, LambdaEvent
 
 from eligibility_signposting_api import repos, services
-from eligibility_signposting_api.config import config, init_logging
+from eligibility_signposting_api.config.config import config, init_logging
 from eligibility_signposting_api.error_handler import handle_exception
 from eligibility_signposting_api.views import eligibility_blueprint
 
diff --git a/src/eligibility_signposting_api/config/__init__.py b/src/eligibility_signposting_api/config/__init__.py
diff --git a/src/eligibility_signposting_api/config/config.py b/src/eligibility_signposting_api/config/config.py
diff --git a/src/eligibility_signposting_api/config/contants.py b/src/eligibility_signposting_api/config/contants.py
@@ -0,0 +1 @@
+MAGIC_COHORT_LABEL = "elid_all_people"
diff --git a/src/eligibility_signposting_api/model/rules.py b/src/eligibility_signposting_api/model/rules.py
@@ -10,6 +10,8 @@
 
 from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator
 
+from eligibility_signposting_api.config.contants import MAGIC_COHORT_LABEL
+
 if typing.TYPE_CHECKING:  # pragma: no cover
     from pydantic import SerializationInfo
 
@@ -107,14 +109,14 @@ class IterationRule(BaseModel):
     attribute_target: RuleAttributeTarget | None = Field(None, alias="AttributeTarget")
     rule_stop: RuleStop = Field(RuleStop(False), alias="RuleStop")  # noqa: FBT003
 
+    model_config = {"populate_by_name": True, "extra": "ignore"}
+
     @field_validator("rule_stop", mode="before")
     def parse_yn_to_bool(cls, v: str | bool) -> bool:  # noqa: N805
         if isinstance(v, str):
             return v.upper() == "Y"
         return v
 
-    model_config = {"populate_by_name": True, "extra": "ignore"}
-
 
 class Iteration(BaseModel):
     id: IterationID = Field(..., alias="ID")
@@ -142,6 +144,17 @@ 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")
diff --git a/src/eligibility_signposting_api/repos/factory.py b/src/eligibility_signposting_api/repos/factory.py
@@ -7,7 +7,7 @@
 from wireup import Inject, service
 from yarl import URL
 
-from eligibility_signposting_api.config import AwsAccessKey, AwsRegion, AwsSecretAccessKey
+from eligibility_signposting_api.config.config import AwsAccessKey, AwsRegion, AwsSecretAccessKey
 
 logger = logging.getLogger(__name__)
 
diff --git a/src/eligibility_signposting_api/services/calculators/eligibility_calculator.py b/src/eligibility_signposting_api/services/calculators/eligibility_calculator.py
@@ -7,6 +7,8 @@
 from itertools import groupby
 from typing import TYPE_CHECKING, Any
 
+from eligibility_signposting_api.config.contants import MAGIC_COHORT_LABEL
+
 if TYPE_CHECKING:
     from eligibility_signposting_api.model.rules import Iteration, IterationCohort
 
@@ -23,7 +25,6 @@
 from eligibility_signposting_api.services.calculators.rule_calculator import RuleCalculator
 
 Row = Collection[Mapping[str, Any]]
-magic_cohort = "elid_all_people"
 
 
 @service
@@ -62,11 +63,31 @@ def person_cohorts(self) -> set[str]:
         return set(cohorts_row.get("COHORT_MAP", {}).get("cohorts", {}).get("M", {}).keys())
 
     @staticmethod
-    def get_best_cohort(cohort_results: dict[str, CohortResult]) -> tuple[Status, list[CohortResult]]:
+    def get_the_best_cohort_memberships(cohort_results: dict[str, CohortResult]) -> tuple[Status, list[CohortResult]]:
+        """
+        1. Get all the cohorts with the best status
+        2. Case 1: Ignore magic cohort if other cohorts have better status
+           Case 2: If no cohorts have the better status than magic cohort,
+                    then response excludes cohort memberships but will have actions/suitability rules
+                    also, excludes the cohorts with no positive or negative description
+        """
         if not cohort_results:
             return eligibility.Status.not_eligible, []
         best_status = eligibility.Status.best(*[result.status for result in cohort_results.values()])
         best_cohorts = [result for result in cohort_results.values() if result.status == best_status]
+        if all(cc.cohort_code and str(cc.cohort_code.upper()) == MAGIC_COHORT_LABEL.upper() for cc in best_cohorts):
+            # Update the magic cohort to have no cohort membership information
+            best_cohorts = [
+                CohortResult(cohort_code="", status=best_status, reasons=best_cohorts[0].reasons, description="")
+            ]
+        else:
+            best_cohorts = [
+                cc
+                for cc in best_cohorts
+                if str(cc.cohort_code.upper()) != MAGIC_COHORT_LABEL.upper()
+                and cc.description
+                and cc.cohort_code.strip()
+            ]
         return best_status, best_cohorts
 
     @staticmethod
@@ -102,9 +123,10 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
                 cohort_results: dict[str, CohortResult] = {}
 
                 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.cohort_label == magic_cohort:
+                    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
@@ -120,7 +142,8 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
                         )
 
                 # Determine Result between cohorts - get the best
-                status, best_cohorts = self.get_best_cohort(cohort_results)
+                status, best_cohorts = self.get_the_best_cohort_memberships(cohort_results)
+
                 iteration_results[active_iteration.name] = IterationResult(status, best_cohorts)
 
             # Determine results between iterations - get the best
diff --git a/tests/fixtures/builders/model/rule.py b/tests/fixtures/builders/model/rule.py
@@ -73,6 +73,10 @@ def fix_iteration_date_invariants(iterations: list[rules.Iteration], start_date:
 
 
 # Iteration cohort factories
+class MagicCohortFactory(IterationCohortFactory):
+    cohort_label = rules.CohortLabel("elid_all_people")
+
+
 class Rsv75RollingCohortFactory(IterationCohortFactory):
     cohort_label = rules.CohortLabel("rsv_75_rolling")
     cohort_group = rules.CohortGroup("rsv_age_range")
@@ -120,7 +124,18 @@ class PostcodeSuppressionRuleFactory(IterationRuleFactory):
     comparator = rules.RuleComparator("SW19")
 
 
-class ICBSuppressionRuleFactory(IterationRuleFactory):
+class DetainedEstateSuppressionRuleFactory(IterationRuleFactory):
+    type = rules.RuleType.suppression
+    name = rules.RuleName("Detained - Suppress Individuals In Detained Estates")
+    description = rules.RuleDescription("Suppress where individual is identified as being in a Detained Estate")
+    priority = rules.RulePriority(160)
+    attribute_level = rules.RuleAttributeLevel.PERSON
+    attribute_name = rules.RuleAttributeName("DE_FLAG")
+    operator = rules.RuleOperator.equals
+    comparator = rules.RuleComparator("Y")
+
+
+class ICBFilterRuleFactory(IterationRuleFactory):
     type = rules.RuleType.filter
     name = rules.RuleName("Not in QE1")
     description = rules.RuleDescription("Not in QE1")
diff --git a/tests/unit/services/calculators/test_eligibility_calculator.py b/tests/unit/services/calculators/test_eligibility_calculator.py
@@ -631,7 +631,7 @@ def test_base_eligible_and_icb_example(
             target="RSV",
             iterations=[
                 rule_builder.IterationFactory.build(
-                    iteration_rules=[rule_builder.ICBSuppressionRuleFactory.build(type=rule_type)],
+                    iteration_rules=[rule_builder.ICBFilterRuleFactory.build(type=rule_type)],
                     iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")],
                 )
             ],
@@ -1301,3 +1301,102 @@ def test_grouped_description_if_the_cohorts_in_group_have_different_descriptions
         ),
         test_comment,
     )
+
+
+@pytest.mark.parametrize(
+    ("person_rows", "expected_status", "expected_description", "expected_cohort_code", "test_comment"),
+    [
+        (
+            person_rows_builder(nhs_number="123", cohorts=[], postcode="AC01", de=True, icb="QE1"),
+            Status.not_eligible,
+            "rsv_age_range negative description",
+            "rsv_age_range",
+            "all the cohorts are not-eligible, so the result will not include magic cohort, but all other cohorts",
+        ),
+        (
+            person_rows_builder(nhs_number="123", cohorts=[], postcode="SW19", de=False, icb="QE1"),
+            Status.not_actionable,
+            "rsv_age_range positive description",
+            "rsv_age_range",
+            "all the cohorts are not-actionable, so the result will not include magic cohort, but all other cohorts",
+        ),
+        (
+            person_rows_builder(nhs_number="123", cohorts=[], postcode="AC01", de=False, icb="QE1"),
+            Status.actionable,
+            "rsv_age_range positive description",
+            "rsv_age_range",
+            "all the cohorts are actionable, so the result will not include magic cohort, but all other cohorts",
+        ),
+        (
+            person_rows_builder(nhs_number="123", cohorts=[], postcode="AC01", de=False, icb="NOT_QE1"),
+            Status.actionable,
+            "",
+            "",
+            "magic_cohort is actionable, but not other, so removed cohort membership info from result",
+        ),
+        (
+            person_rows_builder(nhs_number="123", cohorts=[], postcode="SW19", de=False, icb="NOT_QE1"),
+            Status.not_actionable,
+            "",
+            "",
+            "magic_cohort is not-actionable,but others are not eligible,so removed cohort membership info from result",
+        ),
+    ],
+)
+def test_grouped_description_if_magic_cohort_has_same_status_as_other_cohorts(
+    person_rows: list[dict[str, Any]],
+    expected_status: str,
+    expected_description: str,
+    expected_cohort_code: str,
+    test_comment: str,
+):
+    # Given
+    campaign_configs = [
+        rule_builder.CampaignConfigFactory.build(
+            target="RSV",
+            iterations=[
+                rule_builder.IterationFactory.build(
+                    iteration_cohorts=[
+                        rule_builder.MagicCohortFactory.build(),
+                        rule_builder.Rsv75RollingCohortFactory.build(),
+                    ],
+                    iteration_rules=[
+                        # common rules
+                        rule_builder.DetainedEstateSuppressionRuleFactory.build(type=rules.RuleType.filter),
+                        rule_builder.PostcodeSuppressionRuleFactory.build(
+                            comparator=rules.RuleComparator("SW19"),
+                        ),
+                        # rules for specific cohorts
+                        rule_builder.ICBFilterRuleFactory.build(
+                            cohort_label=rules.CohortLabel("rsv_75_rolling"),
+                        ),
+                    ],
+                )
+            ],
+        )
+    ]
+
+    calculator = EligibilityCalculator(person_rows, campaign_configs)
+
+    # When
+    actual = calculator.evaluate_eligibility()
+
+    # Then
+    assert_that(
+        actual,
+        is_eligibility_status().with_conditions(
+            has_items(
+                is_condition()
+                .with_condition_name(ConditionName("RSV"))
+                .and_cohort_results(
+                    contains_exactly(
+                        is_cohort_result()
+                        .with_status(expected_status)
+                        .and_cohort_code(expected_cohort_code)
+                        .and_description(expected_description)
+                    )
+                )
+            )
+        ),
+        test_comment,
+    )
diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py
@@ -3,7 +3,7 @@
 import pytest
 from yarl import URL
 
-from eligibility_signposting_api.config import LOG_LEVEL, AwsAccessKey, AwsRegion, AwsSecretAccessKey, config
+from eligibility_signposting_api.config.config import LOG_LEVEL, AwsAccessKey, AwsRegion, AwsSecretAccessKey, config
 from eligibility_signposting_api.repos.campaign_repo import BucketName
 from eligibility_signposting_api.repos.person_repo import TableName
 
