Search Fails with unrecognised typo

Author: nhsdevws

backend/Makefile

@@ -6,10 +6,10 @@ package: build
 	docker run --rm -v $(shell pwd)/build:/build imms-lambda-build
 
 test:
-	@PYTHONPATH=src:tests python -m unittest
+	@PYTHONPATH=src:tests:tests/utils python -m unittest
 
 coverage-run:
-	@PYTHONPATH=src:tests coverage run -m unittest discover
+	@PYTHONPATH=src:tests:tests/utils coverage run -m unittest discover
 
 coverage-report:
 	coverage report -m 

backend/poetry.lock

@@ -23,3 +23,16 @@ class Urls:
 
 
 GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE = "Unable to process request. Issue may be transient."
+
+SEARCH_IMMUNIZATIONS_PARAMETERS = [
+    "patient.identifier",
+    "-immunization.target",
+    "_include",
+    "-date.from",
+    "-date.to"
+]
+
+SEARCH_IMMUNIZATION_BY_IDENTIFIER_PARAMETERS = [
+    "identifier",
+    "_elements"
+]

backend/src/constants.py

@@ -11,8 +11,10 @@
 from models.errors import Severity, Code, create_operation_outcome
 from constants import GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE
 from log_structure import function_info
-import base64
-import urllib.parse
+from search_parameter_validator import (
+    is_immunization_by_identifier,
+    get_parsed_body
+)
 
 logging.basicConfig(level="INFO")
 logger = logging.getLogger()
