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
43 changes: 42 additions & 1 deletion tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
import string
from random import choice, randint

import pytest
from faker import Faker
from faker.providers import BaseProvider
from flask import Flask
from flask.testing import FlaskClient

Expand All @@ -18,4 +22,41 @@ def client(app) -> FlaskClient:

@pytest.fixture(scope="session")
def faker() -> Faker:
return Faker("en_UK")
faker = Faker("en_UK")
faker.add_provider(PersonDetailProvider)
return faker


class PersonDetailProvider(BaseProvider):
def nhs_number(self) -> str:
return f"5{randint(1, 999999999):09}"

def icb(self) -> str | None:
if randint(0, 3):
return f"{choice(string.ascii_uppercase)}{choice(string.ascii_uppercase)}{choice(string.digits)}"
return None

def gp_practice(self) -> str | None:
if randint(0, 3):
return f"{choice(string.ascii_uppercase)}{randint(1, 99999):05}"
return None

def pcn(self) -> str | None:
if randint(0, 3):
return f"{choice(string.ascii_uppercase)}{randint(1, 99999):05}"
return None

def comissioning_region(self) -> str | None:
if randint(0, 3):
return f"{choice(string.ascii_uppercase)}{randint(1, 99):02}"
return None

def msoa(self) -> str | None:
if randint(0, 3):
return f"{choice(string.ascii_uppercase)}{randint(1, 99999999):08}"
return None

def lsoa(self) -> str | None:
if randint(0, 3):
return f"{choice(string.ascii_uppercase)}{randint(1, 99999999):08}"
return None
11 changes: 11 additions & 0 deletions tests/fixtures/builders/model/rule.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,3 +54,14 @@ class PostcodeSuppressionRuleFactory(IterationRuleFactory):
attribute_level = rules.RuleAttributeLevel.PERSON
attribute_name = rules.RuleAttributeName("POSTCODE")
comparator = rules.RuleComparator("SW19")


class ICBSuppressionRuleFactory(IterationRuleFactory):
type = rules.RuleType.filter
name = rules.RuleName("Not in QE1")
description = rules.RuleDescription("Not in QE1")
priority = rules.RulePriority(10)
operator = rules.RuleOperator.ne
attribute_level = rules.RuleAttributeLevel.PERSON
attribute_name = rules.RuleAttributeName("ICB")
comparator = rules.RuleComparator("QE1")
69 changes: 52 additions & 17 deletions tests/fixtures/builders/repos/person.py
Original file line number Diff line number Diff line change
@@ -1,33 +1,67 @@
import random
from collections.abc import Sequence
from typing import Any
from datetime import date
from random import choice, shuffle
from typing import Any, Literal, get_args

from faker import Faker

from eligibility_signposting_api.model import eligibility
from tests.conftest import PersonDetailProvider

Gender = Literal["0", "1", "2", "9"] # 0 - Not known, 1- Male, 2 - Female, 9 - Not specified. I know, right?

