VED-1013 Part 3/4 Add support for running new e2e tests in PR environments (#1140)

Author: dlzhry2nhs

tests/e2e_automation/.env.example.pr

@@ -0,0 +1,23 @@
+auth_url=https://internal-dev.api.service.nhs.uk/oauth2-mock/authorize
+token_url=https://internal-dev.api.service.nhs.uk/oauth2-mock/token
+callback_url=https://oauth.pstmn.io/v1/callback
+# Obtain value from dev/testing team
+STATUS_API_KEY=
+
+username=aal3
+scope=nhs-cis2
+
+# Internal-dev PR environment
+baseUrl=https://internal-dev.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4-pr-123
+aws_token_refresh=False
+aws_profile_name=345594581768_DEV-IMMS-Devops
+
+S3_env=pr-123
+# LOCAL_RUN_WITHOUT_S3_UPLOAD = True
+LOCAL_RUN_FILE_NAME=HPV_Vaccinations_v5_V0V8L_20251111T16304982.csv
+AWS_DOMAIN_NAME=https://pr-123.imms.dev.vds.platform.nhs.uk
+
+PROXY_NAME=immunisation-fhir-api-pr-123
+# See README for details on how to obtain this
+APIGEE_ACCESS_TOKEN={use-the-apigee-get-token-utility}
+APIGEE_USERNAME={your-apigee-developer-email}

tests/e2e_automation/README.md

@@ -59,3 +59,22 @@ This directory contains End-to-end Automation Tests for the Immunisation FHIR AP
 - `make test-batch-full` - run Batch tests
 - `make test-batch-smoke` - run Batch smoke tests only (quicker)
 - `make collect-only` - check that all tests are discovered
+
+## Running e2e_automation tests against PR environments
+
+The environment variables define a client ID and client secret for each of the Apigee test apps we use in static
+environments such as `internal-dev`, `internal-qa` and so on.
+
+However, creating pull requests will spin up a dynamic Apigee proxy and AWS backend which lives for the duration of the PR.
+To minimise admin overhead, the automation tests create dynamic applications for the duration of a test run rather than
+us having to manually create new apps each time we produce a pull request.
+
+These tests are run seamlessly in the pipeline. But if you are doing some local changes and want to test against your
+PR environment, please follow these pre-requisites to get it working:
+
+1. [Install](https://docs.apigee.com/api-platform/system-administration/auth-tools#install) and run the Apigee [get_token](https://docs.apigee.com/api-platform/system-administration/using-gettoken) tool to obtain an access token.
+2. Set this value against the `APIGEE_ACCESS_TOKEN` in your .env file.
+3. Finally, use the [.env.example.pr](./.env.example.pr) as your baseline for your .env file and fill all of the required values.
+
+Note: the `get_token` tool is only supported in Linux environments, so if you are using a Windows environment, you will
+at least need to run the operation in WSL to obtain the access token.

tests/e2e_automation/features/conftest.py

@@ -7,6 +7,9 @@
 from utilities.api_fhir_immunization_helper import empty_folder
 from utilities.api_gen_token import get_tokens
 from utilities.api_get_header import get_delete_url_header
+from utilities.apigee.apigee_env_helpers import is_pr_env
+from utilities.apigee.ApigeeApp import ApigeeApp
+from utilities.apigee.ApigeeOnDemandAppManager import ApigeeOnDemandAppManager
 from utilities.aws_token import refresh_sso_token, set_aws_session_token
 from utilities.context import ScenarioContext
 from utilities.enums import SupplierNameWithODSCode
@@ -53,8 +56,25 @@ def global_context():
     ).strip().lower() == "true" else set_aws_session_token()
 
 
+@pytest.fixture(scope="session")
+def temp_apigee_apps():
+    if is_pr_env():
+        apigee_app_mgr = ApigeeOnDemandAppManager()
+        created_apps = apigee_app_mgr.setup_apps_and_product()
+
+        for test_app in created_apps:
+            os.environ[f"{test_app.supplier}_client_Id"] = test_app.client_id
+            os.environ[f"{test_app.supplier}_client_Secret"] = test_app.client_secret
+
+        yield created_apps
+
+        apigee_app_mgr.teardown_apps_and_product()
+    else:
+        yield None
+
+
 @pytest.fixture
-def context(request, global_context) -> ScenarioContext:
+def context(request, global_context, temp_apigee_apps: list[ApigeeApp] | None) -> ScenarioContext:
     ctx = ScenarioContext()
     ctx.aws_profile_name = os.getenv("aws_profile_name")
 

tests/e2e_automation/utilities/apigee/ApigeeApp.py

@@ -0,0 +1,11 @@
+"""Simple data class to hold the required attributes of an Apigee App"""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class ApigeeApp:
+    callback_url: str
+    client_id: str
+    client_secret: str
+    supplier: str

tests/e2e_automation/utilities/apigee/ApigeeOnDemandAppManager.py

@@ -0,0 +1,105 @@
+"""Basic client class for managing interactions with the Apigee API"""
+
+import uuid
+
+import requests
+
+from utilities.apigee.apigee_env_helpers import get_apigee_access_token, get_apigee_username, get_proxy_name
+from utilities.apigee.ApigeeApp import ApigeeApp
+
+
+class ApigeeOnDemandAppManager:
+    """Manager class that provides required Apigee functionality for PR env e2e tests. E.g. creating an app, subscribing it to
+    products and teardown"""
+
+    # We only use the Apigee API in the non-prod organisation and the internal-dev environment
+    _BASE_URL = "https://api.enterprise.apigee.com/v1/organizations/nhsd-nonprod"
+    _APPS_PATH = "apps"
+    _DEVELOPERS_PATH = "developers"
+    _PRODUCTS_PATH = "apiproducts"
+    _INTERNAL_DEV_ENV_NAME = "internal-dev"
+    _TEST_APP_SUPPLIERS = ("EMIS", "MAVIS", "MEDICUS", "Postman_Auth", "RAVS", "SONAR", "TPP")
+
+    def __init__(self):
+        self.pr_proxy_name = get_proxy_name()
+        self.created_product_name_uuid: str = ""
+        self.created_app_name_uuids = []
+        self.display_name = f"test-{self.pr_proxy_name}"
+
+        self.logged_in_username = get_apigee_username()
+        self.access_token = get_apigee_access_token()
+
+        self.requests_session = requests.Session()
+        self.requests_session.headers.update({"Authorization": f"Bearer {self.access_token}"})
+
+    def _create_app(self, target_product_name: str, supplier_name: str) -> ApigeeApp:
+        app_name_uuid = str(uuid.uuid4())
+        app_data = {
+            "name": app_name_uuid,
+            "callbackUrl": "https://oauth.pstmn.io/v1/callback",
+            "status": "approved",
+            "attributes": [
+                {"name": "DisplayName", "value": f"{self.display_name}-{supplier_name}"},
+                {"name": "SupplierSystem", "value": supplier_name},
+            ],
+            "apiProducts": [target_product_name, "identity-service-internal-dev"],
+        }
+
+        response = self.requests_session.post(
+            url=f"{self._BASE_URL}/{self._DEVELOPERS_PATH}/{self.logged_in_username}/{self._APPS_PATH}", json=app_data
+        )
+        response.raise_for_status()
+
+        self.created_app_name_uuids.append(app_name_uuid)
+        response_dict = response.json()
+
+        return ApigeeApp(
+            callback_url=response_dict.get("callbackUrl"),
+            client_id=response_dict["credentials"][0]["consumerKey"],
+            client_secret=response_dict["credentials"][0]["consumerSecret"],
+            supplier=supplier_name,
+        )
+
+    def _create_product(self) -> str:
+        product_name_uuid = str(uuid.uuid4())
+        apigee_product_data = {
+            "name": product_name_uuid,
+            "apiResources": [],
+            "approvalType": "auto",
+            "description": "Autogenerated API product for E2E tests",
+            "displayName": self.display_name,
+            "environments": [self._INTERNAL_DEV_ENV_NAME],
+            "proxies": [self.pr_proxy_name],
+            "scopes": [
+                f"urn:nhsd:apim:app:level3:{self.pr_proxy_name}",
+                f"urn:nhsd:apim:user-nhs-cis2:aal3:{self.pr_proxy_name}",
+            ],
+        }
+
+        response = self.requests_session.post(
+            url=f"{self._BASE_URL}/{self._PRODUCTS_PATH}",
+            json=apigee_product_data,
+        )
+        response.raise_for_status()
+
+        self.created_product_name_uuid = product_name_uuid
+        return product_name_uuid
+
+    def setup_apps_and_product(self) -> list[ApigeeApp]:
+        """Orchestration method to setup the required product and on-demand apps required for PR testing"""
+        created_apps: list[ApigeeApp] = []
+        product_name_uuid = self._create_product()
+
+        for supplier_name in self._TEST_APP_SUPPLIERS:
+            created_apps.append(self._create_app(product_name_uuid, supplier_name))
+
+        return created_apps
+
+    def teardown_apps_and_product(self):
+        """Orchestration method to remove the Apigee resources in a teardown step"""
+        for created_app_name_uuid in self.created_app_name_uuids:
+            self.requests_session.delete(
+                url=f"{self._BASE_URL}/{self._DEVELOPERS_PATH}/{self.logged_in_username}/{self._APPS_PATH}/{created_app_name_uuid}"
+            )
+
+        self.requests_session.delete(url=f"{self._BASE_URL}/{self._PRODUCTS_PATH}/{self.created_product_name_uuid}")

tests/e2e_automation/utilities/apigee/__init__.py

@@ -0,0 +1,28 @@
+import os
+
+
+def get_env_var(var_name: str) -> str:
+    value = os.getenv(var_name)
+
+    if not value:
+        raise EnvironmentError(f"{var_name} environment variable is required")
+
+    return value
+
+
+def get_apigee_username() -> str:
+    return get_env_var("APIGEE_USERNAME")
+
+
+def get_proxy_name() -> str:
+    return get_env_var("PROXY_NAME")
+
+
+def is_pr_env() -> bool:
+    """Checks if the tests are running against a dynamic PR environment"""
+    proxy_name = get_proxy_name()
+    return proxy_name.startswith("immunisation-fhir-api-pr-")
+
+
+def get_apigee_access_token() -> str:
+    return get_env_var("APIGEE_ACCESS_TOKEN")