Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import logging
from collections import defaultdict
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
Expand Down Expand Up @@ -31,6 +32,8 @@
)
from eligibility_signposting_api.model.person import Person

logger = logging.getLogger(__name__)


@service
class EligibilityCalculatorFactory:
Expand Down Expand Up @@ -84,6 +87,9 @@ def get_eligibility_status(self, include_actions: str, conditions: list[str], ca
for condition_name, campaign_group in requested_grouped_campaigns:
best_iteration_result = self.get_best_iteration_result(campaign_group)

if best_iteration_result is None:
continue

matched_action_detail = self.action_rule_handler.get_actions(
self.person,
best_iteration_result.active_iteration,
Expand All @@ -107,28 +113,31 @@ def get_eligibility_status(self, include_actions: str, conditions: list[str], ca
# Consolidate all the results and return
return eligibility_status.EligibilityStatus(conditions=final_result)

def get_best_iteration_result(self, campaign_group: list[CampaignConfig]) -> BestIterationResult:
def get_best_iteration_result(self, campaign_group: list[CampaignConfig]) -> BestIterationResult | None:
iteration_results = self.get_iteration_results(campaign_group)

if iteration_results:
(best_iteration_name, best_iteration_result) = max(
iteration_results.items(),
key=lambda item: next(iter(item[1].cohort_results.values())).status.value
# Below handles the case where there are no cohort results
if item[1].cohort_results
else -1,
)
else:
iteration_result = IterationResult(eligibility_status.Status.not_eligible, [], [])
best_iteration_result = BestIterationResult(iteration_result, None, None, None, {})
if not iteration_results:
return None

(best_iteration_name, best_iteration_result) = max(
iteration_results.items(),
key=lambda item: next(iter(item[1].cohort_results.values())).status.value
# Below handles the case where there are no cohort results
if item[1].cohort_results
else -1,
)

return best_iteration_result

def get_iteration_results(self, campaign_group: list[CampaignConfig]) -> dict[IterationName, BestIterationResult]:
iteration_results: dict[IterationName, BestIterationResult] = {}

for cc in campaign_group:
active_iteration = cc.current_iteration
try:
active_iteration = cc.current_iteration
except StopIteration:
logger.info("Skipping campaign ID %s as no active iteration was found.", cc.id)
continue
cohort_results: dict[CohortLabel, CohortGroupResult] = self.rule_processor.get_cohort_group_results(
self.person, active_iteration
)
Expand Down
102 changes: 88 additions & 14 deletions tests/unit/services/calculators/test_eligibility_calculator.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,16 @@
import datetime
import logging
from typing import Any

import pytest
from faker import Faker
from flask import Flask
from freezegun import freeze_time
from hamcrest import assert_that, contains_exactly, contains_inanyorder, has_item, has_items, is_, is_in
from pydantic import HttpUrl

from eligibility_signposting_api.model import campaign_config as rules_model
from eligibility_signposting_api.model import eligibility_status
from eligibility_signposting_api.model.campaign_config import (
AvailableAction,
CohortLabel,
Description,
RuleAttributeLevel,
Expand Down Expand Up @@ -585,19 +584,94 @@ def test_cohort_group_descriptions_are_selected_based_on_priority_when_cohorts_h
)


book_nbs_comms = AvailableAction(
ActionType="ButtonAuthLink",
ExternalRoutingCode="BookNBS",
ActionDescription="Action description",
UrlLink=HttpUrl("https://www.nhs.uk/book-rsv"),
UrlLabel="Continue to booking",
)
@freeze_time("2025-04-25")
def test_no_active_iteration_returns_empty_conditions_with_single_active_campaign(faker: Faker):
# Given
person_rows = person_rows_builder(NHSNumber(faker.nhs_number()))
campaign_configs = [
rule_builder.CampaignConfigFactory.build(
target="RSV",
iterations=[
rule_builder.IterationFactory.build(
name="inactive iteration",
iteration_rules=[],
iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")],
)
],
)
]
# Need to set the iteration date to override CampaignConfigFactory.fix_iteration_date_invariants behavior
campaign_configs[0].iterations[0].iteration_date = datetime.date(2025, 5, 10)

default_comms_detail = AvailableAction(
ActionType="CareCardWithText",
ExternalRoutingCode="BookLocal",
ActionDescription="You can get an RSV vaccination at your GP surgery",
)
calculator = EligibilityCalculator(person_rows, campaign_configs)

# When
actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL")

# Then
assert_that(actual, is_eligibility_status().with_conditions([]))


@pytest.mark.usefixtures("caplog")
@freeze_time("2025-04-25")
def test_returns_no_condition_data_for_campaign_without_active_iteration(faker: Faker, caplog):
# Given
person_rows = person_rows_builder(NHSNumber(faker.nhs_number()))
campaign_configs = [
rule_builder.CampaignConfigFactory.build(
target="RSV",
iterations=[
rule_builder.IterationFactory.build(
name="inactive iteration",
iteration_rules=[],
iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")],
)
],
),
rule_builder.CampaignConfigFactory.build(
target="COVID",
iterations=[
rule_builder.IterationFactory.build(
name="active iteration",
iteration_rules=[],
iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")],
)
],
),
]
# Need to set the iteration date to override CampaignConfigFactory.fix_iteration_date_invariants behavior
rsv_campaign = campaign_configs[0]
rsv_campaign.iterations[0].iteration_date = datetime.date(2025, 5, 10)

calculator = EligibilityCalculator(person_rows, campaign_configs)

# When
with caplog.at_level(logging.INFO):
actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL")

# Then
condition_names = [condition.condition_name for condition in actual.conditions]

assert ConditionName("RSV") not in condition_names
assert ConditionName("COVID") in condition_names
assert f"Skipping campaign ID {rsv_campaign.id} as no active iteration was found." in caplog.text


@freeze_time("2025-04-25")
def test_no_active_campaign(faker: Faker):
# Given
person_rows = person_rows_builder(NHSNumber(faker.nhs_number()))
campaign_configs = [rule_builder.CampaignConfigFactory.build()]
# Need to set the campaign dates to override CampaignConfigFactory.fix_iteration_date_invariants behavior
campaign_configs[0].start_date = datetime.date(2025, 5, 10)

calculator = EligibilityCalculator(person_rows, campaign_configs)

# When
actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL")

# Then
assert_that(actual, is_eligibility_status().with_conditions([]))


class TestEligibilityResultBuilder:
Expand Down