diff --git a/src/rules_validation_api/README.md b/src/rules_validation_api/README.md new file mode 100644 index 000000000..68d314d1a --- /dev/null +++ b/src/rules_validation_api/README.md @@ -0,0 +1,26 @@ +# ๐Ÿงช Campaign-config Validation + +This Python script is designed to validate a campaign configuration JSON file. + +## ๐Ÿ›  Requirements + +- Python 3.13 +- `rules_validation_api` must be installed and accessible +- Campaign configuration JSON file to verify + +## Steps to verify + +- Get to the `rules_validation_api` folder +- Run `python app.py --config_path ` + +## Results + +- `On success`: + + ```text + "Valid config" is printed + +- `On Failure`: + + ```text + "Errors" is printed diff --git a/src/rules_validation_api/__init__.py b/src/rules_validation_api/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/rules_validation_api/app.py b/src/rules_validation_api/app.py new file mode 100644 index 000000000..feb9464ba --- /dev/null +++ b/src/rules_validation_api/app.py @@ -0,0 +1,29 @@ +import argparse +import json +import sys +from pathlib import Path + +from rules_validation_api.validators.rules_validator import RulesValidation + +GREEN = "\033[92m" # pragma: no cover +RESET = "\033[0m" # pragma: no cover +YELLOW = "\033[93m" # pragma: no cover +RED = "\033[91m" # pragma: no cover + + +def main() -> None: # pragma: no cover + parser = argparse.ArgumentParser(description="Validate campaign configuration.") + parser.add_argument("--config_path", required=True, help="Path to the campaign config JSON file") + args = parser.parse_args() + + try: + with Path(args.config_path).open() as file: + json_data = json.load(file) + RulesValidation(**json_data) + sys.stdout.write(f"{GREEN}Valid Config{RESET}\n") + except ValueError as e: + sys.stderr.write(f"{YELLOW}Validation Error:{RESET} {RED}{e}{RESET}\n") + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/rules_validation_api/validators/__init__.py b/src/rules_validation_api/validators/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/rules_validation_api/validators/actions_mapper_validator.py b/src/rules_validation_api/validators/actions_mapper_validator.py new file mode 100644 index 000000000..a6eafaf87 --- /dev/null +++ b/src/rules_validation_api/validators/actions_mapper_validator.py @@ -0,0 +1,13 @@ +from pydantic import model_validator + +from eligibility_signposting_api.model.campaign_config import ActionsMapper + + +class ActionsMapperValidator(ActionsMapper): + @model_validator(mode="after") + def validate_keys(self) -> "ActionsMapperValidator": + invalid_keys = [key for key in self.root if key is None or key == ""] + if invalid_keys: + msg = f"Invalid keys found in ActionsMapper: {invalid_keys}" + raise ValueError(msg) + return self diff --git a/src/rules_validation_api/validators/available_action_validator.py b/src/rules_validation_api/validators/available_action_validator.py new file mode 100644 index 000000000..c8bf788d4 --- /dev/null +++ b/src/rules_validation_api/validators/available_action_validator.py @@ -0,0 +1,5 @@ +from eligibility_signposting_api.model.campaign_config import AvailableAction + + +class AvailableActionValidation(AvailableAction): + pass diff --git a/src/rules_validation_api/validators/campaign_config_validator.py b/src/rules_validation_api/validators/campaign_config_validator.py new file mode 100644 index 000000000..f25d740ff --- /dev/null +++ b/src/rules_validation_api/validators/campaign_config_validator.py @@ -0,0 +1,11 @@ +from pydantic import field_validator + +from eligibility_signposting_api.model.campaign_config import CampaignConfig, Iteration +from rules_validation_api.validators.iteration_validator import IterationValidation + + +class CampaignConfigValidation(CampaignConfig): + @classmethod + @field_validator("iterations") + def validate_iterations(cls, iterations: list[Iteration]) -> list[IterationValidation]: + return [IterationValidation(**i.model_dump()) for i in iterations] diff --git a/src/rules_validation_api/validators/iteration_rules_validator.py b/src/rules_validation_api/validators/iteration_rules_validator.py new file mode 100644 index 000000000..95d8dce66 --- /dev/null +++ b/src/rules_validation_api/validators/iteration_rules_validator.py @@ -0,0 +1,5 @@ +from eligibility_signposting_api.model.campaign_config import IterationRule + + +class IterationRuleValidation(IterationRule): + pass diff --git a/src/rules_validation_api/validators/iteration_validator.py b/src/rules_validation_api/validators/iteration_validator.py new file mode 100644 index 000000000..3bfd3fec5 --- /dev/null +++ b/src/rules_validation_api/validators/iteration_validator.py @@ -0,0 +1,37 @@ +import typing + +from pydantic import ValidationError, field_validator, model_validator +from pydantic_core import InitErrorDetails + +from eligibility_signposting_api.model.campaign_config import ActionsMapper, Iteration, IterationRule +from rules_validation_api.validators.actions_mapper_validator import ActionsMapperValidator +from rules_validation_api.validators.iteration_rules_validator import IterationRuleValidation + + +class IterationValidation(Iteration): + @classmethod + @field_validator("iteration_rules") + def validate_iterations(cls, iteration_rules: list[IterationRule]) -> list[IterationRuleValidation]: + return [IterationRuleValidation(**i.model_dump()) for i in iteration_rules] + + @classmethod + @field_validator("actions_mapper", mode="after") + def transform_actions_mapper(cls, action_mapper: ActionsMapper) -> ActionsMapper: + ActionsMapperValidator.model_validate(action_mapper.model_dump()) + return action_mapper + + @model_validator(mode="after") + def validate_default_comms_routing_in_actions_mapper(self) -> typing.Self: + default_routing = self.default_comms_routing + actions_mapper = self.actions_mapper.root.keys() + + if default_routing and (not actions_mapper or default_routing not in actions_mapper): + error = InitErrorDetails( + type="value_error", + loc=("actions_mapper",), + input=actions_mapper, + ctx={"error": f"Missing entry for DefaultCommsRouting '{default_routing}' in ActionsMapper"}, + ) + raise ValidationError.from_exception_data(title="IterationValidation", line_errors=[error]) + + return self diff --git a/src/rules_validation_api/validators/rules_validator.py b/src/rules_validation_api/validators/rules_validator.py new file mode 100644 index 000000000..8d52a545b --- /dev/null +++ b/src/rules_validation_api/validators/rules_validator.py @@ -0,0 +1,11 @@ +from pydantic import field_validator + +from eligibility_signposting_api.model.campaign_config import CampaignConfig, Rules +from rules_validation_api.validators.campaign_config_validator import CampaignConfigValidation + + +class RulesValidation(Rules): + @classmethod + @field_validator("campaign_config") + def validate_campaign_config(cls, campaign_config: CampaignConfig) -> CampaignConfig: + return CampaignConfigValidation(**campaign_config.model_dump()) diff --git a/tests/test_data/test_config/test_config.json b/tests/test_data/test_config/test_config.json index 643cbce2a..fe7b41ed6 100644 --- a/tests/test_data/test_config/test_config.json +++ b/tests/test_data/test_config/test_config.json @@ -5,9 +5,9 @@ "Name": "Test Config", "Type": "V", "Target": "RSV", - "Manager": "person@test.com", - "Approver": "person@test.com", - "Reviewer": "person@test.com", + "Manager": ["person@test.com"], + "Approver": ["person@test.com"], + "Reviewer": ["person@test.com"], "IterationFrequency": "X", "IterationType": "M", "IterationTime": "07:00:00", @@ -20,8 +20,8 @@ "DefaultNotEligibleRouting": "INTERNALCONTACTGP1", "ActionsMapper": { "INTERNALCONTACTGP1": {"ExternalRoutingCode": "CONTACTGP","ActionDescription":"Contact GP Text1 description", "ActionType":"text1"}, - "INTERNALCONTACTGP2": {"ExternalRoutingCode": "CONTACTGP","ActionDescription":"Contact GP Link description", "ActionType":"link", "UrlLink": "link123", "UrlLabel": "link label"}, - "INTERNALTESCO": {"ExternalRoutingCode": "TESCO","ActionDescription":"Tesco description", "ActionType":"link", "UrlLink": "tesco link", "UrlLabel": "link label"}, + "INTERNALCONTACTGP2": {"ExternalRoutingCode": "CONTACTGP","ActionDescription":"Contact GP Link description", "ActionType":"link", "UrlLink": "https://www.link123.example", "UrlLabel": "link label"}, + "INTERNALTESCO": {"ExternalRoutingCode": "TESCO","ActionDescription":"Tesco description", "ActionType":"link", "UrlLink": "https://www.tesco_link.example", "UrlLabel": "link label"}, "INTERNALFINDWALKIN": {"ExternalRoutingCode": "FINDWALKIN","ActionDescription":"Find walkin description", "ActionType":"button"}, "XRULEID1": {"ExternalRoutingCode": "FINDWALKIN","ActionDescription":"Find walkin description", "ActionType":"button"}, diff --git a/tests/unit/validation/__init__.py b/tests/unit/validation/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/validation/conftest.py b/tests/unit/validation/conftest.py new file mode 100644 index 000000000..efd625215 --- /dev/null +++ b/tests/unit/validation/conftest.py @@ -0,0 +1,69 @@ +import pytest + + +@pytest.fixture +def valid_campaign_config_with_only_mandatory_fields(): + return { + "ID": "CAMP001", + "Version": 1, + "Name": "Spring Campaign", + "Type": "V", + "Target": "COVID", + "IterationFrequency": "M", + "IterationType": "A", + "StartDate": "20250101", + "EndDate": "20250331", + "Iterations": [ + { + "ID": "ITER001", + "Version": 1, + "Name": "Mid-January Push", + "IterationDate": "20250101", + "IterationNumber": 1, + "ApprovalMinimum": 10, + "ApprovalMaximum": 100, + "Type": "A", + "DefaultCommsRouting": "BOOK_NBS", + "DefaultNotEligibleRouting": "RouteB", + "DefaultNotActionableRouting": "RouteC", + "IterationCohorts": [], + "IterationRules": [], + "ActionsMapper": { + "BOOK_NBS": { + "ExternalRoutingCode": "BookNBS", + "ActionDescription": "", + "ActionType": "ButtonWithAuthLink", + "UrlLink": "http://www.nhs.uk/book-rsv", + "UrlLabel": "Continue to booking", + } + }, + } + ], + } + + +@pytest.fixture +def valid_iteration_rule_with_only_mandatory_fields(): + return { + "Type": "F", + "Name": "Assure only already vaccinated taken from magic cohort", + "Description": "Exclude anyone who has NOT been given a dose of RSV Vaccination from the magic cohort", + "Operator": "is_empty", + "Comparator": "", + "AttributeTarget": "RSV", + "AttributeLevel": "TARGET", + "AttributeName": "LAST_SUCCESSFUL_DATE", + "CohortLabel": "elid_all_people", + "Priority": 100, + } + + +@pytest.fixture +def valid_available_action(): + return { + "ExternalRoutingCode": "BookNBS", + "ActionDescription": "", + "ActionType": "ButtonWithAuthLink", + "UrlLink": "http://www.nhs.uk/book-rsv", + "UrlLabel": "Continue to booking", + } diff --git a/tests/unit/validation/test_actions_mapper_validator.py b/tests/unit/validation/test_actions_mapper_validator.py new file mode 100644 index 000000000..e14989e90 --- /dev/null +++ b/tests/unit/validation/test_actions_mapper_validator.py @@ -0,0 +1,49 @@ +import pytest +from pydantic import ValidationError + +from eligibility_signposting_api.model.campaign_config import AvailableAction +from rules_validation_api.validators.actions_mapper_validator import ActionsMapperValidator + + +@pytest.fixture +def valid_available_action(): + return { + "ExternalRoutingCode": "BookNBS", + "ActionDescription": "", + "ActionType": "ButtonWithAuthLink", + "UrlLink": "http://www.nhs.uk/book-rsv", + "UrlLabel": "Continue to booking", + } + + +class TestBUCValidations: + def make_action(self, data: dict) -> AvailableAction: + return AvailableAction(**data) + + def test_valid_actions_mapper(self, valid_available_action): + data = { + "action1": self.make_action(valid_available_action), + "action2": self.make_action({**valid_available_action, "ExternalRoutingCode": "AltCode"}), + } + mapper = ActionsMapperValidator(root=data) + + expected_action_count = 2 + assert isinstance(mapper, ActionsMapperValidator) + assert len(mapper.root) == expected_action_count + + def test_invalid_actions_mapper_empty_key(self, valid_available_action): + data = {"": self.make_action(valid_available_action), "action2": self.make_action(valid_available_action)} + with pytest.raises(ValidationError) as exc_info: + ActionsMapperValidator(root=data) + assert "Invalid keys found in ActionsMapper" in str(exc_info.value) + assert "['']" in str(exc_info.value) + + @pytest.mark.parametrize("bad_key", [""]) + def test_invalid_keys_parametrized(self, bad_key, valid_available_action): + data = { + bad_key: self.make_action(valid_available_action), + "valid_key": self.make_action(valid_available_action), + } + with pytest.raises(ValidationError) as exc_info: + ActionsMapperValidator(root=data) + assert "Invalid keys found in ActionsMapper" in str(exc_info.value) diff --git a/tests/unit/validation/test_available_action_validator.py b/tests/unit/validation/test_available_action_validator.py new file mode 100644 index 000000000..468ead7b9 --- /dev/null +++ b/tests/unit/validation/test_available_action_validator.py @@ -0,0 +1,55 @@ +import copy + +import pytest +from pydantic import ValidationError + +from rules_validation_api.validators.available_action_validator import AvailableActionValidation + + +# ๐Ÿ” Mandatory Fields +class TestMandatoryFieldsSchemaValidations: + def test_valid_minimal_input(self, valid_available_action): + data = copy.deepcopy(valid_available_action) + data.pop("ActionDescription") + data.pop("UrlLink") + data.pop("UrlLabel") + action = AvailableActionValidation(**data) + assert action.action_type == "ButtonWithAuthLink" + assert action.action_code == "BookNBS" + assert action.action_description is None + assert action.url_link is None + assert action.url_label is None + + def test_missing_required_fields(self, valid_available_action): + data = copy.deepcopy(valid_available_action) + data.pop("ActionType") + data.pop("ExternalRoutingCode") + with pytest.raises(ValidationError) as exc_info: + AvailableActionValidation(**data) + error_msg = str(exc_info.value) + assert "ActionType" in error_msg + assert "ExternalRoutingCode" in error_msg + + +# ๐Ÿ” Optional Fields +class TestOptionalFieldsSchemaValidations: + def test_valid_full_input(self, valid_available_action): + action = AvailableActionValidation(**valid_available_action) + assert action.action_type == "ButtonWithAuthLink" + assert action.action_code == "BookNBS" + assert action.action_description == "" + assert str(action.url_link) == "http://www.nhs.uk/book-rsv" + assert action.url_label == "Continue to booking" + + def test_empty_string_is_valid_for_optional_fields(self, valid_available_action): + action = AvailableActionValidation(**valid_available_action) + assert action.action_description == "" + assert action.url_label == "Continue to booking" + + @pytest.mark.parametrize("bad_url", ["not-a-url", "ftp://bad", "123"]) + def test_invalid_url_raises_validation_error(self, valid_available_action, bad_url): + data = copy.deepcopy(valid_available_action) + data["UrlLink"] = bad_url + with pytest.raises(ValidationError) as exc_info: + AvailableActionValidation(**data) + assert "UrlLink" in str(exc_info.value) diff --git a/tests/unit/validation/test_campaign_config_validator.py b/tests/unit/validation/test_campaign_config_validator.py new file mode 100644 index 000000000..61bb75ca7 --- /dev/null +++ b/tests/unit/validation/test_campaign_config_validator.py @@ -0,0 +1,235 @@ +import pytest +from pydantic import ValidationError + +from rules_validation_api.validators.campaign_config_validator import CampaignConfigValidation + + +class TestMandatoryFieldsSchemaValidations: + def test_campaign_config_with_only_mandatory_fields_configuration( + self, valid_campaign_config_with_only_mandatory_fields + ): + try: + CampaignConfigValidation(**valid_campaign_config_with_only_mandatory_fields) + except ValidationError as e: + pytest.fail(f"Unexpected error during model instantiation: {e}") + + @pytest.mark.parametrize( + "mandatory_field", + [ + "ID", + "Version", + "Name", + "Type", + "Target", + "IterationFrequency", + "IterationType", + "StartDate", + "EndDate", + "Iterations", + ], + ) + def test_missing_mandatory_fields(self, mandatory_field, valid_campaign_config_with_only_mandatory_fields): + data = valid_campaign_config_with_only_mandatory_fields.copy() + data.pop(mandatory_field, None) # Simulate missing field + with pytest.raises(ValidationError): + CampaignConfigValidation(**data) + + # ID + @pytest.mark.parametrize("id_value", ["CAMP001", "12345", "X001"]) + def test_valid_id(self, id_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "ID": id_value} + model = CampaignConfigValidation(**data) + assert model.id == id_value + + # Version + @pytest.mark.parametrize("version_value", [1, 2, 100]) + def test_valid_version(self, version_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Version": version_value} + model = CampaignConfigValidation(**data) + assert model.version == version_value + + # Name + @pytest.mark.parametrize("name_value", ["Spring Campaign", "COVID-Alert", "Mass Outreach"]) + def test_valid_name(self, name_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Name": name_value} + model = CampaignConfigValidation(**data) + assert model.name == name_value + + # Type + @pytest.mark.parametrize("type_value", ["V", "S"]) + def test_valid_type(self, type_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Type": type_value} + model = CampaignConfigValidation(**data) + assert model.type == type_value + + @pytest.mark.parametrize("type_value", ["X", "", None]) + def test_invalid_type(self, type_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Type": type_value} + with pytest.raises(ValidationError): + CampaignConfigValidation(**data) + + # Target + @pytest.mark.parametrize("target_value", ["COVID", "FLU", "MMR", "RSV"]) + def test_valid_target(self, target_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Target": target_value} + model = CampaignConfigValidation(**data) + assert model.target == target_value + + @pytest.mark.parametrize("target_value", ["XYZ", "ABC", "", None]) + def test_invalid_target(self, target_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Target": target_value} + with pytest.raises(ValidationError): + CampaignConfigValidation(**data) + + # IterationFrequency + @pytest.mark.parametrize("freq_value", ["X", "D", "W", "M", "Q", "A"]) + def test_valid_iteration_frequency(self, freq_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "IterationFrequency": freq_value} + model = CampaignConfigValidation(**data) + assert model.iteration_frequency == freq_value + + @pytest.mark.parametrize("freq_value", ["Z", "", None]) + def test_invalid_iteration_frequency(self, freq_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "IterationFrequency": freq_value} + with pytest.raises(ValidationError): + CampaignConfigValidation(**data) + + # IterationType + @pytest.mark.parametrize("iter_type", ["A", "M", "S", "O"]) + def test_valid_iteration_type(self, iter_type, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "IterationType": iter_type} + model = CampaignConfigValidation(**data) + assert model.iteration_type == iter_type + + @pytest.mark.parametrize("iter_type", ["B", "", None]) + def test_invalid_iteration_type(self, iter_type, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "IterationType": iter_type} + with pytest.raises(ValidationError): + CampaignConfigValidation(**data) + + # StartDate + @pytest.mark.parametrize( + "start_date", + [ + "", # empty string + "invalid-date", # malformed value + ], + ) + def test_invalid_start_date(self, start_date, valid_campaign_config_with_only_mandatory_fields): + data = valid_campaign_config_with_only_mandatory_fields.copy() + data["StartDate"] = start_date + + with pytest.raises(ValidationError) as exc_info: + CampaignConfigValidation(**data) + + errors = exc_info.value.errors() + for error in errors: + assert error["loc"][0] == "StartDate" + + # EndDates + @pytest.mark.parametrize( + "end_date", + [ + "", # empty string + "31032025", # malformed value + ], + ) + def test_invalid_end_date(self, end_date, valid_campaign_config_with_only_mandatory_fields): + data = valid_campaign_config_with_only_mandatory_fields.copy() + data["EndDate"] = end_date + + with pytest.raises(ValidationError) as exc_info: + CampaignConfigValidation(**data) + + errors = exc_info.value.errors() + for error in errors: + assert error["loc"][0] == "EndDate" + + +class TestOptionalFieldsSchemaValidations: + @pytest.mark.parametrize("manager", [["alice"], ["bob"], ["carol"]]) + def test_manager_field(self, manager, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Manager": manager} + model = CampaignConfigValidation(**data) + assert model.manager == manager + + @pytest.mark.parametrize("approver", [["alice"], ["bob"], ["carol"]]) + def test_approver_field(self, approver, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Approver": approver} + model = CampaignConfigValidation(**data) + assert model.approver == approver + + @pytest.mark.parametrize("reviewer", [["alice"], ["bob"], ["carol"]]) + def test_reviewer_field(self, reviewer, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Reviewer": reviewer} + model = CampaignConfigValidation(**data) + assert model.reviewer == reviewer + + @pytest.mark.parametrize("iteration_time", ["14:00", "09:30", "18:45"]) + def test_iteration_time_field(self, iteration_time, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "IterationTime": iteration_time} + model = CampaignConfigValidation(**data) + assert model.iteration_time == iteration_time + + @pytest.mark.parametrize("routing", ["email", "sms", "push"]) + def test_default_comms_routing_field(self, routing, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "DefaultCommsRouting": routing} + model = CampaignConfigValidation(**data) + assert model.default_comms_routing == routing + + @pytest.mark.parametrize("min_approval", [0, 1, 2]) + def test_approval_minimum_field(self, min_approval, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "ApprovalMinimum": min_approval} + model = CampaignConfigValidation(**data) + assert model.approval_minimum == min_approval + + @pytest.mark.parametrize("max_approval", [5, 10, 15]) + def test_approval_maximum_field(self, max_approval, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "ApprovalMaximum": max_approval} + model = CampaignConfigValidation(**data) + assert model.approval_maximum == max_approval + + +class TestBUCValidations: + # StartDate and EndDates + @pytest.mark.parametrize( + ("start_date", "end_date"), + [ + ("20250101", "20250331"), # valid range + ("20250601", "20250630"), # valid short range + ("20250101", "20250101"), # same day + ], + ) + def test_valid_start_and_end_dates_and_iteration_dates_relation( + self, start_date, end_date, valid_campaign_config_with_only_mandatory_fields + ): + data = valid_campaign_config_with_only_mandatory_fields.copy() + data["StartDate"] = start_date + data["EndDate"] = end_date + data["Iterations"][0]["IterationDate"] = "20241231" + CampaignConfigValidation(**data) + + @pytest.mark.parametrize( + ("start_date", "end_date"), + [ + ("20241230", "20250101"), # campaign start date is after the iteration date + ("20250331", "20250101"), # end before start + ], + ) + def test_invalid_start_and_end_dates_and_iteration_dates_relation( + self, start_date, end_date, valid_campaign_config_with_only_mandatory_fields + ): + data = valid_campaign_config_with_only_mandatory_fields.copy() + data["StartDate"] = start_date + data["EndDate"] = end_date + data["Iterations"][0]["IterationDate"] = "20241231" + with pytest.raises(ValidationError): + CampaignConfigValidation(**data) + + # Iteration + def test_validate_iterations_non_empty(self, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields, "Iterations": []} + with pytest.raises(ValidationError) as error: + CampaignConfigValidation(**data) + errors = error.value.errors() + assert any(e["loc"][-1] == "Iterations" for e in errors), "Expected validation error on 'Iterations'" diff --git a/tests/unit/validation/test_iteration_rules_validator.py b/tests/unit/validation/test_iteration_rules_validator.py new file mode 100644 index 000000000..af7405cce --- /dev/null +++ b/tests/unit/validation/test_iteration_rules_validator.py @@ -0,0 +1,223 @@ +import pytest +from pydantic import ValidationError + +from rules_validation_api.validators.iteration_validator import IterationRuleValidation + + +class TestMandatoryFieldsSchemaValidations: + def test_campaign_config_with_only_mandatory_fields_configuration( + self, valid_iteration_rule_with_only_mandatory_fields + ): + try: + IterationRuleValidation(**valid_iteration_rule_with_only_mandatory_fields) + except ValidationError as e: + pytest.fail(f"Unexpected error during model instantiation: {e}") + + @pytest.mark.parametrize( + "mandatory_field", + ["Type", "Name", "Description", "Priority", "AttributeLevel", "Operator", "Comparator"], + ) + def test_missing_mandatory_fields(self, mandatory_field, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data.pop(mandatory_field, None) # Simulate missing field + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + assert mandatory_field.lower() + + @pytest.mark.parametrize("type_value", ["F", "S", "R", "X", "Y"]) + def test_valid_type(self, type_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Type"] = type_value + result = IterationRuleValidation(**data) + assert result.type.value == type_value + + @pytest.mark.parametrize("type_value", ["Z", 123, None]) + def test_invalid_type(self, type_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Type"] = type_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + @pytest.mark.parametrize("name_value", ["", "ValidName", "Test_Rule_01"]) + def test_valid_name(self, name_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Name"] = name_value + result = IterationRuleValidation(**data) + assert result.name == name_value + + @pytest.mark.parametrize("name_value", [None, 42]) + def test_invalid_name(self, name_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Name"] = name_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + @pytest.mark.parametrize("description_value", ["", "A rule description", "Sample text"]) + def test_valid_description(self, description_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Description"] = description_value + result = IterationRuleValidation(**data) + assert result.description == description_value + + @pytest.mark.parametrize("description_value", [None]) + def test_invalid_description(self, description_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Description"] = description_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + @pytest.mark.parametrize("priority_value", [-1, -5, 1, 100, 999]) + def test_valid_priority(self, priority_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Priority"] = priority_value + result = IterationRuleValidation(**data) + assert result.priority == priority_value + + @pytest.mark.parametrize("priority_value", ["high", None]) + def test_invalid_priority(self, priority_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Priority"] = priority_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + @pytest.mark.parametrize("attribute_level", ["PERSON", "TARGET", "COHORT"]) + def test_valid_attribute_level(self, attribute_level, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["AttributeLevel"] = attribute_level + result = IterationRuleValidation(**data) + assert result.attribute_level == attribute_level + + @pytest.mark.parametrize("attribute_level", ["", None, 42, "basic", "BASIC"]) + def test_invalid_attribute_level(self, attribute_level, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["AttributeLevel"] = attribute_level + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + @pytest.mark.parametrize("operator_value", ["=", "!=", ">", "<=", "contains", "is_true"]) + def test_valid_operator(self, operator_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Operator"] = operator_value + result = IterationRuleValidation(**data) + assert result.operator.value == operator_value + + @pytest.mark.parametrize("operator_value", ["approx", "", None]) + def test_invalid_operator(self, operator_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Operator"] = operator_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + @pytest.mark.parametrize("comparator_value", ["status", "true", "0"]) + def test_valid_comparator(self, comparator_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Comparator"] = comparator_value + result = IterationRuleValidation(**data) + assert result.comparator == comparator_value + + @pytest.mark.parametrize("comparator_value", [None, 123]) + def test_invalid_comparator(self, comparator_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["Comparator"] = comparator_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + +class TestOptionalFieldsSchemaValidations: + # AttributeName + @pytest.mark.parametrize("attr_name", ["status", "user_type", None]) + def test_valid_attribute_name(self, attr_name, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["AttributeName"] = attr_name + result = IterationRuleValidation(**data) + assert result.attribute_name == attr_name + + @pytest.mark.parametrize("attr_name", [123, {}, []]) + def test_invalid_attribute_name(self, attr_name, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["AttributeName"] = attr_name + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + # CohortLabel + @pytest.mark.parametrize("label", ["Cohort_A", "Segment_2025", None, ""]) + def test_valid_cohort_label(self, label, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["CohortLabel"] = label + result = IterationRuleValidation(**data) + assert result.cohort_label == label + + @pytest.mark.parametrize("label", [123, [], {}]) + def test_invalid_cohort_label(self, label, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["CohortLabel"] = label + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + # AttributeTarget + @pytest.mark.parametrize("target", ["target_value", None]) + def test_valid_attribute_target(self, target, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["AttributeTarget"] = target + result = IterationRuleValidation(**data) + assert result.attribute_target == target + + @pytest.mark.parametrize("target", [123, [], {}]) + def test_invalid_attribute_target(self, target, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["AttributeTarget"] = target + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + # RuleStop + @pytest.mark.parametrize("rule_stop_value", [True, False, "Y", "N", "YES", "NO", "YEAH", "ONE"]) + def test_valid_rule_stop(self, rule_stop_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["RuleStop"] = rule_stop_value + result = IterationRuleValidation(**data) + assert isinstance(result.rule_stop, bool) + + @pytest.mark.parametrize("rule_stop_value", [{}, None]) + def test_invalid_rule_stop(self, rule_stop_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["RuleStop"] = rule_stop_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + # CommsRouting + @pytest.mark.parametrize("routing_value", ["route_A", None]) + def test_valid_comms_routing(self, routing_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["CommsRouting"] = routing_value + result = IterationRuleValidation(**data) + assert result.comms_routing == routing_value + + @pytest.mark.parametrize("routing_value", [123, [], {}]) + def test_invalid_comms_routing(self, routing_value, valid_iteration_rule_with_only_mandatory_fields): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["CommsRouting"] = routing_value + with pytest.raises(ValidationError): + IterationRuleValidation(**data) + + +class TestBUCValidations: + @pytest.mark.parametrize( + ("rule_stop_input", "expected_bool"), + [ + (True, True), + (False, False), + ("Y", True), + ("N", False), + ("YES", False), + ("NO", False), + ("YEAH", False), + ("ONE", False), + ], + ) + def test_rule_stop_boolean_resolution( + self, rule_stop_input, expected_bool, valid_iteration_rule_with_only_mandatory_fields + ): + data = valid_iteration_rule_with_only_mandatory_fields.copy() + data["RuleStop"] = rule_stop_input + result = IterationRuleValidation(**data) + assert result.rule_stop is expected_bool diff --git a/tests/unit/validation/test_iteration_validator.py b/tests/unit/validation/test_iteration_validator.py new file mode 100644 index 000000000..8e58e5a38 --- /dev/null +++ b/tests/unit/validation/test_iteration_validator.py @@ -0,0 +1,192 @@ +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from rules_validation_api.validators.iteration_validator import IterationValidation + + +class TestMandatoryFieldsSchemaValidations: + def test_campaign_config_with_only_mandatory_fields_configuration( + self, valid_campaign_config_with_only_mandatory_fields + ): + try: + IterationValidation(**(valid_campaign_config_with_only_mandatory_fields["Iterations"][0])) + except ValidationError as e: + pytest.fail(f"Unexpected error during model instantiation: {e}") + + @pytest.mark.parametrize( + "mandatory_field", + [ + "ID", + "Version", + "Name", + "IterationDate", + "Type", + "DefaultCommsRouting", + "DefaultNotEligibleRouting", + "DefaultNotActionableRouting", + "IterationCohorts", + "IterationRules", + "ActionsMapper", + ], + ) + def test_missing_mandatory_fields(self, mandatory_field, valid_campaign_config_with_only_mandatory_fields): + data = valid_campaign_config_with_only_mandatory_fields["Iterations"][0].copy() + data.pop(mandatory_field, None) # Simulate missing field + with pytest.raises(ValidationError): + IterationValidation(**data) + assert mandatory_field.lower() + + # ID + @pytest.mark.parametrize("id_value", ["ITER001", "X123", "IT01"]) + def test_valid_id(self, id_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields["Iterations"][0], "ID": id_value} + model = IterationValidation(**data) + assert model.id == id_value + + # Version + @pytest.mark.parametrize("version_value", [1, 2, 100]) + def test_valid_version(self, version_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields["Iterations"][0], "Version": version_value} + model = IterationValidation(**data) + assert model.version == version_value + + # Name + @pytest.mark.parametrize("name_value", ["Mid-January Push", "Spring Surge", "Early Outreach"]) + def test_valid_name(self, name_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields["Iterations"][0], "Name": name_value} + model = IterationValidation(**data) + assert model.name == name_value + + # IterationDate + @pytest.mark.parametrize("date_value", ["20250101", "20250215", "20250301"]) + def test_valid_iteration_date(self, date_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields["Iterations"][0], "IterationDate": date_value} + model = IterationValidation(**data) + expected_date = datetime.strptime(str(date_value), "%Y%m%d").replace(tzinfo=UTC).date() + assert model.iteration_date == expected_date, f"Expected {expected_date}, got {model.iteration_date}" + + # Type + @pytest.mark.parametrize("type_value", ["A", "M", "S", "O"]) + def test_valid_type(self, type_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields["Iterations"][0], "Type": type_value} + model = IterationValidation(**data) + assert model.type == type_value + + @pytest.mark.parametrize("type_value", ["", "Z", None]) + def test_invalid_type(self, type_value, valid_campaign_config_with_only_mandatory_fields): + data = {**valid_campaign_config_with_only_mandatory_fields["Iterations"][0], "Type": type_value} + with pytest.raises(ValidationError): + IterationValidation(**data) + + # DefaultCommsRouting + @pytest.mark.parametrize("routing_value", ["BOOK_NBS"]) + def test_valid_default_comms_routing(self, routing_value, valid_campaign_config_with_only_mandatory_fields): + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "DefaultCommsRouting": routing_value, + } + model = IterationValidation(**data) + assert model.default_comms_routing == routing_value + + # DefaultNotEligibleRouting + @pytest.mark.parametrize("routing_value", ["RouteB", "NotEligComm", "NoComms"]) + def test_valid_default_not_eligible_routing(self, routing_value, valid_campaign_config_with_only_mandatory_fields): + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "DefaultNotEligibleRouting": routing_value, + } + model = IterationValidation(**data) + assert model.default_not_eligible_routing == routing_value + + # DefaultNotActionableRouting + @pytest.mark.parametrize("routing_value", ["RouteC", "HoldComm", "Inactive"]) + def test_valid_default_not_actionable_routing( + self, routing_value, valid_campaign_config_with_only_mandatory_fields + ): + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "DefaultNotActionableRouting": routing_value, + } + model = IterationValidation(**data) + assert model.default_not_actionable_routing == routing_value + + def test_invalid_actions_mapper_empty_key( + self, valid_campaign_config_with_only_mandatory_fields, valid_available_action + ): + actions_mapper = {"": valid_available_action, "action2": valid_available_action} + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "ActionsMapper": actions_mapper, + } + with pytest.raises(ValidationError): + IterationValidation(**data) + + +class TestOptionalFieldsSchemaValidations: + @pytest.mark.parametrize("iteration_number", [1, 5, 10]) + def test_iteration_number(self, iteration_number, valid_campaign_config_with_only_mandatory_fields): + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "IterationNumber": iteration_number, + } + model = IterationValidation(**data) + assert model.iteration_number == iteration_number + + @pytest.mark.parametrize("approval_minimum", [0, 25, 99]) + def test_approval_minimum(self, approval_minimum, valid_campaign_config_with_only_mandatory_fields): + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "ApprovalMinimum": approval_minimum, + } + model = IterationValidation(**data) + assert model.approval_minimum == approval_minimum + + @pytest.mark.parametrize("approval_maximum", [100, 250, 999]) + def test_approval_maximum(self, approval_maximum, valid_campaign_config_with_only_mandatory_fields): + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "ApprovalMaximum": approval_maximum, + } + model = IterationValidation(**data) + assert model.approval_maximum == approval_maximum + + +class TestIterationCohortsSchemaValidations: + def test_valid_iteration_if_actions_mapper_has_entry_for_the_provided_default_routing_key( + self, valid_campaign_config_with_only_mandatory_fields + ): + expected_action = { + "ExternalRoutingCode": "BookLocal", + "ActionDescription": "##Getting the vaccine\n" + "You can get an RSV vaccination at your GP surgery.\n" + "Your GP surgery may contact you about getting the RSV vaccine. " + "This may be by letter, text, phone call, email or through the NHS App. " + "You do not need to wait to be contacted before booking your vaccination.", + "ActionType": "InfoText", + } + + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "DefaultCommsRouting": "BOOK_LOCAL", + "ActionsMapper": {"BOOK_LOCAL": expected_action}, + } + IterationValidation(**data) + + def test_invalid_iteration_if_actions_mapper_has_no_entry_for_the_provided_default_routing_key( + self, valid_campaign_config_with_only_mandatory_fields + ): + data = { + **valid_campaign_config_with_only_mandatory_fields["Iterations"][0], + "DefaultCommsRouting": "BOOK_LOCAL", + "ActionsMapper": {}, + } # Missing BOOK_LOCAL in ActionsMapper + + with pytest.raises(ValidationError) as error: + IterationValidation(**data) + + errors = error.value.errors() + assert any(e["loc"][-1] == "actions_mapper" and "BOOK_LOCAL" in str(e["msg"]) for e in errors), ( + "Expected validation error for missing BOOK_LOCAL entry in ActionsMapper" + ) diff --git a/tests/unit/validation/test_rule_validator.py b/tests/unit/validation/test_rule_validator.py new file mode 100644 index 000000000..a24419824 --- /dev/null +++ b/tests/unit/validation/test_rule_validator.py @@ -0,0 +1,17 @@ +import pytest +from pydantic import ValidationError + +from rules_validation_api.validators.rules_validator import RulesValidation + + +def test_valid_campaign_config(valid_campaign_config_with_only_mandatory_fields): + config_data = {"campaign_config": valid_campaign_config_with_only_mandatory_fields} + validated = RulesValidation(**config_data) + assert validated.campaign_config.name is not None + + +def test_invalid_campaign_config_missing_field(): + invalid_data = {} + + with pytest.raises(ValidationError): + RulesValidation(**invalid_data)