@@ -24,33 +26,13 @@ def search_imms_handler(event: events.APIGatewayProxyEventV1, _context: context_
 
 def search_imms(event: events.APIGatewayProxyEventV1, controller: FhirController):
     try:
+
         query_params = event.get("queryStringParameters", {})
         body = event.get("body")
-        body_has_immunization_identifier = False
-        query_string_has_immunization_identifier = False
-        query_string_has_element = False
-        body_has_immunization_element = False
-        if not (query_params == None and body == None):
-            if query_params:
-                query_string_has_immunization_identifier = "identifier" in event.get(
-                    "queryStringParameters", {}
-                )
-                query_string_has_element = "_elements" in event.get("queryStringParameters", {})
-            # Decode body from base64
-            if body:
-                decoded_body = base64.b64decode(body).decode("utf-8")
-                # Parse the URL encoded body
-                parsed_body = urllib.parse.parse_qs(decoded_body)
-
-                # Check for 'identifier' in body
-                body_has_immunization_identifier = "identifier" in parsed_body
-                body_has_immunization_element = "_elements" in parsed_body
-            if (
-                query_string_has_immunization_identifier
-                or body_has_immunization_identifier
-                or query_string_has_element
-                or body_has_immunization_element
-            ):
+        has_body = body is not None
+        has_query_params = query_params is not None and query_params != {}
+        if has_query_params or has_body:
+            if is_immunization_by_identifier(query_params, get_parsed_body(body)):
                 return controller.get_immunization_by_identifier(event)
         response = controller.search_immunizations(event)
 
@@ -66,6 +48,16 @@ def search_imms(event: events.APIGatewayProxyEventV1, controller: FhirController
             )
             return FhirController.create_response(400, exp_error)
         return response
+            
+    except ValueError as ve:
+        logger.exception("ValueError occurred")
+        exp_error = create_operation_outcome(
+            resource_id=str(uuid.uuid4()),
+            severity=Severity.error,
+            code=Code.invalid,
+            diagnostics=str(ve)
+        )
+        return FhirController.create_response(400, exp_error)
     except Exception:  # pylint: disable = broad-exception-caught
         logger.exception("Unhandled exception")
         exp_error = create_operation_outcome(

backend/src/search_imms_handler.py

@@ -0,0 +1,55 @@
+import base64
+import urllib.parse
+from constants import SEARCH_IMMUNIZATION_BY_IDENTIFIER_PARAMETERS, SEARCH_IMMUNIZATIONS_PARAMETERS
+
+
+def body_to_dict(body):
+    """
+    Converts a body array of {'key': ..., 'value': ...} to a dict.
+    """
+    if isinstance(body, list):
+        return {item['key']: item['value'] for item in body if 'key' in item and 'value' in item}
+    return body if isinstance(body, dict) else {}
+
+
+def check_route_parameters(query_params, body, valid_params: list, second_list: list = None):
+    try:
+
+        query_params = query_params or {}
+        body_dict = body_to_dict(body)
+        # merge query and body parameters
+        all_params = {**query_params, **body_dict}
+
+        found = False
+        for param in all_params:
+            if param in valid_params:
+                found = True
+                break
+        
+        for param in all_params:
+            if param in valid_params or param in second_list:
+                continue
+            elif param == "id":
+                continue
+            else:
+                raise ValueError(f"Invalid body parameter: {param}")
+        return found
+    except Exception as e:
+        raise ValueError(f"Error checking route parameters: {e}")
+
+
+def is_immunization_by_identifier(query_params, body):
+    # check the parameters indicate search by identifier
+    return check_route_parameters(query_params, body, SEARCH_IMMUNIZATION_BY_IDENTIFIER_PARAMETERS,
+                                  SEARCH_IMMUNIZATIONS_PARAMETERS)
+
+# def is_search_immunizations(query_params, body):
+#     # check the parameters indicate search for immunizations
+#     return check_route_parameters(query_params, body, SEARCH_IMMUNIZATIONS_PARAMETERS)
+
+def get_parsed_body(body):
+    if body:
+        decoded_body = base64.b64decode(body).decode("utf-8")
+        # Parse the URL encoded body
+        return urllib.parse.parse_qs(decoded_body)
+    return None

backend/src/search_parameter_validator.py

@@ -129,7 +129,10 @@ def test_search_immunizations_get_id_from_body_imms_identifer(self):
 
     def test_search_immunizations_lambda_size_limit(self):
         """it should return 400 as search returned too many results."""
-        lambda_event = {"pathParameters": {"id": "an-id"}, "body": None}
+        lambda_event = {
+            "pathParameters": {"id": "an-id"},
+            "body": None,
+            }
         request_file = script_location / "sample_data" / "sample_input_search_imms.json"
         with open(request_file) as f:
             exp_res = json.load(f)
@@ -163,6 +166,42 @@ def test_search_handle_exception(self):
         # Then
         act_body = json.loads(act_res["body"])
 
-        self.assertEqual(exp_error["issue"][0]["code"], act_body["issue"][0]["code"])
-        self.assertEqual(exp_error["issue"][0]["severity"], act_body["issue"][0]["severity"])
+        exp_issue = exp_error["issue"][0]
+        act_issue = act_body["issue"][0]
+        self.assertEqual(exp_issue["code"], act_issue["code"])
+        self.assertEqual(exp_issue["severity"], act_issue["severity"])
         self.assertEqual(act_res["statusCode"], 500)
+
+    def test_search_immunizations_invalid_params(self):
+        """it should return 400 if invalid parameters are provided"""
+        lambda_event = {
+            "pathParameters": {"id": "an-id"},
+            "queryStringParameters": {
+                "identifier": "https://supplierABC/identifiers/vacc|f10b59b3-fc73-4616-99c9-9e882ab31184",
+                "_elephants": "id,meta",
+            },
+            "body": None,
+        }
+        # When
+        act_res = search_imms(lambda_event, self.controller)
+
+        # Then
+        self.assertEqual(act_res["statusCode"], 400)
+    
+
+    def test_search_immunizations_invalid_params(self):
+        """it should return 400 if only invalid parameters are provided"""
+        lambda_event = {
+            "pathParameters": {"id": "an-id"},
+            "queryStringParameters": {
+                "_elephants": "id,meta",
+            },
+            "body": None,
+        }
+
+        # When
+        act_res = search_imms(lambda_event, self.controller)
+
+        # Then
+        self.assertEqual(act_res["statusCode"], 400)
+    

backend/tests/test_search_imms.py

@@ -1,6 +1,6 @@
 import unittest
 from unittest.mock import patch
-from src.models.utils.permissions import get_supplier_permissions
+from models.utils.permissions import get_supplier_permissions
 
 
 class TestPermissions(unittest.TestCase):