Merge branch 'main' into chore/eja-adding-main-branch-scanning-for-sonarcube

Author: eddalmond1

infrastructure/stacks/api-layer/eventbridge.tf

@@ -73,17 +73,21 @@ resource "aws_cloudwatch_event_target" "firehose_target" {
       reason     = "$.detail.state.reason"
     }
 
-    input_template = jsonencode({
-      time = "<time>"
-      source = "elid-${var.environment}:cloudwatch:alarm"
-      sourcetype = "aws:cloudwatch:alarm"
-      event = {
-        alarm_name  = "<alarm_name>"
-        new_state   = "<new_state>"
-        old_state   = "<old_state>"
-        reason      = "<reason>"
-        region      = "<region>"
-      }
-    })
+    # Use a heredoc string so EventBridge placeholders like <time> are not JSON-escaped
+    # (jsonencode would turn < and > into \u003c/\u003e, preventing substitution).
+    input_template = <<TEMPLATE
+{
+  "time": "<time>",
+  "source": "elid-${var.environment}:cloudwatch:alarm",
+  "sourcetype": "aws:cloudwatch:alarm",
+  "event": {
+    "alarm_name": "<alarm_name>",
+    "new_state": "<new_state>",
+    "old_state": "<old_state>",
+    "reason": "<reason>",
+    "region": "<region>"
+  }
+}
+TEMPLATE
   }
 }

src/eligibility_signposting_api/audit/audit_context.py

@@ -64,12 +64,12 @@ def append_audit_condition(
         condition_name: ConditionName,
         best_iteration_result: BestIterationResult,
         action_detail: MatchedActionDetail,
-        cohort_results: list[CohortGroupResult],
     ) -> None:
         audit_eligibility_cohorts, audit_eligibility_cohort_groups, audit_actions = [], [], []
         best_active_iteration = best_iteration_result.active_iteration
         best_candidate = best_iteration_result.iteration_result
         best_cohort_results = best_iteration_result.cohort_results
+        filter_audit_rules, suitability_audit_rules = [], []
 
         if best_cohort_results:
             for cohort_label, result in sorted(best_cohort_results.items(), key=lambda item: item[1].cohort_code):
@@ -84,13 +84,7 @@ def append_audit_condition(
                     )
                 )
 
-        filter_audit_rules, suitability_audit_rules = [], []
-        for result in cohort_results:
-            for rule in result.audit_rules:
-                if rule.rule_type == RuleType.filter:
-                    filter_audit_rules.append(rule)
-                if rule.rule_type == RuleType.suppression:
-                    suitability_audit_rules.append(rule)
+                AuditContext.get_audit_rules(filter_audit_rules, suitability_audit_rules, result)
 
         audit_filter_rule = AuditContext.create_audit_filter_rule(filter_audit_rules)
         audit_suitability_rule = AuditContext.create_audit_suitability_rule(suitability_audit_rules)
