tests passing, added 400 error case

Author: kevinmason-nhs

tests/conftest.py

@@ -83,8 +83,7 @@ def asid(is_mocked_environment):
 @pytest.fixture(scope="session")
 def apim_app_flow_vars(allowListodsCode=None):
     if allowListodsCode is not None:
-        print(f"Using allowListodsCode: {allowListodsCode}")
-        return {"ers": {"allowListodsCode": str(allowListodsCode)}}
+        return {"ers": {"allowListodsCode": allowListodsCode}}
 
 
 @pytest.fixture(scope="session")
@@ -264,9 +263,7 @@ async def _make_product(product_scopes, additional_attributes=None):
         ]
 
         if additional_attributes is not None:
-            print(f"Adding additional attributes: {additional_attributes}")
             attributes.extend(additional_attributes)
-            print(f"Attributes now: {attributes}")
 
         body = {
             "proxies": proxies,
@@ -292,8 +289,7 @@ async def user_restricted_app(
 ):
     # Setup
     if apim_app_flow_vars is not None:
-        odslist = json.dumps({"ers": {"allowListodsCode": str(apim_app_flow_vars)}})
-        print(f"Using apim_app_flow_vars: {odslist}")
+        odslist = json.dumps({"ers": {"allowListodsCode": apim_app_flow_vars}})
         app = await make_app(
             user_restricted_product,
             {"asid": asid, "apim-app-flow-vars": odslist},
@@ -318,8 +314,6 @@ async def _make_app(product, custom_attributes={}):
         devAppAPI = DeveloperAppsAPI(client=client)
         app_name = f"apim-auto-{uuid4()}"
 
-        print(f"Custom attributes: {custom_attributes}")
-
         attributes = [
             {"name": key, "value": value} for key, value in custom_attributes.items()
         ]

tests/integration/test_user_restricted.py

@@ -42,12 +42,12 @@ class TestUserRestricted:
     async def test_user_restricted_valid_ods_code(
         self,
         authenticate_user,
+        service_url,
         user: Actor,
+        asid,
         endpoint_url,
         is_fhir_4,
-        service_url,
         apim_app_flow_vars,
-        update_user_restricted_product,
     ):
         access_code = await authenticate_user(user)
 
@@ -65,14 +65,11 @@ async def test_user_restricted_valid_ods_code(
         }
 
         # Make the API call
-
-        # Make request with user with ODS code not in allow list (e.g. R69)
         response = requests.get(
             f"{service_url}{endpoint_url}", headers=client_request_headers
         )
 
         # Verify the status
-        # Verify 403 response with appropriate error message
         assert (
             response.status_code == 200
         ), "Expected a 200 when accessing the api but got " + str(response.status_code)
@@ -89,12 +86,12 @@ async def test_user_restricted_valid_ods_code(
     async def test_user_restricted_invalid_ods_code(
         self,
         authenticate_user,
-        endpoint_url,
+        service_url,
         user: Actor,
+        asid,
+        endpoint_url,
         is_fhir_4,
-        service_url,
         apim_app_flow_vars,
-        update_user_restricted_product,
     ):
         access_code = await authenticate_user(user)
 
@@ -112,14 +109,179 @@ async def test_user_restricted_invalid_ods_code(
         }
 
         # Make the API call
-
-        # Make request with user with ODS code not in allow list (e.g. R69)
         response = requests.get(
             f"{service_url}{endpoint_url}", headers=client_request_headers
         )
-
         # Verify the status
-        # Verify 403 response with appropriate error message
         assert (
             response.status_code == 403
         ), "Expected a 403 when accessing the api but got " + str(response.status_code)
+        # Verify the OperationOutcome payload
+        response_data = response.json()
+        assert response_data["resourceType"] == "OperationOutcome"
+        assert response_data["meta"]["lastUpdated"] is not None
+        assert len(response_data["meta"]["profile"]) == 1
+        assert response_data["meta"]["profile"][0] == (
+            "https://www.hl7.org/fhir/R4/operationoutcome.html"
+            if is_fhir_4
+            else "https://fhir.nhs.uk/STU3/StructureDefinition/eRS-OperationOutcome-1"
+        )
+        assert len(response_data["issue"]) == 1
+        issue = response_data["issue"][0]
+        assert issue["severity"] == "error"
+        assert issue["code"] == "security" if is_fhir_4 else "forbidden"
+        assert issue["diagnostics"] == (
+            "Unauthorised ODS code provided in NHSD-End-User-Organisation-ODS header"
+        )
+        assert len(issue["details"]["coding"]) == 1
+        issue_details = issue["details"]["coding"][0]
+        assert (
+            issue_details["system"]
+            == "https://fhir.nhs.uk/CodeSystem/NHSD-API-ErrorOrWarningCode"
+            if is_fhir_4
+            else "https://fhir.nhs.uk/STU3/CodeSystem/eRS-APIErrorCode-1"
+        )
+        assert (
+            issue_details["code"] == "ACCESS_DENIED" if is_fhir_4 else "ACCESS_DENIED"
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "endpoint_url, is_fhir_4, user ,apim_app_flow_vars",
+        [
+            ("", False, Actor.RC_DEV, [Actor.RC_DEV.org_code]),
+            ("/FHIR/R4/", True, Actor.RC_DEV, [Actor.RC_DEV.org_code]),
+            ("/FHIR/STU3/", False, Actor.RC_DEV, [Actor.RC_DEV.org_code]),
+        ],
+    )
+    async def test_user_restricted_missing_ods_header(
+        self,
+        authenticate_user,
+        service_url,
+        user: Actor,
+        asid,
+        endpoint_url,
+        is_fhir_4,
+        apim_app_flow_vars,
+    ):
+        access_code = await authenticate_user(user)
+
+        client_request_headers = {
+            _HEADER_ECHO: "",  # enable echo target
+            _HEADER_AUTHORIZATION: "Bearer " + access_code,
+            _HEADER_REQUEST_ID: "DUMMY-VALUE",
+            RenamedHeader.REFERRAL_ID.original: _EXPECTED_REFERRAL_ID,
+            RenamedHeader.CORRELATION_ID.original: _EXPECTED_CORRELATION_ID,
+            RenamedHeader.BUSINESS_FUNCTION.original: user.business_function,
+            RenamedHeader.FILENAME.original: _EXPECTED_FILENAME,
+            RenamedHeader.COMM_RULE_ORG.original: _EXPECTED_COMM_RULE_ORG,
+            RenamedHeader.OBO_USER_ID.original: _EXPECTED_OBO_USER_ID,
+        }
+
+        # Make the API call
+        response = requests.get(
+            f"{service_url}{endpoint_url}", headers=client_request_headers
+        )
+        # Verify the status
+        assert (
+            response.status_code == 400
+        ), "Expected a 400 when accessing the api but got " + str(response.status_code)
+        # Verify the OperationOutcome payload
+        response_data = response.json()
+        assert response_data["resourceType"] == "OperationOutcome"
+        assert response_data["meta"]["lastUpdated"] is not None
+        assert len(response_data["meta"]["profile"]) == 1
+        assert response_data["meta"]["profile"][0] == (
+            "https://www.hl7.org/fhir/R4/operationoutcome.html"
+            if is_fhir_4
+            else "https://fhir.nhs.uk/STU3/StructureDefinition/eRS-OperationOutcome-1"
+        )
+        assert len(response_data["issue"]) == 1
+        issue = response_data["issue"][0]
+        assert issue["severity"] == "error"
+        assert issue["code"] == "invalid" if is_fhir_4 else "invalid"
+        assert issue["diagnostics"] == (
+            "Missing or Empty NHSD-End-User-Organisation-ODS header."
+        )
+        assert len(issue["details"]["coding"]) == 1
+        issue_details = issue["details"]["coding"][0]
+        assert (
+            issue_details["system"]
+            == "https://fhir.nhs.uk/CodeSystem/NHSD-API-ErrorOrWarningCode"
+            if is_fhir_4
+            else "https://fhir.nhs.uk/STU3/CodeSystem/eRS-APIErrorCode-1"
+        )
+        assert (
+            issue_details["code"] == "MISSING_HEADER" if is_fhir_4 else "MISSING_HEADER"
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "endpoint_url, is_fhir_4, user ,apim_app_flow_vars",
+        [
+            ("", False, Actor.RC_DEV, [Actor.RC_DEV.org_code]),
+            ("/FHIR/R4/", True, Actor.RC_DEV, [Actor.RC_DEV.org_code]),
+            ("/FHIR/STU3/", False, Actor.RC_DEV, [Actor.RC_DEV.org_code]),
+        ],
+    )
+    async def test_user_restricted_missing_ods_code(
+        self,
+        authenticate_user,
+        service_url,
+        user: Actor,
+        asid,
+        endpoint_url,
+        is_fhir_4,
+        apim_app_flow_vars,
+    ):
+        access_code = await authenticate_user(user)
+
+        client_request_headers = {
+            _HEADER_ECHO: "",  # enable echo target
+            _HEADER_AUTHORIZATION: "Bearer " + access_code,
+            _HEADER_REQUEST_ID: "DUMMY-VALUE",
+            RenamedHeader.REFERRAL_ID.original: _EXPECTED_REFERRAL_ID,
+            RenamedHeader.CORRELATION_ID.original: _EXPECTED_CORRELATION_ID,
+            RenamedHeader.BUSINESS_FUNCTION.original: user.business_function,
+            RenamedHeader.ODS_CODE.original: "",
+            RenamedHeader.FILENAME.original: _EXPECTED_FILENAME,
+            RenamedHeader.COMM_RULE_ORG.original: _EXPECTED_COMM_RULE_ORG,
+            RenamedHeader.OBO_USER_ID.original: _EXPECTED_OBO_USER_ID,
+        }
+
+        # Make the API call
+        response = requests.get(
+            f"{service_url}{endpoint_url}", headers=client_request_headers
+        )
+        # Verify the status
+        assert (
+            response.status_code == 400
+        ), "Expected a 400 when accessing the api but got " + str(response.status_code)
+        # Verify the OperationOutcome payload
+        response_data = response.json()
+        assert response_data["resourceType"] == "OperationOutcome"
+        assert response_data["meta"]["lastUpdated"] is not None
+        assert len(response_data["meta"]["profile"]) == 1
+        assert response_data["meta"]["profile"][0] == (
+            "https://www.hl7.org/fhir/R4/operationoutcome.html"
+            if is_fhir_4
+            else "https://fhir.nhs.uk/STU3/StructureDefinition/eRS-OperationOutcome-1"
+        )
+        assert len(response_data["issue"]) == 1
+        issue = response_data["issue"][0]
+        assert issue["severity"] == "error"
+        assert issue["code"] == "invalid" if is_fhir_4 else "invalid"
+        assert issue["diagnostics"] == (
+            "Missing or Empty NHSD-End-User-Organisation-ODS header."
+        )
+        assert len(issue["details"]["coding"]) == 1
+        issue_details = issue["details"]["coding"][0]
+        assert (
+            issue_details["system"]
+            == "https://fhir.nhs.uk/CodeSystem/NHSD-API-ErrorOrWarningCode"
+            if is_fhir_4
+            else "https://fhir.nhs.uk/STU3/CodeSystem/eRS-APIErrorCode-1"
+        )
+        assert (
+            issue_details["code"] == "MISSING_HEADER" if is_fhir_4 else "MISSING_HEADER"
+        )