VED-967 Fix internal-qa url (#1065)

Author: dlzhry2nhs

lambdas/backend/src/controller/fhir_controller.py

@@ -36,6 +36,7 @@
 from service.fhir_service import FhirService, get_service_url
 
 IMMUNIZATION_ENV = os.getenv("IMMUNIZATION_ENV")
+IMMUNIZATION_BASE_PATH = os.getenv("IMMUNIZATION_BASE_PATH")
 
 
 def make_controller(
@@ -51,7 +52,7 @@ def make_controller(
 
 class FhirController:
     _IMMUNIZATION_ID_PATTERN = r"^[A-Za-z0-9\-.]{1,64}$"
-    _API_SERVICE_URL = get_service_url()
+    _API_SERVICE_URL = get_service_url(IMMUNIZATION_ENV, IMMUNIZATION_BASE_PATH)
 
     def __init__(
         self,

lambdas/backend/src/service/constants.py

@@ -0,0 +1,4 @@
+"""Constants for the fhir_service layer"""
+
+DEFAULT_BASE_PATH = "immunisation-fhir-api/FHIR/R4"
+PR_ENV_PREFIX = "pr-"

lambdas/backend/src/service/fhir_service.py

@@ -2,7 +2,6 @@
 import datetime
 import logging
 import os
-import urllib.parse
 import uuid
 from typing import Any, Optional, cast
 from uuid import uuid4
@@ -43,11 +42,10 @@
     validate_identifiers_match,
     validate_resource_versions_match,
 )
-from controller.constants import IMMUNIZATION_TARGET_LEGACY_KEY_NAME, ImmunizationSearchParameterName
-from controller.parameter_parser import PATIENT_IDENTIFIER_SYSTEM
 from filter import Filter
 from models.errors import UnauthorizedVaxError
 from repository.fhir_repository import ImmunizationRepository
+from service.search_url_helper import create_url_for_bundle_link, get_service_url
 
 logging.basicConfig(level="INFO")
 logger = logging.getLogger()
@@ -59,24 +57,6 @@
 IMMUNIZATION_VALIDATOR = ImmunizationValidator()
 
 
-def get_service_url(
-    service_env: str = IMMUNIZATION_ENV,
-    service_base_path: str = IMMUNIZATION_BASE_PATH,
-) -> str:
-    if not service_base_path:
-        service_base_path = "immunisation-fhir-api/FHIR/R4"
-
-    non_prod = ["internal-dev", "int", "sandbox"]
-    if service_env in non_prod:
-        subdomain = f"{service_env}."
-    elif service_env == "prod":
-        subdomain = ""
-    else:
-        subdomain = "internal-dev."
-
-    return f"https://{subdomain}api.service.nhs.uk/{service_base_path}"
-
-
 class FhirService:
     _DATA_MISSING_DATE_TIME_ERROR_MSG = (
         "Data quality issue - immunisation with ID %s was found containing no occurrenceDateTime"
@@ -98,7 +78,7 @@ def get_immunization_by_identifier(
         """
         Get an Immunization by its ID. Returns a FHIR Bundle containing the search results.
         """
-        base_url = f"{get_service_url()}/Immunization"
+        base_url = f"{get_service_url(IMMUNIZATION_ENV, IMMUNIZATION_BASE_PATH)}/Immunization"
         resource, resource_metadata = self.immunization_repo.get_immunization_by_identifier(identifier)
 
         if not resource:
@@ -251,7 +231,7 @@ def search_immunizations(
             BundleEntry(
                 resource=Immunization.parse_obj(imms),
                 search=BundleEntrySearch(mode="match"),
-                fullUrl=f"{get_service_url()}/Immunization/{imms['id']}",
+                fullUrl=f"{get_service_url(IMMUNIZATION_ENV, IMMUNIZATION_BASE_PATH)}/Immunization/{imms['id']}",
             )
             for imms in processed_resources
         ]
@@ -288,7 +268,15 @@ def search_immunizations(
             link=[
                 BundleLink(
                     relation="self",
-                    url=self.create_url_for_bundle_link(permitted_vacc_types, nhs_number, date_from, date_to, include),
+                    url=create_url_for_bundle_link(
+                        permitted_vacc_types,
+                        nhs_number,
+                        date_from,
+                        date_to,
+                        include,
+                        IMMUNIZATION_ENV,
+                        IMMUNIZATION_BASE_PATH,
+                    ),
                 )
             ],
             total=len(processed_resources),
@@ -408,29 +396,3 @@ def process_patient_for_bundle(patient: dict):
             new_patient["id"] = new_patient["identifier"][0].get("value")
 
         return new_patient
-
-    @staticmethod
-    def create_url_for_bundle_link(
-        immunization_targets: set[str],
-        patient_nhs_number: str,
-        date_from: Optional[datetime.date],
-        date_to: Optional[datetime.date],
-        include: Optional[str],
-    ) -> str:
-        """Creates url for the searchset Bundle Link."""
-        params = {
-            # Temporarily maintaining this for backwards compatibility with imms history, but we should remove it
-            IMMUNIZATION_TARGET_LEGACY_KEY_NAME: ",".join(immunization_targets),
-            ImmunizationSearchParameterName.IMMUNIZATION_TARGET: ",".join(immunization_targets),
-            ImmunizationSearchParameterName.PATIENT_IDENTIFIER: f"{PATIENT_IDENTIFIER_SYSTEM}|{patient_nhs_number}",
-        }
-
-        if date_from:
-            params[ImmunizationSearchParameterName.DATE_FROM] = date_from.isoformat()
-        if date_to:
-            params[ImmunizationSearchParameterName.DATE_TO] = date_to.isoformat()
-        if include:
-            params[ImmunizationSearchParameterName.INCLUDE] = include
-
-        query = urllib.parse.urlencode(params)
-        return f"{get_service_url()}/Immunization?{query}"

lambdas/backend/src/service/search_url_helper.py

@@ -0,0 +1,60 @@
+"""Module containing helper functions for the constructions of Immunisation FHIR API search URLs"""
+
+import datetime
+import urllib.parse
+from typing import Optional
+
+from controller.constants import IMMUNIZATION_TARGET_LEGACY_KEY_NAME, ImmunizationSearchParameterName
+from controller.parameter_parser import PATIENT_IDENTIFIER_SYSTEM
+from service.constants import DEFAULT_BASE_PATH, PR_ENV_PREFIX
+
+
+def get_service_url(service_env: Optional[str], service_base_path: Optional[str]) -> str:
+    """Sets the service URL based on service parameters derived from env vars. PR environments use internal-dev while
+    we also default to this environment. The only other exceptions are preprod which maps to the Apigee int environment
+    and prod which does not have a subdomain."""
+    if not service_base_path:
+        service_base_path = DEFAULT_BASE_PATH
+
+    if service_env is None or is_pr_env(service_env):
+        subdomain = "internal-dev."
+    elif service_env == "preprod":
+        subdomain = "int."
+    elif service_env == "prod":
+        subdomain = ""
+    else:
+        subdomain = f"{service_env}."
+
+    return f"https://{subdomain}api.service.nhs.uk/{service_base_path}"
+
+
+def is_pr_env(service_env: Optional[str]) -> bool:
+    return service_env is not None and service_env.startswith(PR_ENV_PREFIX)
+
+
+def create_url_for_bundle_link(
+    immunization_targets: set[str],
+    patient_nhs_number: str,
+    date_from: Optional[datetime.date],
+    date_to: Optional[datetime.date],
+    include: Optional[str],
+    service_env: Optional[str],
+    service_base_path: Optional[str],
+) -> str:
+    """Creates url for the searchset Bundle Link."""
+    params = {
+        # Temporarily maintaining this for backwards compatibility with imms history, but we should remove it
+        IMMUNIZATION_TARGET_LEGACY_KEY_NAME: ",".join(immunization_targets),
+        ImmunizationSearchParameterName.IMMUNIZATION_TARGET: ",".join(immunization_targets),
+        ImmunizationSearchParameterName.PATIENT_IDENTIFIER: f"{PATIENT_IDENTIFIER_SYSTEM}|{patient_nhs_number}",
+    }
+
+    if date_from:
+        params[ImmunizationSearchParameterName.DATE_FROM] = date_from.isoformat()
+    if date_to:
+        params[ImmunizationSearchParameterName.DATE_TO] = date_to.isoformat()
+    if include:
+        params[ImmunizationSearchParameterName.INCLUDE] = include
+
+    query = urllib.parse.urlencode(params)
+    return f"{get_service_url(service_env, service_base_path)}/Immunization?{query}"

lambdas/backend/tests/service/test_fhir_service.py

@@ -22,7 +22,7 @@
 from common.models.immunization_record_metadata import ImmunizationRecordMetadata
 from models.errors import UnauthorizedVaxError
 from repository.fhir_repository import ImmunizationRepository
-from service.fhir_service import FhirService, get_service_url
+from service.fhir_service import FhirService
 from test_common.testing_utils.generic_utils import load_json_data
 from test_common.testing_utils.immunization_utils import (
     VALID_NHS_NUMBER,
@@ -46,43 +46,14 @@ def setUp(self):
         self.mock_redis_getter = self.redis_getter_patcher.start()
         self.logger_info_patcher = patch("logging.Logger.info")
         self.mock_logger_info = self.logger_info_patcher.start()
+        self.imms_env_patcher = patch("service.fhir_service.IMMUNIZATION_ENV", "internal-dev")
+        self.imms_env_patcher.start()
 
     def tearDown(self):
         super().tearDown()
         patch.stopall()
 
 
-class TestServiceUrl(unittest.TestCase):
-    def setUp(self):
-        self.logger_info_patcher = patch("logging.Logger.info")
-        self.mock_logger_info = self.logger_info_patcher.start()
-
-    def tearDown(self):
-        patch.stopall()
-
-    def test_get_service_url(self):
-        """it should create service url"""
-        env = "int"
-        base_path = "immunisation-fhir-api/FHIR/R4"
-        url = get_service_url(env, base_path)
-        self.assertEqual(url, f"https://{env}.api.service.nhs.uk/{base_path}")
-        # default should be internal-dev
-        env = "it-does-not-exist"
-        base_path = "immunisation-fhir-api/FHIR/R4"
-        url = get_service_url(env, base_path)
-        self.assertEqual(url, f"https://internal-dev.api.service.nhs.uk/{base_path}")
-        # prod should not have a subdomain
-        env = "prod"
-        base_path = "immunisation-fhir-api/FHIR/R4"
-        url = get_service_url(env, base_path)
-        self.assertEqual(url, f"https://api.service.nhs.uk/{base_path}")
-        # any other env should fall back to internal-dev (like pr-xx or per-user)
-        env = "pr-42"
-        base_path = "immunisation-fhir-api/FHIR/R4"
-        url = get_service_url(env, base_path)
-        self.assertEqual(url, f"https://internal-dev.api.service.nhs.uk/{base_path}")
-
-
 class TestGetImmunization(TestFhirServiceBase):
     """Tests for FhirService.get_immunization_by_id"""
 

lambdas/backend/tests/service/test_search_url_helper.py

@@ -0,0 +1,29 @@
+"""Tests for the search_url_helper file"""
+
+import unittest
+
+from service.search_url_helper import get_service_url
+
+
+class TestServiceUrl(unittest.TestCase):
+    def test_get_service_url(self):
+        """it should create service url"""
+        test_cases = [
+            ("pr-123", "https://internal-dev.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"),
+            (None, "https://internal-dev.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"),
+            ("preprod", "https://int.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"),
+            ("prod", "https://api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"),
+            ("ref", "https://ref.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"),
+            ("internal-dev", "https://internal-dev.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"),
+            ("internal-qa", "https://internal-qa.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"),
+        ]
+        mock_base_path = "immunisation-fhir-api/FHIR/R4"
+
+        for mock_env, expected in test_cases:
+            with self.subTest(mock_env=mock_env, expected=expected):
+                self.assertEqual(get_service_url(mock_env, mock_base_path), expected)
+
+    def test_get_service_url_uses_default_path_when_not_provided(self):
+        self.assertEqual(
+            get_service_url(None, None), "https://internal-dev.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4"
+        )