ELI-223: Fixes review comments

Author: shweta-nhs

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/services/calculators/eligibility_calculator.py

@@ -4,7 +4,7 @@
 from collections import defaultdict
 from dataclasses import dataclass, field
 from itertools import chain
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING
 
 from wireup import service
 
@@ -37,7 +37,6 @@
 
 
 logger = logging.getLogger(__name__)
-T = TypeVar("T")
 
 
 @service

src/eligibility_signposting_api/services/processors/token_parser.py

@@ -2,59 +2,70 @@
 from dataclasses import dataclass
 
 
-class InvalidTokenError(ValueError):
-    def __init__(self, message: str = "Invalid token.") -> None:
-        super().__init__(message)
-
-
-class InvalidTokenFormatError(ValueError):
-    def __init__(self, message: str = "Invalid token format.") -> None:
-        super().__init__(message)
-
-
 @dataclass
 class ParsedToken:
-    attribute_level: str  # example: "PERSON" or "TARGET"
-    attribute_name: str  # example: "POSTCODE" or "RSV"
-    attribute_value: str | None  # example: "LAST_SUCCESSFUL_DATE" if attribute_level is TARGET
-    format: str | None  # example: "%d %B %Y" if DATE formatting is used
+    """
+    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:
-        token_body = token[2:-2]  # Strip the surrounding [[ ]]
-        # Check for empty body after stripping, e.g., '[[]]'
+        """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:
-            raise InvalidTokenError
+            message = "Invalid token."
+            raise ValueError(message)
 
         token_parts = token_body.split(".")
 
-        # Check for empty parts created by leading/trailing dots or tokens with no dot
         if len(token_parts) < TokenParser.MIN_TOKEN_PARTS or not all(token_parts):
-            raise InvalidTokenError
+            message = "Invalid token."
+            raise ValueError(message)
 
         token_level = token_parts[0].upper()
         token_name = token_parts[-1]
 
-        # Check if the name contains a date format
         format_match = re.search(r":DATE\(([^()]*)\)", token_name, re.IGNORECASE)
         if not format_match and len(token_name.split(":")) > 1:
-            raise InvalidTokenFormatError
+            message = "Invalid token format."
+            raise ValueError(message)
 
         format_str = format_match.group(1) if format_match else None
 
-        # Remove the date format from the last part
         last_part = re.sub(r":DATE\(.*?\)", "", token_name, flags=re.IGNORECASE)
 
         if len(token_parts) == TokenParser.MIN_TOKEN_PARTS:
-            # Person token, example, [[PERSON.AGE]]
             name = last_part.upper()
             value = None
         else:
-            # Target token, example, [[TARGET.RSV.LAST_SUCCESSFUL_DATE]]
             name = token_parts[1].upper()
             value = last_part.upper()
 

src/eligibility_signposting_api/services/processors/token_processor.py

@@ -5,6 +5,7 @@
 
 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
 
@@ -13,7 +14,17 @@
 
 TARGET_ATTRIBUTE_LEVEL = "TARGET"
 PERSON_ATTRIBUTE_LEVEL = "PERSON"
-ALLOWED_CONDITION = "RSV"
+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
@@ -59,18 +70,6 @@ def replace_token(text: str, person: Person) -> str:
 
         pattern = r"\[\[.*?\]\]"
         all_tokens = re.findall(pattern, text, re.IGNORECASE)
-        allowed_target_attributes = {
-            "NHS_NUMBER",
-            "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",
-        }
         present_attributes = [attribute.get("ATTRIBUTE_TYPE") for attribute in person.data]
 
         for token in all_tokens:
