diff --git a/README.md b/README.md index 28b9197d3..d27ef9560 100644 --- a/README.md +++ b/README.md @@ -189,15 +189,25 @@ graph TB direction TB App["app.py (WireUp DI)"] Config["config.py, error_handler.py"] + subgraph "Audit Layer" + direction TB + Audit["audit/audit_service.py"] + AuditModels["audit/audit_models.py"] + end + subgraph "Validation Layer" + direction TB + Validator["common/request_validator.py"] + ApiErrResp["common/api_error_response.py"] + end subgraph "Presentation Layer" direction TB View["views/eligibility.py"] - ResponseModel["views/response_model/eligibility.py"] + ResponseModel["views/response_model/eligibility_response.py"] end subgraph "Business Logic Layer" direction TB Service["services/eligibility_services.py"] - Operators["services/rules/operators.py"] + Operators["services/operators/operators.py"] end subgraph "Data Access Layer" direction TB @@ -207,24 +217,30 @@ graph TB end subgraph "Models" direction TB - ModelElig["model/eligibility.py"] - ModelRules["model/rules.py"] + ModelElig["model/eligibility_status.py"] + ModelRules["model/campaign_config.py"] end end Lambda -->|"loads"| App App -->|injects| View View -->|calls| Service + View -->|validates via| Validator + View -->|audits via| Audit + View -->|uses| RespModel + Audit -->|uses| AuditModels + Validator -->|uses| ApiErrResp + Service -->|calls| Operators Service -->|calls| PersonRepo Service -->|calls| CampaignRepo PersonRepo -->|uses| DynamoDB CampaignRepo -->|uses| S3Bucket - View -->|uses| ResponseModel App -->|reads| Config + App -->|wires| Factory + Service -->|uses| ModelElig Operators -->|uses| ModelRules - App -->|wires| Factory ``` diff --git a/tests/unit/services/calculators/test_eligibility_calculator.py b/tests/unit/services/calculators/test_eligibility_calculator.py index 11ea20273..b1da841ff 100644 --- a/tests/unit/services/calculators/test_eligibility_calculator.py +++ b/tests/unit/services/calculators/test_eligibility_calculator.py @@ -3,44 +3,42 @@ import pytest from faker import Faker -from flask import Flask, g +from flask import Flask from freezegun import freeze_time -from hamcrest import assert_that, contains_exactly, contains_inanyorder, equal_to, has_item, has_items, is_in -from pydantic import HttpUrl, ValidationError +from hamcrest import assert_that, contains_exactly, contains_inanyorder, has_item, has_items, is_, is_in +from pydantic import HttpUrl -from eligibility_signposting_api.audit.audit_models import AuditAction, AuditEvent from eligibility_signposting_api.model import campaign_config as rules_model +from eligibility_signposting_api.model import eligibility_status from eligibility_signposting_api.model.campaign_config import ( - ActionsMapper, AvailableAction, CohortLabel, Description, - IterationCohort, RuleAttributeLevel, RuleAttributeName, RuleAttributeTarget, RuleComparator, RuleName, RuleOperator, - RuleStop, RuleType, ) from eligibility_signposting_api.model.eligibility_status import ( ActionCode, ActionDescription, ActionType, + CohortGroupResult, ConditionName, DateOfBirth, InternalActionCode, + IterationResult, NHSNumber, Postcode, + Reason, RuleDescription, + RulePriority, Status, SuggestedAction, - UrlLabel, - UrlLink, ) -from eligibility_signposting_api.model.person import Person 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 @@ -48,7 +46,6 @@ is_cohort_result, is_condition, is_eligibility_status, - is_reason, ) @@ -57,39 +54,6 @@ def app(): return Flask(__name__) -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")], - iteration_rules=[], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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"), [ @@ -136,55 +100,6 @@ def test_base_eligible_with_when_magic_cohort_is_present( ) -@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")], - iteration_rules=[], - ) - ], - 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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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"], @@ -214,123 +129,6 @@ def test_campaigns_with_applicable_iteration_types_in_campaign_level_considered( ) -@pytest.mark.parametrize( - "iteration_type", - ["A", "M", "S", "O"], -) -def test_campaigns_with_applicable_iteration_types_in_iteration_level_considered(iteration_type: str, faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=[]) - campaign_configs = [ - rule_builder.CampaignConfigFactory.build( - target="RSV", iterations=[rule_builder.IterationFactory.build(type=iteration_type)] - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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])) - ), - ), - ) - - -@pytest.mark.parametrize( - "iteration_type", - ["NA", "N", "FAKE", "F"], -) -def test_invalid_iteration_types_in_campaign_level_raises_validation_error(iteration_type: str): - with pytest.raises(ValidationError): - rule_builder.CampaignConfigFactory.build(target="RSV", iteration_type=iteration_type) - - -@pytest.mark.parametrize( - "iteration_type", - ["NA", "N", "FAKE", "F"], -) -def test_invalid_iteration_types_in_iteration_level_raises_validation_error(iteration_type: str): - with pytest.raises(ValidationError): - rule_builder.CampaignConfigFactory.build( - target="RSV", iterations=[rule_builder.IterationFactory.build(type=iteration_type)] - ) - - -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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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 @@ -378,86 +176,6 @@ def test_simple_rule_only_excludes_from_live_iteration(faker: Faker): ) -@pytest.mark.parametrize( - ("rule_type", "expected_status"), - [(rules_model.RuleType.suppression, Status.not_actionable), (rules_model.RuleType.filter, Status.not_eligible)], -) -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.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_item( - is_condition().with_condition_name(ConditionName("RSV")).and_status(expected_status).and_actions([]) - ) - ), - ) - - -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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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"), [ @@ -548,287 +266,23 @@ def test_rules_with_same_priority_must_all_match_to_exclude( ) -def test_multiple_conditions_where_both_are_actionable(faker: Faker): - # Given - nhs_number = NHSNumber(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"], icb="QE1") - 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()], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={"rule_1_comms_routing": book_nbs_comms, "defaultcomms": default_comms_detail} - ), - ) - ], - ), - 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(), - rule_builder.ICBRedirectRuleFactory.build(), - ], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={"ActionCode1": book_nbs_comms, "defaultcomms": default_comms_detail} - ), - ) - ], - ), - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(Status.actionable) - .and_actions( - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), - url_link=None, - url_label=None, - ) - ] - ), - is_condition() - .with_condition_name(ConditionName("COVID")) - .and_status(Status.actionable) - .and_actions( - [ - SuggestedAction( - internal_action_code=InternalActionCode("ActionCode1"), - action_type=ActionType("ButtonAuthLink"), - action_code=ActionCode("BookNBS"), - action_description=ActionDescription("Action description"), - url_link=UrlLink(HttpUrl("https://www.nhs.uk/book-rsv")), - url_label=UrlLabel("Continue to booking"), - ) - ] - ), - ) - ), - ) - - -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.get_eligibility_status("N", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(Status.actionable) - .and_actions(None), - is_condition() - .with_condition_name(ConditionName("COVID")) - .and_status(Status.not_actionable) - .and_actions(None), - is_condition() - .with_condition_name(ConditionName("FLU")) - .and_status(Status.not_eligible) - .and_actions(None), - ) - ), - ) - - -@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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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 -): +@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()) @@ -895,74 +349,6 @@ def test_status_on_target_based_on_last_successful_date( ) -@pytest.mark.parametrize( - ("attribute_name", "expected_status", "test_comment"), - [ - ( - 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", - ), - ], -) -def test_status_on_cohort_attribute_level( - attribute_name: RuleAttributeName, expected_status: Status, test_comment: str, faker: Faker -): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_row: Person = person_rows_builder(nhs_number, cohorts=["cohort1", "covid_eligibility_complaint_list"]) - - person_row_with_extra_items_in_cohort_row = Person(person_row.data) - for row in person_row_with_extra_items_in_cohort_row.data: - if row.get("ATTRIBUTE_TYPE", "") == "COHORTS": - row["LOCATION"] = "HP1" - - 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=RuleType.filter, - name=RuleName("Exclude those in a complaint cohort"), - description=RuleDescription( - "Ensure anyone who has registered a complaint is not shown as eligible" - ), - priority=15, - operator=RuleOperator.member_of, - attribute_level=RuleAttributeLevel.COHORT, - attribute_name=attribute_name, - comparator=RuleComparator("covid_eligibility_complaint_list"), - ) - ], - ) - ], - ) - ] - - calculator = EligibilityCalculator(person_row_with_extra_items_in_cohort_row, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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"), [ @@ -1010,231 +396,55 @@ def test_status_if_iteration_rules_contains_cohort_label_field( @pytest.mark.parametrize( - ("rule_stop", "expected_reason_results", "test_comment"), # Changed expected_reasons to expected_reason_results + ("person_rows", "expected_status", "expected_cohort_group_and_description", "test_comment"), [ ( - RuleStop(True), # noqa: FBT003 + 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, [ - RuleDescription("reason 1"), - RuleDescription("reason 2"), + ("magic cohort group", "magic positive description"), + ("rsv_age_range", "rsv_age_range positive description"), ], - "rule_stop is True, last rule should not run", + "all the cohorts are not-actionable", ), ( - RuleStop(False), # noqa: FBT003 + person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling"], postcode="AC01", de=False, icb="QE1"), + Status.actionable, [ - RuleDescription("reason 1"), - RuleDescription("reason 2"), - RuleDescription("reason 3"), + ("magic cohort group", "magic positive description"), + ("rsv_age_range", "rsv_age_range positive description"), ], - "rule_stop is False, last rule should run", - ), - ], -) -def test_rules_stop_behavior( - rule_stop: 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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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", + "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", ), ], ) @@ -1300,111 +510,51 @@ def test_cohort_groups_and_their_descriptions_when_magic_cohort_is_present( ) -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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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"), + ("person_rows", "expected_description", "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", + person_rows_builder(nhs_number="123", cohorts=[]), + "rsv_age_range negative description 1", + "status - 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", + person_rows_builder(nhs_number="123", cohorts=["rsv_75_rolling", "rsv_75to79_2024"], postcode="SW19"), + "rsv_age_range positive description 1", + "status - not actionable", ), ( - ["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", + 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_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, +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 - 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(), + rule_builder.Rsv75to79CohortFactory.build( + positive_description=Description("rsv_age_range positive description 2"), + negative_description=Description("rsv_age_range negative description 2"), + priority=2, + ), + rule_builder.Rsv75RollingCohortFactory.build( + positive_description=Description("rsv_age_range positive description 1"), + negative_description=Description("rsv_age_range negative description 1"), + priority=1, + ), ], iteration_rules=[rule_builder.PostcodeSuppressionRuleFactory.build()], ) @@ -1424,1592 +574,264 @@ def test_cohort_groups_and_their_descriptions_and_the_collection_of_s_rules_when 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 - ] + is_cohort_result().with_cohort_code("rsv_age_range").with_description(expected_description) ) - ), + ) ) ), 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", - ), - ], +book_nbs_comms = AvailableAction( + ActionType="ButtonAuthLink", + ExternalRoutingCode="BookNBS", + ActionDescription="Action description", + UrlLink=HttpUrl("https://www.nhs.uk/book-rsv"), + UrlLabel="Continue to booking", ) -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.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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=Description("rsv_age_range positive description 2"), - negative_description=Description("rsv_age_range negative description 2"), - priority=2, - ), - rule_builder.Rsv75RollingCohortFactory.build( - positive_description=Description("rsv_age_range positive description 1"), - negative_description=Description("rsv_age_range negative description 1"), - priority=1, - ), - ], - iteration_rules=[rule_builder.PostcodeSuppressionRuleFactory.build()], - ) - ], - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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[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=RuleType.filter), - rule_builder.DetainedEstateSuppressionRuleFactory.build(), - ], - ) - ], - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # 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, - ) - - -book_nbs_comms = AvailableAction( - ActionType="ButtonAuthLink", - ExternalRoutingCode="BookNBS", - ActionDescription="Action description", - UrlLink=HttpUrl("https://www.nhs.uk/book-rsv"), - UrlLabel="Continue to booking", -) - -default_comms_detail = AvailableAction( - ActionType="CareCardWithText", - ExternalRoutingCode="BookLocal", - ActionDescription="You can get an RSV vaccination at your GP surgery", -) - - -@pytest.mark.parametrize( - ("test_comment", "default_comms_routing", "comms_routing", "actions_mapper", "expected_actions"), - [ - ( - """Rule match: default_comms_routing present, action_mapper present, - return actions from matching comms from rule""", - "defaultcomms", - "InternalBookNBS", - {"InternalBookNBS": book_nbs_comms, "defaultcomms": default_comms_detail}, - [ - SuggestedAction( - internal_action_code=InternalActionCode("InternalBookNBS"), - action_type=ActionType("ButtonAuthLink"), - action_code=ActionCode("BookNBS"), - action_description=ActionDescription("Action description"), - url_link=UrlLink(HttpUrl("https://www.nhs.uk/book-rsv")), - url_label=UrlLabel("Continue to booking"), - ) - ], - ), - ( - """Rule match: default_comms_routing has multiple values, - comms missing in rule, all default comms should be returned in actions""", - "defaultcomms1|defaultcomms2", - None, - {"defaultcomms1": default_comms_detail, "defaultcomms2": default_comms_detail}, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms1"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), - url_link=None, - url_label=None, - ), - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms2"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), - url_link=None, - url_label=None, - ), - ], - ), - ( - """Rule match: default_comms_routing has multiple values, - comms is empty string, all default comms should be returned in actions""", - "defaultcomms1", - "", - {"defaultcomms1": default_comms_detail}, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms1"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), - url_link=None, - url_label=None, - ) - ], - ), - ( - """Rule match: default_comms_routing present, - action_mapper missing for matching comms, return default_comms in actions""", - "defaultcomms", - "InternalBookNBS", - {"defaultcomms": default_comms_detail}, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), - url_link=None, - url_label=None, - ) - ], - ), - ( - """Rule match: default_comms_routing present, - rule has an incorrect comms key, return default_comms in actions""", - "defaultcomms", - "InvalidCode", - {"defaultcomms": default_comms_detail}, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), - url_link=None, - url_label=None, - ) - ], - ), - ( - """Rule match: action_mapper present without url, - return actions from matching comms from rule""", - "defaultcomms", - "InternalBookNBS", - { - "InternalBookNBS": AvailableAction( - ActionType=book_nbs_comms.action_type, - ExternalRoutingCode=book_nbs_comms.action_code, - ActionDescription=book_nbs_comms.action_description, - ) - }, - [ - SuggestedAction( - internal_action_code=InternalActionCode("InternalBookNBS"), - action_type=ActionType(book_nbs_comms.action_type), - action_code=ActionCode(book_nbs_comms.action_code), - action_description=ActionDescription(book_nbs_comms.action_description), - url_link=None, - url_label=None, - ) - ], - ), - ( - """Rule match: default_comms_routing missing, - comms present in rule, action_mapper missing, return no actions""", - "", - "InternalBookNBS", - {}, - [], - ), - ( - """Rule match: default_comms_routing missing, but action_mapper present, - return actions from matching comms from rule""", - "", - "InternalBookNBS", - {"InternalBookNBS": book_nbs_comms}, - [ - SuggestedAction( - internal_action_code=InternalActionCode("InternalBookNBS"), - action_type=ActionType("ButtonAuthLink"), - action_code=ActionCode("BookNBS"), - action_description=ActionDescription("Action description"), - url_link=UrlLink(HttpUrl("https://www.nhs.uk/book-rsv")), - url_label=UrlLabel("Continue to booking"), - ) - ], - ), - ( - """Rule match: default_comms_routing present, - comms present in rule, but action_mapper missing, return no actions""", - "defaultcommskeywithoutactionmapper", - "InternalBookNBS", - {}, - [], - ), - ( - """Rule match: default_comms_routing has multiple values, - one of the value is invalid, valid values should be returned in actions""", - "defaultcomms1|invaliddefault", - None, - {"defaultcomms1": default_comms_detail}, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms1"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), - url_link=None, - url_label=None, - ) - ], - ), - ], -) -def test_correct_actions_determined_from_redirect_r_rules( # noqa: PLR0913 - test_comment: str, - default_comms_routing: str, - comms_routing: str, - actions_mapper: ActionsMapper, - expected_actions: list[SuggestedAction], - faker: Faker, -): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb="QE1") - - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_comms_routing=default_comms_routing, - actions_mapper=rule_builder.ActionsMapperFactory.build(root=actions_mapper), - iteration_rules=[rule_builder.ICBRedirectRuleFactory.build(comms_routing=comms_routing)], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.actionable)) - .and_actions(equal_to(expected_actions)) - ) - ), - test_comment, - ) - - -@pytest.mark.parametrize( - ("test_comment", "redirect_r_rule_cohort_label"), - [ - ("cohort_label matches person cohort, result action ActionCode1", "cohort1"), - ("cohort_label NOT matches person cohort, result action ActionCode1", "cohort2"), - ], -) -def test_cohort_label_not_supported_used_in_r_rules(test_comment: str, redirect_r_rule_cohort_label: str, faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb="QE1") - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "ActionCode1": book_nbs_comms, - "defaultcomms": default_comms_detail, - } - ), - iteration_rules=[ - rule_builder.ICBRedirectRuleFactory.build( - cohort_label=CohortLabel(redirect_r_rule_cohort_label) - ) - ], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.actionable)) - .and_actions( - equal_to( - [ - SuggestedAction( - internal_action_code=InternalActionCode("ActionCode1"), - action_type=ActionType("ButtonAuthLink"), - action_code=ActionCode("BookNBS"), - action_description=ActionDescription("Action description"), - url_link=UrlLink(HttpUrl("https://www.nhs.uk/book-rsv")), - url_label=UrlLabel("Continue to booking"), - ) - ] - ) - ) - ) - ), - test_comment, - ) - - -def test_multiple_r_rules_match_with_same_priority(faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb="QE1") - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "rule_1_comms_routing": book_nbs_comms, - "rule_2_comms_routing": book_nbs_comms, - "rule_3_comms_routing": book_nbs_comms, - "defaultcomms": default_comms_detail, - } - ), - iteration_rules=[ - rule_builder.ICBRedirectRuleFactory.build(comms_routing="rule_1_comms_routing"), - rule_builder.ICBRedirectRuleFactory.build(comms_routing="rule_2_comms_routing"), - rule_builder.ICBRedirectRuleFactory.build( - priority=2, - attribute_name=RuleAttributeName("ICBMismatch"), - comms_routing="rule_3_comms_routing", - ), - ], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.actionable)) - .and_actions( - equal_to( - [ - SuggestedAction( - internal_action_code=InternalActionCode("rule_1_comms_routing"), - action_type=ActionType("ButtonAuthLink"), - action_code=ActionCode("BookNBS"), - action_description=ActionDescription("Action description"), - url_link=UrlLink(HttpUrl("https://www.nhs.uk/book-rsv")), - url_label=UrlLabel("Continue to booking"), - ) - ] - ) - ) - ) - ), - ) - - -def test_multiple_r_rules_with_same_priority_one_rule_mismatch_should_return_default_comms(faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb="QE1") - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "rule_1_comms_routing": book_nbs_comms, - "rule_2_comms_routing": book_nbs_comms, - "rule_3_comms_routing": book_nbs_comms, - "defaultcomms": default_comms_detail, - } - ), - iteration_rules=[ - rule_builder.ICBRedirectRuleFactory.build(comms_routing="rule_1_comms_routing"), - rule_builder.ICBRedirectRuleFactory.build(comms_routing="rule_2_comms_routing"), - rule_builder.ICBRedirectRuleFactory.build( - attribute_name=RuleAttributeName("ICBMismatch"), - comms_routing="rule_3_comms_routing", - ), - ], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.actionable)) - .and_actions( - equal_to( - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultcomms"), - action_type=ActionType("CareCardWithText"), - action_code=ActionCode("BookLocal"), - action_description=ActionDescription( - "You can get an RSV vaccination at your GP surgery" - ), - url_link=None, - url_label=None, - ) - ] - ) - ) - ) - ), - ) - - -def test_only_highest_priority_rule_is_applied_and_return_actions_only_for_that_rule(faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb="QE1") - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "rule_1_comms_routing": AvailableAction( - ActionType="ButtonAuthLink", - ExternalRoutingCode="BookNBS", - ActionDescription="Action description", - ), - "rule_2_comms_routing": AvailableAction( - ActionType="AuthLink", - ExternalRoutingCode="BookNBS", - ActionDescription="Action description", - UrlLink=HttpUrl("https://www.nhs.uk/book-rsv"), - UrlLabel="Continue to booking", - ), - "defaultcomms": default_comms_detail, - } - ), - iteration_rules=[ - rule_builder.ICBRedirectRuleFactory.build(priority=2, comms_routing="rule_2_comms_routing"), - rule_builder.ICBRedirectRuleFactory.build(priority=1, comms_routing="rule_1_comms_routing"), - ], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - expected_actions = SuggestedAction( - internal_action_code=InternalActionCode("rule_1_comms_routing"), - action_type=ActionType("ButtonAuthLink"), - action_code=ActionCode("BookNBS"), - action_description=ActionDescription("Action description"), - url_link=None, - url_label=None, - ) - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.actionable)) - .and_actions(equal_to([expected_actions])) - ) - ), - ) - - -def test_should_include_actions_when_include_actions_flag_is_true_when_status_is_actionable(faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb="QE1") - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "book_nbs": book_nbs_comms, - "defaultcomms": default_comms_detail, - } - ), - iteration_rules=[ - rule_builder.ICBRedirectRuleFactory.build(priority=2, comms_routing="book_nbs"), - ], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.actionable)) - .and_actions( - equal_to( - [ - SuggestedAction( - internal_action_code=InternalActionCode("book_nbs"), - action_type=ActionType("ButtonAuthLink"), - action_code=ActionCode("BookNBS"), - action_description=ActionDescription("Action description"), - url_link=UrlLink(HttpUrl("https://www.nhs.uk/book-rsv")), - url_label=UrlLabel("Continue to booking"), - ) - ] - ) - ) - ) - ), - ) - - -def test_should_not_include_actions_when_include_actions_flag_is_false_when_status_is_actionable(faker: Faker): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb="QE1") - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_comms_routing="defaultcomms", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "book_nbs": book_nbs_comms, - "defaultcomms": default_comms_detail, - } - ), - iteration_rules=[ - rule_builder.ICBRedirectRuleFactory.build(priority=2, comms_routing="book_nbs"), - ], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - actual = calculator.get_eligibility_status("N", ["ALL"], "ALL") - - # Then - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.actionable)) - .and_actions(equal_to(None)) - ) - ), - ) - - -@pytest.mark.parametrize( - ( - "test_comment", - "person_icb", - "default_comms_routing", - "comms_routing", - "actions_mapper", - "expected_actions", - "expected_audit_actions", - "expected_rule_priority", - "expected_rule_name", - ), - [ - ( - """Not eligible person with matching NonEligibleActionRule""", - "QE1", - "", - "ActionCode1", - { - "ActionCode1": AvailableAction( - ActionType="InfoText", - ExternalRoutingCode="HealthcareProInfo", - ActionDescription="""Speak to your healthcare professional if you think - you should be offered this vaccination.""", - ) - }, - [ - SuggestedAction( - internal_action_code=InternalActionCode("ActionCode1"), - action_type=ActionType("InfoText"), - action_code=ActionCode("HealthcareProInfo"), - action_description=ActionDescription( - """Speak to your healthcare professional if you think - you should be offered this vaccination.""" - ), - url_link=None, - url_label=None, - ) - ], - [ - AuditAction( - internal_action_code="ActionCode1", - action_code="HealthcareProInfo", - action_type="InfoText", - action_description="""Speak to your healthcare professional if you think - you should be offered this vaccination.""", - action_url=None, - action_url_label=None, - ) - ], - "20", - "In QE1", - ), - ( - """Not eligible person with NON matching NonEligibleActionRule""", - "WS3", - "defaultCommsCode", - "ActionCode1", - { - "defaultCommsCode": AvailableAction( - ActionType="DefaultInfoText", - ExternalRoutingCode="DefaultHealthcareProInfo", - ActionDescription="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - ) - }, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultCommsCode"), - action_type=ActionType("DefaultInfoText"), - action_code=ActionCode("DefaultHealthcareProInfo"), - action_description=ActionDescription( - """Default Speak to your healthcare professional if you think - you should be offered this vaccination.""" - ), - url_link=None, - url_label=None, - ) - ], - [ - AuditAction( - internal_action_code="defaultCommsCode", - action_code="DefaultHealthcareProInfo", - action_type="DefaultInfoText", - action_description="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - action_url=None, - action_url_label=None, - ) - ], - None, - None, - ), - ( - """Not eligible person with matching but missing NonEligibleActionRule, fall back to default comms""", - "QE1", - "defaultCommsCode", - "ActionCode1", - { - "defaultCommsCode": AvailableAction( - ActionType="DefaultInfoText", - ExternalRoutingCode="DefaultHealthcareProInfo", - ActionDescription="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - ) - }, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultCommsCode"), - action_type=ActionType("DefaultInfoText"), - action_code=ActionCode("DefaultHealthcareProInfo"), - action_description=ActionDescription( - """Default Speak to your healthcare professional if you think - you should be offered this vaccination.""" - ), - url_link=None, - url_label=None, - ) - ], - [ - AuditAction( - internal_action_code="defaultCommsCode", - action_code="DefaultHealthcareProInfo", - action_type="DefaultInfoText", - action_description="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - action_url=None, - action_url_label=None, - ) - ], - "20", - "In QE1", - ), - ], +default_comms_detail = AvailableAction( + ActionType="CareCardWithText", + ExternalRoutingCode="BookLocal", + ActionDescription="You can get an RSV vaccination at your GP surgery", ) -def test_correct_actions_determined_from_not_eligible_action_rules( # noqa: PLR0913 - app, - test_comment, - person_icb, - default_comms_routing, - comms_routing, - actions_mapper, - expected_actions, - expected_audit_actions, - expected_rule_priority, - expected_rule_name, - faker: Faker, -): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["NotEligibleCohort"], icb=person_icb) - - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_not_eligible_routing=default_comms_routing, - actions_mapper=rule_builder.ActionsMapperFactory.build(root=actions_mapper), - iteration_rules=[ - rule_builder.ICBNonEligibleActionRuleFactory.build(comms_routing=comms_routing) - ], - ) - ], - ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - with app.app_context(): - g.audit_log = AuditEvent() - - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.not_eligible)) - .and_actions(equal_to(expected_actions)) - ) - ), - test_comment, - ) - - cond = g.audit_log.response.condition[0] - assert cond.actions == expected_audit_actions - - assert getattr(cond.action_rule, "rule_priority", None) == expected_rule_priority - assert getattr(cond.action_rule, "rule_name", None) == expected_rule_name - - -def test_no_actions_returned_when_non_eligible_actions_and_defaultcomms_not_given( - app, - faker: Faker, -): - """ - ELI-295 - Campaign config without NonEligibleActions (X rules) should not return - any actions/default actions for NonEligible status - """ - # Given - nhs_number = NHSNumber(faker.nhs_number()) - person_rows = person_rows_builder(nhs_number, cohorts=["NotEligibleCohort"]) +class TestEligibilityResultBuilder: + def test_build_condition_results_empty_input(self): + condition_results = {} + result = EligibilityCalculator.build_condition_results(condition_results) + assert_that(result, is_([])) - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - actions_mapper={}, - iteration_rules=[], - ) - ], + def test_build_condition_results_single_condition_single_cohort_actionable(self): + cohort_group_results = [CohortGroupResult("COHORT_A", Status.actionable, [], "Cohort A Description", [])] + suggested_actions = [ + SuggestedAction( + internal_action_code=InternalActionCode("default_action_code"), + action_type=ActionType("CareCardWithText"), + action_code=ActionCode("BookLocal"), + action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), + url_link=None, + url_label=None, ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - with app.app_context(): - g.audit_log = AuditEvent() - - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - expected_actions = [] - expected_audit_action = [] - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.not_eligible)) - .and_actions(equal_to(expected_actions)) - ) - ), - ) - - cond = g.audit_log.response.condition[0] - assert cond.actions == expected_audit_action + ] + iteration_result = IterationResult(Status.actionable, cohort_group_results, suggested_actions) + condition_results = {ConditionName("RSV"): iteration_result} -def test_actions_returned_when_non_eligible_actions_not_given_and_defaultcomms_given( - app, - faker: Faker, -): - """ - ELI-295 - Campaign config without NonEligibleActions (X rules) but with default comms routing - should return the default comms actions - """ + result = EligibilityCalculator.build_condition_results(condition_results) - # Given - nhs_number = NHSNumber(faker.nhs_number()) + assert_that(len(result), is_(1)) + assert_that(result[0].condition_name, is_(ConditionName("RSV"))) + assert_that(result[0].status, is_(Status.actionable)) + assert_that(result[0].actions, is_(suggested_actions)) + assert_that(result[0].status_text, is_(Status.actionable.get_status_text(ConditionName("RSV")))) - person_rows = person_rows_builder(nhs_number, cohorts=["NotEligibleCohort"]) + assert_that(len(result[0].cohort_results), is_(1)) + deduplicated_cohort = result[0].cohort_results[0] + assert_that(deduplicated_cohort.cohort_code, is_("COHORT_A")) + assert_that(deduplicated_cohort.status, is_(Status.actionable)) + assert_that(deduplicated_cohort.reasons, is_([])) + assert_that(deduplicated_cohort.description, is_("Cohort A Description")) + assert_that(deduplicated_cohort.audit_rules, is_([])) - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_not_eligible_routing="defaultCommsCode", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "defaultCommsCode": AvailableAction( - ActionType="DefaultInfoText", - ExternalRoutingCode="DefaultHealthcareProInfo", - ActionDescription="Default Speak to your healthcare professional.", - ) - } - ), - iteration_rules=[], - ) - ], + def test_build_condition_results_single_condition_single_cohort_not_eligible_with_reasons(self): + cohort_group_results = [CohortGroupResult("COHORT_A", Status.not_eligible, [], "Cohort A Description", [])] + suggested_actions = [ + SuggestedAction( + internal_action_code=InternalActionCode("default_action_code"), + action_type=ActionType("CareCardWithText"), + action_code=ActionCode("BookLocal"), + action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), + url_link=None, + url_label=None, ) - ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - with app.app_context(): - g.audit_log = AuditEvent() - - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - expected_actions = [ + ] + iteration_result = IterationResult(Status.not_eligible, cohort_group_results, suggested_actions) + + condition_results = {ConditionName("RSV"): iteration_result} + + result = EligibilityCalculator.build_condition_results(condition_results) + + assert_that(len(result), is_(1)) + assert_that(result[0].condition_name, is_(ConditionName("RSV"))) + assert_that(result[0].status, is_(Status.not_eligible)) + assert_that(result[0].actions, is_(suggested_actions)) + assert_that(result[0].status_text, is_(Status.not_eligible.get_status_text(ConditionName("RSV")))) + + assert_that(len(result[0].cohort_results), is_(1)) + deduplicated_cohort = result[0].cohort_results[0] + assert_that(deduplicated_cohort.cohort_code, is_("COHORT_A")) + assert_that(deduplicated_cohort.status, is_(Status.not_eligible)) + assert_that(deduplicated_cohort.reasons, is_([])) + assert_that(deduplicated_cohort.description, is_("Cohort A Description")) + assert_that(deduplicated_cohort.audit_rules, is_([])) + + def test_build_condition_results_single_condition_multiple_cohorts_same_cohort_code_same_status(self): + reason_1 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 1"), + RulePriority("1"), + RuleDescription("Filter Rule Description 2"), + matcher_matched=True, + ) + reason_2 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 2"), + RulePriority("2"), + RuleDescription("Filter Rule Description 2"), + matcher_matched=True, + ) + cohort_group_results = [ + CohortGroupResult("COHORT_A", Status.not_eligible, [reason_1], "", []), + # The below description will be picked up as the first one is empty + CohortGroupResult("COHORT_A", Status.not_eligible, [reason_2], "Cohort A Description 2", []), + CohortGroupResult("COHORT_A", Status.not_eligible, [], "Cohort A Description 3", []), + ] + suggested_actions = [ SuggestedAction( - internal_action_code=InternalActionCode("defaultCommsCode"), - action_type=ActionType("DefaultInfoText"), - action_code=ActionCode("DefaultHealthcareProInfo"), - action_description=ActionDescription("Default Speak to your healthcare professional."), + internal_action_code=InternalActionCode("default_action_code"), + action_type=ActionType("CareCardWithText"), + action_code=ActionCode("BookLocal"), + action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), url_link=None, url_label=None, ) ] - expected_audit_action = [ - AuditAction( - internal_action_code="defaultCommsCode", - action_code="DefaultHealthcareProInfo", - action_type="DefaultInfoText", - action_description="Default Speak to your healthcare professional.", - action_url=None, - action_url_label=None, + iteration_result = IterationResult(Status.not_eligible, cohort_group_results, suggested_actions) + + condition_results = {ConditionName("RSV"): iteration_result} + + result = EligibilityCalculator.build_condition_results(condition_results) + + assert_that(len(result), is_(1)) + condition = result[0] + assert_that(len(condition.cohort_results), is_(1)) + + deduplicated_cohort = condition.cohort_results[0] + assert_that(deduplicated_cohort.cohort_code, is_("COHORT_A")) + assert_that(deduplicated_cohort.status, is_(Status.not_eligible)) + assert_that(deduplicated_cohort.reasons, contains_inanyorder(reason_1, reason_2)) + assert_that(deduplicated_cohort.description, is_("Cohort A Description 2")) + assert_that(deduplicated_cohort.audit_rules, is_([])) + + def test_build_condition_results_multiple_cohorts_different_cohort_code_same_status(self): + reason_1 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 1"), + RulePriority("1"), + RuleDescription("Filter Rule Description 2"), + matcher_matched=True, + ) + reason_2 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 2"), + RulePriority("2"), + RuleDescription("Filter Rule Description 2"), + matcher_matched=True, + ) + cohort_group_results = [ + CohortGroupResult("COHORT_X", Status.not_eligible, [reason_1], "Cohort X Description", []), + CohortGroupResult("COHORT_Y", Status.not_eligible, [reason_2], "Cohort Y Description", []), + ] + suggested_actions = [ + SuggestedAction( + internal_action_code=InternalActionCode("default_action_code"), + action_type=ActionType("CareCardWithText"), + action_code=ActionCode("BookLocal"), + action_description=ActionDescription("You can get an RSV vaccination at your GP surgery"), + url_link=None, + url_label=None, ) ] - assert_that( - actual, - is_eligibility_status().with_conditions( - has_items( - is_condition() - .with_condition_name(ConditionName("RSV")) - .and_status(equal_to(Status.not_eligible)) - .and_actions(equal_to(expected_actions)) - ) - ), - ) - - cond = g.audit_log.response.condition[0] - assert cond.actions == expected_audit_action - - -@pytest.mark.parametrize( - ( - "test_comment", - "person_icb", - "default_comms_routing", - "comms_routing", - "actions_mapper", - "expected_actions", - "expected_audit_actions", - ), - [ - ( - """Not actionable person with matching NonActionableActionRule""", - "QE1", - "", - "ActionCode1", - { - "ActionCode1": AvailableAction( - ActionType="InfoText", - ExternalRoutingCode="HealthcareProInfo", - ActionDescription="""Speak to your healthcare professional if you think - you should be offered this vaccination.""", - ) - }, - [ - SuggestedAction( - internal_action_code=InternalActionCode("ActionCode1"), - action_type=ActionType("InfoText"), - action_code=ActionCode("HealthcareProInfo"), - action_description=ActionDescription( - """Speak to your healthcare professional if you think - you should be offered this vaccination.""" - ), - url_link=None, - url_label=None, - ) - ], - [ - AuditAction( - internal_action_code="ActionCode1", - action_code="HealthcareProInfo", - action_type="InfoText", - action_description="""Speak to your healthcare professional if you think - you should be offered this vaccination.""", - action_url=None, - action_url_label=None, - ) - ], - ), - ( - """Not actionable person with NON matching NonActionableActionRule""", - "WS3", - "defaultCommsCode", - "ActionCode1", - { - "defaultCommsCode": AvailableAction( - ActionType="DefaultInfoText", - ExternalRoutingCode="DefaultHealthcareProInfo", - ActionDescription="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - ) - }, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultCommsCode"), - action_type=ActionType("DefaultInfoText"), - action_code=ActionCode("DefaultHealthcareProInfo"), - action_description=ActionDescription( - """Default Speak to your healthcare professional if you think - you should be offered this vaccination.""" - ), - url_link=None, - url_label=None, - ) - ], - [ - AuditAction( - internal_action_code="defaultCommsCode", - action_code="DefaultHealthcareProInfo", - action_type="DefaultInfoText", - action_description="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - action_url=None, - action_url_label=None, - ) - ], - ), - ( - """Not actionable person with matching but missing NonActionableActionRule, fall back to default comms""", - "QE1", - "defaultCommsCode", - "ActionCode1", - { - "defaultCommsCode": AvailableAction( - ActionType="DefaultInfoText", - ExternalRoutingCode="DefaultHealthcareProInfo", - ActionDescription="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - ) - }, - [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultCommsCode"), - action_type=ActionType("DefaultInfoText"), - action_code=ActionCode("DefaultHealthcareProInfo"), - action_description=ActionDescription( - """Default Speak to your healthcare professional if you think - you should be offered this vaccination.""" - ), - url_link=None, - url_label=None, - ) - ], - [ - AuditAction( - internal_action_code="defaultCommsCode", - action_code="DefaultHealthcareProInfo", - action_type="DefaultInfoText", - action_description="""Default Speak to your healthcare professional if you think - you should be offered this vaccination.""", - action_url=None, - action_url_label=None, - ) - ], - ), - ], -) -def test_correct_actions_determined_from_not_actionable_action_rules( # noqa: PLR0913 - app, - test_comment, - person_icb, - default_comms_routing, - comms_routing, - actions_mapper, - expected_actions, - expected_audit_actions, - faker: Faker, -): - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], icb=person_icb, de=True) + iteration_result = IterationResult(Status.not_eligible, cohort_group_results, suggested_actions) - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_not_actionable_routing=default_comms_routing, - actions_mapper=rule_builder.ActionsMapperFactory.build(root=actions_mapper), - iteration_rules=[ - rule_builder.DetainedEstateSuppressionRuleFactory.build(), - rule_builder.ICBNonActionableActionRuleFactory.build(comms_routing=comms_routing), - ], - ) - ], - ) - ) - ] + condition_results = {ConditionName("RSV"): iteration_result} - calculator = EligibilityCalculator(person_rows, campaign_configs) + result = EligibilityCalculator.build_condition_results(condition_results) - # When - with app.app_context(): - g.audit_log = AuditEvent() - - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - 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_actions(equal_to(expected_actions)) - ) - ), - test_comment, - ) + assert_that(len(result), is_(1)) + condition = result[0] + assert_that(len(condition.cohort_results), is_(2)) - cond = g.audit_log.response.condition[0] - assert cond.actions == expected_audit_actions + expected_deduplicated_cohorts = [ + CohortGroupResult("COHORT_X", Status.not_eligible, [reason_1], "Cohort X Description", []), + CohortGroupResult("COHORT_Y", Status.not_eligible, [reason_2], "Cohort Y Description", []), + ] + assert_that(condition.cohort_results, contains_inanyorder(*expected_deduplicated_cohorts)) + + def test_build_condition_results_cohorts_status_not_matching_iteration_status(self): + reason_1 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 1"), + RulePriority("1"), + RuleDescription("Matching"), + matcher_matched=True, + ) + reason_2 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 2"), + RulePriority("2"), + RuleDescription("Not matching"), + matcher_matched=True, + ) + cohort_group_results = [ + CohortGroupResult("COHORT_X", Status.not_eligible, [reason_1], "Cohort X Description", []), + CohortGroupResult("COHORT_Y", Status.not_actionable, [reason_2], "Cohort Y Description", []), + ] + iteration_result = IterationResult(Status.not_eligible, cohort_group_results, []) -def test_no_actions_returned_when_non_actionable_actions_and_defaultcomms_not_given( - app, - faker: Faker, -): - """ - ELI-295 - Campaign config without NonActionableActions (Y rules) should not return - any actions/default actions for NonActionable status - """ + condition_results = {ConditionName("RSV"): iteration_result} - # Given - nhs_number = NHSNumber(faker.nhs_number()) + result = EligibilityCalculator.build_condition_results(condition_results) - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], de=True) + assert_that(len(result), is_(1)) + condition = result[0] + assert_that(len(condition.cohort_results), is_(1)) + assert_that(condition.cohort_results[0].cohort_code, is_("COHORT_X")) + assert_that(condition.cohort_results[0].status, is_(Status.not_eligible)) - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - actions_mapper={}, - iteration_rules=[rule_builder.DetainedEstateSuppressionRuleFactory.build()], - ) - ], - ) + def test_build_condition_results_multiple_conditions(self): + reason_1 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 1"), + RulePriority("1"), + RuleDescription("Filter Rule Description 2"), + matcher_matched=True, ) - ] - - calculator = EligibilityCalculator(person_rows, campaign_configs) - - # When - with app.app_context(): - g.audit_log = AuditEvent() - - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") - - # Then - expected_actions = [] - expected_audit_action = [] - 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_actions(equal_to(expected_actions)) - ) - ), + reason_2 = Reason( + RuleType.filter, + eligibility_status.RuleName("Filter Rule 2"), + RulePriority("2"), + RuleDescription("Filter Rule Description 2"), + matcher_matched=True, ) + cohort_group_result1 = [CohortGroupResult("RSV_COHORT", Status.not_eligible, [reason_1], "RSV Desc", [])] + cohort_group_result2 = [CohortGroupResult("COVID_COHORT", Status.not_actionable, [reason_2], "Covid Desc", [])] - cond = g.audit_log.response.condition[0] - assert cond.actions == expected_audit_action - - -def test_actions_returned_when_non_actionable_actions_not_given_and_defaultcomms_given( - app, - faker: Faker, -): - """ - ELI-295 - Campaign config without NonActionableActions (Y rules) with default comms routing - should return default comms actions - """ - - # Given - nhs_number = NHSNumber(faker.nhs_number()) - - person_rows = person_rows_builder(nhs_number, cohorts=["cohort1"], de=True) - - campaign_configs = [ - ( - rule_builder.CampaignConfigFactory.build( - target="RSV", - iterations=[ - rule_builder.IterationFactory.build( - iteration_cohorts=[rule_builder.IterationCohortFactory.build(cohort_label="cohort1")], - default_not_actionable_routing="defaultCommsCode", - actions_mapper=rule_builder.ActionsMapperFactory.build( - root={ - "defaultCommsCode": AvailableAction( - ActionType="DefaultInfoText", - ExternalRoutingCode="DefaultHealthcareProInfo", - ActionDescription="Default Speak to your healthcare professional.", - ) - } - ), - iteration_rules=[rule_builder.DetainedEstateSuppressionRuleFactory.build()], - ) - ], - ) - ) - ] + iteration_result1 = IterationResult(Status.not_eligible, cohort_group_result1, []) - calculator = EligibilityCalculator(person_rows, campaign_configs) + iteration_result2 = IterationResult(Status.not_actionable, cohort_group_result2, []) - # When - with app.app_context(): - g.audit_log = AuditEvent() + condition_results = { + ConditionName("RSV"): iteration_result1, + ConditionName("COVID"): iteration_result2, + } - actual = calculator.get_eligibility_status("Y", ["ALL"], "ALL") + result = EligibilityCalculator.build_condition_results(condition_results) - # Then - expected_actions = [ - SuggestedAction( - internal_action_code=InternalActionCode("defaultCommsCode"), - action_type=ActionType("DefaultInfoText"), - action_code=ActionCode("DefaultHealthcareProInfo"), - action_description=ActionDescription("Default Speak to your healthcare professional."), - url_link=None, - url_label=None, - ) - ] - expected_audit_action = [ - AuditAction( - internal_action_code="defaultCommsCode", - action_code="DefaultHealthcareProInfo", - action_type="DefaultInfoText", - action_description="Default Speak to your healthcare professional.", - action_url=None, - action_url_label=None, - ) - ] - 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_actions(equal_to(expected_actions)) - ) - ), - ) + rsv = next((c for c in result if c.condition_name == ConditionName("RSV")), None) + assert_that(rsv.status, is_(Status.not_eligible)) + assert_that(len(rsv.cohort_results), is_(1)) + assert_that(rsv.cohort_results[0].cohort_code, is_("RSV_COHORT")) + assert_that(rsv.cohort_results[0].reasons, is_([reason_1])) - cond = g.audit_log.response.condition[0] - assert cond.actions == expected_audit_action + covid = next((c for c in result if c.condition_name == ConditionName("COVID")), None) + assert_that(covid.status, is_(Status.not_actionable)) + assert_that(len(covid.cohort_results), is_(1)) + assert_that(covid.cohort_results[0].cohort_code, is_("COVID_COHORT")) + assert_that(covid.cohort_results[0].reasons, is_([reason_2])) diff --git a/tests/unit/services/processors/test_action_rule_handler.py b/tests/unit/services/processors/test_action_rule_handler.py index bc8d04ca1..13e9d4592 100644 --- a/tests/unit/services/processors/test_action_rule_handler.py +++ b/tests/unit/services/processors/test_action_rule_handler.py @@ -79,6 +79,23 @@ def test_get_action_rules_components_not_eligible_actions_type(): assert_that(default_comms, is_("default_not_eligible")) +def test_get_action_rules_components_not_actionable_actions_type(): + iteration = rule_builder.IterationFactory.build( + default_comms_routing="default_redirect", + default_not_eligible_routing="default_not_eligible", + default_not_actionable_routing="default_not_actionable", + actions_mapper=ActionsMapperFactory.build(), + iteration_rules=[rule_builder.ICBNonActionableActionRuleFactory.build(name="NonActionableRule")], + ) + rules_found, mapper, default_comms = ActionRuleHandler._get_action_rules_components( + iteration, RuleType.not_actionable_actions + ) + assert_that(len(rules_found), is_(1)) + assert_that(rules_found[0].name, is_(RuleName("NonActionableRule"))) + assert_that(mapper, is_(iteration.actions_mapper)) + assert_that(default_comms, is_("default_not_actionable")) + + def test_get_action_rules_components_no_matching_rules(): iteration = rule_builder.IterationFactory.build( iteration_rules=[rule_builder.PersonAgeSuppressionRuleFactory.build()] @@ -171,7 +188,7 @@ def test_handle_actions_no_matching_rules_returns_default( @patch("eligibility_signposting_api.services.processors.action_rule_handler.RuleCalculator") @patch.object(ActionRuleHandler, "_get_actions_from_comms") @patch.object(ActionRuleHandler, "_get_action_rules_components") -def test_handle_actions_matching_rule_overrides_default( +def test_handle_actions_matching_redirect_rule_overrides_default( mock_get_action_rules_components, mock_get_actions_from_comms, mock_rule_calculator_class, @@ -234,6 +251,138 @@ def test_handle_actions_matching_rule_overrides_default( mock_rule_calculator_class.assert_called_once_with(person=MOCK_PERSON, rule=matching_rule) +@patch("eligibility_signposting_api.services.processors.action_rule_handler.RuleCalculator") +@patch.object(ActionRuleHandler, "_get_actions_from_comms") +@patch.object(ActionRuleHandler, "_get_action_rules_components") +def test_handle_actions_matching_not_eligible_rule_overrides_default( + mock_get_action_rules_components, + mock_get_actions_from_comms, + mock_rule_calculator_class, + handler: ActionRuleHandler, +): + matching_rule = rule_builder.ICBNonEligibleActionRuleFactory.build( + priority=10, comms_routing="rule_specific_action", name="RuleSpecificAction" + ) + active_iteration = rule_builder.IterationFactory.build( + default_not_eligible_routing="default_not_eligible", + actions_mapper=ActionsMapperFactory.build( + root={"default_not_eligible": DEFAULT_COMMS_DETAIL, "rule_specific_action": BOOK_NBS_COMMS} + ), + iteration_rules=[matching_rule], + ) + mock_get_action_rules_components.return_value = ( + (matching_rule,), + active_iteration.actions_mapper, + active_iteration.default_not_eligible_routing, + ) + + mock_get_actions_from_comms.side_effect = [ + [ + SuggestedAction( + internal_action_code=InternalActionCode("default_not_eligible"), + action_type=ActionType(DEFAULT_COMMS_DETAIL.action_type), + action_code=ActionCode(DEFAULT_COMMS_DETAIL.action_code), + action_description=ActionDescription(DEFAULT_COMMS_DETAIL.action_description), + url_link=DEFAULT_COMMS_DETAIL.url_link, + url_label=DEFAULT_COMMS_DETAIL.url_label, + ) + ], + [ + SuggestedAction( + internal_action_code=InternalActionCode("rule_specific_action"), + action_type=ActionType(BOOK_NBS_COMMS.action_type), + action_code=ActionCode(BOOK_NBS_COMMS.action_code), + action_description=ActionDescription(BOOK_NBS_COMMS.action_description), + url_link=BOOK_NBS_COMMS.url_link, + url_label=BOOK_NBS_COMMS.url_label, + ) + ], + ] + + mock_rule_instance = Mock() + mock_rule_instance.evaluate_exclusion.return_value = (Status.actionable, Mock(matcher_matched=True)) + mock_rule_calculator_class.return_value = mock_rule_instance + + matched_action_detail = handler._handle(MOCK_PERSON, active_iteration, RuleType.not_eligible_actions) + + assert_that(len(matched_action_detail.actions), is_(1)) + assert_that(matched_action_detail.actions[0].internal_action_code, is_(InternalActionCode("rule_specific_action"))) + assert_that(matched_action_detail.rule_priority, is_(RulePriority(10))) + assert_that(matched_action_detail.rule_name, is_(RuleName("RuleSpecificAction"))) + + mock_get_action_rules_components.assert_called_once_with(active_iteration, RuleType.not_eligible_actions) + assert_that(mock_get_actions_from_comms.call_count, is_(2)) + mock_get_actions_from_comms.assert_any_call(active_iteration.actions_mapper, "default_not_eligible") + mock_get_actions_from_comms.assert_any_call(active_iteration.actions_mapper, "rule_specific_action") + mock_rule_calculator_class.assert_called_once_with(person=MOCK_PERSON, rule=matching_rule) + + +@patch("eligibility_signposting_api.services.processors.action_rule_handler.RuleCalculator") +@patch.object(ActionRuleHandler, "_get_actions_from_comms") +@patch.object(ActionRuleHandler, "_get_action_rules_components") +def test_handle_actions_matching_not_actionable_rule_overrides_default( + mock_get_action_rules_components, + mock_get_actions_from_comms, + mock_rule_calculator_class, + handler: ActionRuleHandler, +): + matching_rule = rule_builder.ICBNonActionableActionRuleFactory.build( + priority=10, comms_routing="rule_specific_action", name="RuleSpecificAction" + ) + active_iteration = rule_builder.IterationFactory.build( + default_not_actionable_routing="default_not_actionable", + actions_mapper=ActionsMapperFactory.build( + root={"default_not_actionable": DEFAULT_COMMS_DETAIL, "rule_specific_action": BOOK_NBS_COMMS} + ), + iteration_rules=[matching_rule], + ) + mock_get_action_rules_components.return_value = ( + (matching_rule,), + active_iteration.actions_mapper, + active_iteration.default_not_actionable_routing, + ) + + mock_get_actions_from_comms.side_effect = [ + [ + SuggestedAction( + internal_action_code=InternalActionCode("default_not_actionable"), + action_type=ActionType(DEFAULT_COMMS_DETAIL.action_type), + action_code=ActionCode(DEFAULT_COMMS_DETAIL.action_code), + action_description=ActionDescription(DEFAULT_COMMS_DETAIL.action_description), + url_link=DEFAULT_COMMS_DETAIL.url_link, + url_label=DEFAULT_COMMS_DETAIL.url_label, + ) + ], + [ + SuggestedAction( + internal_action_code=InternalActionCode("rule_specific_action"), + action_type=ActionType(BOOK_NBS_COMMS.action_type), + action_code=ActionCode(BOOK_NBS_COMMS.action_code), + action_description=ActionDescription(BOOK_NBS_COMMS.action_description), + url_link=BOOK_NBS_COMMS.url_link, + url_label=BOOK_NBS_COMMS.url_label, + ) + ], + ] + + mock_rule_instance = Mock() + mock_rule_instance.evaluate_exclusion.return_value = (Status.actionable, Mock(matcher_matched=True)) + mock_rule_calculator_class.return_value = mock_rule_instance + + matched_action_detail = handler._handle(MOCK_PERSON, active_iteration, RuleType.not_actionable_actions) + + assert_that(len(matched_action_detail.actions), is_(1)) + assert_that(matched_action_detail.actions[0].internal_action_code, is_(InternalActionCode("rule_specific_action"))) + assert_that(matched_action_detail.rule_priority, is_(RulePriority(10))) + assert_that(matched_action_detail.rule_name, is_(RuleName("RuleSpecificAction"))) + + mock_get_action_rules_components.assert_called_once_with(active_iteration, RuleType.not_actionable_actions) + assert_that(mock_get_actions_from_comms.call_count, is_(2)) + mock_get_actions_from_comms.assert_any_call(active_iteration.actions_mapper, "default_not_actionable") + mock_get_actions_from_comms.assert_any_call(active_iteration.actions_mapper, "rule_specific_action") + mock_rule_calculator_class.assert_called_once_with(person=MOCK_PERSON, rule=matching_rule) + + @patch("eligibility_signposting_api.services.processors.action_rule_handler.RuleCalculator") @patch.object(ActionRuleHandler, "_get_actions_from_comms") @patch.object(ActionRuleHandler, "_get_action_rules_components") diff --git a/tests/unit/services/processors/test_rule_processor.py b/tests/unit/services/processors/test_rule_processor.py index 1531c3a39..b97dea167 100644 --- a/tests/unit/services/processors/test_rule_processor.py +++ b/tests/unit/services/processors/test_rule_processor.py @@ -1,10 +1,10 @@ from unittest.mock import Mock, patch import pytest -from hamcrest import assert_that, empty, has_length, is_ +from hamcrest import assert_that, empty, is_ -from eligibility_signposting_api.model.campaign_config import RuleType -from eligibility_signposting_api.model.eligibility_status import CohortGroupResult, Reason, Status +from eligibility_signposting_api.model.campaign_config import CohortLabel, RuleType +from eligibility_signposting_api.model.eligibility_status import CohortGroupResult, Reason, RuleName, Status from eligibility_signposting_api.model.person import Person from eligibility_signposting_api.services.processors.person_data_reader import PersonDataReader from eligibility_signposting_api.services.processors.rule_processor import RuleProcessor @@ -116,8 +116,8 @@ def test_evaluate_rules_priority_group_one_not_eligible(mock_rule_calculator_cla status, reasons, is_rule_stop = rule_processor.evaluate_rules_priority_group(MOCK_PERSON_DATA, rules_group) assert_that(status, is_(Status.actionable)) - assert_that(reasons, has_length(1)) - assert_that(reasons[0].rule_name, is_("ExclusionReason")) + assert_that(len(reasons), is_(1)) + assert_that(reasons[0].rule_name, is_(RuleName("ExclusionReason"))) assert_that(is_rule_stop, is_(False)) assert_that(mock_rule_calculator_class.call_count, is_(2)) @@ -140,7 +140,7 @@ def test_evaluate_rules_priority_group_with_rule_stop(mock_rule_calculator_class status, reasons, is_rule_stop = rule_processor.evaluate_rules_priority_group(MOCK_PERSON_DATA, rules_group) assert_that(status, is_(Status.actionable)) - assert_that(reasons, has_length(1)) + assert_that(len(reasons), is_(1)) assert_that(is_rule_stop, is_(True)) @@ -180,7 +180,7 @@ def test_is_eligible_by_filter_rules_not_eligible( is_eligible = rule_processor.is_eligible(MOCK_PERSON_DATA, cohort, cohort_results, filter_rules) assert_that(is_eligible, is_(False)) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.not_eligible)) assert_that(cohort_results["COHORT_A"].description, is_("Not Eligible")) assert_that(cohort_results["COHORT_A"].audit_rules, is_([mock_reason])) @@ -202,7 +202,7 @@ def test_evaluate_suppression_rules_actionable( rule_processor.is_actionable(MOCK_PERSON_DATA, cohort, cohort_results, suppression_rules) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.actionable)) assert_that(cohort_results["COHORT_A"].description, is_("Actionable")) assert_that(cohort_results["COHORT_A"].reasons, is_([])) @@ -228,7 +228,7 @@ def test_evaluate_suppression_rules_not_actionable( rule_processor.is_actionable(MOCK_PERSON_DATA, cohort, cohort_results, suppression_rules) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.not_actionable)) assert_that(cohort_results["COHORT_A"].description, is_("Positive Description")) assert_that(cohort_results["COHORT_A"].reasons, is_([mock_reason])) @@ -260,7 +260,7 @@ def test_evaluate_suppression_rules_stops_on_rule_stop( rule_processor.is_actionable(MOCK_PERSON_DATA, cohort, cohort_results, suppression_rules) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.not_actionable)) assert_that(cohort_results["COHORT_A"].reasons, is_([mock_reason_p1])) assert_that(cohort_results["COHORT_A"].audit_rules, is_([mock_reason_p1])) @@ -291,7 +291,7 @@ def test_evaluate_suppression_rules_does_not_stop_on_rule_stop_when_status_is_ac rule_processor.is_actionable(MOCK_PERSON_DATA, cohort, cohort_results, suppression_rules) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.not_actionable)) assert_that(cohort_results["COHORT_A"].reasons, is_([mock_reason_p2])) assert_that(cohort_results["COHORT_A"].audit_rules, is_([mock_reason_p2])) @@ -405,7 +405,7 @@ def mock_evaluate_side_effect(person, rules_group): # noqa: ARG001 is_eligible = rule_processor.is_eligible(MOCK_PERSON_DATA, cohort, cohort_results, filter_rules) assert_that(is_eligible, is_(False)) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.not_eligible)) assert_that(cohort_results["COHORT_A"].description, is_("Not Eligible")) assert_that(cohort_results["COHORT_A"].audit_rules, is_([mock_reason])) @@ -427,7 +427,7 @@ def test_is_actionable_by_suppression_rules( rule_processor.is_actionable(MOCK_PERSON_DATA, cohort, cohort_results, suppression_rules) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.actionable)) assert_that(cohort_results["COHORT_A"].description, is_("Actionable")) assert_that(cohort_results["COHORT_A"].reasons, is_(empty())) @@ -463,10 +463,133 @@ def mock_evaluate_side_effect(person, rules_group): # noqa: ARG001 rule_processor.is_actionable(MOCK_PERSON_DATA, cohort, cohort_results, suppression_rules) - assert_that(cohort_results, has_length(1)) + assert_that(len(cohort_results), is_(1)) assert_that(cohort_results["COHORT_A"].status, is_(Status.not_actionable)) assert_that(cohort_results["COHORT_A"].description, is_("Positive Description")) assert_that(cohort_results["COHORT_A"].reasons, is_([mock_reason])) assert_that(cohort_results["COHORT_A"].audit_rules, is_([mock_reason])) mock_get_exclusion_rules.assert_called_once_with(cohort, suppression_rules) mock_evaluate_rules_priority_group.assert_called_once() + + +@patch.object(RuleProcessor, "get_rules_by_type") +@patch("eligibility_signposting_api.services.processors.rule_processor.BaseEligibilityHandler") +@patch("eligibility_signposting_api.services.processors.rule_processor.FilterRuleHandler") +@patch("eligibility_signposting_api.services.processors.rule_processor.SuppressionRuleHandler") +def test_get_cohort_group_results( + mock_suppression_handler_class, + mock_filter_handler_class, + mock_base_handler_class, + mock_get_rules_by_type, + rule_processor, +): + mock_base_handler_instance = mock_base_handler_class.return_value + mock_filter_handler_instance = mock_filter_handler_class.return_value + mock_suppression_handler_instance = mock_suppression_handler_class.return_value + + mock_base_handler_instance.next.return_value = mock_filter_handler_instance + mock_filter_handler_instance.next.return_value = mock_suppression_handler_instance + + cohort_a = rule_builder.IterationCohortFactory.build( + cohort_label="COHORT_A", priority=1, cohort_group="common_cohort" + ) + cohort_b = rule_builder.IterationCohortFactory.build( + cohort_label="COHORT_B", priority=2, cohort_group="common_cohort" + ) + active_iteration = rule_builder.IterationFactory.build( + iteration_cohorts=[cohort_a, cohort_b], + iteration_rules=[ + rule_builder.IterationRuleFactory.build(type=RuleType.filter, priority=1), + rule_builder.IterationRuleFactory.build(type=RuleType.suppression, priority=1), + ], + ) + + filter_rules = (rule_builder.IterationRuleFactory.build(type=RuleType.filter),) + suppression_rules = (rule_builder.IterationRuleFactory.build(type=RuleType.suppression),) + mock_get_rules_by_type.return_value = (filter_rules, suppression_rules) + + def mock_handle_side_effect(person, cohort, cohort_results_dict, rule_processor_instance): # noqa: ARG001 + if cohort.cohort_label == CohortLabel("COHORT_A"): + cohort_results_dict[CohortLabel("COHORT_A")] = CohortGroupResult( + cohort_code=cohort.cohort_group, + status=Status.actionable, + reasons=[], + description="Cohort A Description", + audit_rules=[], + ) + elif cohort.cohort_label == CohortLabel("COHORT_B"): + cohort_results_dict[CohortLabel("COHORT_B")] = CohortGroupResult( + cohort_code=cohort.cohort_group, + status=Status.not_eligible, + reasons=[], + description="Cohort B Description", + audit_rules=[], + ) + + mock_base_handler_instance.handle.side_effect = mock_handle_side_effect + + result = rule_processor.get_cohort_group_results(MOCK_PERSON_DATA, active_iteration) + + mock_get_rules_by_type.assert_called_once_with(active_iteration) + + mock_base_handler_class.assert_called_once_with() + mock_filter_handler_class.assert_called_once_with(filter_rules=filter_rules) + mock_suppression_handler_class.assert_called_once_with(suppression_rules=suppression_rules) + + mock_base_handler_instance.next.assert_called_once_with(mock_filter_handler_instance) + mock_filter_handler_instance.next.assert_called_once_with(mock_suppression_handler_instance) + + assert_that(mock_base_handler_instance.handle.call_count, is_(2)) + calls = mock_base_handler_instance.handle.call_args_list + assert_that(calls[0].args[1], is_(cohort_a)) + assert_that(calls[1].args[1], is_(cohort_b)) + + assert_that(len(result), is_(2)) + expected_result = { + CohortLabel("COHORT_A"): CohortGroupResult( + cohort_code=cohort_a.cohort_group, + status=Status.actionable, + reasons=[], + description="Cohort A Description", + audit_rules=[], + ), + CohortLabel("COHORT_B"): CohortGroupResult( + cohort_code=cohort_b.cohort_group, + status=Status.not_eligible, + reasons=[], + description="Cohort B Description", + audit_rules=[], + ), + } + assert_that(result, is_(expected_result)) + + assert_that(result[CohortLabel("COHORT_A")].status, is_(Status.actionable)) + assert_that(result[CohortLabel("COHORT_B")].status, is_(Status.not_eligible)) + + assert_that(result[CohortLabel("COHORT_A")].status, is_(Status.actionable)) + assert_that(result[CohortLabel("COHORT_B")].status, is_(Status.not_eligible)) + + +@patch.object(RuleProcessor, "get_rules_by_type", return_value=((), ())) +@patch("eligibility_signposting_api.services.processors.rule_processor.BaseEligibilityHandler") +@patch("eligibility_signposting_api.services.processors.rule_processor.FilterRuleHandler") +@patch("eligibility_signposting_api.services.processors.rule_processor.SuppressionRuleHandler") +def test_get_cohort_group_results_no_rules_no_cohorts( + mock_suppression_handler_class, + mock_filter_handler_class, + mock_base_handler_class, + mock_get_rules_by_type, + rule_processor, +): + mock_base_handler_instance = mock_base_handler_class.return_value + active_iteration = rule_builder.IterationFactory.build(iteration_cohorts=[], iteration_rules=[]) + + result = rule_processor.get_cohort_group_results(MOCK_PERSON_DATA, active_iteration) + + mock_get_rules_by_type.assert_called_once_with(active_iteration) + mock_base_handler_class.assert_called_once_with() + mock_filter_handler_class.assert_called_once_with(filter_rules=()) + mock_suppression_handler_class.assert_called_once_with(suppression_rules=()) + + mock_base_handler_instance.handle.assert_not_called() + assert_that(result, is_({}))