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
3 changes: 2 additions & 1 deletion src/eligibility_signposting_api/model/rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
RuleDescription = NewType("RuleDescription", str)
RulePriority = NewType("RulePriority", int)
RuleAttributeName = NewType("RuleAttributeName", str)
RuleAttributeTarget = NewType("RuleAttributeTarget", str)
RuleComparator = NewType("RuleComparator", str)
StartDate = NewType("StartDate", date)
EndDate = NewType("EndDate", date)
Expand Down Expand Up @@ -96,7 +97,7 @@ class IterationRule(BaseModel):
cohort_label: CohortLabel | None = Field(None, alias="CohortLabel")
operator: RuleOperator = Field(..., alias="Operator")
comparator: RuleComparator = Field(..., alias="Comparator")
attribute_target: str | None = Field(None, alias="AttributeTarget")
attribute_target: RuleAttributeTarget | None = Field(None, alias="AttributeTarget")

model_config = {"populate_by_name": True, "extra": "ignore"}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,11 @@ def get_attribute_value(self) -> str | None:
(r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == "PERSON"), None
)
attribute_value = person.get(self.rule.attribute_name) if person else None
case rules.RuleAttributeLevel.TARGET:
target: Mapping[str, str | None] | None = next(
(r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == self.rule.attribute_target), None
)
attribute_value = target.get(self.rule.attribute_name) if target else None
case _: # pragma: no cover
msg = f"{self.rule.attribute_level} not implemented"
raise NotImplementedError(msg)
Expand Down
4 changes: 3 additions & 1 deletion tests/fixtures/builders/repos/person.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,9 @@ def person_rows_builder( # noqa:PLR0913
{
"NHS_NUMBER": key,
"ATTRIBUTE_TYPE": vaccine,
"LAST_SUCCESSFUL_DATE": last_successful_date.strftime("%Y%m%d"),
"LAST_SUCCESSFUL_DATE": (
last_successful_date.strftime("%Y%m%d") if last_successful_date else last_successful_date
),
"OPTOUT": choice(["Y", "N"]),
"LAST_INVITE_DATE": faker.past_date("-5y").strftime("%Y%m%d"),
}
Expand Down
70 changes: 70 additions & 0 deletions tests/unit/services/calculators/test_eligibility_calculator.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from freezegun import freeze_time
from hamcrest import assert_that, contains_exactly, empty, has_item, has_items

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.services.calculators.eligibility_calculator import EligibilityCalculator
Expand Down Expand Up @@ -616,3 +617,72 @@ def test_base_eligible_and_icb_example(
has_item(is_condition().with_condition_name(ConditionName("RSV")).and_status(expected_status))
),
)


@pytest.mark.parametrize(
("last_successful_date", "expected_status", "test_comment"),
[
("20240101", Status.not_actionable, "last_successful_date is a past date"),
("20250101", Status.not_actionable, "last_successful_date is today"),
# Below is a non-ideal situation (might be due to a data entry error), so considered as actionable.
("20260101", Status.actionable, "last_successful_date is a future date"),
("", Status.actionable, "last_successful_date is empty"),
(None, Status.actionable, "last_successful_date is none"),
],
)
@freeze_time("2025-01-01")
def test_not_actionable_status_on_target_when_last_successful_date_lte_today(
last_successful_date, expected_status, test_comment, faker: Faker
):
# Given
nhs_number = NHSNumber(faker.nhs_number())

target_rows = person_rows_builder(
nhs_number,
cohorts=["cohort1"],
vaccines=[
(
"RSV",
datetime.datetime.strptime(last_successful_date, "%Y%m%d").replace(tzinfo=datetime.UTC)
if last_successful_date
else None,
)
],
)

campaign_configs = [
rule_builder.CampaignConfigFactory.build(
target="RSV",
iterations=[
rule_builder.IterationFactory.build(
iteration_rules=[
rule_builder.IterationRuleFactory.build(
type=rules.RuleType.suppression,
name=rules.RuleName("You have already been vaccinated against RSV"),
description=rules.RuleDescription("Exclude anyone Completed RSV Vaccination"),
operator=rules.RuleOperator.day_lte,
attribute_level=rules.RuleAttributeLevel.TARGET,
attribute_name=rules.RuleAttributeName("LAST_SUCCESSFUL_DATE"),
comparator=rules.RuleComparator("0"),
attribute_target=rules.RuleAttributeTarget("RSV"),
)
],
iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")],
)
],
)
]

calculator = EligibilityCalculator(target_rows, campaign_configs)

# 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,
)
Loading