@@ -86,18 +85,18 @@ def replace_token(text: str, person: Person) -> str:
 
             if (
                 parsed_token.attribute_level == TARGET_ATTRIBUTE_LEVEL
-                and parsed_token.attribute_name == ALLOWED_CONDITION
-                and parsed_token.attribute_value in allowed_target_attributes
+                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_target_rsv = parsed_token.attribute_name.upper() == ALLOWED_CONDITION
+                    is_allowed_target = parsed_token.attribute_name.upper() in ALLOWED_CONDITIONS.__args__
 
-                    if (is_target_rsv or is_person_attribute) and key_to_find in attribute:
+                    if (is_allowed_target or is_person_attribute) and key_to_find in attribute:
                         found_attribute = attribute
                         key_to_replace = key_to_find
                         break

tests/unit/services/processors/test_token_processor.py

@@ -166,16 +166,16 @@ def test_missing_patient_vaccine_data_on_target_attribute_should_replace_with_em
 
         assert actual.condition_name == "Last successful date: "
 
-    def test_non_rsv_target_token_should_raise_error(self):
+    def test_not_allowed_target_conditions_token_should_raise_error(self):
         person = Person(
             [
                 {"ATTRIBUTE_TYPE": "PERSON", "AGE": "30"},
-                {"ATTRIBUTE_TYPE": "COVID", "CONDITION_NAME": "COVID", "LAST_SUCCESSFUL_DATE": "20250101"},
+                {"ATTRIBUTE_TYPE": "YELLOW_FEVER", "CONDITION_NAME": "COVID", "LAST_SUCCESSFUL_DATE": "20250101"},
             ]
         )
 
         condition = Condition(
-            condition_name=ConditionName("Last successful date: [[TARGET.COVID.LAST_SUCCESSFUL_DATE]]"),
+            condition_name=ConditionName("Last successful date: [[TARGET.YELLOW_FEVER.LAST_SUCCESSFUL_DATE]]"),
             status=Status.actionable,
             status_text=StatusText("Some status"),
             cohort_results=[],
@@ -184,7 +184,7 @@ def test_non_rsv_target_token_should_raise_error(self):
         )
 
         expected_error = re.escape(
-            "Invalid attribute name 'LAST_SUCCESSFUL_DATE' in token '[[TARGET.COVID.LAST_SUCCESSFUL_DATE]]'."
+            "Invalid attribute name 'LAST_SUCCESSFUL_DATE' in token '[[TARGET.YELLOW_FEVER.LAST_SUCCESSFUL_DATE]]'."
         )
         with pytest.raises(ValueError, match=expected_error):
             TokenProcessor.find_and_replace_tokens(person, condition)
@@ -266,7 +266,7 @@ def test_valid_token_valid_format_should_replace_with_date_formatting(self):
 
         condition = Condition(
             condition_name=ConditionName(
-                "You had your RSV vaccine on [[TARGET.RSV.LAST_SUCCESSFUL_DATE:DATE(%d %B %Y)]]"
+                "You had your COVID vaccine on [[TARGET.COVID.LAST_SUCCESSFUL_DATE:DATE(%d %B %Y)]]"
             ),
             status=Status.actionable,
             status_text=StatusText("Your birthday is on [[PERSON.DATE_OF_BIRTH:DATE(%-d %B %Y)]]"),
@@ -276,7 +276,7 @@ def test_valid_token_valid_format_should_replace_with_date_formatting(self):
         )
 
         expected = Condition(
-            condition_name=ConditionName("You had your RSV vaccine on 01 January 2025"),
+            condition_name=ConditionName("You had your COVID vaccine on 01 January 2025"),
             status=Status.actionable,
             status_text=StatusText("Your birthday is on 27 March 1990"),
             cohort_results=[],
@@ -334,12 +334,12 @@ def test_valid_token_invalid_format_should_raise_error(self, token_format: str):
     def test_valid_date_format(self, token_format: str, expected: str):
         person = Person(
             [
-                {"ATTRIBUTE_TYPE": "RSV", "CONDITION_NAME": "RSV", "LAST_SUCCESSFUL_DATE": "19900327"},
+                {"ATTRIBUTE_TYPE": "MMR", "CONDITION_NAME": "MMR", "LAST_SUCCESSFUL_DATE": "19900327"},
             ]
         )
 
         condition = Condition(
-            condition_name=ConditionName(f"Date: [[TARGET.RSV.LAST_SUCCESSFUL_DATE{token_format}]]"),
+            condition_name=ConditionName(f"Date: [[TARGET.MMR.LAST_SUCCESSFUL_DATE{token_format}]]"),
             status=Status.actionable,
             status_text=StatusText("Some text"),
             cohort_results=[],
@@ -358,22 +358,22 @@ def test_valid_date_format(self, token_format: str, expected: str):
             ("[[PERSON.date_of_birth:DATE(%d %B %Y)]]", "27 March 1990"),
             ("[[PERSON.DATE_OF_BIRTH:date(%d %B %Y)]]", "27 March 1990"),
             ("[[pErSoN.DATE_OF_BIRTH:DATE(%d %B %Y)]]", "27 March 1990"),
-            ("[[target.RSV.LAST_SUCCESSFUL_DATE:DATE(%-d %B %Y)]]", "1 January 2025"),
-            ("[[TARGET.rsv.LAST_SUCCESSFUL_DATE:DATE(%-d %B %Y)]]", "1 January 2025"),
-            ("[[TARGET.RSV.last_successful_date:DATE(%-d %B %Y)]]", "1 January 2025"),
-            ("[[TARGET.RSV.last_successful_date:date(%-d %B %Y)]]", "1 January 2025"),
+            ("[[target.FLU.LAST_SUCCESSFUL_DATE:DATE(%-d %B %Y)]]", "1 January 2025"),
+            ("[[TARGET.FLU.LAST_SUCCESSFUL_DATE:DATE(%-d %B %Y)]]", "1 January 2025"),
+            ("[[TARGET.FLU.last_successful_date:DATE(%-d %B %Y)]]", "1 January 2025"),
+            ("[[TARGET.FLU.last_successful_date:date(%-d %B %Y)]]", "1 January 2025"),
         ],
     )
     def test_token_replace_is_case_insensitive(self, token: str, expected: str):
         person = Person(
             [
                 {"ATTRIBUTE_TYPE": "PERSON", "AGE": "30", "DATE_OF_BIRTH": "19900327"},
-                {"ATTRIBUTE_TYPE": "RSV", "CONDITION_NAME": "RSV", "LAST_SUCCESSFUL_DATE": "20250101"},
+                {"ATTRIBUTE_TYPE": "FLU", "CONDITION_NAME": "FLU", "LAST_SUCCESSFUL_DATE": "20250101"},
             ]
         )
 
         condition = Condition(
-            condition_name=ConditionName(f"RSV vaccine on: {token}."),
+            condition_name=ConditionName(f"FLU vaccine on: {token}."),
             status=Status.actionable,
             status_text=StatusText(f"Your DOB is: {token}."),
             cohort_results=[],
@@ -384,4 +384,4 @@ def test_token_replace_is_case_insensitive(self, token: str, expected: str):
         result = TokenProcessor.find_and_replace_tokens(person, condition)
 
         assert result.status_text == f"Your DOB is: {expected}."
-        assert result.condition_name == f"RSV vaccine on: {expected}."
+        assert result.condition_name == f"FLU vaccine on: {expected}."