Merge branch 'main' into feature/te-auto-test-deployments

Author: TOEL2

infrastructure/stacks/iams-developer-roles/github_actions_policies.tf

@@ -65,9 +65,12 @@ resource "aws_iam_policy" "lambda_management" {
           "lambda:GetPolicy",
           "lambda:GetAlias",
           "lambda:GetFunction",
-          "lambda:GetProvisionedConcurrencyConfig",
           "lambda:GetLayerVersion",
-          "lambda:PutProvisionedConcurrencyConfig"
+          "lambda:GetProvisionedConcurrencyConfig",
+          "lambda:PutProvisionedConcurrencyConfig",
+          "lambda:DeleteProvisionedConcurrencyConfig",
+          "lambda:ListProvisionedConcurrencyConfigs",
+
         ],
         Resource = [
           "arn:aws:lambda:*:${data.aws_caller_identity.current.account_id}:function:eligibility_signposting_api",
@@ -109,7 +112,7 @@ resource "aws_iam_policy" "dynamodb_management" {
         }
       ],
       # to create test users in preprod
-      var.environment == "preprod" ? [
+        var.environment == "preprod" ? [
         {
           Effect = "Allow",
           Action = [

infrastructure/stacks/iams-developer-roles/iams_permissions_boundary.tf

@@ -152,9 +152,11 @@ data "aws_iam_policy_document" "permissions_boundary" {
       "lambda:RemovePermission",
       "lambda:GetPolicy",
       "lambda:GetAlias",
-      "lambda:GetProvisionedConcurrencyConfig",
       "lambda:GetLayerVersion",
+      "lambda:GetProvisionedConcurrencyConfig",
       "lambda:PutProvisionedConcurrencyConfig",
+      "lambda:DeleteProvisionedConcurrencyConfig",
+      "lambda:ListProvisionedConcurrencyConfigs",
 
       # CloudWatch Logs - log management
       "logs:CreateLogGroup",

src/eligibility_signposting_api/common/error_handler.py

@@ -6,19 +6,26 @@
 from werkzeug.exceptions import HTTPException
 
 from eligibility_signposting_api.common.api_error_response import INTERNAL_SERVER_ERROR
+from eligibility_signposting_api.services.processors.token_processor import TokenError
 
 logger = logging.getLogger(__name__)
 
 
 def handle_exception(e: Exception) -> ResponseReturnValue | HTTPException:
-    logger.exception("Unexpected Exception", exc_info=e)
-
-    # Let Flask handle its own exceptions for now.
     if isinstance(e, HTTPException):
         return e
 
-    full_traceback = "".join(traceback.format_exception(e))
+    if isinstance(e, TokenError):
+        tb = traceback.extract_tb(e.__traceback__)
+        clean_traceback = "".join(traceback.format_list(tb))
+        logger.error("A ValueError occurred (value redacted). Traceback follows:\n%s", clean_traceback)
+        log_msg = f"An unexpected error occurred: {clean_traceback}"
+    else:
+        full_traceback = "".join(traceback.format_exception(e))
+        logger.exception("Unexpected Exception", exc_info=e)
+        log_msg = f"An unexpected error occurred: {full_traceback}"
+
     response = INTERNAL_SERVER_ERROR.log_and_generate_response(
-        log_message=f"An unexpected error occurred: {full_traceback}", diagnostics="An unexpected error occurred."
+        log_message=log_msg, diagnostics="An unexpected error occurred."
     )
     return make_response(response.get("body"), response.get("statusCode"), response.get("headers"))

src/eligibility_signposting_api/services/processors/token_parser.py

@@ -60,7 +60,7 @@ def parse(token: str) -> ParsedToken:
 
         format_str = format_match.group(1) if format_match else None
 
-        last_part = re.sub(r":DATE\(.*?\)", "", token_name, flags=re.IGNORECASE)
+        last_part = re.sub(r":DATE\([^)]*\)", "", token_name, flags=re.IGNORECASE)
 
         if len(token_parts) == TokenParser.MIN_TOKEN_PARTS:
             name = last_part.upper()

src/eligibility_signposting_api/services/processors/token_processor.py

@@ -1,17 +1,14 @@
 import re
 from dataclasses import Field, fields, is_dataclass
 from datetime import UTC, datetime
-from typing import Any, Never, TypeVar
+from typing import Any, Never
 
 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 = {
@@ -27,10 +24,14 @@
 }
 
 
+class TokenError(Exception):
+    """Person value error."""
+
+
 @service
 class TokenProcessor:
     @staticmethod
-    def find_and_replace_tokens(person: Person, data_class: T) -> T:
+    def find_and_replace_tokens[T](person: Person, data_class: T) -> T:
         if not is_dataclass(data_class):
             return data_class
         for class_field in fields(data_class):
@@ -95,8 +96,9 @@ def replace_token(text: str, person: Person) -> str:
                 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__
+                    is_correct_target = parsed_token.attribute_name.upper() == attribute.get("ATTRIBUTE_TYPE")
 
-                    if (is_allowed_target or is_person_attribute) and key_to_find in attribute:
+                    if ((is_allowed_target and is_correct_target) or is_person_attribute) and key_to_find in attribute:
                         found_attribute = attribute
                         key_to_replace = key_to_find
                         break
@@ -120,9 +122,9 @@ def handle_token_not_found(parsed_token: ParsedToken, token: str) -> Never:
         raise ValueError(message)
 
     @staticmethod
-    def apply_formatting(attribute: dict[str, T], attribute_value: str, date_format: str | None) -> str:
+    def apply_formatting[T](attributes: dict[str, T], attribute_name: str, date_format: str | None) -> str:
         try:
-            attribute_data = attribute.get(attribute_value)
+            attribute_data = attributes.get(attribute_name)
             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))
@@ -132,3 +134,6 @@ def apply_formatting(attribute: dict[str, T], attribute_value: str, date_format:
         except AttributeError as error:
             message = "Invalid token format"
             raise AttributeError(message) from error
+        except ValueError as error:
+            message = "Invalid value error"
+            raise TokenError(message) from error

tests/e2e/data/configs/inProgressTestConfigs/ELI-376-405/AUTO_RSV_ELI-376-02.json

@@ -0,0 +1,88 @@
+{
+  "CampaignConfig": {
+    "ID": "AUTO_RSV_ELI-223-01-Campaign-ID",
+    "Version": 1,
+    "Name": "ELI-223-01-Iteration-Config-Name",
+    "Type": "V",
+    "Target": "RSV",
+    "Manager": [
+      "[email protected]"
+    ],
+    "Approver": [
+      "[email protected]"
+    ],
+    "Reviewer": [
+      "[email protected]"
+    ],
+    "IterationFrequency": "X",
+    "IterationType": "O",
+    "IterationTime": "07:00:00",
+    "StartDate": "20250717",
+    "EndDate": "20350717",
+    "ApprovalMinimum": 0,
+    "ApprovalMaximum": 0,
+    "DefaultCommsRouting": "BOOK_NBS",
+    "Iterations": [
+      {
+        "ID": "AUTO_RSV_ELI-223-01-Iteration-ID",
+        "DefaultCommsRouting": "TEST_ACTION",
+        "DefaultNotActionableRouting": "TEST_NOT_ACTION",
+        "DefaultNotEligibleRouting": "TEST_NOT_ELI",
+        "Version": 1,
+        "Name": "ELI-223-01-Iteration-Config-Name",
+        "IterationDate": "20240808",
+        "IterationNumber": 1,
+        "CommsType": "I",
+        "ApprovalMinimum": 0,
+        "ApprovalMaximum": 0,
+        "Type": "O",
+        "IterationCohorts": [
+          {
+            "CohortLabel": "rsv_eli_223_cohort_1",
+            "CohortGroup": "rsv_eli_223_cohort_group",
+            "PositiveDescription": "are a member of eli_223_cohort_group",
+            "NegativeDescription": "are not a member of eli_223_cohort_group",
+            "Priority": 0
+          }
+        ],
+        "IterationRules": [
+          {
+            "Type": "S",
+            "Name": "Already Vaccinated",
+            "Description": "## You've had your RSV vaccination\\n\\nWe believe you had the RSV vaccination on [[TARGET.RSV.LAST_SUCCESSFUL_DATE:DATE(%d %B %Y)]]",
+            "Priority": 200,
+            "AttributeLevel": "TARGET",
+            "AttributeTarget": "RSV",
+            "AttributeName": "LAST_SUCCESSFUL_DATE",
+            "Operator": "Y>=",
+            "Comparator": "-25",
+            "RuleStop": "Y"
+          }
+        ],
+        "ActionsMapper": {
+          "TEST_ACTION": {
+            "ExternalRoutingCode": "TestActionDefault",
+            "ActionDescription": "TestActionDefault Description",
+            "ActionType": "ButtonWithAuthLink",
+            "UrlLink": "http://www.nhs.uk/book-rsv",
+            "UrlLabel": "TestActionDefault"
+          },
+          "TEST_NOT_ACTION": {
+            "ExternalRoutingCode": "TestNotAction",
+            "ActionDescription": "TestNotAction Description",
+            "ActionType": "ButtonWithAuthLink",
+            "UrlLink": null,
+            "UrlLabel": ""
+          },
+          "AMEND_NBS": {
+            "ExternalRoutingCode": "AmendNBS",
+            "ActionDescription": "## You have an RSV vaccination appointment\n You can view, change or cancel your appointment below.",
+            "ActionType": "ButtonWithAuthLink",
+            "UrlLink": "http://www.nhs.uk/book-rsv",
+            "UrlLabel": "Manage your appointment"
+          }
+        }
+      }
+    ]
+  }
+}