def person_rows_builder(
nhs_number: eligibility.NHSNumber,

def person_rows_builder( # noqa:PLR0913
nhs_number: str,
*,
date_of_birth: eligibility.DateOfBirth | None = None,
postcode: eligibility.Postcode | None = None,
cohorts: Sequence[str] | None = None,
vaccines: Sequence[str] | None = None,
date_of_birth: date | None = ...,
gender: Gender | None = ...,
postcode: str | None = ...,
cohorts: Sequence[str] | None = ...,
vaccines: Sequence[tuple[str, date]] | None = ...,
icb: str | None = ...,
gp_practice: str | None = ...,
pcn: str | None = ...,
comissioning_region: str | None = ...,
thirteen_q: bool | None = ...,
care_home: bool | None = ...,
de: bool | None = ...,
msoa: str | None = ...,
lsoa: str | None = ...,
) -> list[dict[str, Any]]:
faker = Faker("en_UK")
faker.add_provider(PersonDetailProvider)

key = f"PERSON#{nhs_number}"
date_of_birth = date_of_birth or eligibility.DateOfBirth(faker.date_of_birth(minimum_age=18, maximum_age=99))
postcode = postcode or eligibility.Postcode(faker.postcode())
cohorts = cohorts if cohorts is not None else ["cohort-a", "cohort-b"]
vaccines = vaccines if vaccines is not None else ["RSV", "COVID"]
date_of_birth = date_of_birth if date_of_birth is not ... else faker.date_of_birth(minimum_age=18, maximum_age=99)
gender = gender if gender is not ... else choice(get_args(Gender))
postcode = postcode if postcode is not ... else faker.postcode()
cohorts = cohorts if cohorts is not ... else ["cohort-a", "cohort-b"]
vaccines = vaccines if vaccines is not ... else [("RSV", faker.past_date("-5y")), ("COVID", faker.past_date("-5y"))]
icb = icb if icb is not ... else faker.icb()
gp_practice = gp_practice if gp_practice is not ... else faker.gp_practice()
pcn = pcn if pcn is not ... else faker.pcn()
comissioning_region = comissioning_region if comissioning_region is not ... else faker.comissioning_region()
thirteen_q = thirteen_q if thirteen_q is not ... else faker.boolean()
care_home = care_home if care_home is not ... else faker.boolean()
de = de if de is not ... else faker.boolean()
msoa = msoa if msoa is not ... else faker.msoa()
lsoa = lsoa if lsoa is not ... else faker.lsoa()
rows: list[dict[str, Any]] = [
{
"NHS_NUMBER": key,
"ATTRIBUTE_TYPE": "PERSON",
"DATE_OF_BIRTH": date_of_birth.strftime("%Y%m%d"),
"GENDER": gender,
"POSTCODE": postcode,
"ICB": icb,
"GP_PRACTICE": gp_practice,
"PCN": pcn,
"COMISSIONING_REGION": comissioning_region,
"13Q_FLAG": "Y" if thirteen_q else "N",
"CARE_HOME_FLAG": "Y" if care_home else "N",
"DE_FLAG": "Y" if de else "N",
"MSOA": msoa,
"LSOA": lsoa,
},
{
"NHS_NUMBER": key,
Expand All @@ -45,11 +79,12 @@ def person_rows_builder(
{
"NHS_NUMBER": key,
"ATTRIBUTE_TYPE": vaccine,
"LAST_SUCCESSFUL_DATE": faker.past_date().strftime("%Y%m%d"),
"OPTOUT": random.choice(["Y", "N"]),
"LAST_INVITE_DATE": faker.past_date().strftime("%Y%m%d"),
"LAST_SUCCESSFUL_DATE": last_successful_date.strftime("%Y%m%d"),
"OPTOUT": choice(["Y", "N"]),
"LAST_INVITE_DATE": faker.past_date("-5y").strftime("%Y%m%d"),
}
for vaccine in vaccines
for vaccine, last_successful_date in vaccines
)

shuffle(rows)
return rows
4 changes: 2 additions & 2 deletions tests/integration/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,7 @@ def person_table(dynamodb_resource: ServiceResource) -> Generator[Any]:

@pytest.fixture
def persisted_person(person_table: Any, faker: Faker) -> Generator[eligibility.NHSNumber]:
nhs_number = eligibility.NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = eligibility.NHSNumber(faker.nhs_number())
date_of_birth = eligibility.DateOfBirth(faker.date_of_birth(minimum_age=18, maximum_age=65))

for row in (rows := person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1"])):
Expand All @@ -229,7 +229,7 @@ def persisted_person(person_table: Any, faker: Faker) -> Generator[eligibility.N

@pytest.fixture
def persisted_77yo_person(person_table: Any, faker: Faker) -> Generator[eligibility.NHSNumber]:
nhs_number = eligibility.NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = eligibility.NHSNumber(faker.nhs_number())
date_of_birth = eligibility.DateOfBirth(faker.date_of_birth(minimum_age=77, maximum_age=77))

for row in (rows := person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1", "cohort2"])):
Expand Down
2 changes: 1 addition & 1 deletion tests/integration/lambda/test_app_running_as_lambda.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ def test_install_and_call_flask_lambda_with_unknown_nhs_number(
):
"""Given lambda installed into localstack, run it via http, with a nonexistent NHS number specified"""
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())

# When
response = httpx.get(str(flask_function_url / "eligibility" / nhs_number))
Expand Down
2 changes: 1 addition & 1 deletion tests/integration/repo/test_person_repo.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ def test_person_found(person_table: Any, persisted_person: NHSNumber):

def test_person_not_found(person_table: Any, faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())
repo = PersonRepo(person_table)

# When, Then
Expand Down
70 changes: 58 additions & 12 deletions tests/unit/services/calculators/test_eligibility_calculator.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

def test_not_base_eligible(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())

person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"])
campaign_configs = [
Expand Down Expand Up @@ -48,7 +48,7 @@ def test_not_base_eligible(faker: Faker):
@freeze_time("2025-04-25")
def test_only_live_campaigns_considered(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())

person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"])
campaign_configs = [
Expand Down Expand Up @@ -95,7 +95,7 @@ def test_only_live_campaigns_considered(faker: Faker):

def test_base_eligible_and_simple_rule_includes(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())
date_of_birth = DateOfBirth(faker.date_of_birth(minimum_age=76, maximum_age=79))

person_rows = person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1"])
Expand Down Expand Up @@ -127,7 +127,7 @@ def test_base_eligible_and_simple_rule_includes(faker: Faker):

def test_base_eligible_but_simple_rule_excludes(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
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"])
Expand Down Expand Up @@ -160,7 +160,7 @@ def test_base_eligible_but_simple_rule_excludes(faker: Faker):
@freeze_time("2025-04-25")
def test_simple_rule_only_excludes_from_live_iteration(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())
date_of_birth = DateOfBirth(faker.date_of_birth(minimum_age=66, maximum_age=74))

person_rows = person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1"])
Expand Down Expand Up @@ -207,7 +207,7 @@ def test_simple_rule_only_excludes_from_live_iteration(faker: Faker):
@freeze_time("2025-04-25")
def test_campaign_with_no_active_iteration_not_considered(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())

person_rows = person_rows_builder(nhs_number)
campaign_configs = [
Expand Down Expand Up @@ -240,7 +240,7 @@ def test_campaign_with_no_active_iteration_not_considered(faker: Faker):
)
def test_rule_types_cause_correct_statuses(rule_type: rules_model.RuleType, expected_status: Status, faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
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"])
Expand Down Expand Up @@ -272,7 +272,7 @@ def test_rule_types_cause_correct_statuses(rule_type: rules_model.RuleType, expe

def test_multiple_rule_types_cause_correct_status(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
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"])
Expand Down Expand Up @@ -369,7 +369,7 @@ def test_rules_with_same_priority_must_all_match_to_exclude(
faker: Faker,
):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())
date_of_birth = DateOfBirth(faker.date_of_birth(minimum_age=66, maximum_age=74))

person_rows = person_rows_builder(
Expand Down Expand Up @@ -404,7 +404,7 @@ def test_rules_with_same_priority_must_all_match_to_exclude(

def test_multiple_conditions_where_both_are_actionable(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())
date_of_birth = DateOfBirth(faker.date_of_birth(minimum_age=76, maximum_age=78))

person_rows = person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1"])
Expand Down Expand Up @@ -448,7 +448,7 @@ def test_multiple_conditions_where_both_are_actionable(faker: Faker):

def test_multiple_conditions_where_all_give_unique_statuses(faker: Faker):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())
date_of_birth = DateOfBirth(faker.date_of_birth(minimum_age=76, maximum_age=78))

person_rows = person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1"])
Expand Down Expand Up @@ -551,7 +551,7 @@ def test_multiple_campaigns_for_single_condition(
test_comment: str, campaign1: rules_model.CampaignConfig, campaign2: rules_model.CampaignConfig, faker: Faker
):
# Given
nhs_number = NHSNumber(f"5{faker.random_int(max=999999999):09d}")
nhs_number = NHSNumber(faker.nhs_number())
date_of_birth = DateOfBirth(faker.date_of_birth(minimum_age=76, maximum_age=78))

person_rows = person_rows_builder(nhs_number, date_of_birth=date_of_birth, cohorts=["cohort1"])
Expand All @@ -570,3 +570,49 @@ def test_multiple_campaigns_for_single_condition(
),
test_comment,
)


@pytest.mark.parametrize(
("icb", "rule_type", "expected_status"),
[
("QE1", rules_model.RuleType.suppression, Status.actionable),
("QWU", rules_model.RuleType.suppression, Status.not_actionable),
("", rules_model.RuleType.suppression, Status.not_actionable),
(None, rules_model.RuleType.suppression, Status.not_actionable),
("QE1", rules_model.RuleType.filter, Status.actionable),
("QWU", rules_model.RuleType.filter, Status.not_eligible),
("", rules_model.RuleType.filter, Status.not_eligible),
(None, rules_model.RuleType.filter, Status.not_eligible),
],
)
def test_base_eligible_and_icb_example(
icb: str | None, rule_type: rules_model.RuleType, expected_status: Status, faker: Faker
):
# Given
nhs_number = NHSNumber(faker.nhs_number())

person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb=icb)
campaign_configs = [
rule_builder.CampaignConfigFactory.build(
target="RSV",
iterations=[
rule_builder.IterationFactory.build(
iteration_rules=[rule_builder.ICBSuppressionRuleFactory.build(type=rule_type)],
iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")],
)
],
)
]

calculator = EligibilityCalculator(person_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))
),
)
Loading