[ERSSUP-78354]-[]-[Initial inclusion of apim nhsd_apim_auth_headers p…

azure/templates/ers-integration-tests.yml

@@ -8,7 +8,7 @@ steps:
       export OAUTH_BASE_URI="https://$(ENVIRONMENT).api.service.nhs.uk"
       export JWT_PRIVATE_KEY_ABSOLUTE_PATH="$(Pipeline.Workspace)/secrets/$(JWT_TESTING_PRIVATE_KEY)"
 
-      poetry run pytest -v tests/integration --junitxml=tests/ers-test-integration-report.xml
+      poetry run pytest -v tests/integration --junitxml=tests/ers-test-integration-report.xml --api-name=e-referrals-service-api --proxy-name=$(FULLY_QUALIFIED_SERVICE_NAME)
     displayName: Run eRS integration tests
     workingDirectory: "$(Pipeline.Workspace)/s/$(SERVICE_NAME)/$(SERVICE_ARTIFACT_NAME)"
   - task: PublishTestResults@2

manifest_template.yml

@@ -79,7 +79,7 @@ ACCESS_MODES:
   - name: healthcare-worker
     nameSuffix: -healthcare-worker
     displayName: Healthcare Worker
-    scopes: ['urn:nhsd:apim:user-nhs-id:aal3:e-referrals-service-api']
+    scopes: ['urn:nhsd:apim:user-nhs-id:aal3:e-referrals-service-api', 'urn:nhsd:apim:user-nhs-cis2:aal3:e-referrals-service-api']
     requireCallbackUrl: true
     description: User restricted
 

proxies/live/apiproxy/policies/OAuthV2.VerifyAccessToken.xml

@@ -1,4 +1,4 @@
 <OAuthV2 async="false" continueOnError="false" enabled="true" name="OauthV2.VerifyAccessToken">
-  <Operation>VerifyAccessToken</Operation>
-  <Scope>urn:nhsd:apim:app:level3:e-referrals-service-api urn:nhsd:apim:user-nhs-id:aal3:e-referrals-service-api</Scope>
+    <Operation>VerifyAccessToken</Operation>
+    <Scope>urn:nhsd:apim:app:level3:e-referrals-service-api urn:nhsd:apim:user-nhs-id:aal3:e-referrals-service-api urn:nhsd:apim:user-nhs-cis2:aal3:e-referrals-service-api</Scope>
 </OAuthV2>

pyproject.toml

@@ -45,4 +45,4 @@ pytest-nhsd-apim = "^3.4.2"
 [tool.poetry.scripts]
 
 [tool.pytest.ini_options]
-markers = ["smoke_test", "integration_test", "sandbox"]
+markers = ["smoke_test", "integration_test", "sandbox", "authentication_type"]

tests/asserts.py

@@ -2,7 +2,7 @@
 from typing import Dict, Iterable
 from pytest_check import check
 from requests import Response
-from data import RenamedHeader
+from .data import RenamedHeader
 
 # Headers which are expected for all requests
 _generic_headers = {

tests/conftest.py

@@ -1,8 +1,9 @@
-import os
 import pytest
 import pytest_asyncio
+import warnings
 
 from uuid import uuid4
+from .utils import get_env
 
 from pytest_nhsd_apim.identity_service import (
     AuthorizationCodeConfig,
@@ -18,18 +19,12 @@
 )
 
 
-from data import Actor
+from .data import Actor
 
 
-def get_env(variable_name: str) -> str:
-    """Returns an environment variable"""
-    try:
-        var = os.environ[variable_name]
-        if not var:
-            raise RuntimeError(f"Variable is null, Check {variable_name}.")
-        return var
-    except KeyError:
-        raise RuntimeError(f"Variable is not set, Check {variable_name}.")
+def _create_apigee_client():
+    config = ApigeeNonProdCredentials()
+    return ApigeeClient(config=config)
 
 
 @pytest.fixture(scope="session")
@@ -102,7 +97,6 @@ def app_restricted_user_id(is_mocked_environment):
 
 
 @pytest.fixture(
-    scope="session",
     params=["PROVIDER_AUTHORISED_APPLICATION", "REFERRER_AUTHORISED_APPLICATION"],
 )
 def app_restricted_business_function(request):
@@ -111,8 +105,7 @@ def app_restricted_business_function(request):
 
 @pytest.fixture()
 def client():
-    config = ApigeeNonProdCredentials()
-    return ApigeeClient(config=config)
+    return _create_apigee_client()
 
 
 @pytest_asyncio.fixture
@@ -304,3 +297,82 @@ def app_restricted_access_code(
     token_response = authenticator.get_token()
     assert "access_token" in token_response
     return token_response["access_token"]
+
+
+@pytest.fixture
+def _pre_authentication(
+    _create_test_app,
+    request,
+    asid,
+    app_restricted_ods_code,
+    app_restricted_user_id,
+    app_restricted_business_function,
+):
+    """
+    Fixture adding custom attributes to the application created by the pytest_nhsd_apim module's fixtures. This is required as custom attributes are not publically exposed by the module itself.
+    """
+
+    warnings.warn("Pre authentication fixture:")
+
+    created_app = _create_test_app
+    if not created_app:
+        raise ValueError("No app has been initialised.")
+
+    warnings.warn(f"created app={created_app}")
+
+    api = DeveloperAppsAPI(client=_create_apigee_client())
+
+    marker = request.node.get_closest_marker("authentication_type")
+    if not marker:
+        raise ValueError(
+            "No pytest.mark.authentication_type included with request. Have you used the user_restricted_access or app_restricted_access decorators?"
+        )
+
+    # Update the attributes of the created application to add in the ASID attribute.
+    additional_attributes = [{"name": "asid", "value": asid}]
+
+    if marker.args and marker.args[0]:
+        type = marker.args[0]
+    elif marker.kwargs and marker.kwargs["type"]:
+        type = marker.kwargs["type"]
+    else:
+        raise ValueError("No type provided with pytest.mark.authentication_type marker")
+
+    if type == "app-restricted":
+        additional_attributes = additional_attributes + [
+            {"name": "app-restricted-ods-code", "value": app_restricted_ods_code},
+            {"name": "app-restricted-user-id", "value": app_restricted_user_id},
+            {
+                "name": "app-restricted-business-function",
+                "value": app_restricted_business_function,
+            },
+        ]
+
+    original_attributes = created_app["attributes"]
+    modified_attributes = original_attributes + additional_attributes
+    created_app["attributes"] = modified_attributes
+
+    warnings.warn(f"updated app={created_app}")
+
+    yield api.put_app_by_name(
+        email="apm-testing-internal-dev@nhs.net",
+        app_name=created_app["name"],
+        body=created_app,
+    )
+
+    # Reset the app back to its original attributes so it can be updated by a subsequent test
+    created_app["attributes"] = original_attributes
+    api.put_app_by_name(
+        email="apm-testing-internal-dev@nhs.net",
+        app_name=created_app["name"],
+        body=created_app,
+    )
+
+
+@pytest.fixture
+def _nhsd_apim_auth_token_data(_pre_authentication, _nhsd_apim_auth_token_data):
+    """
+    Override of the pytest_nhsd_apim _nhsd_apim_auth_token_data fixture to invoke _pre_authenication before being executed.
+    """
+
+    return _nhsd_apim_auth_token_data

tests/decorators.py

@@ -0,0 +1,86 @@
+import pytest
+
+from functools import wraps
+from typing import Callable
+from asyncio import iscoroutinefunction
+
+from .data import Actor
+from .utils import get_env
+
+
+def _calculate_default_user() -> Actor:
+    return (
+        Actor.RC_DEV
+        if get_env("ENVIRONMENT") == "internal-dev"
+        and "ft" in get_env("SERVICE_BASE_PATH")
+        else Actor.RC
+    )
+
+
+_DEFAULT_USER: Actor = _calculate_default_user()
+
+
+def user_restricated_access(function: Callable = None, user: Actor = _DEFAULT_USER):
+    """
+    Decorator indicating that a given function should be authenticated with User Restricted access with a supplied user.
+    This will lead to a fixture named 'nhsd_apim_auth_headers' being provided to the function as a dictionary, including the headers required to authenticate as the supplied user.
+
+    :param user: An Actor indicating the user to authenticate as. If no user is provided _DEFAULT_USER will be used instead.
+    """
+
+    def decorator(func):
+        auth_args = {
+            "access": "healthcare_worker",
+            "level": "aal3",
+            "login_form": {"username": user.user_id},
+            # Force a token to always be created to ensure no app details are cached
+            "force_new_token": True,
+        }
+
+        @pytest.mark.authentication_type("user-restricted")
+        @pytest.mark.nhsd_apim_authorization(auth_args)
+        @wraps(func)
+        async def async_wrapper(*args, **kwargs):
+            return await func(*args, **kwargs)
+
+        @pytest.mark.authentication_type("user-restricted")
+        @pytest.mark.nhsd_apim_authorization(auth_args)
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            return func(*args, **kwargs)
+
+        # if the decorated function is async, return an async function, else return a synchronous version.
+        return async_wrapper if iscoroutinefunction(func) else wrapper
+
+    # If a function is provided the decorator is being called without any parameters and we need to call our decorator supplying this as the function.
+    # Otherwise the decorator is being called with arguments and can be returned directly.
+    return decorator(function) if function else decorator
+
+
+def app_restricted_access(func):
+    """
+    Decorator indicating that the given function should be authenticated with Application Restricted access with a list of set types.
+    This will lead to a fixture named 'nhsd_apim_auth_headers' being provided to the function as a dictionary, including the headers required to authenticate as the default application.
+    """
+
+    auth_args = {
+        "access": "application",
+        "level": "level3",
+        # Force a token to always be created to ensure no app details are cached
+        "force_new_token": True,
+    }
+
+    @pytest.mark.authentication_type("app-restricted")
+    @pytest.mark.nhsd_apim_authorization(auth_args)
+    @wraps(func)
+    async def async_wrapper(*args, **kwargs):
+        return await func(*args, **kwargs)
+
+    @pytest.mark.authentication_type("app-restricted")
+    @pytest.mark.nhsd_apim_authorization(auth_args)
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        return func(*args, **kwargs)
+
+    # if the decorated function is async, return an async function as a decorator, otherwise use a synchronous decorator.
+    return async_wrapper if iscoroutinefunction(func) else wrapper

tests/integration/test_app_restricted.py

@@ -1,8 +1,9 @@
 import pytest
 import requests
 from requests import Response
-from data import RenamedHeader
-from asserts import assert_ok_response, assert_error_response
+from tests.data import RenamedHeader
+from tests.asserts import assert_ok_response, assert_error_response
+from tests.decorators import app_restricted_access
 
 _HEADER_AUTHORIZATION = "Authorization"
 _HEADER_ECHO = "echo"  # enable echo target
@@ -86,21 +87,24 @@ def test_authorised_application_header_rejected(
 
         assert_error_response(response, _EXPECTED_CORRELATION_ID, 403)
 
+    @app_restricted_access
     def test_headers_on_echo_target(
         self,
-        app_restricted_access_code,
+        nhsd_apim_auth_headers,
         service_url,
         asid,
         app_restricted_ods_code,
         app_restricted_user_id,
         app_restricted_business_function,
     ):
-        client_request_headers = {
-            _HEADER_ECHO: "",  # enable echo target
-            _HEADER_AUTHORIZATION: "Bearer " + app_restricted_access_code,
-            RenamedHeader.CORRELATION_ID.original: _EXPECTED_CORRELATION_ID,
-            _HEADER_REQUEST_ID: "DUMMY",  # this must be less than 10 characters
-        }
+        client_request_headers = dict(
+            nhsd_apim_auth_headers,
+            **{
+                _HEADER_ECHO: "",  # enable echo target
+                RenamedHeader.CORRELATION_ID.original: _EXPECTED_CORRELATION_ID,
+                _HEADER_REQUEST_ID: "DUMMY",  # this must be less than 10 characters
+            },
+        )
 
         # Make the API call
         response = requests.get(service_url, headers=client_request_headers)

tests/integration/test_headers.py

@@ -1,8 +1,9 @@
 import pytest
 import requests
 from requests import Response
-from data import RenamedHeader
-from asserts import assert_ok_response
+from tests.data import RenamedHeader
+from tests.asserts import assert_ok_response
+from tests.decorators import user_restricated_access
 
 _HEADER_AUTHORIZATION = "Authorization"
 _HEADER_ECHO = "echo"  # enable echo target
@@ -31,24 +32,27 @@
 
 @pytest.mark.integration_test
 class TestHeaders:
+
     @pytest.mark.asyncio
+    @user_restricated_access
     async def test_headers_on_echo_target(
-        self, authenticate_user, service_url, referring_clinician, asid
+        self, nhsd_apim_auth_headers, service_url, referring_clinician, asid
     ):
-        access_code = await authenticate_user(referring_clinician)
 
-        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: referring_clinician.business_function,
-            RenamedHeader.ODS_CODE.original: referring_clinician.org_code,
-            RenamedHeader.FILENAME.original: _EXPECTED_FILENAME,
-            RenamedHeader.COMM_RULE_ORG.original: _EXPECTED_COMM_RULE_ORG,
-            RenamedHeader.OBO_USER_ID.original: _EXPECTED_OBO_USER_ID,
-        }
+        client_request_headers = dict(
+            nhsd_apim_auth_headers,
+            **{
+                _HEADER_ECHO: "",  # enable echo target
+                _HEADER_REQUEST_ID: "DUMMY-VALUE",
+                RenamedHeader.REFERRAL_ID.original: _EXPECTED_REFERRAL_ID,
+                RenamedHeader.CORRELATION_ID.original: _EXPECTED_CORRELATION_ID,
+                RenamedHeader.BUSINESS_FUNCTION.original: referring_clinician.business_function,
+                RenamedHeader.ODS_CODE.original: referring_clinician.org_code,
+                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(service_url, headers=client_request_headers)

tests/sandbox/SandboxTest.py

@@ -5,10 +5,10 @@
 from pytest_check import check
 
 import pytest
-import asserts
+from tests import asserts
 
-from data import Actor, RenamedHeader
-from utils import HttpMethod
+from tests.data import Actor, RenamedHeader
+from .utils import HttpMethod
 
 
 @pytest.mark.sandbox

tests/sandbox/conftest.py

@@ -1,13 +1,11 @@
-import os
-
 import pytest
 import json
 import requests
 
 from typing import Dict, Callable
 
-from data import Actor, RenamedHeader
-from utils import HttpMethod
+from tests.data import Actor, RenamedHeader
+from .utils import HttpMethod
 
 
 @pytest.fixture(scope="session")