Review ready

Author: nhsdevws

backend/poetry.lock

@@ -1,5 +1,6 @@
 import base64
 import datetime
+import urllib.parse
 from dataclasses import dataclass
 
 from aws_lambda_typing.events import APIGatewayProxyEventV1
@@ -9,6 +10,8 @@
 from clients import redis_client, logger
 from models.errors import ParameterException
 from models.constants import Constants
+from constants import SEARCH_IMMUNIZATION_BY_IDENTIFIER_PARAMETERS, SEARCH_IMMUNIZATIONS_PARAMETERS
+
 
 ParamValue = list[str]
 ParamContainer = dict[str, ParamValue]
@@ -162,3 +165,53 @@ def create_query_string(search_params: SearchParams) -> str:
     ]
     search_params_qs = urlencode(sorted(params, key=lambda x: x[0]), safe=",")
     return search_params_qs
+
+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):
+    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
+        
+        # check if any params are not in the valid list
+        for param in all_params:
+            if param in valid_params:
+                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)
+
+
+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/parameter_parser.py

@@ -11,8 +11,9 @@
 from models.errors import Severity, Code, create_operation_outcome
 from constants import GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE
 from log_structure import function_info
-from search_parameter_validator import (
+from parameter_parser import (
     is_immunization_by_identifier,
+    is_search_immunizations,
     get_parsed_body
 )
 
@@ -32,8 +33,14 @@ def search_imms(event: events.APIGatewayProxyEventV1, controller: FhirController
         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)):
+            parsed_body = get_parsed_body(body)
+            if is_immunization_by_identifier(query_params, parsed_body):
                 return controller.get_immunization_by_identifier(event)
+            elif is_search_immunizations(query_params, parsed_body):
+                return controller.search_immunizations(event)
+            else:
+                raise ValueError("Missing search parameters")
+
         response = controller.search_immunizations(event)
 
         result_json = json.dumps(response)
