Eli 150/campaign config validation

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 <path_to_config>`
+
+## Results
+
+- `On success`:
+
+  ```text
+  "Valid config" is printed
+
+- `On Failure`:
+
+  ```text
+  "Errors" is printed

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()

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

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

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]

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

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

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())

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"},

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",
+    }

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)

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)