@@ -116,6 +110,14 @@ def append_audit_condition(
 
         g.audit_log.response.condition.append(audit_condition)
 
+    @staticmethod
+    def get_audit_rules(filter_audit_rules: list, suitability_audit_rules: list, result: CohortGroupResult) -> None:
+        for rule in result.audit_rules:
+            if rule.rule_type == RuleType.filter:
+                filter_audit_rules.append(rule)
+            if rule.rule_type == RuleType.suppression:
+                suitability_audit_rules.append(rule)
+
     @staticmethod
     def add_rule_name_and_priority_to_audit(
         best_candidate: IterationResult | None,

src/eligibility_signposting_api/config/contants.py

@@ -1,3 +1,6 @@
+from typing import Literal
+
 MAGIC_COHORT_LABEL = "elid_all_people"
 RULE_STOP_DEFAULT = False
 NHS_NUMBER_HEADER = "nhs-login-nhs-number"
+ALLOWED_CONDITIONS = Literal["COVID", "FLU", "MMR", "RSV"]

src/eligibility_signposting_api/model/campaign_config.py

@@ -11,7 +11,7 @@
 
 from pydantic import BaseModel, Field, HttpUrl, RootModel, field_serializer, field_validator, model_validator
 
-from eligibility_signposting_api.config.contants import MAGIC_COHORT_LABEL, RULE_STOP_DEFAULT
+from eligibility_signposting_api.config.contants import ALLOWED_CONDITIONS, MAGIC_COHORT_LABEL, RULE_STOP_DEFAULT
 
 if typing.TYPE_CHECKING:  # pragma: no cover
     from pydantic import SerializationInfo
@@ -184,7 +184,7 @@ class CampaignConfig(BaseModel):
     version: CampaignVersion = Field(..., alias="Version")
     name: CampaignName = Field(..., alias="Name")
     type: Literal["V", "S"] = Field(..., alias="Type")
-    target: Literal["COVID", "FLU", "MMR", "RSV"] = Field(..., alias="Target")
+    target: ALLOWED_CONDITIONS = Field(..., alias="Target")
     manager: list[str] | None = Field(None, alias="Manager")
     approver: list[str] | None = Field(None, alias="Approver")
     reviewer: list[str] | None = Field(None, alias="Reviewer")

src/eligibility_signposting_api/repos/person_repo.py

@@ -38,14 +38,11 @@ def __init__(self, table: Annotated[Any, Inject(qualifier="person_table")]) -> N
 
     def get_eligibility_data(self, nhs_number: NHSNumber) -> Person:
         response = self.table.query(KeyConditionExpression=Key("NHS_NUMBER").eq(nhs_number))
-        logger.debug("response %r for %r", response, nhs_number, extra={"response": response, "nhs_number": nhs_number})
 
         if not (items := response.get("Items")) or not next(
             (item for item in items if item.get("ATTRIBUTE_TYPE") == "PERSON"), None
         ):
             message = f"Person not found with nhs_number {nhs_number}"
             raise NotFoundError(message)
 
-        logger.debug("returning items %s", items, extra={"items": items})
-
         return Person(data=items)

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -23,6 +23,7 @@
 from eligibility_signposting_api.services.processors.action_rule_handler import ActionRuleHandler
 from eligibility_signposting_api.services.processors.campaign_evaluator import CampaignEvaluator
 from eligibility_signposting_api.services.processors.rule_processor import RuleProcessor
+from eligibility_signposting_api.services.processors.token_processor import TokenProcessor
 
 if TYPE_CHECKING:
     from collections.abc import Collection
@@ -34,6 +35,7 @@
     )
     from eligibility_signposting_api.model.person import Person
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -99,19 +101,22 @@ def get_eligibility_status(self, include_actions: str, conditions: list[str], ca
                 include_actions_flag=include_actions_flag,
             )
 
+            best_iteration_result = TokenProcessor.find_and_replace_tokens(self.person, best_iteration_result)
+            matched_action_detail = TokenProcessor.find_and_replace_tokens(self.person, matched_action_detail)
+
             condition_results[condition_name] = best_iteration_result.iteration_result
             condition_results[condition_name].actions = matched_action_detail.actions
 
             condition: Condition = self.build_condition(
                 iteration_result=condition_results[condition_name], condition_name=condition_name
             )
+
             final_result.append(condition)
 
             AuditContext.append_audit_condition(
                 condition_name,
                 best_iteration_result,
                 matched_action_detail,
-                condition_results[condition_name].cohort_results,
             )
 
         # Consolidate all the results and return

src/eligibility_signposting_api/services/eligibility_services.py

@@ -42,15 +42,6 @@ def get_eligibility_status(
             try:
                 person_data = self.person_repo.get_eligibility_data(nhs_number)
                 campaign_configs = list(self.campaign_repo.get_campaign_configs())
-                logger.debug(
-                    "got person_data for %r",
-                    nhs_number,
-                    extra={
-                        "campaign_configs": [c.model_dump(by_alias=True) for c in campaign_configs],
-                        "person_data": person_data,
-                        "nhs_number": nhs_number,
-                    },
-                )
             except NotFoundError as e:
                 raise UnknownPersonError from e
             else:

src/eligibility_signposting_api/services/processors/token_parser.py

@@ -0,0 +1,72 @@
+import re
+from dataclasses import dataclass
+
+
+@dataclass
+class ParsedToken:
+    """
+    A class to represent a parsed token.
+    ...
+    Attributes
+    ----------
+    attribute_level : str
+        Example: "PERSON" or "TARGET"
+    attribute_name : str
+        Example: "POSTCODE" or "RSV"
+    attribute_value : int
+        Example: "LAST_SUCCESSFUL_DATE" if attribute_level is TARGET
+    format : str
+        Example: "%d %B %Y" if DATE formatting is used
+    """
+
+    attribute_level: str
+    attribute_name: str
+    attribute_value: str | None
+    format: str | None
+
+
+class TokenParser:
+    MIN_TOKEN_PARTS = 2
+
+    @staticmethod
+    def parse(token: str) -> ParsedToken:
+        """Parses a token into its parts.
+        Steps:
+        Strip the surrounding [[ ]]
+        Check for empty body after stripping, e.g., '[[]]'
+        Check for empty parts created by leading/trailing dots or tokens with no dot
+        Check if the name contains a date format
+        Return a ParsedToken object
+        """
+
+        token_body = token[2:-2]
+        if not token_body:
+            message = "Invalid token."
+            raise ValueError(message)
+
+        token_parts = token_body.split(".")
+
+        if len(token_parts) < TokenParser.MIN_TOKEN_PARTS or not all(token_parts):
+            message = "Invalid token."
+            raise ValueError(message)
+
+        token_level = token_parts[0].upper()
+        token_name = token_parts[-1]
+
+        format_match = re.search(r":DATE\(([^()]*)\)", token_name, re.IGNORECASE)
+        if not format_match and len(token_name.split(":")) > 1:
+            message = "Invalid token format."
+            raise ValueError(message)
+
+        format_str = format_match.group(1) if format_match else None
+
+        last_part = re.sub(r":DATE\(.*?\)", "", token_name, flags=re.IGNORECASE)
+
+        if len(token_parts) == TokenParser.MIN_TOKEN_PARTS:
+            name = last_part.upper()
+            value = None
+        else:
+            name = token_parts[1].upper()
+            value = last_part.upper()
+
+        return ParsedToken(attribute_level=token_level, attribute_name=name, attribute_value=value, format=format_str)

src/eligibility_signposting_api/services/processors/token_processor.py

@@ -0,0 +1,134 @@
+import re
+from dataclasses import Field, fields, is_dataclass
+from datetime import UTC, datetime
+from typing import Any, Never, TypeVar
+
+from wireup import service
+
+from eligibility_signposting_api.config.contants import ALLOWED_CONDITIONS
+from eligibility_signposting_api.model.person import Person
+from eligibility_signposting_api.services.processors.token_parser import ParsedToken, TokenParser
+
+T = TypeVar("T")
+
+
+TARGET_ATTRIBUTE_LEVEL = "TARGET"
+PERSON_ATTRIBUTE_LEVEL = "PERSON"
+ALLOWED_TARGET_ATTRIBUTES = {
+    "ATTRIBUTE_TYPE",
+    "VALID_DOSES_COUNT",
+    "INVALID_DOSES_COUNT",
+    "LAST_SUCCESSFUL_DATE",
+    "LAST_VALID_DOSE_DATE",
+    "BOOKED_APPOINTMENT_DATE",
+    "BOOKED_APPOINTMENT_PROVIDER",
+    "LAST_INVITE_DATE",
+    "LAST_INVITE_STATUS",
+}
+
+
+@service
+class TokenProcessor:
+    @staticmethod
+    def find_and_replace_tokens(person: Person, data_class: T) -> T:
+        if not is_dataclass(data_class):
+            return data_class
+        for class_field in fields(data_class):
+            value = getattr(data_class, class_field.name)
+            if isinstance(value, str):
+                setattr(data_class, class_field.name, TokenProcessor.replace_token(value, person))
+            elif isinstance(value, list):
+                TokenProcessor.process_list(class_field, data_class, person, value)
+            elif isinstance(value, dict):
+                TokenProcessor.process_dict(class_field, data_class, person, value)
+            elif is_dataclass(value):
+                setattr(data_class, class_field.name, TokenProcessor.find_and_replace_tokens(person, value))
+        return data_class
+
+    @staticmethod
+    def process_dict(class_field: Field, data_class: object, person: Person, value: dict[Any, Any]) -> None:
+        for key, dict_value in value.items():
+            if isinstance(dict_value, str):
+                value[key] = TokenProcessor.replace_token(dict_value, person)
+            elif is_dataclass(dict_value):
+                value[key] = TokenProcessor.find_and_replace_tokens(person, dict_value)
+        setattr(data_class, class_field.name, value)
+
+    @staticmethod
+    def process_list(class_field: Field, data_class: object, person: Person, value: list[Any]) -> None:
+        for i, item in enumerate(value):
+            if is_dataclass(item):
+                value[i] = TokenProcessor.find_and_replace_tokens(person, item)
+            elif isinstance(item, str):
+                value[i] = TokenProcessor.replace_token(item, person)
+            setattr(data_class, class_field.name, value)
+
+    @staticmethod
+    def replace_token(text: str, person: Person) -> str:
+        if not isinstance(text, str):
+            return text
+
+        pattern = r"\[\[.*?\]\]"
+        all_tokens = re.findall(pattern, text, re.IGNORECASE)
+        present_attributes = [attribute.get("ATTRIBUTE_TYPE") for attribute in person.data]
+
+        for token in all_tokens:
+            parsed_token = TokenParser.parse(token)
+            found_attribute, key_to_replace, replace_with = None, None, None
+
+            attribute_level_map = {
+                TARGET_ATTRIBUTE_LEVEL: parsed_token.attribute_value,
+                PERSON_ATTRIBUTE_LEVEL: parsed_token.attribute_name,
+            }
+
+            key_to_find = attribute_level_map.get(parsed_token.attribute_level)
+
+            if (
+                parsed_token.attribute_level == TARGET_ATTRIBUTE_LEVEL
+                and parsed_token.attribute_name in ALLOWED_CONDITIONS.__args__
+                and parsed_token.attribute_value in ALLOWED_TARGET_ATTRIBUTES
+                and parsed_token.attribute_name not in present_attributes
+            ):
+                replace_with = ""
+
+            if replace_with != "":
+                for attribute in person.data:
+                    is_person_attribute = attribute.get("ATTRIBUTE_TYPE") == PERSON_ATTRIBUTE_LEVEL
+                    is_allowed_target = parsed_token.attribute_name.upper() in ALLOWED_CONDITIONS.__args__
+
+                    if (is_allowed_target or is_person_attribute) and key_to_find in attribute:
+                        found_attribute = attribute
+                        key_to_replace = key_to_find
+                        break
+
+                if not found_attribute or key_to_replace is None:
+                    TokenProcessor.handle_token_not_found(parsed_token, token)
+
+                replace_with = TokenProcessor.apply_formatting(found_attribute, key_to_replace, parsed_token.format)
+            text = text.replace(token, str(replace_with))
+        return text
+
+    @staticmethod
+    def handle_token_not_found(parsed_token: ParsedToken, token: str) -> Never:
+        if parsed_token.attribute_level == TARGET_ATTRIBUTE_LEVEL:
+            message = f"Invalid attribute name '{parsed_token.attribute_value}' in token '{token}'."
+            raise ValueError(message)
+        if parsed_token.attribute_level == PERSON_ATTRIBUTE_LEVEL:
+            message = f"Invalid attribute name '{parsed_token.attribute_name}' in token '{token}'."
+            raise ValueError(message)
+        message = f"Invalid attribute level '{parsed_token.attribute_level}' in token '{token}'."
+        raise ValueError(message)
+
+    @staticmethod
+    def apply_formatting(attribute: dict[str, T], attribute_value: str, date_format: str | None) -> str:
+        try:
+            attribute_data = attribute.get(attribute_value)
+            if (date_format or date_format == "") and attribute_data:
+                replace_with_date_object = datetime.strptime(str(attribute_data), "%Y%m%d").replace(tzinfo=UTC)
+                replace_with = replace_with_date_object.strftime(str(date_format))
+            else:
+                replace_with = attribute_data if attribute_data else ""
+            return str(replace_with)
+        except AttributeError as error:
+            message = "Invalid token format"
+            raise AttributeError(message) from error