diff --git a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_006.json b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_006.json index 728250475..f1101ccc4 100644 --- a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_006.json +++ b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_006.json @@ -39,7 +39,7 @@ { "NHS_NUMBER": "5000000006", "ATTRIBUTE_TYPE": "RSV", - "LAST_SUCCESSFUL_DATE": "<>" + "LAST_SUCCESSFUL_DATE": "<>" } ] } diff --git a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_007.json b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_007.json index fc9d4afb0..3c9c46c5e 100644 --- a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_007.json +++ b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_007.json @@ -39,7 +39,7 @@ { "NHS_NUMBER": "5000000007", "ATTRIBUTE_TYPE": "RSV", - "LAST_SUCCESSFUL_DATE": "<>" + "LAST_SUCCESSFUL_DATE": "<>" } ] } diff --git a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_012.json b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_012.json index 922748b12..b7ff528f2 100644 --- a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_012.json +++ b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_012.json @@ -29,7 +29,7 @@ { "NHS_NUMBER": "5000000012", "ATTRIBUTE_TYPE": "RSV", - "LAST_SUCCESSFUL_DATE": "<>" + "LAST_SUCCESSFUL_DATE": "<>" } ] } diff --git a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_013.json b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_013.json index 2381f3b6f..c0f69b5dc 100644 --- a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_013.json +++ b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_013.json @@ -39,7 +39,7 @@ { "NHS_NUMBER": "50000000014", "ATTRIBUTE_TYPE": "RSV", - "LAST_SUCCESSFUL_DATE": "<>" + "LAST_SUCCESSFUL_DATE": "<>" } ] } diff --git a/tests/e2e/data/generate_dynamo_data.py b/tests/e2e/data/generate_dynamo_data.py new file mode 100644 index 000000000..ea4e8f599 --- /dev/null +++ b/tests/e2e/data/generate_dynamo_data.py @@ -0,0 +1,116 @@ +import json +import logging +import os +import re +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +OUTPUT_ROOT = "out" +DATE_FORMAT = "%Y%m%d" +VAR_PATTERN = re.compile(r"<<([^<>]+)>>") +REQUIRED_TOKEN_PARTS = 3 + +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s") + + +class DateVariableResolver: + def __init__(self, today: datetime | None = None): + self.today = today or datetime.now(tz=timezone.UTC) + + def resolve(self, token: str) -> str: + logger.debug("Resolving variable: %s", token) + parts = token.split("_") + if len(parts) < REQUIRED_TOKEN_PARTS or parts[0].upper() != "DATE": + msg = f"Unsupported variable format: {token}" + raise ValueError(msg) + _, unit, value = parts[0], parts[1].lower(), parts[2] + try: + offset = int(value) + except ValueError as err: + msg = f"Invalid offset value: {value}" + raise ValueError(msg) from err + if unit == "day": + return (self.today + timedelta(days=offset)).strftime(DATE_FORMAT) + if unit == "week": + return (self.today + timedelta(weeks=offset)).strftime(DATE_FORMAT) + if unit == "year": + return (self.today.replace(year=self.today.year + offset)).strftime(DATE_FORMAT) + if unit == "age": + try: + birth_date = self.today.replace(year=self.today.year - offset) + except ValueError: + birth_date = self.today.replace(month=2, day=28, year=self.today.year - offset) + return birth_date.strftime(DATE_FORMAT) + msg = f"Unsupported calculation unit: {unit}" + raise ValueError(msg) + + +class JsonTestDataProcessor: + def __init__(self, input_dir: Path, output_dir: Path, resolver: DateVariableResolver): + self.input_dir = input_dir + self.output_dir = output_dir + self.resolver = resolver + + def resolve_placeholders(self, obj: Any) -> Any: + if isinstance(obj, dict): + return {k: self.resolve_placeholders(v) for k, v in obj.items()} + if isinstance(obj, list): + return [self.resolve_placeholders(item) for item in obj] + if isinstance(obj, str): + return VAR_PATTERN.sub(self._replace_token, obj) + return obj + + def _replace_token(self, match: re.Match) -> str: + token = match.group(1) + try: + return self.resolver.resolve(token) + except ValueError: + logger.warning("Failed to resolve variable: %s", token) + return match.group(0) + + def process_file(self, file_path: Path): + logger.info("Processing file: %s", file_path) + try: + with file_path.open() as f: + content = json.load(f) + except Exception: + logger.exception("Failed to read file: %s", file_path) + return + try: + resolved = self.resolve_placeholders(content) + except Exception: + logger.exception("Failed to resolve placeholders in file: %s", file_path) + return + if "data" not in resolved: + logger.error("Missing 'data' key in file: %s", file_path) + return + relative_path = file_path.relative_to(self.input_dir) + output_path = self.output_dir / relative_path + output_path.parent.mkdir(parents=True, exist_ok=True) + try: + with output_path.open("w") as f: + json.dump(resolved["data"], f, indent=2) + logger.info("Written resolved file: %s", output_path) + except Exception: + logger.exception("Failed to write output to: %s", output_path) + + +def main(): + input_dir = Path() + output_dir = Path(OUTPUT_ROOT) + resolver = DateVariableResolver() + processor = JsonTestDataProcessor(input_dir, output_dir, resolver) + logger.info("Scanning for JSON files in directory: %s", input_dir) + for root, _, files in os.walk(input_dir): + for file in files: + file_path = Path(root) / file + if file.endswith(".json"): + processor.process_file(file_path) + else: + logger.debug("Skipping non-JSON file: %s", file) + + +if __name__ == "__main__": + main() diff --git a/tests/e2e/data/generate_test_data.py b/tests/e2e/data/generate_test_data.py deleted file mode 100644 index f1e40ea79..000000000 --- a/tests/e2e/data/generate_test_data.py +++ /dev/null @@ -1,1610 +0,0 @@ -import datetime -from typing import Any - -import pytest -from faker import Faker -from freezegun import freeze_time -from hamcrest import assert_that, contains_exactly, contains_inanyorder, equal_to, has_item, has_items, is_in - -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, - RuleDescription, - Status, -) -from eligibility_signposting_api.services.calculators.eligibility_calculator import EligibilityCalculator -from tests.fixtures.builders.model import rule as rule_builder -from tests.fixtures.builders.repos.person import person_rows_builder -from tests.fixtures.matchers.eligibility import ( - is_cohort_result, - is_condition, - is_eligibility_status, - is_reason, -) - - -def test_not_base_eligible(faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"]) - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort2")] - ) - ], - ) - ) - ] - - 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(Status.not_eligible)) - ), - ) - - -@pytest.mark.parametrize( - ("person_cohorts", "iteration_cohorts", "status", "test_comment"), - [ - (["cohort1"], ["elid_all_people"], Status.actionable, "Only magic cohort present"), - (["cohort1"], ["elid_all_people", "cohort1"], Status.actionable, "Magic cohort with other cohorts"), - (["cohort1"], ["cohort2"], Status.not_eligible, "No magic cohort. No matching person cohort"), - ([], ["elid_all_people"], Status.actionable, "No person cohorts. Only magic cohort present"), - ], -) -def test_base_eligible_with_when_magic_cohort_is_present( - faker: Faker, person_cohorts: list[str], iteration_cohorts: list[str], status: Status, test_comment: str -): - # Given - 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=person_cohorts) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.IterationCohortFactory.build(cohort_label=label) for label in iteration_cohorts - ], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - ) - ], - ) - ] - - 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(status)) - ), - test_comment, - ) - - -@freeze_time("2025-04-25") -def test_only_live_campaigns_considered(faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - name="Live", - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort2")], - ) - ], - start_date=datetime.date(2025, 4, 20), - end_date=datetime.date(2025, 4, 30), - ), - rule_builder.CampaignConfigFactory.build( - name="No longer live", - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.IterationCohortFactory.build(cohort_label="cohort1"), - rule_builder.IterationCohortFactory.build(cohort_label="cohort2"), - ], - ) - ], - start_date=datetime.date(2025, 4, 1), - end_date=datetime.date(2025, 4, 24), - ), - ] - - 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(Status.not_eligible)) - ), - ) - - -@pytest.mark.parametrize( - "iteration_type", - ["A", "M", "S", "O"], -) -def test_campaigns_with_applicable_iteration_types_considered(iteration_type: str, faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number) - campaign_configs = [rule_builder.CampaignConfigFactory.build(target="RSV", iteration_type=iteration_type)] - - 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(is_in([Status.actionable, Status.not_actionable, Status.not_eligible])) - ), - ), - ) - - -def test_base_eligible_and_simple_rule_includes(faker: Faker): - # Given - 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"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - 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(Status.actionable)) - ), - ) - - -def test_base_eligible_but_simple_rule_excludes(faker: Faker): - # Given - 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"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - 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(Status.not_actionable)) - ), - ) - - -@freeze_time("2025-04-25") -def test_simple_rule_only_excludes_from_live_iteration(faker: Faker): - # Given - 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"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - name="old iteration - would not exclude 74 year old", - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build(comparator="-65")], - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_date=datetime.date(2025, 4, 10), - ), - rule_builder.IterationFactory.build( - name="current - would exclude 74 year old", - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_date=datetime.date(2025, 4, 20), - ), - rule_builder.IterationFactory.build( - name="next iteration - would not exclude 74 year old", - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build(comparator="-65")], - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_date=datetime.date(2025, 4, 30), - ), - ], - ) - ] - - 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(Status.not_actionable)) - ), - ) - - -@pytest.mark.parametrize( - ("rule_type", "expected_status"), - [ - (rules_model.RuleType.suppression, Status.not_actionable), - (rules_model.RuleType.filter, Status.not_eligible), - (rules_model.RuleType.redirect, Status.actionable), - ], -) -def test_rule_types_cause_correct_statuses(rule_type: rules_model.RuleType, expected_status: Status, faker: Faker): - # Given - 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"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.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)) - ), - ) - - -def test_multiple_rule_types_cause_correct_status(faker: Faker): - # Given - 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"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_rules=[ - rule_builder.PersonAgeSuppressionRuleFactory.build( - priority=rules_model.RulePriority(5), type=rules_model.RuleType.suppression - ), - rule_builder.PersonAgeSuppressionRuleFactory.build( - priority=rules_model.RulePriority(10), type=rules_model.RuleType.filter - ), - rule_builder.PersonAgeSuppressionRuleFactory.build( - priority=rules_model.RulePriority(15), type=rules_model.RuleType.suppression - ), - ], - 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(Status.not_eligible)) - ), - ) - - -@pytest.mark.parametrize( - ("test_comment", "rule1", "rule2", "expected_status"), - [ - ( - "two rules, both exclude, same priority, should exclude", - rule_builder.PersonAgeSuppressionRuleFactory.build(priority=rules_model.RulePriority(5)), - rule_builder.PostcodeSuppressionRuleFactory.build(priority=rules_model.RulePriority(5)), - Status.not_actionable, - ), - ( - "two rules, rule 1 excludes, same priority, should allow", - rule_builder.PersonAgeSuppressionRuleFactory.build(priority=rules_model.RulePriority(5)), - rule_builder.PostcodeSuppressionRuleFactory.build( - priority=rules_model.RulePriority(5), comparator=rules_model.RuleComparator("NW1") - ), - Status.actionable, - ), - ( - "two rules, rule 2 excludes, same priority, should allow", - rule_builder.PersonAgeSuppressionRuleFactory.build( - priority=rules_model.RulePriority(5), comparator=rules_model.RuleComparator("-65") - ), - rule_builder.PostcodeSuppressionRuleFactory.build(priority=rules_model.RulePriority(5)), - Status.actionable, - ), - ( - "two rules, rule 1 excludes, different priority, should exclude", - rule_builder.PersonAgeSuppressionRuleFactory.build(priority=rules_model.RulePriority(5)), - rule_builder.PostcodeSuppressionRuleFactory.build( - priority=rules_model.RulePriority(10), comparator=rules_model.RuleComparator("NW1") - ), - Status.not_actionable, - ), - ( - "two rules, rule 2 excludes, different priority, should exclude", - rule_builder.PersonAgeSuppressionRuleFactory.build( - priority=rules_model.RulePriority(5), comparator=rules_model.RuleComparator("-65") - ), - rule_builder.PostcodeSuppressionRuleFactory.build(priority=rules_model.RulePriority(10)), - Status.not_actionable, - ), - ( - "two rules, both excludes, different priority, should exclude", - rule_builder.PersonAgeSuppressionRuleFactory.build(priority=rules_model.RulePriority(5)), - rule_builder.PostcodeSuppressionRuleFactory.build(priority=rules_model.RulePriority(10)), - Status.not_actionable, - ), - ], -) -def test_rules_with_same_priority_must_all_match_to_exclude( - test_comment: str, - rule1: rules_model.IterationRule, - rule2: rules_model.IterationRule, - expected_status: Status, - faker: Faker, -): - # Given - 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, postcode=Postcode("SW19 2BH"), cohorts=["cohort1"] - ) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_rules=[rule1, rule2], - 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)) - ), - test_comment, - ) - - -def test_multiple_conditions_where_both_are_actionable(faker: Faker): - # Given - 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"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - ) - ], - ), - rule_builder.CampaignConfigFactory.build( - target="COVID", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - ) - ], - ), - ] - - 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_status(Status.actionable), - is_condition().with_condition_name(ConditionName("COVID")).and_status(Status.actionable), - ) - ), - ) - - -def test_multiple_conditions_where_all_give_unique_statuses(faker: Faker): - # Given - 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"]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - ) - ], - ), - rule_builder.CampaignConfigFactory.build( - target="COVID", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build(comparator="-85")], - ) - ], - ), - rule_builder.CampaignConfigFactory.build( - target="FLU", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort2")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build(comparator="-85")], - ) - ], - ), - ] - - 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_status(Status.actionable), - is_condition().with_condition_name(ConditionName("COVID")).and_status(Status.not_actionable), - is_condition().with_condition_name(ConditionName("FLU")).and_status(Status.not_eligible), - ) - ), - ) - - -@pytest.mark.parametrize( - ("test_comment", "campaign1", "campaign2"), - [ - ( - "1st campaign allows, 2nd excludes", - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - ) - ], - ), - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build(comparator="-85")], - ) - ], - ), - ), - ( - "1st campaign excludes, 2nd allows", - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build(comparator="-85")], - ) - ], - ), - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()], - ) - ], - ), - ), - ], -) -def test_multiple_campaigns_for_single_condition( - test_comment: str, campaign1: rules_model.CampaignConfig, campaign2: rules_model.CampaignConfig, faker: Faker -): - # Given - 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"]) - campaign_configs = [campaign1, campaign2] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.evaluate_eligibility() - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - contains_exactly(is_condition().with_condition_name(ConditionName("RSV")).and_status(Status.actionable)) - ), - 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.ICBFilterRuleFactory.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)) - ), - ) - - -@pytest.mark.parametrize( - ("vaccine", "last_successful_date", "expected_status", "test_comment"), - [ - ("RSV", "20240601", Status.not_actionable, "last_successful_date is a past date"), - ("RSV", "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. - ("RSV", "20260101", Status.actionable, "last_successful_date is a future date"), - ("RSV", "20230601", Status.actionable, "last_successful_date is a long past"), - ("RSV", "", Status.actionable, "last_successful_date is empty"), - ("RSV", None, Status.actionable, "last_successful_date is none"), - ("COVID", "20240601", Status.actionable, "No RSV row"), - ], -) -@freeze_time("2025-01-01") -def test_status_on_target_based_on_last_successful_date( - vaccine: str, last_successful_date: str, expected_status: Status, test_comment: str, faker: Faker -): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - target_rows = person_rows_builder( - nhs_number, - cohorts=["cohort1"], - vaccines=[ - ( - vaccine, - 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 in the last year"), - description=rules.RuleDescription( - "Exclude anyone Completed RSV Vaccination in the last year" - ), - priority=10, - operator=rules.RuleOperator.day_gte, - attribute_level=rules.RuleAttributeLevel.TARGET, - attribute_name=rules.RuleAttributeName("LAST_SUCCESSFUL_DATE"), - comparator=rules.RuleComparator("-365"), - attribute_target=rules.RuleAttributeTarget("RSV"), - ), - rule_builder.IterationRuleFactory.build( - type=rules.RuleType.suppression, - name=rules.RuleName("You have a vaccination date in the future for RSV"), - description=rules.RuleDescription("Exclude anyone with future Completed RSV Vaccination"), - priority=10, - 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, - ) - - -@pytest.mark.parametrize( - ("attribute_name", "expected_status", "test_comment"), - [ - ( - rules.RuleAttributeName("COHORT_LABEL"), - Status.not_eligible, - "cohort label provided", - ), - ( - None, - Status.not_eligible, - "cohort label is the default attribute name for the cohort attribute level", - ), - ( - rules.RuleAttributeName("LOCATION"), - Status.actionable, - "attribute name that is not cohort label", - ), - ], -) -def test_status_on_cohort_attribute_level( - attribute_name: rules.RuleAttributeName, expected_status: Status, test_comment: str, faker: Faker -): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_row: list[dict[str, Any]] = person_rows_builder( - nhs_number, cohorts=["cohort1", "covid_eligibility_complaint_list"] - ) - person_row_with_extra_items_in_cohort_row = [ - {**r, "LOCATION": "HP1"} for r in person_row if r.get("ATTRIBUTE_TYPE", "") == "COHORTS" - ] - - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - iteration_rules=[ - rule_builder.IterationRuleFactory.build( - type=rules.RuleType.filter, - name=rules.RuleName("Exclude those in a complaint cohort"), - description=rules.RuleDescription( - "Ensure anyone who has registered a complaint is not shown as eligible" - ), - priority=15, - operator=rules.RuleOperator.member_of, - attribute_level=rules.RuleAttributeLevel.COHORT, - attribute_name=attribute_name, - comparator=rules.RuleComparator("covid_eligibility_complaint_list"), - ) - ], - ) - ], - ) - ] - - calculator = EligibilityCalculator(person_row_with_extra_items_in_cohort_row, 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, - ) - - -@pytest.mark.parametrize( - ("person_cohorts", "expected_status", "test_comment"), - [ - (["cohort1", "cohort2"], Status.actionable, "cohort1 is not actionable, cohort 2 is actionable"), - (["cohort3", "cohort2"], Status.actionable, "cohort3 is not eligible, cohort 2 is actionable"), - (["cohort1"], Status.not_actionable, "cohort1 is not actionable"), - ], -) -def test_status_if_iteration_rules_contains_cohort_label_field( - person_cohorts, expected_status: Status, test_comment: str, faker: Faker -): - # Given - 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=person_cohorts) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.IterationCohortFactory.build(cohort_label="cohort1"), - rule_builder.IterationCohortFactory.build(cohort_label="cohort2"), - ], - iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.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_items(is_condition().with_condition_name(ConditionName("RSV")).and_status(expected_status)) - ), - test_comment, - ) - - -@pytest.mark.parametrize( - ("rule_stop", "expected_reason_results", "test_comment"), # Changed expected_reasons to expected_reason_results - [ - ( - rules.RuleStop(True), # noqa: FBT003 - [ - RuleDescription("reason 1"), - RuleDescription("reason 2"), - ], - "rule_stop is True, last rule should not run", - ), - ( - rules.RuleStop(False), # noqa: FBT003 - [ - RuleDescription("reason 1"), - RuleDescription("reason 2"), - RuleDescription("reason 3"), - ], - "rule_stop is False, last rule should run", - ), - ], -) -def test_rules_stop_behavior( - rule_stop: rules.RuleStop, expected_reason_results: list[RuleDescription], test_comment: str, faker: Faker -) -> None: - # Given - 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"]) - - # Build campaign configuration - campaign_config = rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_rules=[ - rule_builder.PersonAgeSuppressionRuleFactory.build( - priority=10, description="reason 1", rule_stop=rule_stop - ), - rule_builder.PersonAgeSuppressionRuleFactory.build(priority=10, description="reason 2"), - rule_builder.PersonAgeSuppressionRuleFactory.build(priority=15, description="reason 3"), - ], - iteration_cohorts=[ - rule_builder.IterationCohortFactory.build(cohort_group="cohort_group1", cohort_label="cohort1") - ], - ) - ], - ) - - calculator = EligibilityCalculator(person_rows, [campaign_config]) - - # When - actual = calculator.evaluate_eligibility() - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.not_actionable)) - .and_cohort_results( - has_items( - is_cohort_result().with_reasons( - contains_inanyorder( - *[ - is_reason().with_rule_description(equal_to(result)) - for result in expected_reason_results - ] - ) - ) - ) - ) - ) - ), - test_comment, - ) - - -@pytest.mark.parametrize( - ("person_cohorts", "iteration_cohorts", "expected_status", "expected_cohorts"), - [ - ( - ["covid_cohort", "flu_cohort"], - ["rsv_clinical_cohort", "rsv_75_rolling"], - Status.not_eligible, - ["rsv_clinical_cohort_group", "rsv_75_rolling_group"], - ), - ( - ["rsv_clinical_cohort", "rsv_75_rolling"], - ["rsv_clinical_cohort", "rsv_75_rolling"], - Status.actionable, - ["rsv_clinical_cohort_group"], - ), - ( - ["covid_cohort", "rsv_75_rolling"], - ["rsv_clinical_cohort", "rsv_75_rolling"], - Status.not_actionable, - ["rsv_75_rolling_group"], - ), - ( - ["covid_cohort", "rsv_clinical_cohort"], - ["rsv_clinical_cohort", "rsv_75_rolling"], - Status.actionable, - ["rsv_clinical_cohort_group"], - ), - ( - ["rsv_75to79_2024", "rsv_75_rolling"], - ["rsv_75to79_2024", "rsv_75_rolling"], - Status.not_actionable, - ["rsv_75_rolling_group", "rsv_75to79_2024_group"], - ), - ], -) -def test_eligibility_results_when_multiple_cohorts( - person_cohorts: list[str], - iteration_cohorts: list[str], - expected_status: Status, - expected_cohorts: list[str], - faker: Faker, -): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - dob_person_less_than_75 = DateOfBirth(faker.date_of_birth(minimum_age=66, maximum_age=74)) - - person_rows = person_rows_builder(nhs_number, date_of_birth=dob_person_less_than_75, cohorts=person_cohorts) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.IterationCohortFactory.build( - cohort_group=f"{cohorts}_group", - cohort_label=cohorts, - positive_description="positive description", - negative_description="negative description", - ) - for cohorts in iteration_cohorts - ], - iteration_rules=[ - rule_builder.PersonAgeSuppressionRuleFactory.build(cohort_label="rsv_75_rolling"), - rule_builder.PersonAgeSuppressionRuleFactory.build(cohort_label="rsv_75to79_2024"), - ], - ) - ], - ) - ] - - 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_status(equal_to(expected_status)) - .and_cohort_results( - contains_inanyorder( - *[ - is_cohort_result().with_cohort_code(equal_to(cohort_label)) - for cohort_label in expected_cohorts - ] - ) - ) - ) - ), - ) - - -@pytest.mark.parametrize( - ("person_rows", "expected_status", "expected_cohort_group_and_description", "test_comment"), - [ - ( - person_rows_builder(nhs_number="123", cohorts=[], postcode="AC01", de=True, icb="QE1"), - Status.not_eligible, - [ - ("magic cohort group", "magic negative description"), - ("rsv_age_range", "rsv_age_range negative description"), - ], - "rsv_75_rolling is not base-eligible & magic cohort group not eligible by F rules ", - ), - ( - 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"), - ("rsv_age_range", "rsv_age_range positive description"), - ], - "all the cohorts are not-actionable", - ), - ( - person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling"], postcode="AC01", de=False, icb="QE1"), - Status.actionable, - [ - ("magic cohort group", "magic positive description"), - ("rsv_age_range", "rsv_age_range positive description"), - ], - "all the cohorts are actionable", - ), - ( - 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=["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_present( - person_rows: list[dict[str, Any]], - expected_status: str, - expected_cohort_group_and_description: list[tuple[str, str]], - test_comment: str, -): - # Given - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.Rsv75RollingCohortFactory.build(), - rule_builder.MagicCohortFactory.build(), - ], - iteration_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"), - ), - ], - ) - ], - ) - ] - - 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_cohort_code(item[0]) - .with_description(item[1]) - .with_status(expected_status) - for item in expected_cohort_group_and_description - ] - ) - ) - ) - ), - test_comment, - ) - - -def test_cohort_groups_and_their_descriptions_when_best_status_is_not_eligible( - faker: Faker, -): - # Given - 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=[]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.Rsv75RollingCohortFactory.build(), - rule_builder.Rsv75to79CohortFactory.build(), - rule_builder.RsvPretendClinicalCohortFactory.build(), - ], - iteration_rules=[rule_builder.PostcodeSuppressionRuleFactory.build()], - ) - ], - ) - ] - - 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_status(Status.not_eligible) - .and_cohort_results( - contains_exactly( - is_cohort_result() - .with_cohort_code("rsv_age_range") - .with_description("rsv_age_range negative description"), - is_cohort_result() - .with_cohort_code("rsv_clinical_cohort") - .with_description("rsv_clinical_cohort negative description"), - ) - ) - ) - ), - ) - - -@pytest.mark.parametrize( - ("person_cohorts", "expected_cohort_group_and_description_and_s_rule_names", "test_comment"), - [ - ( - ["rsv_75_rolling"], - [("rsv_age_range", "rsv_age_range positive description", ["Excluded postcode In SW19"])], - "rsv_75_rolling is not-actionable, others are not-eligible", - ), - ( - ["rsv_75_rolling", "rsv_75to79_2024"], - [ - ( - "rsv_age_range", - "rsv_age_range positive description", - ["Excluded postcode In SW19", "Excluded postcode In SW19"], - ) - ], - "rsv_75_rolling, rsv_75to79_2024 is not-actionable, rsv_pretend_clinical_cohort are not-eligible", - ), - ( - ["rsv_75_rolling", "rsv_75to79_2024", "rsv_pretend_clinical_cohort"], - [ - ( - "rsv_age_range", - "rsv_age_range positive description", - ["Excluded postcode In SW19", "Excluded postcode In SW19"], - ), - ("rsv_clinical_cohort", "rsv_clinical_cohort positive description", ["Excluded postcode In SW19"]), - ], - "all are not-actionable", - ), - ], -) -def test_cohort_groups_and_their_descriptions_and_the_collection_of_s_rules_when_best_status_is_not_actionable( - person_cohorts: list[str], - expected_cohort_group_and_description_and_s_rule_names: list[tuple[str, str, list[str]]], - test_comment: str, - faker: Faker, -): - # Given - 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=person_cohorts, postcode="SW19") - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.Rsv75RollingCohortFactory.build(), - rule_builder.Rsv75to79CohortFactory.build(), - rule_builder.RsvPretendClinicalCohortFactory.build(), - ], - iteration_rules=[rule_builder.PostcodeSuppressionRuleFactory.build()], - ) - ], - ) - ] - - 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_status(Status.not_actionable) - .and_cohort_results( - contains_exactly( - *[ - is_cohort_result() - .with_cohort_code(item[0]) - .and_description(item[1]) - .and_reasons( - contains_exactly(*[is_reason().with_rule_name(rule_name) for rule_name in item[2]]) - ) - for item in expected_cohort_group_and_description_and_s_rule_names - ] - ) - ), - ) - ), - test_comment, - ) - - -@pytest.mark.parametrize( - ("person_cohorts", "expected_cohort_group_and_description", "test_comment"), - [ - ( - ["rsv_75_rolling"], - [("rsv_age_range", "rsv_age_range positive description")], - "rsv_75_rolling is actionable, others are not-eligible", - ), - ( - ["rsv_75_rolling", "rsv_75to79_2024"], - [("rsv_age_range", "rsv_age_range positive description")], - "rsv_75_rolling, rsv_75to79_2024 is actionable, rsv_pretend_clinical_cohort are not-eligible", - ), - ( - ["rsv_75_rolling", "rsv_75to79_2024", "rsv_pretend_clinical_cohort"], - [ - ("rsv_age_range", "rsv_age_range positive description"), - ("rsv_clinical_cohort", "rsv_clinical_cohort positive description"), - ], - "all are actionable", - ), - ], -) -def test_cohort_group_and_descriptions_when_best_status_is_actionable( - person_cohorts: list[str], - expected_cohort_group_and_description: list[tuple[str, str]], - test_comment: str, - faker: Faker, -): - # Given - 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=person_cohorts, postcode="hp") - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.Rsv75RollingCohortFactory.build(), - rule_builder.Rsv75to79CohortFactory.build(), - rule_builder.RsvPretendClinicalCohortFactory.build(), - ], - iteration_rules=[rule_builder.PostcodeSuppressionRuleFactory.build()], - ) - ], - ) - ] - - 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_status(Status.actionable) - .and_cohort_results( - contains_exactly( - *[ - is_cohort_result().with_cohort_code(item[0]).with_description(item[1]) - for item in expected_cohort_group_and_description - ] - ) - ) - ) - ), - test_comment, - ) - - -@pytest.mark.parametrize( - ("person_rows", "expected_description", "test_comment"), - [ - ( - person_rows_builder(nhs_number="123", cohorts=[]), - "rsv_age_range negative description 1", - "status - not eligible", - ), - ( - person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling", "rsv_75to79_2024"], postcode="SW19"), - "rsv_age_range positive description 1", - "status - not actionable", - ), - ( - person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling", "rsv_75to79_2024"], postcode="hp"), - "rsv_age_range positive description 1", - "status - actionable", - ), - ( - person_rows_builder(nhs_number="123", cohorts=["rsv_75to79_2024"], postcode="hp"), - "rsv_age_range positive description 2", - "rsv_75to79_2024 - actionable and rsv_75_rolling is not eligible", - ), - ], -) -def test_cohort_group_descriptions_are_selected_based_on_priority_when_cohorts_have_different_non_empty_descriptions( - person_rows: list[dict[str, Any]], expected_description: str, test_comment: str -): - # Given - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[ - rule_builder.Rsv75to79CohortFactory.build( - positive_description=rules.Description("rsv_age_range positive description 2"), - negative_description=rules.Description("rsv_age_range negative description 2"), - priority=2, - ), - rule_builder.Rsv75RollingCohortFactory.build( - positive_description=rules.Description("rsv_age_range positive description 1"), - negative_description=rules.Description("rsv_age_range negative description 1"), - priority=1, - ), - ], - iteration_rules=[rule_builder.PostcodeSuppressionRuleFactory.build()], - ) - ], - ) - ] - - 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_cohort_code("rsv_age_range").with_description(expected_description) - ) - ) - ) - ), - test_comment, - ) - - -@pytest.mark.parametrize( - ("person_rows", "iteration_cohorts", "expected_cohort_group_and_description", "expected_status", "test_comment"), - [ - ( - person_rows_builder("123", postcode="SW19", cohorts=[], de=False), - [rule_builder.Rsv75to79CohortFactory.build(negative_description=None, priority=2)], - [("rsv_age_range", "")], - Status.not_eligible, - "if group has one cohort, with no description, expect no description", - ), - ( - person_rows_builder("123", postcode="SW19", cohorts=["rsv_75to79_2024", "rsv_75_rolling"], de=False), - [rule_builder.Rsv75to79CohortFactory.build(negative_description=None, priority=2)], - [("rsv_age_range", "")], - Status.not_eligible, - "if group has one cohort, with no description, expect no description", - ), - ( - person_rows_builder("123", postcode="HP1", cohorts=["rsv_75to79_2024", "rsv_75_rolling"], de=True), - [rule_builder.Rsv75to79CohortFactory.build(positive_description=None, priority=2)], - [("rsv_age_range", "")], - Status.not_actionable, - "if group has one cohort, with no description, expect no description", - ), - ( - person_rows_builder("123", postcode="HP1", cohorts=["rsv_75to79_2024", "rsv_75_rolling"], de=False), - [rule_builder.Rsv75to79CohortFactory.build(positive_description=None, priority=2)], - [("rsv_age_range", "")], - Status.actionable, - "if group has one cohort, with no description, expect no description", - ), - ( - person_rows_builder("123", postcode="SW19", cohorts=[], de=False), - [ - rule_builder.Rsv75to79CohortFactory.build(priority=2, negative_description=None), - rule_builder.Rsv75RollingCohortFactory.build(priority=3, negative_description="rsv age range -ve 1"), - rule_builder.Rsv75RollingCohortFactory.build( - cohort_label="rsv_75_rolling_2", priority=4, negative_description="rsv age range -ve 2" - ), - ], - [("rsv_age_range", "rsv age range -ve 1")], - Status.not_eligible, - "if group has more than one cohort, at least one has description, expect first non empty description", - ), - ( - person_rows_builder("123", postcode="SW19", cohorts=["rsv_75to79_2024", "rsv_75_rolling"], de=False), - [ - rule_builder.Rsv75to79CohortFactory.build(priority=2, negative_description=None), - rule_builder.Rsv75RollingCohortFactory.build(priority=3, negative_description="rsv age range -ve 1"), - rule_builder.Rsv75RollingCohortFactory.build( - cohort_label="rsv_75_rolling_2", priority=4, negative_description="rsv age range -ve 2" - ), - ], - [("rsv_age_range", "rsv age range -ve 1")], - Status.not_eligible, - "if group has more than one cohort, at least one has description, expect first non empty description", - ), - ( - person_rows_builder("123", postcode="HP1", cohorts=["rsv_75to79_2024", "rsv_75_rolling"], de=True), - [ - rule_builder.Rsv75to79CohortFactory.build(priority=2, positive_description=None), - rule_builder.Rsv75RollingCohortFactory.build(priority=3, positive_description="rsv age range +ve 1"), - rule_builder.Rsv75RollingCohortFactory.build( - cohort_label="rsv_75_rolling_2", priority=4, positive_description="rsv age range +ve 2" - ), - ], - [("rsv_age_range", "rsv age range +ve 1")], - Status.not_actionable, - "if group has more than one cohort, at least one has description, expect first non empty description", - ), - ( - person_rows_builder("123", postcode="HP1", cohorts=["rsv_75to79_2024", "rsv_75_rolling"], de=False), - [ - rule_builder.Rsv75to79CohortFactory.build(priority=2, positive_description=None), - rule_builder.Rsv75RollingCohortFactory.build(priority=3, positive_description="rsv age range +ve 1"), - rule_builder.Rsv75RollingCohortFactory.build( - cohort_label="rsv_75_rolling_2", priority=4, positive_description="rsv age range +ve 2" - ), - ], - [("rsv_age_range", "rsv age range +ve 1")], - Status.actionable, - "if group has more than one cohort, at least one has description, expect first non empty description", - ), - ], -) -def test_cohort_group_descriptions_pick_first_non_empty_if_available( - person_rows: list[dict[str, Any]], - iteration_cohorts: list[rules.IterationCohort], - expected_cohort_group_and_description: list[tuple[str, str]], - expected_status: Status, - test_comment: str, -): - # Given - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=iteration_cohorts, - iteration_rules=[ - rule_builder.PostcodeSuppressionRuleFactory.build(type=rules.RuleType.filter), - rule_builder.DetainedEstateSuppressionRuleFactory.build(), - ], - ) - ], - ) - ] - - 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_status(expected_status) - .and_cohort_results( - contains_exactly( - *[ - is_cohort_result() - .with_cohort_code(item[0]) - .with_description(item[1]) - .with_status(expected_status) - for item in expected_cohort_group_and_description - ] - ) - ) - ) - ), - test_comment, - )