Added test using new test plugin

tests/api_tests.py

@@ -147,46 +147,32 @@ async def test_immunization_no_auth_bearer_token_provided(
 
 @pytest.mark.e2e
 @pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "test_app",
-    _add_authorised_targets_to_request_params(
-        [
-            {
-                "suffixes": ["-user-restricted"],
-                "requested_proofing_level": "P9",
-                "identity_proofing_level": "P9",
-            }
-        ]
-    ),
-    indirect=True,
+@pytest.mark.nhsd_apim_authorization(
+    access="patient",
+    level="P9",
+    login_form={"username": "9912003071"}
 )
-async def test_bad_nhs_number(test_app, api_client: APISessionClient):
+async def test_bad_nhs_number(nhsd_apim_proxy_url, _nhsd_apim_auth_token_data):
     await asyncio.sleep(1)  # Add delay to tests to avoid 429 on service callout
 
-    subject_token_claims = {
-        "identity_proofing_level": test_app.request_params["identity_proofing_level"]
-    }
-    token_response = await conftest.get_token_nhs_login_token_exchange(
-        test_app, subject_token_claims=subject_token_claims
-    )
-    token = token_response["access_token"]
+    token = _nhsd_apim_auth_token_data["access_token"]
 
     correlation_id = _generate_correlation_id('test_bad_nhs_number')
     headers = {"Authorization": f"Bearer {token}", "X-Correlation-ID": correlation_id}
 
-    async with api_client.get(
-        _valid_uri_procedure_below("90000000009", "90640007"), headers=headers, allow_retries=True
-    ) as resp:
-        assert resp.status == 400
-        body = await resp.json()
-        assert body["resourceType"] == "OperationOutcome", body
-        issue = next(
-            (i for i in body.get("issue", []) if i.get("severity") == "error"), None
-        )
-        assert (
-            issue.get("diagnostics")
-            == "Missing required request parameters: [patient.identifier]"
-        ), body
+    resp = requests.get(
+        f'{nhsd_apim_proxy_url}/{_valid_uri_procedure_below("90000000009", "90640007")}', headers=headers
+    )
+    assert resp.status_code == 400
+    body = resp.json()
+    assert body["resourceType"] == "OperationOutcome", body
+    issue = next(
+        (i for i in body.get("issue", []) if i.get("severity") == "error"), None
+    )
+    assert (
+        issue.get("diagnostics")
+        == "Missing required request parameters: [patient.identifier]"
+    ), body
 
 
 @pytest.mark.e2e
@@ -270,38 +256,17 @@ async def test_token_exchange_happy_path(test_app, api_client: APISessionClient)
         assert len(body["entry"]) == 3, body
 
 
-@pytest.mark.e2e
-@pytest.mark.asyncio
+@pytest.mark.nhsd_apim_authorization(
+    access="application",
+    level="level3",
+)
 @pytest.mark.parametrize(
-    "test_app",
-    _add_authorised_targets_to_request_params(
-        [
-            {
-                "suffixes": ["-application-restricted"],
-                "requested_proofing_level": "P9",
-                "identity_proofing_level": "P9",
-            },
-            {
-                "suffixes": ["-application-restricted"],
-                "requested_proofing_level": "P5",
-                "identity_proofing_level": "P9",
-            },
-            {
-                "suffixes": ["-application-restricted"],
-                "requested_proofing_level": "P5",
-                "identity_proofing_level": "P5",
-            },
-        ]
-    ),
-    indirect=True,
+    "nhs_login_id", ["9912003072", "9912003071"]
 )
-async def test_token_exchange_sad_path(test_app, api_client: APISessionClient):
-    subject_token_claims = {
-        "identity_proofing_level": test_app.request_params["identity_proofing_level"]
-    }
-    await conftest.check_for_unauthorised_token_exchange(
-        test_app, subject_token_claims=subject_token_claims
-    )
+def test_token_exchange_sad_path(_test_app_credentials, environment, _jwt_keys, _keycloak_client_credentials,
+                                       nhs_login_id):
+    conftest.check_for_unauthorised_token_exchange(
+        _test_app_credentials, environment, _jwt_keys, _keycloak_client_credentials, nhs_login_id)
 
 
 @pytest.mark.e2e

tests/conftest.py

@@ -13,6 +13,13 @@
 from api_test_utils.fixtures import api_client  # pylint: disable=unused-import
 from api_test_utils.apigee_api_products import ApigeeApiProducts
 
+from pytest_nhsd_apim.identity_service import (
+    KeycloakUserConfig,
+    KeycloakUserAuthenticator,
+    TokenExchangeConfig,
+    TokenExchangeAuthenticator
+)
+
 
 def get_env(variable_name: str) -> str:
     """Returns a environment variable"""
@@ -198,44 +205,38 @@ async def get_token_nhs_login_token_exchange(
     return token_resp["body"]
 
 
-async def check_for_unauthorised_token_exchange(
-    test_app: ApigeeApiDeveloperApps,
-    subject_token_claims: dict = None,
-    client_assertion_jwt: dict = None,
+def check_for_unauthorised_token_exchange(
+    _test_app_credentials, environment, _jwt_keys, _keycloak_client_credentials, nhs_login_id: str
 ):
-    """Call identity server to get an access token"""
-    if client_assertion_jwt is not None:
-        client_assertion_jwt = test_app.oauth.create_jwt(
-            kid="test-1", claims=client_assertion_jwt
-        )
-    else:
-        client_assertion_jwt = test_app.oauth.create_jwt(kid="test-1")
+    keycloak_user_config = KeycloakUserConfig(
+        realm=f"NHS-Login-mock-{environment}",
+        client_id=_keycloak_client_credentials["nhs-login"]["client_id"],
+        client_secret=_keycloak_client_credentials["nhs-login"]["client_secret"],
+        login_form={"username": nhs_login_id},
+    )
 
-    if subject_token_claims is not None:
-        id_token_jwt = nhs_login_id_token(
-            test_app=test_app, id_token_claims=subject_token_claims
-        )
-    else:
-        id_token_jwt = nhs_login_id_token(test_app=test_app)
+    authenticator = KeycloakUserAuthenticator(config=keycloak_user_config)
 
-    # When
-    token_resp = await test_app.oauth.get_token_response(
-        grant_type="token_exchange",
-        data={
-            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
-            "subject_token_type": "urn:ietf:params:oauth:token-type:id_token",
-            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
-            "subject_token": id_token_jwt,
-            "client_assertion": client_assertion_jwt,
-        },
-    )
-    assert token_resp["status_code"] == 401
-    assert token_resp["body"]["error"] == "unauthorized_client"
-    assert (
-        token_resp["body"]["error_description"]
-        == "you have tried to request authorization but your application is not configured to use this authorization grant type"
+    id_token = authenticator.get_token()["id_token"]
+
+    token_exchange_config = TokenExchangeConfig(
+        environment=environment,
+        identity_service_base_url=f"https://{environment}.api.service.nhs.uk/oauth2-mock",
+        client_id=_test_app_credentials["consumerKey"],
+        jwt_private_key=_jwt_keys["private_key_pem"],
+        jwt_kid="test-1",
+        id_token=id_token,
     )
 
+    authenticator = TokenExchangeAuthenticator(config=token_exchange_config)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        authenticator.get_token()
+
+    print(str(exc_info.value))
+    # token_resp["body"]["error_description"]
+    # == "you have tried to request authorization but your application is not configured to use this authorization grant type"
+
 
 @pytest.fixture(scope="session")
 def api_test_config() -> APITestSessionConfig:
@@ -259,7 +260,7 @@ def service_url(environment):
     return f"{base_url}/{service_base_path}"
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture()
 def test_app(request):
     """Setup & Teardown an app-restricted app for this api"""
     request_params = request.param