From 5340a45e71df6af9dae6c5b455d3397dc42bbfa8 Mon Sep 17 00:00:00 2001 From: ivano mannella Date: Tue, 24 Jun 2025 09:56:19 +0100 Subject: [PATCH 1/4] from DATE_TOMORROW to DATE_DAY_+1 to be consistent --- tests/e2e/data/dynamoDB/AUTO_RSV_SB_004.json | 2 +- tests/e2e/data/dynamoDB/AUTO_RSV_SB_005.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_004.json b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_004.json index ca15b178e..ed6764ede 100644 --- a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_004.json +++ b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_004.json @@ -39,7 +39,7 @@ { "NHS_NUMBER": "5000000004", "ATTRIBUTE_TYPE": "RSV", - "BOOKED_APPOINTMENT_DATE": "<>", + "BOOKED_APPOINTMENT_DATE": "<>", "BOOKED_APPOINTMENT_PROVIDER": "NBS" } ] diff --git a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_005.json b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_005.json index a0f76ddd3..1892dd217 100644 --- a/tests/e2e/data/dynamoDB/AUTO_RSV_SB_005.json +++ b/tests/e2e/data/dynamoDB/AUTO_RSV_SB_005.json @@ -39,7 +39,7 @@ { "NHS_NUMBER": "5000000005", "ATTRIBUTE_TYPE": "RSV", - "BOOKED_APPOINTMENT_DATE": "<>", + "BOOKED_APPOINTMENT_DATE": "<>", "BOOKED_APPOINTMENT_PROVIDER": "ACC" } ] From 57d741fc3f1813b964cfeb04aef0bfe2234eef44 Mon Sep 17 00:00:00 2001 From: ivano mannella Date: Tue, 24 Jun 2025 09:56:40 +0100 Subject: [PATCH 2/4] +.gitignore changes for e2e --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index 6098a015b..b863bfb62 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,5 @@ sandbox/specification/* /sandbox/specification/* /integration-test-results.xml /specification/tmp/* +/tests/e2e/data/out* +/tests/e2e/reports/* From c084190af2c67648d22c5b24fe38f107a0cdbdf9 Mon Sep 17 00:00:00 2001 From: ivano mannella Date: Tue, 24 Jun 2025 10:20:04 +0100 Subject: [PATCH 3/4] +changes to features/environment.py --- poetry.lock | 32 ++++---- pyproject.toml | 2 + tests/e2e/features/environment.py | 117 +++++++++++++----------------- 3 files changed, 69 insertions(+), 82 deletions(-) diff --git a/poetry.lock b/poetry.lock index 55167e136..269c8ac2f 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.1.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand. [[package]] name = "aiohappyeyeballs" @@ -220,18 +220,18 @@ cryptography = "*" [[package]] name = "awscli" -version = "1.40.39" +version = "1.40.41" description = "Universal Command Line Environment for AWS." optional = false python-versions = ">=3.9" groups = ["dev"] files = [ - {file = "awscli-1.40.39-py3-none-any.whl", hash = "sha256:3bb17cef517d0e7ff03b91c1a85d6b7d3d294587a76def938fa93293b8fdded4"}, - {file = "awscli-1.40.39.tar.gz", hash = "sha256:82985feb7aff8e7fbfcbff836ce9c1aec0ad5a949a77545b54da0b8130f4d5a9"}, + {file = "awscli-1.40.41-py3-none-any.whl", hash = "sha256:d75cc6c654418ac4d30eb996081033e90024fa7a661db8ab40de4b5a545eaa79"}, + {file = "awscli-1.40.41.tar.gz", hash = "sha256:553c3a3ba7879be18c5db219f9a710daf90d750044eb604297b25805b05ebc42"}, ] [package.dependencies] -botocore = "1.38.40" +botocore = "1.38.42" colorama = ">=0.2.5,<0.4.7" docutils = ">=0.18.1,<=0.19" PyYAML = ">=3.10,<6.1" @@ -313,18 +313,18 @@ files = [ [[package]] name = "boto3" -version = "1.38.37" +version = "1.38.42" description = "The AWS SDK for Python" optional = false python-versions = ">=3.9" groups = ["main", "dev"] files = [ - {file = "boto3-1.38.37-py3-none-any.whl", hash = "sha256:46a512b1fbc4c51a9abfef8e2130db0806cb00ef137e161f6f751421c78a7c0c"}, - {file = "boto3-1.38.37.tar.gz", hash = "sha256:4ccd700a2a36de0cd63bd8c79cca6164cb684e34fc1126de5c41525e4d0bfaee"}, + {file = "boto3-1.38.42-py3-none-any.whl", hash = "sha256:a9b4c7021bf5adee985523fc87db27a7200de161c094cb8f709b93a81797dc8a"}, + {file = "boto3-1.38.42.tar.gz", hash = "sha256:2cb783c668ae4f2a86b6497b47251b9baf9a16db8fff863b57eae683276b9e1f"}, ] [package.dependencies] -botocore = ">=1.38.37,<1.39.0" +botocore = ">=1.38.42,<1.39.0" jmespath = ">=0.7.1,<2.0.0" s3transfer = ">=0.13.0,<0.14.0" @@ -333,14 +333,14 @@ crt = ["botocore[crt] (>=1.21.0,<2.0a0)"] [[package]] name = "botocore" -version = "1.38.40" +version = "1.38.42" description = "Low-level, data-driven core of boto 3." optional = false python-versions = ">=3.9" groups = ["main", "dev"] files = [ - {file = "botocore-1.38.40-py3-none-any.whl", hash = "sha256:7528f47945502bf4226e629337c2ac2e454e661ac8fd1dc0fbf7f38082930f3f"}, - {file = "botocore-1.38.40.tar.gz", hash = "sha256:aefbfe835a7ebe9bbdd88df3999b0f8f484dd025af4ebb3f3387541316ce4349"}, + {file = "botocore-1.38.42-py3-none-any.whl", hash = "sha256:fbbeac30c045b5c19f1c3bb063ea2b6315ce2d6fcb3d898e87d1c1846297961c"}, + {file = "botocore-1.38.42.tar.gz", hash = "sha256:3a14188e48f6e26be561164373d34150fa9cb39f7ad32cc745dcd3ab05f43683"}, ] [package.dependencies] @@ -2632,14 +2632,14 @@ six = ">=1.5" [[package]] name = "python-dotenv" -version = "1.1.0" +version = "1.1.1" description = "Read key-value pairs from a .env file and set them as environment variables" optional = false python-versions = ">=3.9" groups = ["dev"] files = [ - {file = "python_dotenv-1.1.0-py3-none-any.whl", hash = "sha256:d7c01d9e2293916c18baf562d95698754b0dbbb5e74d457c45d4f6561fb9d55d"}, - {file = "python_dotenv-1.1.0.tar.gz", hash = "sha256:41f90bc6f5f177fb41f53e87666db362025010eb28f60a01c9143bfa33a2b2d5"}, + {file = "python_dotenv-1.1.1-py3-none-any.whl", hash = "sha256:31f23644fe2602f88ff55e1f5c79ba497e01224ee7737937930c448e4d0e24dc"}, + {file = "python_dotenv-1.1.1.tar.gz", hash = "sha256:a8a6399716257f45be6a007360200409fce5cda2661e3dec71d23dc15f6189ab"}, ] [package.extras] @@ -3478,4 +3478,4 @@ propcache = ">=0.2.1" [metadata] lock-version = "2.1" python-versions = "^3.13" -content-hash = "0ac8be9dfc1718f5a9b8f616d1baa5e82f5dbcbe2633b71bfadcf42ea3f4da11" +content-hash = "350611ee13ef6aefb16bb0a69766b37d5691ba38f74e6dc6f4c08be2b0223b26" diff --git a/pyproject.toml b/pyproject.toml index 3de4a7453..b55498847 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,6 +60,8 @@ moto = "^5.1.5" requests = "^2.31.0" jsonschema = "^4.24.0" behave = "^1.2.6" +boto3 = "^1.38.42" +python-dotenv = "^1.1.1" [tool.poetry-plugin-lambda-build] docker-image = "public.ecr.aws/sam/build-python3.13:1.139-x86_64" # See https://gallery.ecr.aws/search?searchTerm=%22python%22&architecture=x86-64&popularRegistries=amazon&verified=verified&operatingSystems=Linux diff --git a/tests/e2e/features/environment.py b/tests/e2e/features/environment.py index 388fcfbef..cfb2becac 100644 --- a/tests/e2e/features/environment.py +++ b/tests/e2e/features/environment.py @@ -4,100 +4,108 @@ from pathlib import Path import boto3 +from botocore.exceptions import BotoCoreError from dotenv import load_dotenv -# Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger("behave.environment") def _load_environment_variables(context): - """Load environment variables and set up context.""" load_dotenv(dotenv_path=".env") - - # API configuration context.base_url = os.getenv("BASE_URL") context.api_key = os.getenv("API_KEY") context.valid_nhs_number = os.getenv("VALID_NHS_NUMBER", "50000000004") - - # AWS configuration context.aws_region = os.getenv("AWS_REGION", "eu-west-2") context.inserted_items = [] context.abort_on_aws_error = os.getenv("ABORT_ON_AWS_FAILURE", "false").lower() == "true" context.keep_seed = os.getenv("KEEP_SEED", "false").lower() == "true" - - # S3 configuration context.s3_bucket = os.getenv("S3_BUCKET_NAME") context.s3_upload_dir = os.getenv("S3_UPLOAD_DIR", "") context.s3_data_path = Path(os.getenv("S3_JSON_SOURCE_DIR", "./data/s3")).resolve() - - # DynamoDB configuration context.dynamodb_table_name = os.getenv("DYNAMODB_TABLE_NAME", "eligibilty_data_store") - context.dynamo_data_path = Path(os.getenv("DYNAMO_JSON_SOURCE_DIR", "./data/dynamoDB/test_data.json")).resolve() - + context.dynamo_data_path = Path(os.getenv("DYNAMO_JSON_SOURCE_DIR", "./data/out/dynamoDB")).resolve() logger.info("ABORT_ON_AWS_FAILURE=%s", context.abort_on_aws_error) logger.info("KEEP_SEED=%s", context.keep_seed) -def _setup_dynamodb(context): - """Set up DynamoDB connection and seed data.""" +def _connect_to_dynamodb(context): try: context.dynamodb = boto3.resource("dynamodb", region_name=context.aws_region) context.table = context.dynamodb.Table(context.dynamodb_table_name) _ = context.table.table_status - logger.info("Connected to DynamoDB table: %s", context.dynamodb_table_name) - except (boto3.exceptions.Boto3Error, boto3.exceptions.BotoCoreError): + except (boto3.exceptions.Boto3Error, BotoCoreError): logger.exception("DynamoDB not accessible") - if context.abort_on_aws_error: - context.abort_all = True return False + else: + logger.info("Connected to DynamoDB table: %s", context.dynamodb_table_name) + return True - if not context.dynamo_data_path.exists(): - logger.error("Seed file not found: %s", context.dynamo_data_path) - if context.abort_on_aws_error: - context.abort_all = True - return False +def _get_dynamo_seed_files(context): + if not context.dynamo_data_path.exists() or not context.dynamo_data_path.is_dir(): + logger.error("Seed directory not found: %s", context.dynamo_data_path) + return [] + return list(context.dynamo_data_path.glob("*.json")) + + +def _load_seed_file(file_path: Path): try: - with context.dynamo_data_path.open() as f: - items = json.load(f) + with file_path.open() as f: + return json.load(f) except (OSError, json.JSONDecodeError): - logger.exception("Failed to load seed file") - if context.abort_on_aws_error: - context.abort_all = True - return False + logger.exception("Failed to load seed file: %s", file_path) + return [] + - logger.info("Inserting %d items into DynamoDB...", len(items)) +def _insert_dynamodb_items(context, items): for item in items: try: context.table.put_item(Item=item) context.inserted_items.append(item) - except (boto3.exceptions.Boto3Error, boto3.exceptions.BotoCoreError): + except (boto3.exceptions.Boto3Error, BotoCoreError): logger.exception("Failed to insert item %s", item.get("PK", "")) - logger.info("Inserted %d items", len(context.inserted_items)) + +def _setup_dynamodb(context): + if not _connect_to_dynamodb(context): + if context.abort_on_aws_error: + context.abort_all = True + return False + json_files = _get_dynamo_seed_files(context) + if not json_files: + logger.error("No JSON files found in the directory: %s", context.dynamo_data_path) + if context.abort_on_aws_error: + context.abort_all = True + return False + logger.info("Found %d JSON files to insert into DynamoDB", len(json_files)) + for file_path in json_files: + items = _load_seed_file(file_path) + if not items: + if context.abort_on_aws_error: + context.abort_all = True + continue + logger.info("Inserting %d items from %s...", len(items), file_path.name) + _insert_dynamodb_items(context, items) + logger.info("Inserted %d items from %d files", len(context.inserted_items), len(json_files)) return True def _setup_s3(context): - """Upload test data to S3 bucket.""" if not context.s3_bucket: logger.info("Skipping S3 upload — no S3_BUCKET_NAME set.") return True - logger.info( "Uploading JSON files from %s to S3 bucket: %s/%s", context.s3_data_path, context.s3_bucket, context.s3_upload_dir, ) - try: s3_client = boto3.client("s3", region_name=context.aws_region) if not context.s3_data_path.exists(): logger.error("S3 source directory not found: %s", context.s3_data_path) return False - json_files = list(context.s3_data_path.glob("*.json")) upload_success = True for file_path in json_files: @@ -105,100 +113,77 @@ def _setup_s3(context): try: s3_client.upload_file(str(file_path), context.s3_bucket, key) logger.info("Uploaded %s to s3://%s/%s", file_path.name, context.s3_bucket, key) - except (boto3.exceptions.Boto3Error, boto3.exceptions.BotoCoreError): + except (boto3.exceptions.Boto3Error, BotoCoreError): logger.exception("Failed to upload %s", file_path.name) upload_success = False - - if upload_success: - return True - except (boto3.exceptions.Boto3Error, boto3.exceptions.BotoCoreError): + except (boto3.exceptions.Boto3Error, BotoCoreError): logger.exception("S3 upload setup failed") if context.abort_on_aws_error: context.abort_all = True return False + else: + return upload_success def before_all(context): - """Initialize test environment before all tests.""" logger.info("Loading .env and initializing AWS fixtures...") - - # Load environment variables _load_environment_variables(context) - - # Set up DynamoDB _setup_dynamodb(context) - - # Set up S3 _setup_s3(context) def before_scenario(context, scenario): if getattr(context, "abort_all", False): scenario.skip("Skipping scenario due to setup failure") - if "requires_dynamodb" in scenario.tags and not context.inserted_items: scenario.skip("Skipping due to missing seeded DynamoDB data") def _cleanup_dynamodb(context): - """Clean up seeded items from DynamoDB.""" if not context.inserted_items: logger.info("No items were inserted — skipping DynamoDB cleanup.") return - logger.info("Cleaning up seeded items from DynamoDB...") delete_count = 0 for item in context.inserted_items: nhs_number = item.get("NHS_NUMBER") attribute_type = item.get("ATTRIBUTE_TYPE") - if nhs_number and attribute_type: try: context.table.delete_item(Key={"NHS_NUMBER": nhs_number, "ATTRIBUTE_TYPE": attribute_type}) delete_count += 1 - except (boto3.exceptions.Boto3Error, boto3.exceptions.BotoCoreError): + except (boto3.exceptions.Boto3Error, BotoCoreError): logger.exception("Failed to delete item (%s, %s)", nhs_number, attribute_type) else: logger.error("Cannot delete item — missing NHS_NUMBER or ATTRIBUTE_TYPE: %s", item) - logger.info("Deleted %d/%d DynamoDB items", delete_count, len(context.inserted_items)) def _cleanup_s3(context): - """Clean up uploaded files from S3.""" if not (context.s3_bucket and context.s3_data_path.exists()): logger.info("Skipping S3 cleanup — no bucket or source directory not found.") return - logger.info("Cleaning up uploaded files from S3...") try: s3_client = boto3.client("s3", region_name=context.aws_region) json_files = list(context.s3_data_path.glob("*.json")) deleted_files = 0 - for file_path in json_files: key = f"{context.s3_upload_dir}/{file_path.name}" if context.s3_upload_dir else file_path.name try: s3_client.delete_object(Bucket=context.s3_bucket, Key=key) logger.info("Deleted s3://%s/%s", context.s3_bucket, key) deleted_files += 1 - except (boto3.exceptions.Boto3Error, boto3.exceptions.BotoCoreError): + except (boto3.exceptions.Boto3Error, BotoCoreError): logger.exception("Failed to delete s3://%s/%s", context.s3_bucket, key) - logger.info("Deleted %d/%d files from S3", deleted_files, len(json_files)) - except (boto3.exceptions.Boto3Error, boto3.exceptions.BotoCoreError): + except (boto3.exceptions.Boto3Error, BotoCoreError): logger.exception("S3 cleanup failed") def after_all(context): - """Clean up resources after all tests have run.""" - # Early exit if KEEP_SEED is true if getattr(context, "keep_seed", False): logger.info("KEEP_SEED=true — skipping cleanup.") return - - # Clean up DynamoDB _cleanup_dynamodb(context) - - # Clean up S3 _cleanup_s3(context) From 8daa131137dcb2ae5c989b3875a5da4c27d67465 Mon Sep 17 00:00:00 2001 From: ivano mannella Date: Tue, 24 Jun 2025 10:20:47 +0100 Subject: [PATCH 4/4] +generate DynamoDB data tests/e2e/data/generate_test_data.py --- tests/e2e/data/generate_test_data.py | 1610 ++++++++++++++++++++++++++ 1 file changed, 1610 insertions(+) create mode 100644 tests/e2e/data/generate_test_data.py diff --git a/tests/e2e/data/generate_test_data.py b/tests/e2e/data/generate_test_data.py new file mode 100644 index 000000000..f1e40ea79 --- /dev/null +++ b/tests/e2e/data/generate_test_data.py @@ -0,0 +1,1610 @@ +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, + )