eligibility-signposting-api/src/rules_validation_api/validators/iteration_validator.py at 3a1140a06f14b0094f3caa0c1f547f3990f0e2ee · NHSDigital/eligibility-signposting-api · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
import typing

from pydantic import Field, ValidationError, field_validator, model_validator
from pydantic_core import InitErrorDetails

from eligibility_signposting_api.model.campaign_config import (
    ActionsMapper,
    Iteration,
    IterationCohort,
    IterationRule,
    RuleType,
)
from rules_validation_api.validators.actions_mapper_validator import ActionsMapperValidation
from rules_validation_api.validators.iteration_cohort_validator import IterationCohortValidation
from rules_validation_api.validators.iteration_rules_validator import IterationRuleValidation


class IterationValidation(Iteration):
    iteration_cohorts: list[IterationCohort] = Field(..., alias="IterationCohorts")
    iteration_rules: list[IterationRule] = Field(..., alias="IterationRules")
    actions_mapper: ActionsMapper = Field(..., alias="ActionsMapper")

    @field_validator("iteration_rules")
    @classmethod
    def validate_iteration_rules(cls, iteration_rules: list[IterationRule]) -> list[IterationRuleValidation]:
        return [IterationRuleValidation(**i.model_dump()) for i in iteration_rules]

    @field_validator("iteration_cohorts")
    @classmethod
    def validate_iteration_cohorts(cls, iteration_cohorts: list[IterationCohort]) -> list[IterationCohortValidation]:
        return [IterationCohortValidation(**i.model_dump()) for i in iteration_cohorts]

    @field_validator("actions_mapper", mode="after")
    @classmethod
    def transform_actions_mapper(cls, action_mapper: ActionsMapper) -> ActionsMapper:
        ActionsMapperValidation.model_validate(action_mapper.model_dump())
        return action_mapper

    @model_validator(mode="after")
    def action_mapper_validation(self) -> typing.Self:
        all_errors = []

        for validator in [
            self.validate_default_comms_routing_in_actions_mapper,
            self.validate_default_not_eligible_routing_in_actions_mapper,
            self.validate_default_not_actionable_routing_in_actions_mapper,
            self.validate_iteration_rules_against_actions_mapper,
        ]:
            try:
                validator()
            except ValidationError as ve:
                all_errors.extend(ve.errors(include_input=False))

        if all_errors:
            raise ValidationError.from_exception_data(title="IterationValidation", line_errors=all_errors)

        return self

    def validate_default_comms_routing_in_actions_mapper(self) -> typing.Self:
        default_routes = self.default_comms_routing
        actions_keys = list(self.actions_mapper.root.keys())
        line_errors = []

        for routing in default_routes.split("|"):
            cleaned_routing = routing.strip()
            if cleaned_routing and (not actions_keys or cleaned_routing not in actions_keys):
                error = InitErrorDetails(
                    type="value_error",
                    loc=("actions_mapper",),
                    input=actions_keys,
                    ctx={"error": f"Missing entry for DefaultCommsRouting '{cleaned_routing}' in ActionsMapper"},
                )
                line_errors.append(error)

        if line_errors:
            raise ValidationError.from_exception_data(title="IterationValidation", line_errors=line_errors)

        return self

    def validate_default_not_eligible_routing_in_actions_mapper(self) -> typing.Self:
        default_not_eligibile_routes = self.default_not_eligible_routing
        actions_keys = list(self.actions_mapper.root.keys())
        line_errors = []

        for routing in default_not_eligibile_routes.split("|"):
            cleaned_routing = routing.strip()
            if cleaned_routing and (not actions_keys or cleaned_routing not in actions_keys):
                error = InitErrorDetails(
                    type="value_error",
                    loc=("actions_mapper",),
                    input=actions_keys,
                    ctx={"error": f"Missing entry for DefaultNotEligibleRouting '{cleaned_routing}' in ActionsMapper"},
                )
                line_errors.append(error)

        if line_errors:
            raise ValidationError.from_exception_data(title="IterationValidation", line_errors=line_errors)

        return self

    def validate_default_not_actionable_routing_in_actions_mapper(self) -> typing.Self:
        default_not_actionable_routes = self.default_not_actionable_routing
        actions_keys = list(self.actions_mapper.root.keys())
        line_errors = []

        for routing in default_not_actionable_routes.split("|"):
            cleaned_routing = routing.strip()
            if cleaned_routing and (not actions_keys or cleaned_routing not in actions_keys):
                error = InitErrorDetails(
                    type="value_error",
                    loc=("actions_mapper",),
                    input=actions_keys,
                    ctx={
                        "error": f"Missing entry for DefaultNotActionableRouting '{cleaned_routing}' in ActionsMapper"
                    },
                )
                line_errors.append(error)

        if line_errors:
            raise ValidationError.from_exception_data(title="IterationValidation", line_errors=line_errors)

        return self

    def validate_iteration_rules_against_actions_mapper(self) -> typing.Self:
        actions_keys = list(self.actions_mapper.root.keys())
        line_errors = []

        for rule in self.iteration_rules:
            if (
                rule.type
                in [
                    RuleType.redirect,
                    RuleType.not_actionable_actions,
                    RuleType.not_eligible_actions,
                ]
                and rule.comms_routing
            ):
                for routing in rule.comms_routing.split("|"):
                    cleaned_routing = routing.strip()
                    if cleaned_routing and (not actions_keys or cleaned_routing not in actions_keys):
                        error = InitErrorDetails(
                            type="value_error",
                            loc=("iteration_rules",),
                            input=actions_keys,
                            ctx={"error": f"Missing entry for CommsRouting '{cleaned_routing}' in ActionsMapper"},
                        )
                        line_errors.append(error)

        if line_errors:
            raise ValidationError.from_exception_data(title="IterationValidation", line_errors=line_errors)

        return self