@@ -50,7 +57,7 @@ def search_imms(event: events.APIGatewayProxyEventV1, controller: FhirController
         return response
 
     except ValueError as ve:
-        logger.exception("ValueError occurred")
+        logger.exception("Invalid parameters")
         exp_error = create_operation_outcome(
             resource_id=str(uuid.uuid4()),
             severity=Severity.error,

backend/src/search_imms_handler.py

@@ -1,6 +1,7 @@
 import base64
 import unittest
 import datetime
+import urllib.parse
 from unittest.mock import create_autospec, patch
 
 from authorization import Authorization
@@ -13,7 +14,14 @@
     process_search_params,
     create_query_string,
     SearchParams,
+    body_to_dict,
+    check_route_parameters,
+    is_immunization_by_identifier,
+    is_search_immunizations,
+    get_parsed_body,
 )
+from constants import SEARCH_IMMUNIZATION_BY_IDENTIFIER_PARAMETERS, SEARCH_IMMUNIZATIONS_PARAMETERS
+
 
 class TestParameterParser(unittest.TestCase):
     def setUp(self):
@@ -212,3 +220,81 @@ def test_create_query_string_with_multiple_immunization_targets_comma_separated(
         expected = "-immunization.target=b,c&patient.identifier=https%3A%2F%2Ffhir.nhs.uk%2FId%2Fnhs-number%7Ca"
 
         self.assertEqual(expected, query_string)
+
+
+class TestSearchParameterValidator(unittest.TestCase):
+    def test_body_to_dict_with_list(self):
+        body = [{'key': 'identifier_1', 'value': ['id_1']}]
+        result = body_to_dict(body)
+        self.assertEqual(result, {'identifier_1': ['id_1']})
+
+    def test_body_to_dict_with_dict(self):
+        body = {'identifier_2': ['id_2']}
+        result = body_to_dict(body)
+        self.assertEqual(result, {'identifier_2': ['id_2']})
+
+    def test_body_to_dict_with_none(self):
+        body = None
+        result = body_to_dict(body)
+        self.assertEqual(result, {})
+
+    def test_check_route_parameters_valid(self):
+        query_params = {'identifier_3': 'id_3'}
+        body = None
+        valid_params = ['identifier_3']
+        self.assertTrue(check_route_parameters(query_params, body, valid_params))
+
+    def test_check_route_parameters_invalid(self):
+        query_params = {'invalid_4': 'id_4'}
+        body = None
+        valid_params = ['identifier_4']
+        with self.assertRaises(ValueError) as cm:
+            check_route_parameters(query_params, body, valid_params)
+        self.assertIn("Invalid body parameter: invalid_4", str(cm.exception))
+
+    def test_check_route_parameters_valid_body_list(self):
+        query_params = {}
+        body = [{'key': 'identifier_5', 'value': ['id_5']}]
+        valid_params = ['identifier_5']
+        self.assertTrue(check_route_parameters(query_params, body, valid_params))
+
+    def test_check_route_parameters_invalid_body_list(self):
+        query_params = {}
+        body = [{'key': 'badkey_6', 'value': ['id_6']}]
+        valid_params = ['identifier_6']
+        with self.assertRaises(ValueError) as cm:
+            check_route_parameters(query_params, body, valid_params)
+        self.assertIn("Invalid body parameter: badkey_6", str(cm.exception))
+
+    def test_is_immunization_by_identifier_true(self):
+        query_params = {'identifier': 'id_7'}
+        body = None
+        self.assertTrue(is_immunization_by_identifier(query_params, body))
+
+    def test_is_immunization_by_identifier_false(self):
+        query_params = {'badkey_8': 'id_8'}
+        body = None
+        with self.assertRaises(ValueError):
+            is_immunization_by_identifier(query_params, body)
+
+    def test_is_search_immunizations_true(self):
+        query_params = {'patient.identifier': 'id'}
+        body = None
+        self.assertTrue(is_search_immunizations(query_params, body))
+
+    def test_is_search_immunizations_false(self):
+        query_params = {'badkey_10': 'bad_10'}
+        body = None
+        with self.assertRaises(ValueError):
+            is_search_immunizations(query_params, body)
+
+    def test_get_parsed_body_none(self):
+        self.assertIsNone(get_parsed_body(None))
+
+    def test_get_parsed_body_valid(self):
+        data = {'identifier_11': ['id_11'], 'key_11': ['data_11']}
+        encoded = urllib.parse.urlencode({k: v[0] for k, v in data.items()})
+        b64 = base64.b64encode(encoded.encode("utf-8")).decode("utf-8")
+        parsed = get_parsed_body(b64)
+        self.assertEqual(parsed, data)
+        

backend/src/search_parameter_validator.py

@@ -7,6 +7,7 @@
 from search_imms_handler import search_imms
 from pathlib import Path
 from constants import GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE
+from utils.generic_utils import encode_b64
 
 script_location = Path(__file__).absolute().parent
 
@@ -59,11 +60,13 @@ def test_search_immunizations_to_get_imms_id(self):
         self.controller.get_immunization_by_identifier.assert_called_once_with(lambda_event)
         self.assertDictEqual(exp_res, act_res)
 
+
+
     def test_search_immunizations_get_id_from_body(self):
-        """it should return a list of Immunizations"""
+        """it should NOT return a list of Immunizations. ID is not a valid parameter """
         lambda_event = {
             "pathParameters": {"id": "an-id"},
-            "body": "cGF0aWVudC5pZGVudGlmaWVyPWh0dHBzJTNBJTJGJTJGZmhpci5uaHMudWslMkZJZCUyRm5ocy1udW1iZXIlN0M5NjkzNjMyMTA5Ji1pbW11bml6YXRpb24udGFyZ2V0PUNPVklEMTkmX2luY2x1ZGU9SW1tdW5pemF0aW9uJTNBcGF0aWVudCZpZGVudGlmaWVyPWh0dHBzJTNBJTJGJTJGc3VwcGxpZXJBQkMlMkZpZGVudGlmaWVycyUyRnZhY2MlN0NmMTBiNTliMy1mYzczLTQ2MTYtOTljOS05ZTg4MmFiMzExODQmX2VsZW1lbnRzPWlkJTJDbWV0YSZpZD1z",
+            "body": encode_b64('"id"="s"'),
             "queryStringParameters": None,
         }
         exp_res = {"a-key": "a-value"}
@@ -74,8 +77,13 @@ def test_search_immunizations_get_id_from_body(self):
         act_res = search_imms(lambda_event, self.controller)
 
         # Then
-        self.controller.get_immunization_by_identifier.assert_called_once_with(lambda_event)
-        self.assertDictEqual(exp_res, act_res)
+        # ID is not a valid parameter
+        self.controller.get_immunization_by_identifier.assert_not_called()
+        act_body = json.loads(act_res["body"])
+        act_issue = act_body["issue"][0]
+        self.assertEqual(act_issue["code"], Code.invalid)
+        self.assertEqual(act_issue["severity"], Severity.error)
+        self.assertEqual(act_issue["diagnostics"], 'Error checking route parameters: Invalid body parameter: "id"')
 
     def test_search_immunizations_get_id_from_body_passing_none(self):
         """it should enter search_immunizations as both the request params are none"""
@@ -166,10 +174,9 @@ def test_search_handle_exception(self):
         # Then
         act_body = json.loads(act_res["body"])
 
-        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_issue["code"], Code.server_error)
+        self.assertEqual(act_issue["severity"], Severity.error)
         self.assertEqual(act_res["statusCode"], 500)
 
     def test_search_immunizations_invalid_params(self):
@@ -186,22 +193,8 @@ def test_search_immunizations_invalid_params(self):
         act_res = search_imms(lambda_event, self.controller)
 
         # Then
+        act_body = json.loads(act_res["body"])
+        act_issue = act_body["issue"][0]
         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)
-    
+        self.assertEqual(act_issue["severity"], Severity.error)
+        self.assertEqual(act_issue["diagnostics"], 'Error checking route parameters: Invalid body parameter: _elephants')

backend/tests/test_parameter_parser.py

@@ -3,6 +3,7 @@
 import json
 import os
 import unittest
+import base64
 from decimal import Decimal
 from typing import Literal, Any
 from jsonpath_ng.ext import parse
@@ -91,3 +92,12 @@ def update_contained_resource_field(
         {field_to_update: update_value}
     )
     return json_data
+
+
+def encode_b64(data):
+
+    if isinstance(data, str):
+        data = data.encode('utf-8')
+
+    encoded = base64.b64encode(data)
+    return encoded.decode('utf-8')