eli-579 - handling 'unknown' targets when used in derivation functions

Author: eddalmond1

src/eligibility_signposting_api/services/processors/token_processor.py

@@ -94,6 +94,9 @@ def get_token_replacement(token: str, person_data: list[dict], present_attribute
         if parsed_token.function_name:
             return TokenProcessor.get_derived_value(parsed_token, person_data, present_attributes, token)
 
+        # For non-derived tokens, validate that target attributes are allowed
+        TokenProcessor.validate_target_attribute(parsed_token, token)
+
         found_attribute, key_to_replace = TokenProcessor.find_matching_attribute(parsed_token, person_data)
 
         if not found_attribute or not key_to_replace:
@@ -139,17 +142,16 @@ def get_derived_value(
         # For TARGET level tokens, validate the condition is allowed
         if parsed_token.attribute_level == TARGET_ATTRIBUTE_LEVEL:
             is_allowed_condition = parsed_token.attribute_name in ALLOWED_CONDITIONS.__args__
-            is_allowed_target_attr = parsed_token.attribute_value in ALLOWED_TARGET_ATTRIBUTES
 
             # If condition is not allowed, raise error
             if not is_allowed_condition:
                 TokenProcessor.handle_token_not_found(parsed_token, token)
 
-            # If vaccine type is not in person data but is allowed, return empty
+            # If vaccine type is not in person data, return empty string
+            # For derived values, any target attribute name is allowed (e.g., NEXT_BOOKING_AVAILABLE)
+            # since it's just a placeholder that may be surfaced in the future
             if parsed_token.attribute_name not in present_attributes:
-                if is_allowed_target_attr:
-                    return ""
-                TokenProcessor.handle_token_not_found(parsed_token, token)
+                return ""
 
         try:
             target_attribute = parsed_token.attribute_value or parsed_token.attribute_name
@@ -185,6 +187,26 @@ def should_replace_with_empty(parsed_token: ParsedToken, present_attributes: set
 
         return all([is_target_level, is_allowed_condition, is_allowed_target_attr, is_attr_not_present])
 
+    @staticmethod
+    def validate_target_attribute(parsed_token: ParsedToken, token: str) -> None:
+        """Validate that target attribute is allowed for non-derived tokens.
+
+        For regular (non-derived) tokens, only allow known target attributes.
+        Derived values with functions can use any custom target attribute name.
+
+        Args:
+            parsed_token: The parsed token to validate
+            token: The original token string for error messages
+
+        Raises:
+            ValueError: If the target attribute is not in ALLOWED_TARGET_ATTRIBUTES
+        """
+        if (
+            parsed_token.attribute_level == TARGET_ATTRIBUTE_LEVEL
+            and parsed_token.attribute_value not in ALLOWED_TARGET_ATTRIBUTES
+        ):
+            TokenProcessor.handle_token_not_found(parsed_token, token)
+
     @staticmethod
     def find_matching_attribute(parsed_token: ParsedToken, person_data: list[dict]) -> tuple[dict | None, str | None]:
         attribute_level_map = {

tests/integration/conftest.py

@@ -1162,6 +1162,48 @@ def campaign_config_with_multiple_add_days(
     s3_client.delete_object(Bucket=rules_bucket, Key=f"{campaign.name}.json")
 
 
+@pytest.fixture
+def campaign_config_with_custom_target_attributes(
+    s3_client: BaseClient, rules_bucket: BucketName
+) -> Generator[CampaignConfig]:
+    """Campaign config with custom target attribute names for derived values."""
+    campaign: CampaignConfig = rule.CampaignConfigFactory.build(
+        target="COVID",
+        iterations=[
+            rule.IterationFactory.build(
+                default_comms_routing="CUSTOM_BOOKING_DATE",
+                actions_mapper=rule.ActionsMapperFactory.build(
+                    root={
+                        "CUSTOM_BOOKING_DATE": AvailableAction(
+                            ActionType="DataValue",
+                            ExternalRoutingCode="NextBookingAvailable",
+                            ActionDescription=(
+                                "[[TARGET.COVID.NEXT_BOOKING_AVAILABLE:ADD_DAYS(71, LAST_SUCCESSFUL_DATE):"
+                                "DATE(%d %B %Y)]]"
+                            ),
+                        ),
+                    }
+                ),
+                iteration_rules=[],
+                iteration_cohorts=[
+                    rule.IterationCohortFactory.build(
+                        cohort_label="cohort_label1",
+                        cohort_group="cohort_group1",
+                        positive_description="Positive Description",
+                        negative_description="Negative Description",
+                    )
+                ],
+            )
+        ],
+    )
+    campaign_data = {"CampaignConfig": campaign.model_dump(by_alias=True)}
+    s3_client.put_object(
+        Bucket=rules_bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
+    )
+    yield campaign
+    s3_client.delete_object(Bucket=rules_bucket, Key=f"{campaign.name}.json")
+
+
 @pytest.fixture(scope="class")
 def multiple_campaign_configs(s3_client: BaseClient, rules_bucket: BucketName) -> Generator[list[CampaignConfig]]:
     """Create and upload multiple campaign configs to S3, then clean up after tests."""
@@ -1521,6 +1563,20 @@ def consumer_to_active_campaign_config_with_multiple_add_days_mapping(
     s3_client.delete_object(Bucket=consumer_mapping_bucket, Key="consumer_mapping.json")
 
 
+@pytest.fixture
+def consumer_to_active_campaign_config_with_custom_target_attributes_mapping(
+    s3_client: BaseClient,
+    consumer_mapping_bucket: ConsumerMapping,
+    campaign_config_with_custom_target_attributes: CampaignConfig,
+    consumer_id: ConsumerId,
+):
+    consumer_mapping = create_and_put_consumer_mapping_in_s3(
+        campaign_config_with_custom_target_attributes, consumer_id, consumer_mapping_bucket, s3_client
+    )
+    yield consumer_mapping
+    s3_client.delete_object(Bucket=consumer_mapping_bucket, Key="consumer_mapping.json")
+
+
 @pytest.fixture
 def consumer_to_campaign_having_inactive_iteration_mapping(
     s3_client: BaseClient,

tests/integration/in_process/test_derived_values.py

@@ -246,3 +246,65 @@ def test_multiple_actions_with_different_add_days_parameters(
                 has_item(has_entries(actionCode="DateOfNextDoseAt61Days", description="20260330")),
             ),
         )
+
+
+class TestCustomTargetAttributeNames:
+    """Test that custom target attribute names work with derived values in integration."""
+
+    def test_custom_target_attribute_with_derived_value(
+        self,
+        client: FlaskClient,
+        person_with_covid_vaccination: NHSNumber,
+        consumer_id: ConsumerId,
+        consumer_to_active_campaign_config_with_custom_target_attributes_mapping: ConsumerMapping,  # noqa: ARG002
+        secretsmanager_client: BaseClient,  # noqa: ARG002
+    ):
+        """
+        Test that custom target attribute names like NEXT_BOOKING_AVAILABLE work with derived values.
+
+        This tests the issue reported in production where:
+            [[TARGET.COVID.NEXT_BOOKING_AVAILABLE:ADD_DAYS(71, LAST_SUCCESSFUL_DATE):DATE(%d %B %Y)]]
+        was raising a ValueError.
+
+        Given:
+            - A person with COVID vaccination on 2026-01-28
+            - A campaign config using a custom target attribute: NEXT_BOOKING_AVAILABLE
+            - The token: [[TARGET.COVID.NEXT_BOOKING_AVAILABLE:ADD_DAYS(71, LAST_SUCCESSFUL_DATE):DATE(%d %B %Y)]]
+
+        Expected:
+            - Should calculate 2026-01-28 + 71 days = 2026-04-09
+            - Should format as "09 April 2026"
+            - Should NOT raise ValueError about invalid attribute name
+        """
+        # Given
+        headers = {"nhs-login-nhs-number": str(person_with_covid_vaccination), UNIQUE_CONSUMER_HEADER: str(consumer_id)}
+
+        # When
+        response = client.get(
+            f"/patient-check/{person_with_covid_vaccination}?includeActions=Y",
+            headers=headers,
+        )
+
+        # Then
+        assert_that(
+            response,
+            is_response().with_status_code(HTTPStatus.OK).and_text(is_json_that(has_key("processedSuggestions"))),
+        )
+
+        body = response.get_json()
+        assert_that(body, is_not(none()))
+        processed_suggestions = body.get("processedSuggestions", [])
+
+        covid_suggestion = next(
+            (s for s in processed_suggestions if s.get("condition") == "COVID"),
+            None,
+        )
+        assert_that(covid_suggestion, is_not(none()))
+
+        actions = covid_suggestion.get("actions", [])  # type: ignore[union-attr]
+
+        # Verify the custom target attribute with derived value works correctly
+        assert_that(
+            actions,
+            has_item(has_entries(actionCode="NextBookingAvailable", description="09 April 2026")),
+        )

tests/unit/services/processors/test_token_processor.py

@@ -444,3 +444,167 @@ def test_token_replace_is_case_insensitive(self, token: str, expected: str):
 
         assert result.status_text == f"Your DOB is: {expected}."
         assert result.condition_name == f"FLU vaccine on: {expected}."
+
+
+class TestCustomTargetAttributeNames:
+    """Test that custom target attribute names work with derived values."""
+
+    def test_custom_target_attribute_with_add_days_when_data_present(self):
+        """Test that custom target attributes like NEXT_BOOKING_AVAILABLE work when data is present."""
+        person = Person(
+            [
+                {"ATTRIBUTE_TYPE": "COVID", "LAST_SUCCESSFUL_DATE": "20260128"},
+            ]
+        )
+
+        condition = Condition(
+            condition_name=ConditionName("Test"),
+            status=Status.actionable,
+            status_text=StatusText(
+                "Next booking: [[TARGET.COVID.NEXT_BOOKING_AVAILABLE:ADD_DAYS(71, LAST_SUCCESSFUL_DATE)]]"
+            ),
+            cohort_results=[],
+            suitability_rules=[],
+            actions=[],
+        )
+
+        result = TokenProcessor.find_and_replace_tokens(person, condition)
+
+        # 2026-01-28 + 71 days = 2026-04-09
+        assert result.status_text == "Next booking: 20260409"
+
+    def test_custom_target_attribute_with_add_days_and_formatting(self):
+        """Test that custom target attributes work with both ADD_DAYS and DATE formatting."""
+        person = Person(
+            [
+                {"ATTRIBUTE_TYPE": "COVID", "LAST_SUCCESSFUL_DATE": "20260128"},
+            ]
+        )
+
+        condition = Condition(
+            condition_name=ConditionName("Test"),
+            status=Status.actionable,
+            status_text=StatusText(
+                "Date: [[TARGET.COVID.NEXT_BOOKING_AVAILABLE:ADD_DAYS(71, LAST_SUCCESSFUL_DATE):DATE(%d %B %Y)]]"
+            ),
+            cohort_results=[],
+            suitability_rules=[],
+            actions=[],
+        )
+
+        result = TokenProcessor.find_and_replace_tokens(person, condition)
+
+        # 2026-01-28 + 71 days = 2026-04-09, formatted as "09 April 2026"
+        assert result.status_text == "Date: 09 April 2026"
+
+    def test_custom_target_attribute_returns_empty_when_condition_not_present(self):
+        """Test that custom target attributes return empty string when condition data not present."""
+        person = Person(
+            [
+                {"ATTRIBUTE_TYPE": "PERSON", "AGE": "30"},
+                # No COVID data present
+            ]
+        )
+
+        condition = Condition(
+            condition_name=ConditionName("Test"),
+            status=Status.actionable,
+            status_text=StatusText(
+                "Next booking: [[TARGET.COVID.NEXT_BOOKING_AVAILABLE:ADD_DAYS(71, LAST_SUCCESSFUL_DATE)]]"
+            ),
+            cohort_results=[],
+            suitability_rules=[],
+            actions=[],
+        )
+
+        result = TokenProcessor.find_and_replace_tokens(person, condition)
+
+        # Should return empty string when condition data is not present
+        assert result.status_text == "Next booking: "
+
+    def test_multiple_custom_target_attributes_with_different_functions(self):
+        """Test multiple custom target attributes with different parameters."""
+        person = Person(
+            [
+                {"ATTRIBUTE_TYPE": "COVID", "LAST_SUCCESSFUL_DATE": "20260128"},
+            ]
+        )
+
+        condition = Condition(
+            condition_name=ConditionName("Test"),
+            status=Status.actionable,
+            status_text=StatusText(
+                "First: [[TARGET.COVID.CUSTOM_FIELD_A:ADD_DAYS(30, LAST_SUCCESSFUL_DATE)]] "
+                "Second: [[TARGET.COVID.CUSTOM_FIELD_B:ADD_DAYS(60, LAST_SUCCESSFUL_DATE)]]"
+            ),
+            cohort_results=[],
+            suitability_rules=[],
+            actions=[],
+        )
+
+        result = TokenProcessor.find_and_replace_tokens(person, condition)
+
+        # 2026-01-28 + 30 = 2026-02-27, + 60 = 2026-03-29
+        assert result.status_text == "First: 20260227 Second: 20260329"
+
+    def test_custom_target_attribute_raises_error_for_invalid_condition(self):
+        """Test that invalid condition names still raise errors even with custom target attributes."""
+        person = Person(
+            [
+                {"ATTRIBUTE_TYPE": "PERSON", "AGE": "30"},
+            ]
+        )
+
+        condition = Condition(
+            condition_name=ConditionName("Test"),
+            status=Status.actionable,
+            status_text=StatusText("Invalid: [[TARGET.INVALID_CONDITION.CUSTOM_FIELD:ADD_DAYS(30)]]"),
+            cohort_results=[],
+            suitability_rules=[],
+            actions=[],
+        )
+
+        with pytest.raises(ValueError, match="Invalid attribute name 'CUSTOM_FIELD'"):
+            TokenProcessor.find_and_replace_tokens(person, condition)
+
+    def test_non_derived_token_with_invalid_target_attribute_raises_error(self):
+        """Test that non-derived tokens (without functions) validate target attributes strictly."""
+        person = Person(
+            [
+                {"ATTRIBUTE_TYPE": "COVID", "LAST_SUCCESSFUL_DATE": "20260128"},
+            ]
+        )
+
+        condition = Condition(
+            condition_name=ConditionName("Test"),
+            status=Status.actionable,
+            status_text=StatusText("Invalid: [[TARGET.COVID.CUSTOM_INVALID_FIELD]]"),
+            cohort_results=[],
+            suitability_rules=[],
+            actions=[],
+        )
+
+        # Non-derived tokens should only allow ALLOWED_TARGET_ATTRIBUTES
+        with pytest.raises(ValueError, match="Invalid attribute name 'CUSTOM_INVALID_FIELD'"):
+            TokenProcessor.find_and_replace_tokens(person, condition)
+
+    def test_non_derived_token_with_valid_target_attribute_works(self):
+        """Test that non-derived tokens with valid target attributes work correctly."""
+        person = Person(
+            [
+                {"ATTRIBUTE_TYPE": "COVID", "LAST_SUCCESSFUL_DATE": "20260128"},
+            ]
+        )
+
+        condition = Condition(
+            condition_name=ConditionName("Test"),
+            status=Status.actionable,
+            status_text=StatusText("Last date: [[TARGET.COVID.LAST_SUCCESSFUL_DATE]]"),
+            cohort_results=[],
+            suitability_rules=[],
+            actions=[],
+        )
+
+        result = TokenProcessor.find_and_replace_tokens(person, condition)
+
+        assert result.status_text == "Last date: 20260128"