diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d6141cba..1a16455b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,7 @@ See [Legacy Airlock & migration](docs/azure-tre-overview/airlock.md#legacy-airlo * Add weekly and manual builds for Dockerfiles and Porter bundles, with selected pull request builds. Retry a failed target once and report final results. ([#5088](https://github.com/microsoft/AzureTRE/issues/5088)) BUG FIXES: +* Create Airlock prerequisites with explicit consent. Track owned review resources and wait for deletion. Preserve certificates if Nexus cleanup fails. ([#5183](https://github.com/microsoft/AzureTRE/issues/5183)) * Handle expected plain-text Airlock errors and give fixture cleanup a separate, bounded timeout. ([#5174](https://github.com/microsoft/AzureTRE/issues/5174)) * Install and verify required Azure CLI tools before configuring Windows review VMs. ([#5182](https://github.com/microsoft/AzureTRE/issues/5182)) * Use supported Dsv6 sizes for Guacamole Linux and Airlock review VMs. ([#5172](https://github.com/microsoft/AzureTRE/issues/5172)) diff --git a/e2e_tests/airlock/request.py b/e2e_tests/airlock/request.py index 86e8ab4a4..46fadad84 100644 --- a/e2e_tests/airlock/request.py +++ b/e2e_tests/airlock/request.py @@ -4,6 +4,9 @@ import os from urllib.parse import urlparse from azure.storage.blob import BlobClient +from azure.storage.blob.aio import BlobClient as AsyncBlobClient +from azure.core.exceptions import ResourceNotFoundError, ResourceExistsError +from uuid import uuid4 from airlock import strings from e2e_tests.helpers import get_auth_header, get_full_endpoint @@ -83,6 +86,35 @@ async def delete_blob_using_sas(file_path: str, sas_url: str): client.delete_blob() +async def wait_for_draft_container_deletion(sas_url: str, *, timeout=60, interval=2): + """Require ContainerNotFound after Azure's asynchronous container deletion.""" + started = asyncio.get_running_loop().time() + parsed = urlparse(sas_url) + # A new name and overwrite avoid mistaking BlobAlreadyExists for revoked access. + blob_url = parsed._replace(path=f"{parsed.path.rstrip('/')}/revocation-probe-{uuid4().hex}").geturl() + async with AsyncBlobClient.from_blob_url(blob_url, retry_total=0, connection_timeout=5, read_timeout=10) as client: + try: + async with asyncio.timeout(timeout): + while True: + try: + await client.upload_blob(b"post-submission draft probe", overwrite=True) + except ResourceNotFoundError as error: + if getattr(error, "error_code", None) != "ContainerNotFound": + raise + LOGGER.info( + "Draft container deletion confirmed after %.1f seconds (ContainerNotFound)", + asyncio.get_running_loop().time() - started, + ) + return + except ResourceExistsError as error: + # Azure can reject writes while deletion is still in progress. + if getattr(error, "error_code", None) != "ContainerBeingDeleted": + raise + await asyncio.sleep(interval) + except TimeoutError: + raise AssertionError(f"Draft container deletion was not confirmed within {timeout} seconds") from None + + async def wait_for_status(request_status: str, workspace_owner_token, workspace_path, request_id, verify): while True: diff --git a/e2e_tests/conftest.py b/e2e_tests/conftest.py index b77297a15..b0ece17c5 100644 --- a/e2e_tests/conftest.py +++ b/e2e_tests/conftest.py @@ -5,12 +5,14 @@ from typing import Tuple import config import logging -from contextlib import asynccontextmanager, contextmanager +from contextlib import AsyncExitStack, asynccontextmanager, contextmanager from resources.resource import post_resource, disable_and_delete_resource from resources.workspace import get_workspace_auth_details from resources import strings as resource_strings from helpers import get_admin_token, get_template +from e2e_tests.resources.nexus import nexus_prerequisites +from e2e_tests.timeouts import async_test_timeout, cleanup_deadline LOGGER = logging.getLogger(__name__) @@ -72,12 +74,12 @@ def pause_test_timeout_for_cleanup(request): @asynccontextmanager async def resource_cleanup_timeout(resource_path, request=None): with pause_test_timeout_for_cleanup(request): - timeout = asyncio.timeout(CLEANUP_TIMEOUT_SECONDS) + timeout = None try: - async with timeout: + async with cleanup_deadline(CLEANUP_TIMEOUT_SECONDS) as timeout: yield except TimeoutError as error: - if not timeout.expired(): + if timeout is None or not timeout.expired(): raise message = f"Cleanup of {resource_path} exceeded {CLEANUP_TIMEOUT_SECONDS} seconds" LOGGER.error("%s. Check the operation polling logs for its last observed state.", message) @@ -96,6 +98,7 @@ def verify(pytestconfig): return False +@async_test_timeout(60 * 60) async def create_or_get_test_workspace( auth_type: str, verify: bool, @@ -139,13 +142,18 @@ async def create_or_get_test_workspace( # TODO: Temp fix to solve creation of workspaces - https://github.com/microsoft/AzureTRE/issues/2986 await asyncio.sleep(random.uniform(1, 9)) workspace_path, workspace_id = await post_resource( - payload, resource_strings.API_WORKSPACES, access_token=admin_token, verify=verify + payload, + resource_strings.API_WORKSPACES, + access_token=admin_token, + verify=verify, + cleanup_failed_create=True, ) LOGGER.info(f"Workspace {workspace_id} {template_name} created") return workspace_path, workspace_id +@async_test_timeout(60 * 60) async def create_or_get_test_workpace_service( workspace_path, workspace_owner_token, pre_created_workspace_service_id, verify ): @@ -165,6 +173,7 @@ async def create_or_get_test_workpace_service( endpoint=f"/api{workspace_path}/{resource_strings.API_WORKSPACE_SERVICES}", access_token=workspace_owner_token, verify=verify, + cleanup_failed_create=True, ) return workspace_service_path, workspace_service_id @@ -257,19 +266,32 @@ async def get_workspace_owner_token(workspace_id, verify): async def disable_and_delete_ws_resource(resource_path, workspace_id, verify): workspace_owner_token = await get_workspace_owner_token(workspace_id, verify) - await disable_and_delete_resource(f"/api{resource_path}", workspace_owner_token, verify) + await disable_and_delete_resource(f"/api{resource_path}", workspace_owner_token, verify, allow_failed_disable=True) async def disable_and_delete_tre_resource(resource_path, verify): admin_token = await get_admin_token(verify) - await disable_and_delete_resource(f"/api{resource_path}", admin_token, verify) + await disable_and_delete_resource(f"/api{resource_path}", admin_token, verify, allow_failed_disable=True) + + +@pytest.fixture(scope="session") +async def setup_nexus_prerequisites(verify, request): + # Dependencies must outlive review VMs and their workspace finaliser. + resources = AsyncExitStack() + try: + async with asyncio.timeout(60 * 60): + await resources.enter_async_context(nexus_prerequisites(verify)) + yield + finally: + async with resource_cleanup_timeout("Nexus prerequisites", request): + await resources.aclose() # Session scope isn't in effect with python-xdist: https://github.com/microsoft/AzureTRE/issues/2868 @pytest.fixture(scope="session") async def setup_test_airlock_import_review_workspace_and_guacamole_service( - verify, request -) -> Tuple[str, str, str, str, str]: + setup_nexus_prerequisites, verify, request +) -> Tuple[str, str, str, str]: pre_created_workspace_id = config.TEST_AIRLOCK_IMPORT_REVIEW_WORKSPACE_ID # Set up workspace_path, workspace_id = await create_or_get_test_workspace( @@ -279,22 +301,39 @@ async def setup_test_airlock_import_review_workspace_and_guacamole_service( pre_created_workspace_id=pre_created_workspace_id, ) - admin_token = await get_admin_token(verify=verify) - workspace_owner_token, _ = await get_workspace_auth_details( - admin_token=admin_token, workspace_id=workspace_id, verify=verify - ) + workspace_service_path = None pre_created_workspace_service_id = config.TEST_AIRLOCK_IMPORT_REVIEW_WORKSPACE_SERVICE_ID + try: + admin_token = await get_admin_token(verify=verify) + workspace_owner_token, _ = await get_workspace_auth_details( + admin_token=admin_token, workspace_id=workspace_id, verify=verify + ) - workspace_service_path, workspace_service_id = await create_or_get_test_workpace_service( - workspace_path, - workspace_owner_token=workspace_owner_token, - pre_created_workspace_service_id=pre_created_workspace_service_id, - verify=verify, - ) + workspace_service_path, workspace_service_id = await create_or_get_test_workpace_service( + workspace_path, + workspace_owner_token=workspace_owner_token, + pre_created_workspace_service_id=pre_created_workspace_service_id, + verify=verify, + ) - yield workspace_path, workspace_id, workspace_service_path, workspace_service_id + yield workspace_path, workspace_id, workspace_service_path, workspace_service_id - # Tear-down in a cascaded way - await clean_up_test_workspace( - pre_created_workspace_id=pre_created_workspace_id, workspace_path=workspace_path, verify=verify, request=request - ) + finally: + try: + # A reused parent has no cascading cleanup. Remove only a child + # created by this fixture. Preserve any pre-existing service. + if pre_created_workspace_id and workspace_service_path is not None: + await clean_up_test_workspace_service( + pre_created_workspace_service_id, + workspace_service_path, + workspace_id, + verify, + request=request, + ) + finally: + await clean_up_test_workspace( + pre_created_workspace_id=pre_created_workspace_id, + workspace_path=workspace_path, + verify=verify, + request=request, + ) diff --git a/e2e_tests/pytest.ini b/e2e_tests/pytest.ini index 3b4a6478e..1d29fdbcc 100644 --- a/e2e_tests/pytest.ini +++ b/e2e_tests/pytest.ini @@ -9,6 +9,7 @@ markers = performance: marks tests for performance evaluation timeout: used to set test timeout with pytest-timeout airlock: only airlock related + airlock_validation: draft-container deletion and review-VM flow, including Nexus prerequisites and cleanup workspace_services asyncio_mode = auto diff --git a/e2e_tests/resources/nexus.py b/e2e_tests/resources/nexus.py new file mode 100644 index 000000000..05acf9aa9 --- /dev/null +++ b/e2e_tests/resources/nexus.py @@ -0,0 +1,111 @@ +"""Nexus prerequisites shared by Linux and Airlock review VM tests.""" + +from contextlib import AsyncExitStack, asynccontextmanager +from uuid import uuid4 + +from packaging.version import Version + +from e2e_tests import config +from e2e_tests.helpers import get_admin_token +from e2e_tests.resources import strings +from e2e_tests.resources.resource import ( + FAILED_CREATE_CLEANUP_TIMEOUT_SECONDS, + disable_and_delete_resource, + get_resource, + post_resource, + temporary_resource, +) +from e2e_tests.timeouts import cleanup_deadline + + +async def cleanup_certificate(resource_path, access_token, verify): + async with cleanup_deadline(FAILED_CREATE_CLEANUP_TIMEOUT_SECONDS): + services = (await get_resource(strings.API_SHARED_SERVICES, access_token, verify))["sharedServices"] + if any(service["templateName"] == strings.NEXUS_SHARED_SERVICE for service in services): + raise RuntimeError(f"Retaining certificate dependency {resource_path}: Nexus is still present") + await disable_and_delete_resource(f"/api{resource_path}", access_token, verify, allow_failed_disable=True) + + +@asynccontextmanager +async def certificate_prerequisite(payload, endpoint, access_token, verify): + resource_path, _ = await post_resource(payload, endpoint, access_token, verify, cleanup_failed_create=True) + try: + yield resource_path + except BaseException as original_error: + try: + await cleanup_certificate(resource_path, access_token, verify) + except Exception as cleanup_error: + original_error.add_note(f"Cleanup of {resource_path} also failed: {cleanup_error!r}") + raise + else: + await cleanup_certificate(resource_path, access_token, verify) + + +@asynccontextmanager +async def nexus_prerequisites(verify): + """Reuse a suitable Nexus instance or create explicitly authorised prerequisites.""" + admin_token = await get_admin_token(verify) + services = (await get_resource(strings.API_SHARED_SERVICES, admin_token, verify))["sharedServices"] + nexus_services = [service for service in services if service["templateName"] == strings.NEXUS_SHARED_SERVICE] + assert len(nexus_services) <= 1, "This test requires at most one Nexus shared service" + if nexus_services: + nexus = nexus_services[0] + assert nexus["isEnabled"] and nexus["deploymentStatus"] in ("deployed", "updated"), ( + "Nexus must be deployed and enabled" + ) + assert Version(nexus["templateVersion"]) >= Version("3.11.0"), "Upgrade Nexus to bundle 3.11.0 before this test" + yield + return + + assert config.TEST_ACCEPT_NEXUS_EULA, ( + "This test requires Nexus. Deploy Nexus 3.11.0 first, or explicitly accept its CE EULA " + "with TEST_ACCEPT_NEXUS_EULA=true (acceptNexusEula in the branch workflow)." + ) + cert_services = [service for service in services if service["templateName"] == strings.CERTS_SHARED_SERVICE] + assert len(cert_services) <= 1, "This test requires at most one certificate shared service" + async with AsyncExitStack() as resources: + if cert_services: + certs = cert_services[0] + assert certs["isEnabled"] and certs["deploymentStatus"] in ("deployed", "updated"), ( + "Repair the existing Nexus certificate service before this test" + ) + assert certs["properties"].get("domain_prefix") == "nexus", ( + "The existing certificate service belongs to another domain. Deploy Nexus separately before this test." + ) + cert_name = certs["properties"]["cert_name"] + else: + cert_name = f"nexus-e2e-{uuid4().hex[:8]}" + await resources.enter_async_context( + certificate_prerequisite( + { + "templateName": strings.CERTS_SHARED_SERVICE, + "properties": { + "display_name": "E2E Nexus certificate", + "description": "Certificate prerequisite for E2E Nexus", + "domain_prefix": "nexus", + "cert_name": cert_name, + }, + }, + strings.API_SHARED_SERVICES, + admin_token, + verify, + ) + ) + await resources.enter_async_context( + temporary_resource( + { + "templateName": strings.NEXUS_SHARED_SERVICE, + "properties": { + "display_name": "E2E Nexus", + "description": "Nexus prerequisite for E2E workspace resources", + "ssl_cert_name": cert_name, + "accept_nexus_eula": config.TEST_ACCEPT_NEXUS_EULA, + "vm_size": "Standard_D2s_v3", + }, + }, + strings.API_SHARED_SERVICES, + admin_token, + verify, + ) + ) + yield diff --git a/e2e_tests/resources/resource.py b/e2e_tests/resources/resource.py index eff78338c..efea3c8f9 100644 --- a/e2e_tests/resources/resource.py +++ b/e2e_tests/resources/resource.py @@ -1,6 +1,7 @@ import asyncio import logging from contextlib import asynccontextmanager +from e2e_tests.timeouts import cleanup_deadline from httpx import AsyncClient, Timeout from starlette import status from e2e_tests.helpers import assert_status, get_auth_header, get_full_endpoint @@ -10,6 +11,7 @@ LOGGER = logging.getLogger(__name__) TIMEOUT = Timeout(10, read=60) +FAILED_CREATE_CLEANUP_TIMEOUT_SECONDS = 60 * 60 async def get_resource(endpoint, access_token, verify): @@ -24,8 +26,19 @@ async def get_resource(endpoint, access_token, verify): async def post_resource( - payload, endpoint, access_token, verify, method="POST", wait=True, etag="*", access_token_for_wait=None + payload, + endpoint, + access_token, + verify, + method="POST", + wait=True, + etag="*", + access_token_for_wait=None, + *, + cleanup_failed_create=False, ): + if cleanup_failed_create and (method != "POST" or not wait): + raise ValueError("Failed-create cleanup requires a POST with deployment polling") async with AsyncClient(verify=verify, timeout=30.0) as client: full_endpoint = get_full_endpoint(endpoint) auth_headers = get_auth_header(access_token) @@ -46,17 +59,47 @@ async def post_resource( if wait: wait_token = access_token_for_wait if access_token_for_wait is not None else access_token - await wait_for( - check_method, - client, - operation_endpoint, - wait_token, - [strings.RESOURCE_STATUS_DEPLOYMENT_FAILED, strings.RESOURCE_STATUS_UPDATING_FAILED], - ) + try: + await wait_for( + check_method, + client, + operation_endpoint, + wait_token, + [strings.RESOURCE_STATUS_DEPLOYMENT_FAILED, strings.RESOURCE_STATUS_UPDATING_FAILED], + ) + except BaseException as original_error: + if cleanup_failed_create: + cleanup = asyncio.create_task( + cleanup_accepted_create(resource_path, operation_endpoint, client, wait_token, verify) + ) + # Cancellation must not abandon a known resource before its + # fixture can register teardown. Recovery has its own deadline. + while not cleanup.done(): + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError: + continue + except Exception: + break + try: + cleanup.result() + except BaseException as cleanup_error: + original_error.add_note( + f"Cleanup of {resource_path} after {operation_endpoint} also failed: {cleanup_error!r}" + ) + raise return resource_path, resource_id +async def cleanup_accepted_create(resource_path, operation_endpoint, client, access_token, verify): + async with cleanup_deadline(FAILED_CREATE_CLEANUP_TIMEOUT_SECONDS): + # An accepted deployment may still be running when its caller times out. + # Wait for it to finish before sending another provisioning action. + await wait_for(install_done, client, operation_endpoint, access_token, []) + await disable_and_delete_resource(f"/api{resource_path}", access_token, verify, allow_failed_disable=True) + + async def disable_and_delete_resource(endpoint, access_token, verify, *, allow_failed_disable=False): async with AsyncClient(verify=verify, timeout=TIMEOUT) as client: full_endpoint = get_full_endpoint(endpoint) @@ -71,7 +114,18 @@ async def disable_and_delete_resource(endpoint, access_token, verify, *, allow_f # The API persists isEnabled=False before provisioning. Failed installs can # fail again during disable, but deletion is safe once that operation ends. failure_states = [] if allow_failed_disable else [strings.RESOURCE_STATUS_UPDATING_FAILED] - await wait_for(patch_done, client, operation_endpoint, access_token, failure_states) + disable_state = await wait_for(patch_done, client, operation_endpoint, access_token, failure_states) + if disable_state == strings.RESOURCE_STATUS_UPDATING_FAILED: + response = await client.get(full_endpoint, headers=get_auth_header(access_token), timeout=TIMEOUT) + assert_status(response, [status.HTTP_200_OK], "Could not verify disabled resource") + resources = [ + response.json()[key] + for key in ("workspace", "workspaceService", "userResource", "sharedService") + if key in response.json() + ] + assert len(resources) == 1 and resources[0].get("isEnabled") is False, ( + "Refusing deletion after failed disable: the resource is not confirmed disabled" + ) # delete auth_headers = get_auth_header(access_token) @@ -89,28 +143,20 @@ async def disable_and_delete_resource(endpoint, access_token, verify, *, allow_f @asynccontextmanager async def temporary_resource(payload, endpoint, access_token, verify): """Create a test resource and remove it after success or failed provisioning.""" - async with AsyncClient(verify=verify, timeout=TIMEOUT) as client: - response = await client.post(get_full_endpoint(endpoint), headers=get_auth_header(access_token), json=payload) - assert_status(response, [status.HTTP_202_ACCEPTED], "The test resource could not be created") - resource_path = response.json()["operation"]["resourcePath"] + resource_path, _ = await post_resource(payload, endpoint, access_token, verify, cleanup_failed_create=True) + try: + yield resource_path + except BaseException as original_error: try: - await wait_for( - install_done, - client, - response.headers["Location"], - access_token, - [strings.RESOURCE_STATUS_DEPLOYMENT_FAILED], - ) - yield resource_path - except BaseException as original_error: - try: + async with cleanup_deadline(FAILED_CREATE_CLEANUP_TIMEOUT_SECONDS): await disable_and_delete_resource( f"/api{resource_path}", access_token, verify, allow_failed_disable=True ) - except Exception as cleanup_error: - original_error.add_note(f"Cleanup of {resource_path} also failed: {cleanup_error!r}") - raise - else: + except Exception as cleanup_error: + original_error.add_note(f"Cleanup of {resource_path} also failed: {cleanup_error!r}") + raise + else: + async with cleanup_deadline(FAILED_CREATE_CLEANUP_TIMEOUT_SECONDS): await disable_and_delete_resource(f"/api{resource_path}", access_token, verify) @@ -138,3 +184,40 @@ async def wait_for(func, client, operation_endpoint, access_token, failure_state except Exception: LOGGER.exception(f"Failed to deploy. Status message: {message}.\n{operation_steps}") raise + return done_state + + +async def delete_owned_resource_if_present(resource_path, access_token, verify): + """Wait for active operations and remove an owned resource, unless already deleted.""" + terminal = { + "deployed", + "deployment_failed", + "updated", + "updating_failed", + "deleted", + "deleting_failed", + "action_succeeded", + "action_failed", + "pipeline_succeeded", + "pipeline_failed", + } + endpoint = get_full_endpoint(f"/api{resource_path}") + async with AsyncClient(verify=verify, timeout=TIMEOUT) as client: + while True: + response = await client.get(endpoint, headers=get_auth_header(access_token), timeout=TIMEOUT) + if response.status_code == 404: + return + assert_status(response, [status.HTTP_200_OK], "Could not inspect the owned review VM") + state = response.json()["userResource"]["deploymentStatus"] + if state == "deleted": + return + operations = await client.get( + endpoint + "/operations", headers=get_auth_header(access_token), timeout=TIMEOUT + ) + if operations.status_code == 404: + return + assert_status(operations, [status.HTTP_200_OK], "Could not inspect review VM operations") + if state in terminal and all(op["status"] in terminal for op in operations.json()["operations"]): + break + await asyncio.sleep(30) + await disable_and_delete_resource(f"/api{resource_path}", access_token, verify, allow_failed_disable=True) diff --git a/e2e_tests/test_airlock.py b/e2e_tests/test_airlock.py index 4ce8fc1c3..0ca3672fe 100644 --- a/e2e_tests/test_airlock.py +++ b/e2e_tests/test_airlock.py @@ -1,3 +1,4 @@ +from contextlib import asynccontextmanager import os import pytest import asyncio @@ -5,12 +6,20 @@ from httpx import AsyncClient from azure.core.exceptions import ResourceNotFoundError -from airlock.request import post_request, get_request, upload_blob_using_sas, delete_blob_using_sas, wait_for_status -from resources.resource import get_resource, post_resource +from airlock.request import ( + post_request, + get_request, + upload_blob_using_sas, + delete_blob_using_sas, + wait_for_status, + wait_for_draft_container_deletion, +) +from e2e_tests.timeouts import async_test_timeout, cleanup_deadline +from resources.resource import get_resource, post_resource, delete_owned_resource_if_present from resources.workspace import get_workspace_auth_details from airlock import strings as airlock_strings from e2e_tests.conftest import get_workspace_owner_token -from helpers import get_admin_token +from helpers import assert_status, get_admin_token, get_auth_header, get_full_endpoint from e2e_tests.token_provider import TokenProvider @@ -81,17 +90,17 @@ async def submit_airlock_import_request(workspace_path: str, workspace_owner_tok @pytest.mark.timeout(30 * 60) @pytest.mark.airlock -async def test_draft_container_is_sealed_after_submit(setup_test_workspace, verify): - """A SAS handed out while the request was in Draft must stop working once it is submitted, - otherwise a researcher could alter the data after it has been scanned and reviewed.""" +@pytest.mark.airlock_validation +async def test_draft_container_is_deleted_after_submit(setup_test_workspace, verify): + """The draft SAS must stop working after asynchronous deletion completes.""" workspace_path, workspace_id = setup_test_workspace workspace_owner_token = await get_workspace_owner_token(workspace_id, verify) _, container_url = await submit_airlock_import_request(workspace_path, workspace_owner_token, verify) - # Submission copies the data out and deletes the draft container, so the old SAS resolves to nothing. - with pytest.raises(ResourceNotFoundError): - await upload_blob_using_sas(BLOB_FILE_PATH, container_url) + # Azure permits draft access for up to 30 seconds after accepting deletion. + # The submitted copy is separate. Do not interpret a same-name conflict as denial. + await wait_for_draft_container_deletion(container_url) async def create_draft_import_request_with_file(workspace_path: str, workspace_owner_token: str, verify: bool): @@ -250,8 +259,79 @@ async def test_in_progress_data_is_not_reachable_from_the_public_internet(setup_ ) -@pytest.mark.timeout(50 * 60) +@asynccontextmanager +async def managed_review_vm(workspace_path, request_id, workspace_owner_token, review_token, verify): + resource_path, _ = await post_resource( + payload={}, + endpoint=f"/api{workspace_path}/requests/{request_id}/review-user-resource", + access_token=workspace_owner_token, + verify=verify, + method="POST", + wait=True, + access_token_for_wait=review_token, + cleanup_failed_create=True, + ) + original = None + try: + yield resource_path + except BaseException as error: + original = error + raise + finally: + + async def recover(): + async with cleanup_deadline(60 * 60): + await delete_owned_resource_if_present(resource_path, review_token, verify) + + recovery = asyncio.create_task(recover()) + cancellation = None + while not recovery.done(): + try: + await asyncio.shield(recovery) + except asyncio.CancelledError as error: + cancellation = error + except Exception: + break + try: + recovery.result() + except (Exception, asyncio.CancelledError) as error: + failure = original if original is not None else cancellation + if failure is None: + raise + failure.add_note(f"Cleanup of {resource_path} also failed: {error!r}") + if cancellation is not None and original is None: + raise cancellation + + +async def wait_for_review_vm_deletion(resource_path, access_token, verify, *, timeout_seconds=30 * 60): + """Wait for automatic deletion without sending a second delete request.""" + last_state = "not yet observed" + deadline = asyncio.timeout(timeout_seconds) + try: + async with deadline, AsyncClient(verify=verify, timeout=30) as client: + while True: + response = await client.get( + get_full_endpoint(f"/api{resource_path}"), headers=get_auth_header(access_token) + ) + if response.status_code == 404: + return + assert_status(response, [200], "Could not check review VM deletion") + last_state = response.json()["userResource"]["deploymentStatus"] + assert not last_state.endswith("_failed"), f"Review VM deletion failed: {last_state}" + if last_state == "deleted": + return + LOGGER.info("Waiting for review VM deletion: %s, %s", resource_path, last_state) + await asyncio.sleep(30) + except TimeoutError as error: + if not deadline.expired(): + raise + raise TimeoutError(f"Review VM deletion timed out. Last state: {last_state}") from error + + +@pytest.mark.timeout(150 * 60, func_only=True) @pytest.mark.airlock +@pytest.mark.airlock_validation +@async_test_timeout(50 * 60) async def test_airlock_review_vm_flow( setup_test_workspace, setup_test_airlock_import_review_workspace_and_guacamole_service, verify ): @@ -312,35 +392,28 @@ async def test_airlock_review_vm_flow( f"/api/workspaces/{import_review_workspace_id}/requests/{request_id}", import_workspace_owner_token, verify, 404 ) - user_resource_path, user_resource_id = await post_resource( - payload={}, - endpoint=f"/api{workspace_path}/requests/{request_id}/review-user-resource", - access_token=workspace_owner_token, - verify=verify, - method="POST", - wait=True, - access_token_for_wait=import_workspace_owner_token, # needs a different token as is created in a separate workspace - ) - - LOGGER.info(f"Airlock Review VM has been created: {user_resource_path}") - - # Approve request - LOGGER.info("Approving airlock request") - payload = {"approval": "True", "decisionExplanation": "the reason why this request was approved/rejected"} - request_result = await post_request( - payload, f"/api{workspace_path}/requests/{request_id}/review", workspace_owner_token, verify, 200 - ) - assert ( - request_result["airlockRequest"]["reviews"][0]["decisionExplanation"] - == "the reason why this request was approved/rejected" - ) - - await wait_for_status(airlock_strings.APPROVED_STATUS, workspace_owner_token, workspace_path, request_id, verify) - LOGGER.info("Airlock request has been approved") - - # Check that deletion for user resource has started - user_resource = await get_resource(f"/api{user_resource_path}", import_workspace_owner_token, verify) - assert user_resource["userResource"]["deploymentStatus"] == "updating" - LOGGER.info("Review VM has started deletion successfully") + async with managed_review_vm( + workspace_path, request_id, workspace_owner_token, import_workspace_owner_token, verify + ) as user_resource_path: + LOGGER.info(f"Airlock Review VM has been created: {user_resource_path}") + + # Approve request + LOGGER.info("Approving airlock request") + payload = {"approval": "True", "decisionExplanation": "the reason why this request was approved/rejected"} + request_result = await post_request( + payload, f"/api{workspace_path}/requests/{request_id}/review", workspace_owner_token, verify, 200 + ) + assert ( + request_result["airlockRequest"]["reviews"][0]["decisionExplanation"] + == "the reason why this request was approved/rejected" + ) + + await wait_for_status( + airlock_strings.APPROVED_STATUS, workspace_owner_token, workspace_path, request_id, verify + ) + LOGGER.info("Airlock request has been approved") + + await wait_for_review_vm_deletion(user_resource_path, import_workspace_owner_token, verify) + LOGGER.info("Review VM deletion completed") # EXPORT FLOW diff --git a/e2e_tests/test_guacamole_linuxvm.py b/e2e_tests/test_guacamole_linuxvm.py index b6f4020ec..6b37a6353 100644 --- a/e2e_tests/test_guacamole_linuxvm.py +++ b/e2e_tests/test_guacamole_linuxvm.py @@ -1,88 +1,16 @@ """Check that a Guacamole Linux VM finishes its Nexus-backed bootstrap.""" -from contextlib import AsyncExitStack, asynccontextmanager -from uuid import uuid4 - import pytest -from packaging.version import Version -from e2e_tests import config from e2e_tests.conftest import get_workspace_owner_token -from e2e_tests.helpers import get_admin_token from e2e_tests.resources import strings -from e2e_tests.resources.resource import get_resource, temporary_resource +from e2e_tests.resources.resource import temporary_resource +from e2e_tests.resources.nexus import nexus_prerequisites pytestmark = pytest.mark.asyncio(loop_scope="session") -@asynccontextmanager -async def linux_vm_nexus(verify): - """Reuse a suitable Nexus instance or create explicitly authorised prerequisites.""" - admin_token = await get_admin_token(verify) - services = (await get_resource(strings.API_SHARED_SERVICES, admin_token, verify))["sharedServices"] - nexus_services = [service for service in services if service["templateName"] == strings.NEXUS_SHARED_SERVICE] - assert len(nexus_services) <= 1, "This test requires at most one Nexus shared service" - if nexus_services: - nexus = nexus_services[0] - assert nexus["isEnabled"] and nexus["deploymentStatus"] in ("deployed", "updated"), ( - "Nexus must be deployed and enabled" - ) - assert Version(nexus["templateVersion"]) >= Version("3.11.0"), "Upgrade Nexus to bundle 3.11.0 before this test" - yield - return - - assert config.TEST_ACCEPT_NEXUS_EULA, ( - "This test requires Nexus. Deploy Nexus 3.11.0 first, or explicitly accept its CE EULA " - "with TEST_ACCEPT_NEXUS_EULA=true (acceptNexusEula in the branch workflow)." - ) - cert_services = [service for service in services if service["templateName"] == strings.CERTS_SHARED_SERVICE] - assert len(cert_services) <= 1, "This test requires at most one certificate shared service" - async with AsyncExitStack() as resources: - if cert_services: - certs = cert_services[0] - assert certs["isEnabled"] and certs["deploymentStatus"] in ("deployed", "updated"), ( - "Repair the existing Nexus certificate service before this test" - ) - assert certs["properties"].get("domain_prefix") == "nexus", ( - "The existing certificate service belongs to another domain. Deploy Nexus separately before this test." - ) - cert_name = certs["properties"]["cert_name"] - else: - cert_name = f"nexus-linux-e2e-{uuid4().hex[:8]}" - await resources.enter_async_context( - temporary_resource( - { - "templateName": strings.CERTS_SHARED_SERVICE, - "properties": { - "display_name": "Linux E2E Nexus certificate", - "domain_prefix": "nexus", - "cert_name": cert_name, - }, - }, - strings.API_SHARED_SERVICES, - admin_token, - verify, - ) - ) - await resources.enter_async_context( - temporary_resource( - { - "templateName": strings.NEXUS_SHARED_SERVICE, - "properties": { - "display_name": "Linux E2E Nexus", - "ssl_cert_name": cert_name, - "accept_nexus_eula": config.TEST_ACCEPT_NEXUS_EULA, - }, - }, - strings.API_SHARED_SERVICES, - admin_token, - verify, - ) - ) - yield - - @pytest.mark.linux_vm @pytest.mark.timeout(150 * 60) async def test_create_guacamole_linux_vm(setup_test_workspace_and_guacamole_service, verify): @@ -98,7 +26,7 @@ async def test_create_guacamole_linux_vm(setup_test_workspace_and_guacamole_serv }, } - async with linux_vm_nexus(verify): + async with nexus_prerequisites(verify): async with temporary_resource( payload, f"/api{workspace_service_path}/{strings.API_USER_RESOURCES}", workspace_owner_token, verify ): diff --git a/e2e_tests/timeouts.py b/e2e_tests/timeouts.py new file mode 100644 index 000000000..f601093ff --- /dev/null +++ b/e2e_tests/timeouts.py @@ -0,0 +1,40 @@ +"""Cooperative deadlines for asynchronous test bodies.""" + +import asyncio +from functools import wraps +from contextlib import asynccontextmanager +from contextvars import ContextVar + + +def async_test_timeout(seconds): + """Finish cancellation before pytest starts another test or fixture cleanup.""" + + def decorate(test): + @wraps(test) + async def bounded_test(*args, **kwargs): + async with asyncio.timeout(seconds): + return await test(*args, **kwargs) + + return bounded_test + + return decorate + + +_CLEANUP_DEADLINE = ContextVar("e2e_cleanup_deadline", default=None) + + +@asynccontextmanager +async def cleanup_deadline(seconds): + """Share one absolute deadline across nested resource finalisers.""" + loop = asyncio.get_running_loop() + outer = _CLEANUP_DEADLINE.get() + deadline = min(outer, loop.time() + seconds) if outer is not None else loop.time() + seconds + if loop.time() >= deadline: + raise TimeoutError("The shared resource cleanup deadline has expired") + token = _CLEANUP_DEADLINE.set(deadline) + timeout = asyncio.timeout_at(deadline) + try: + async with timeout: + yield timeout + finally: + _CLEANUP_DEADLINE.reset(token) diff --git a/e2e_tests/unit_tests/test_accepted_resource_cleanup.py b/e2e_tests/unit_tests/test_accepted_resource_cleanup.py new file mode 100644 index 000000000..5eb320ea0 --- /dev/null +++ b/e2e_tests/unit_tests/test_accepted_resource_cleanup.py @@ -0,0 +1,206 @@ +"""Retain ownership of accepted resources across setup cancellation.""" + +import asyncio +from contextlib import asynccontextmanager +import json +import unittest +from unittest.mock import AsyncMock, patch + +from httpx import AsyncClient, MockTransport, Response + +from e2e_tests import conftest as fixtures, helpers, test_airlock as airlock +from e2e_tests.resources import resource + + +@asynccontextmanager +async def template(*args): + yield Response(200, json={"properties": {}}) + + +class AcceptedResourceCleanupTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.requests = [] + self.authorization = [] + self.path = "/workspaces/new-workspace" + self.states = ["deploying", "deployed"] + self.disable_status = 202 + self.create_status = 202 + self.block_delete = False + self.poll_started = asyncio.Event() + self.delete_started = asyncio.Event() + self.release_delete = asyncio.Event() + self.enterContext(patch.object(helpers.config, "TRE_URL", "https://tre.example.test")) + self.enterContext(patch.object(fixtures, "get_admin_token", AsyncMock(return_value="admin-token"))) + self.enterContext(patch.object(fixtures, "get_template", template)) + self.enterContext(patch.object(fixtures.random, "uniform", return_value=0)) + self.enterContext(patch.object(fixtures, "post_resource", resource.post_resource)) + self.enterContext( + patch.object( + resource, "AsyncClient", side_effect=lambda **_: AsyncClient(transport=MockTransport(self.handle)) + ) + ) + + async def handle(self, request): + self.requests.append((request.method, request.url.path)) + self.authorization.append(request.headers.get("Authorization")) + code = 200 + phase = request.url.path.rsplit("/", 1)[-1] + if request.method == "POST": + phase, state, code = "install", "deploying", self.create_status + elif request.method == "PATCH": + self.assertEqual(json.loads(request.content), {"isEnabled": False}) + phase, state, code = "disable", "updating", self.disable_status + elif request.method == "DELETE": + self.delete_started.set() + if self.block_delete: + await self.release_delete.wait() + phase, state = "delete", "deleting" + elif phase == "install": + self.poll_started.set() + state = self.states.pop(0) if len(self.states) > 1 else self.states[0] + else: + state = {"disable": "updated", "delete": "deleted"}[phase] + return Response( + code, + headers={"Location": f"/api/operations/{phase}"}, + json={ + "operation": { + "resourcePath": self.path, + "resourceId": self.path.rsplit("/", 1)[-1], + "status": state, + "message": "Mock operation", + "steps": [], + } + }, + ) + + async def create(self, **kwargs): + return await resource.post_resource({}, "/api/workspaces", "token", True, cleanup_failed_create=True, **kwargs) + + async def test_workspace_setup_timeout_removes_the_accepted_workspace(self): + with self.assertRaises(TimeoutError): + async with asyncio.timeout(0.01): + await fixtures.create_or_get_test_workspace("Automatic", True) + self.assertEqual( + [method for method, _ in self.requests], ["POST", "GET", "GET", "PATCH", "GET", "DELETE", "GET"] + ) + self.assertEqual(self.requests[-2], ("DELETE", "/api" + self.path)) + + async def test_service_timeout_in_existing_workspace_deletes_only_the_new_service(self): + self.path = "/workspaces/existing/workspace-services/new-service" + with self.assertRaises(TimeoutError): + async with asyncio.timeout(0.01): + await fixtures.create_or_get_test_workpace_service("/workspaces/existing", "token", "", True) + self.assertEqual([path for method, path in self.requests if method == "DELETE"], ["/api" + self.path]) + + async def test_review_vm_timeout_finishes_child_cleanup_before_workspace_teardown(self): + self.path = "/workspaces/review/workspace-services/guacamole/user-resources/review-vm" + self.block_delete = True + events = [] + + async def post(**kwargs): + if kwargs.get("method") == "PATCH": + return "/workspaces/research", "research" + return await resource.post_resource(**kwargs) + + self.enterContext(patch.object(airlock, "post_resource", post)) + self.enterContext(patch.object(airlock, "get_workspace_owner_token", AsyncMock(return_value="research-token"))) + self.enterContext(patch.object(airlock, "get_admin_token", AsyncMock(return_value="admin-token"))) + self.enterContext( + patch.object(airlock, "get_resource", AsyncMock(return_value={"workspace": {"_etag": "etag"}})) + ) + self.enterContext( + patch.object(airlock, "get_workspace_auth_details", AsyncMock(return_value=("review-token", None))) + ) + self.enterContext( + patch.object(airlock, "submit_airlock_import_request", AsyncMock(return_value=("request", "unused"))) + ) + self.enterContext(patch.object(airlock, "get_request", AsyncMock())) + + async def run_with_teardown(): + try: + async with asyncio.timeout(0.01): + await airlock.test_airlock_review_vm_flow( + ("/workspaces/research", "research"), + ( + "/workspaces/review", + "review", + "/workspaces/review/workspace-services/guacamole", + "guacamole", + ), + True, + ) + finally: + events.append("workspace-teardown") + + task = asyncio.create_task(run_with_teardown()) + try: + await asyncio.wait_for(self.delete_started.wait(), 1) + self.assertEqual(events, []) + self.assertFalse(task.done()) + finally: + self.release_delete.set() + with self.assertRaises(TimeoutError): + await task + self.assertEqual(events, ["workspace-teardown"]) + self.assertEqual(self.requests[0], ("POST", "/api/workspaces/research/requests/request/review-user-resource")) + self.assertEqual(self.requests[-2:], [("DELETE", "/api" + self.path), ("GET", "/api/operations/delete")]) + self.assertEqual( + self.authorization, ["Bearer research-token"] + ["Bearer review-token"] * (len(self.requests) - 1) + ) + + async def test_failed_deployment_still_cleans_up_and_preserves_failure(self): + self.states = ["deployment_failed"] + with self.assertRaises(AssertionError): + await self.create() + self.assertIn(("DELETE", "/api" + self.path), self.requests) + + async def test_rejected_creation_does_not_delete_anything(self): + self.create_status = 403 + with self.assertRaises(AssertionError): + await self.create() + self.assertEqual(self.requests, [("POST", "/api/workspaces")]) + + async def test_pending_deployment_recovery_has_its_own_bound_and_records_resource(self): + self.states = ["deploying"] + with patch.object(resource, "FAILED_CREATE_CLEANUP_TIMEOUT_SECONDS", 0.01): + with self.assertRaises(TimeoutError) as caught: + async with asyncio.timeout(0.01): + await self.create() + self.assertNotIn("PATCH", [method for method, _ in self.requests]) + note = caught.exception.__cause__.__notes__[0] + self.assertIn(self.path, note) + self.assertIn("/api/operations/install", note) + + async def test_cleanup_failure_is_attached_to_the_original_timeout(self): + self.disable_status = 403 + with self.assertRaises(TimeoutError) as caught: + async with asyncio.timeout(0.01): + await self.create() + self.assertIn("also failed", caught.exception.__cause__.__notes__[0]) + self.assertNotIn("DELETE", [method for method, _ in self.requests]) + + async def test_repeated_cancellation_waits_for_cleanup_to_finish(self): + self.block_delete = True + task = asyncio.create_task(self.create()) + await asyncio.wait_for(self.poll_started.wait(), 1) + task.cancel() + await asyncio.wait_for(self.delete_started.wait(), 1) + task.cancel() + await asyncio.sleep(0) + self.assertFalse(task.done()) + self.release_delete.set() + with self.assertRaises(asyncio.CancelledError): + await asyncio.wait_for(task, 1) + self.assertEqual(self.requests[-1], ("GET", "/api/operations/delete")) + + async def test_cleanup_option_rejects_updates_and_unpolled_creates(self): + for options in ({"method": "PATCH"}, {"wait": False}): + with self.subTest(options=options), self.assertRaises(ValueError): + await self.create(**options) + self.assertEqual(self.requests, []) + + async def test_precreated_resources_are_returned_without_mutation(self): + await fixtures.create_or_get_test_workspace("Automatic", True, pre_created_workspace_id="existing") + await fixtures.create_or_get_test_workpace_service("/workspaces/existing", "token", "existing-service", True) + self.assertEqual(self.requests, []) diff --git a/e2e_tests/unit_tests/test_airlock_request.py b/e2e_tests/unit_tests/test_airlock_request.py index 8046385b6..205a42f2f 100644 --- a/e2e_tests/unit_tests/test_airlock_request.py +++ b/e2e_tests/unit_tests/test_airlock_request.py @@ -2,9 +2,10 @@ import json import unittest -from unittest.mock import patch +from unittest.mock import AsyncMock, MagicMock, patch from httpx import AsyncClient, MockTransport, Response +from azure.core.exceptions import ResourceNotFoundError, ResourceExistsError, HttpResponseError from e2e_tests import helpers from e2e_tests.airlock import request @@ -51,5 +52,60 @@ async def test_link_response_does_not_log_sas(self): self.assertNotIn("test-secret-sas", "\n".join(logs.output)) +class DraftDeletionTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.client = AsyncMock() + self.context = MagicMock() + self.context.__aenter__ = AsyncMock(return_value=self.client) + self.context.__aexit__ = AsyncMock(return_value=False) + self.factory = self.enterContext( + patch.object(request.AsyncBlobClient, "from_blob_url", return_value=self.context) + ) + self.sas = "https://storage.example.test/draft?sig=private-sas" + + def missing(self, code="ContainerNotFound"): + error = ResourceNotFoundError("Not found") + error.error_code = code + return error + + async def test_transient_writes_are_retried_until_container_not_found(self): + self.client.upload_blob.side_effect = [{"etag": "written"}, self.missing()] + await request.wait_for_draft_container_deletion(self.sas, interval=0) + self.assertEqual(self.client.upload_blob.await_count, 2) + self.assertTrue(self.client.upload_blob.call_args.kwargs["overwrite"]) + self.assertIn("/draft/revocation-probe-", self.factory.call_args.args[0]) + + async def test_container_being_deleted_is_retried_and_not_accepted_as_complete(self): + error = ResourceExistsError("Deleting") + error.error_code = "ContainerBeingDeleted" + self.client.upload_blob.side_effect = [error, self.missing()] + await request.wait_for_draft_container_deletion(self.sas, interval=0) + self.assertEqual(self.client.upload_blob.await_count, 2) + + async def test_writes_that_continue_past_deadline_fail(self): + self.client.upload_blob.return_value = {"etag": "still writable"} + with self.assertRaisesRegex(AssertionError, "not confirmed within"): + await request.wait_for_draft_container_deletion(self.sas, timeout=0.01, interval=0.001) + + async def test_conflict_authentication_and_wrong_not_found_errors_fail(self): + for error in ( + ResourceExistsError("BlobAlreadyExists"), + HttpResponseError("Forbidden"), + self.missing("BlobNotFound"), + ): + with self.subTest(error=error): + self.client.upload_blob.side_effect = error + with self.assertRaises(type(error)): + await request.wait_for_draft_container_deletion(self.sas, interval=0) + + async def test_probe_names_are_distinct_and_client_closes(self): + self.client.upload_blob.side_effect = self.missing() + await request.wait_for_draft_container_deletion(self.sas) + first = self.factory.call_args.args[0] + await request.wait_for_draft_container_deletion(self.sas) + self.assertNotEqual(first, self.factory.call_args.args[0]) + self.assertEqual(self.context.__aexit__.await_count, 2) + + if __name__ == "__main__": unittest.main() diff --git a/e2e_tests/unit_tests/test_fixture_cleanup.py b/e2e_tests/unit_tests/test_fixture_cleanup.py index 9341d5836..6669fdb67 100644 --- a/e2e_tests/unit_tests/test_fixture_cleanup.py +++ b/e2e_tests/unit_tests/test_fixture_cleanup.py @@ -23,6 +23,20 @@ from pathlib import Path import pytest from e2e_tests import conftest as fixtures +from contextlib import asynccontextmanager + +@asynccontextmanager +async def nexus_prerequisites(verify): + if settings.get("record_nexus"): + event(["nexus", "ready"]) + try: + yield + finally: + if settings.get("record_nexus"): + event(["nexus", "delete"]) + +fixtures.nexus_prerequisites = nexus_prerequisites + settings = json.loads(Path("settings.json").read_text()) fixtures.CLEANUP_TIMEOUT_SECONDS = settings["budget"] @@ -41,6 +55,8 @@ async def auth_details(*args, **kwargs): return "offline-workspace-token", {} async def create_service(*args, **kwargs): + if settings.get("fail_service_setup"): + raise RuntimeError("SERVICE_SETUP_FAILED") return "/workspaces/review/workspace-services/service", "service" async def delete_resource(resource_path, *args): @@ -73,6 +89,7 @@ async def delete_resource(resource_path, *args): fixtures.config.TEST_AIRLOCK_IMPORT_REVIEW_WORKSPACE_SERVICE_ID = "" # Register the actual fixtures and hooks, with only Azure calls replaced. +setup_nexus_prerequisites = fixtures.setup_nexus_prerequisites setup_test_workspace = fixtures.setup_test_workspace setup_test_workspace_and_guacamole_service = fixtures.setup_test_workspace_and_guacamole_service setup_test_aad_workspace = fixtures.setup_test_aad_workspace @@ -123,19 +140,24 @@ def run_pytest( budget=1, durations=None, failure_resource=None, + record_nexus=False, + fail_service_setup=False, slow_report=False, ): settings = { "budget": budget, "durations": durations or {"review": 0.2, "research": 0.2}, "failure_resource": failure_resource, + "record_nexus": record_nexus, + "fail_service_setup": fail_service_setup, "slow_report": slow_report, } with tempfile.TemporaryDirectory() as directory: folder = Path(directory) (folder / "conftest.py").write_text(textwrap.dedent(FIXTURE_STUBS)) (folder / "test_cases.py").write_text( - "import time\nimport pytest\nimport asyncio\nfrom conftest import event\n" + test_source + "import time\nimport pytest\nimport asyncio\nfrom conftest import event\nfrom e2e_tests.timeouts import async_test_timeout\n" + + test_source ) (folder / "settings.json").write_text(json.dumps(settings)) environment = os.environ.copy() @@ -194,6 +216,12 @@ def test_slow_cleanup_does_not_inherit_last_test_budget_in_either_order(self): self.assertEqual(len(report.findall(".//testcase")), 2) self.assert_sequential_cleanup(events) + def test_nexus_outlives_review_workspace_cleanup(self): + result, _, events = self.run_pytest(RESOURCE_TEST, record_nexus=True) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertLess(events.index(["nexus", "ready"]), events.index(["review", "start"])) + self.assertLess(events.index(["review", "complete"]), events.index(["nexus", "delete"])) + def test_restored_timer_does_not_interrupt_teardown_error_reporting(self): result, report, events = self.run_pytest(RESOURCE_TEST, failure_resource="review", slow_report=True) self.assertEqual(result.returncode, 1, result.stdout + result.stderr) @@ -202,6 +230,15 @@ def test_restored_timer_does_not_interrupt_teardown_error_reporting(self): self.assertNotIn("INTERNALERROR", result.stdout) self.assertIn(["research", "complete"], events) + def test_service_setup_failure_still_cleans_up_review_workspace_before_nexus(self): + source = RESOURCE_TEST.replace("@pytest.mark.timeout(0.1)", "@pytest.mark.timeout(1, func_only=True)") + result, report, events = self.run_pytest(source, record_nexus=True, fail_service_setup=True) + self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + self.assertIn("SERVICE_SETUP_FAILED", result.stdout) + self.assertEqual(len(report.findall(".//error")), 1) + self.assertLess(events.index(["review", "complete"]), events.index(["nexus", "delete"])) + self.assertIn(["research", "complete"], events) + def test_cleanup_timeout_finishes_cancellation_before_next_fixture(self): # Keep the independent cleanup deadline short, but leave enough time # for pytest to format its expected error on slower CI runners. @@ -260,6 +297,36 @@ def test_body_timeout(): self.assertIn("Timeout >0.05s", result.stdout) self.assert_sequential_cleanup(events) + def test_async_body_timeout_cancels_polling_before_next_test_and_cleanup(self): + source = ( + RESOURCE_TEST + + """ +@pytest.mark.asyncio(loop_scope="session") +@pytest.mark.timeout(1) +@async_test_timeout(0.02) +async def test_async_timeout(): + try: + await asyncio.sleep(10) + event(["body", "leaked"]) + finally: + await asyncio.sleep(0.03) + event(["body", "cancel_complete"]) + +@pytest.mark.asyncio(loop_scope="session") +async def test_after_timeout(): + event(["next", "start"]) + await asyncio.sleep(0.05) +""" + ) + result, report, events = self.run_pytest(source) + self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + self.assertEqual(len(report.findall(".//failure")), 1) + self.assertEqual(len(report.findall(".//error")), 0) + self.assertIn("TimeoutError", result.stdout) + self.assertEqual(events[:2], [["body", "cancel_complete"], ["next", "start"]]) + self.assert_sequential_cleanup(events[2:]) + self.assertNotIn("Task exception was never retrieved", result.stdout + result.stderr) + def test_setup_timeout_remains_error_and_session_cleanup_finishes(self): source = ( RESOURCE_TEST diff --git a/e2e_tests/unit_tests/test_guacamole_linuxvm_cleanup.py b/e2e_tests/unit_tests/test_guacamole_linuxvm_cleanup.py index 17831a41f..c174446ba 100644 --- a/e2e_tests/unit_tests/test_guacamole_linuxvm_cleanup.py +++ b/e2e_tests/unit_tests/test_guacamole_linuxvm_cleanup.py @@ -10,14 +10,14 @@ from e2e_tests import helpers from e2e_tests import test_guacamole_linuxvm as linuxvm -from e2e_tests.resources import resource, strings +from e2e_tests.resources import nexus, resource, strings class LinuxVmCleanupTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.enterContext(patch.object(helpers.config, "TRE_URL", "https://tre.example.test")) self.enterContext(patch.object(linuxvm, "get_workspace_owner_token", new=AsyncMock(return_value="token"))) - self.enterContext(patch.object(linuxvm, "get_admin_token", new=AsyncMock(return_value="admin-token"))) + self.enterContext(patch.object(nexus, "get_admin_token", new=AsyncMock(return_value="admin-token"))) self.fixture = ("/workspaces/ws", "ws", "/workspaces/ws/workspace-services/service", "service") self.resource_path = self.fixture[2] + "/user-resources/linux" self.requests = [] @@ -35,8 +35,12 @@ def mock_api( disable_status=202, disable_outcome=strings.RESOURCE_STATUS_UPDATED, delete_outcome=strings.RESOURCE_STATUS_DELETED, + resource_disabled=True, ): + install_polls = 0 + def handle(request): + nonlocal install_polls if request.url.path == "/api/shared-services": return Response( 200, @@ -52,6 +56,8 @@ def handle(request): }, ) self.requests.append(request) + if request.method == "GET" and request.url.path == f"/api{self.resource_path}": + return Response(200, json={"userResource": {"isEnabled": not resource_disabled}}) phase = request.url.path.rsplit("/", 1)[-1] response_status = 200 if request.method == "POST": @@ -65,6 +71,10 @@ def handle(request): else: self.assertEqual(request.method, "GET") state = {"install": outcome, "disable": disable_outcome, "delete": delete_outcome}[phase] + if phase == "install": + install_polls += 1 + if isinstance(state, Exception) and install_polls > 1: + state = strings.RESOURCE_STATUS_DEPLOYMENT_FAILED if isinstance(state, Exception): raise state return Response( @@ -86,9 +96,14 @@ def create_client(**_): self.enterContext(patch.object(resource, "AsyncClient", side_effect=create_client)) - def assert_deleted(self): + def assert_deleted(self, failed_disable=False, failed_create=False): self.assertEqual( - [request.method for request in self.requests], ["POST", "GET", "PATCH", "GET", "DELETE", "GET"] + [request.method for request in self.requests], + ["POST", "GET"] + + (["GET"] if failed_create else []) + + ["PATCH", "GET"] + + (["GET"] if failed_disable else []) + + ["DELETE", "GET"], ) self.assertEqual(self.requests[-2].url.path, f"/api{self.resource_path}") @@ -103,16 +118,27 @@ async def test_failed_bootstrap_cleans_up_after_failed_disable(self): ) with self.assertRaises(AssertionError): await linuxvm.test_create_guacamole_linux_vm(self.fixture, True) - self.assert_deleted() + self.assert_deleted(failed_disable=True, failed_create=True) - async def test_polling_error_still_cleans_up(self): + async def test_failed_disable_does_not_delete_an_enabled_resource(self): + self.mock_api( + strings.RESOURCE_STATUS_DEPLOYMENT_FAILED, + disable_outcome=strings.RESOURCE_STATUS_UPDATING_FAILED, + resource_disabled=False, + ) + with self.assertRaises(AssertionError) as caught: + await linuxvm.test_create_guacamole_linux_vm(self.fixture, True) + self.assertIn("not confirmed disabled", str(caught.exception.__notes__)) + self.assertNotIn("DELETE", [request.method for request in self.requests]) + + async def test_transient_polling_error_waits_for_terminal_state_before_cleanup(self): # Use a non-retried error to verify cleanup without delaying the test. failure = RuntimeError("poll failed") self.mock_api(failure) with self.assertRaises(RuntimeError) as caught: await linuxvm.test_create_guacamole_linux_vm(self.fixture, True) self.assertIs(caught.exception, failure) - self.assert_deleted() + self.assert_deleted(failed_create=True) async def test_cleanup_failure_preserves_bootstrap_error(self): failure = RuntimeError("bootstrap failed") @@ -121,7 +147,7 @@ async def test_cleanup_failure_preserves_bootstrap_error(self): await linuxvm.test_create_guacamole_linux_vm(self.fixture, True) self.assertIs(caught.exception, failure) self.assertIn("also failed", caught.exception.__notes__[0]) - self.assert_deleted() + self.assert_deleted(failed_create=True) async def test_cleanup_failure_fails_a_successful_bootstrap(self): self.mock_api(strings.RESOURCE_STATUS_DEPLOYED, delete_outcome=strings.RESOURCE_STATUS_DELETING_FAILED) @@ -140,7 +166,7 @@ async def test_rejected_disable_does_not_delete(self): with self.assertRaises(AssertionError) as caught: await linuxvm.test_create_guacamole_linux_vm(self.fixture, True) self.assertIn("also failed", caught.exception.__notes__[0]) - self.assertEqual([request.method for request in self.requests], ["POST", "GET", "PATCH"]) + self.assertEqual([request.method for request in self.requests], ["POST", "GET", "GET", "PATCH"]) async def test_successful_bootstrap_still_requires_successful_disable(self): self.mock_api(strings.RESOURCE_STATUS_DEPLOYED, disable_outcome=strings.RESOURCE_STATUS_UPDATING_FAILED) diff --git a/e2e_tests/unit_tests/test_linux_vm_nexus.py b/e2e_tests/unit_tests/test_linux_vm_nexus.py index d50050953..303e33a9c 100644 --- a/e2e_tests/unit_tests/test_linux_vm_nexus.py +++ b/e2e_tests/unit_tests/test_linux_vm_nexus.py @@ -1,15 +1,18 @@ """Test Nexus prerequisites, explicit consent and ownership of cleanup resources.""" +import asyncio import json import unittest from pathlib import Path +from types import SimpleNamespace from unittest.mock import AsyncMock, patch from httpx import AsyncClient, MockTransport, Response -from jsonschema import validate +from jsonschema import ValidationError, validate -from e2e_tests import helpers -from e2e_tests import test_guacamole_linuxvm as linuxvm +from api_app.services.schema_service import enrich_shared_service_template +from e2e_tests import helpers, conftest as fixtures +from e2e_tests.resources import nexus from e2e_tests.resources import resource, strings @@ -25,9 +28,13 @@ def setUp(self): ), strings.CERTS_SHARED_SERVICE: json.loads((templates / "certs/template_schema.json").read_text()), } + self.schemas = { + name: enrich_shared_service_template(SimpleNamespace(model_dump=lambda schema=schema, **_: schema)) + for name, schema in self.schemas.items() + } self.enterContext(patch.object(helpers.config, "TRE_URL", "https://tre.example.test")) - self.enterContext(patch.object(linuxvm, "get_admin_token", new=AsyncMock(return_value="admin-token"))) - self.enterContext(patch.object(linuxvm.config, "TEST_ACCEPT_NEXUS_EULA", True)) + self.enterContext(patch.object(nexus, "get_admin_token", new=AsyncMock(return_value="admin-token"))) + self.enterContext(patch.object(nexus.config, "TEST_ACCEPT_NEXUS_EULA", True)) self.enterContext( patch.object( resource, "AsyncClient", side_effect=lambda **_: AsyncClient(transport=MockTransport(self.handle)) @@ -38,6 +45,8 @@ def handle(self, request): self.requests.append(request) if request.method == "GET" and request.url.path == strings.API_SHARED_SERVICES: return Response(200, json={"sharedServices": self.services}) + if request.method == "GET" and request.url.path in ("/api/shared-services/nexus", "/api/shared-services/certs"): + return Response(200, json={"sharedService": {"isEnabled": False}}) response_status = 200 if request.method == "POST": payload = json.loads(request.content) @@ -45,6 +54,7 @@ def handle(self, request): kind = "nexus" if payload["templateName"] == strings.NEXUS_SHARED_SERVICE else "certs" if kind == "nexus": self.assertIs(payload["properties"]["accept_nexus_eula"], True) + self.assertEqual(payload["properties"]["vm_size"], "Standard_D2s_v3") phase, response_status, state = "install", 202, "deploying" elif request.method == "GET": kind, phase = request.url.path.rsplit("/", 1)[-1].split("-") @@ -96,22 +106,22 @@ def deleted(self): async def test_existing_nexus_is_reused_without_consent_or_mutation(self): self.services = [self.nexus()] - with patch.object(linuxvm.config, "TEST_ACCEPT_NEXUS_EULA", False): - async with linuxvm.linux_vm_nexus(True): + with patch.object(nexus.config, "TEST_ACCEPT_NEXUS_EULA", False): + async with nexus.nexus_prerequisites(True): pass self.assertEqual([request.method for request in self.requests], ["GET"]) async def test_missing_nexus_requires_explicit_consent_before_creation(self): - with patch.object(linuxvm.config, "TEST_ACCEPT_NEXUS_EULA", False): + with patch.object(nexus.config, "TEST_ACCEPT_NEXUS_EULA", False): with self.assertRaisesRegex(AssertionError, "explicitly accept"): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.fail("Prerequisites should not be ready") self.assertEqual([request.method for request in self.requests], ["GET"]) async def test_ambiguous_existing_nexus_is_not_mutated(self): self.services = [self.nexus(), self.nexus()] with self.assertRaisesRegex(AssertionError, "at most one Nexus"): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.fail("Ambiguous ownership must stop setup") self.assertEqual([request.method for request in self.requests], ["GET"]) @@ -121,12 +131,12 @@ async def test_old_or_failed_nexus_is_not_silently_reused_or_replaced(self): self.services = [self.nexus(**changes)] self.requests.clear() with self.assertRaises(AssertionError): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.fail("Unsuitable Nexus should be rejected") self.assertEqual([request.method for request in self.requests], ["GET"]) async def test_created_resources_are_deleted_in_reverse_dependency_order(self): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.assertEqual(self.deleted(), []) posts = [json.loads(request.content)["templateName"] for request in self.requests if request.method == "POST"] self.assertEqual(posts, [strings.CERTS_SHARED_SERVICE, strings.NEXUS_SHARED_SERVICE]) @@ -134,25 +144,38 @@ async def test_created_resources_are_deleted_in_reverse_dependency_order(self): async def test_existing_certificate_service_is_preserved(self): self.services = [self.certs()] - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): pass posts = [json.loads(request.content) for request in self.requests if request.method == "POST"] self.assertEqual(len(posts), 1) self.assertEqual(posts[0]["properties"]["ssl_cert_name"], "existing-cert") self.assertEqual(self.deleted(), ["nexus"]) + async def test_both_prerequisites_include_inherited_required_properties(self): + async with nexus.nexus_prerequisites(True): + pass + posts = [json.loads(request.content) for request in self.requests if request.method == "POST"] + self.assertEqual(len(posts), 2) + for payload in posts: + with self.subTest(template=payload["templateName"]): + schema = self.schemas[payload["templateName"]] + validate(payload["properties"], schema) + del payload["properties"]["description"] + with self.assertRaisesRegex(ValidationError, "'description' is a required property"): + validate(payload["properties"], schema) + async def test_unrelated_certificate_service_is_not_replaced(self): # The API allows only one active service per template. self.services = [self.certs(properties={"domain_prefix": "other", "cert_name": "unrelated"})] with self.assertRaisesRegex(AssertionError, "belongs to another domain"): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.fail("An unrelated certificate service must be preserved") self.assertEqual([request.method for request in self.requests], ["GET"]) async def test_failed_existing_certificate_service_requires_repair(self): self.services = [self.certs(deploymentStatus="deployment_failed")] with self.assertRaisesRegex(AssertionError, "Repair the existing Nexus certificate service"): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.fail("A failed certificate service must not be reused") self.assertEqual([request.method for request in self.requests], ["GET"]) @@ -160,14 +183,14 @@ async def test_failed_nexus_deployment_cleans_up_both_owned_resources(self): self.outcomes[("nexus", "install")] = "deployment_failed" self.outcomes[("nexus", "disable")] = "updating_failed" with self.assertRaises(AssertionError): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.fail("A failed Nexus install must not be used") self.assertEqual(self.deleted(), ["nexus", "certs"]) async def test_failed_certificate_deployment_does_not_create_nexus(self): self.outcomes[("certs", "install")] = "deployment_failed" with self.assertRaises(AssertionError): - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): self.fail("A failed certificate install must stop setup") posts = [json.loads(request.content)["templateName"] for request in self.requests if request.method == "POST"] self.assertEqual(posts, [strings.CERTS_SHARED_SERVICE]) @@ -177,11 +200,44 @@ async def test_vm_failure_preserved_while_shared_resources_are_cleaned(self): failure = RuntimeError("VM bootstrap failed") bootstrap_vm = AsyncMock(side_effect=failure) with self.assertRaisesRegex(RuntimeError, "VM bootstrap failed") as caught: - async with linuxvm.linux_vm_nexus(True): + async with nexus.nexus_prerequisites(True): await bootstrap_vm() self.assertIs(caught.exception, failure) self.assertEqual(self.deleted(), ["nexus", "certs"]) + async def test_prompt_nexus_delete_failure_retains_its_certificate(self): + owned = nexus.nexus_prerequisites(True) + await owned.__aenter__() + self.services = [self.nexus(deploymentStatus="deleting_failed")] + self.outcomes[("nexus", "delete")] = "deleting_failed" + with self.assertRaises(AssertionError) as caught: + await owned.__aexit__(None, None, None) + self.assertEqual(self.deleted(), ["nexus"]) + self.assertIn("Retaining certificate dependency", str(caught.exception.__notes__)) + + async def test_certificate_lookup_failure_retains_the_dependency(self): + owned = nexus.nexus_prerequisites(True) + await owned.__aenter__() + with patch.object(nexus, "get_resource", AsyncMock(side_effect=RuntimeError("lookup failed"))): + with self.assertRaisesRegex(RuntimeError, "lookup failed"): + await owned.__aexit__(None, None, None) + self.assertEqual(self.deleted(), ["nexus"]) + + async def test_nested_cleanup_stops_at_one_deadline_and_retains_certificate(self): + owned = nexus.nexus_prerequisites(True) + await owned.__aenter__() + self.outcomes[("nexus", "disable")] = "updating" + with patch.object(fixtures, "CLEANUP_TIMEOUT_SECONDS", 0.01): + with self.assertRaises(TimeoutError): + async with fixtures.resource_cleanup_timeout("Nexus prerequisites"): + await asyncio.wait_for(owned.__aexit__(None, None, None), 0.5) + mutations = [(request.method, request.url.path) for request in self.requests if request.method != "GET"] + self.assertEqual(mutations[-1], ("PATCH", "/api/shared-services/nexus")) + self.assertNotIn(("PATCH", "/api/shared-services/certs"), mutations) + count = len(self.requests) + await asyncio.sleep(0.01) + self.assertEqual(len(self.requests), count) + if __name__ == "__main__": unittest.main() diff --git a/e2e_tests/unit_tests/test_review_vm_cleanup.py b/e2e_tests/unit_tests/test_review_vm_cleanup.py new file mode 100644 index 000000000..84790c8ac --- /dev/null +++ b/e2e_tests/unit_tests/test_review_vm_cleanup.py @@ -0,0 +1,252 @@ +"""Exercise review VM ownership after creation and during asynchronous deletion.""" + +import asyncio +import unittest +from unittest.mock import AsyncMock, patch + +from httpx import AsyncClient, MockTransport, Response +from e2e_tests import helpers, test_airlock as airlock +from e2e_tests.resources import resource + + +class ReviewVmCleanupTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.requests = [] + self.state = "deployed" + self.active_operation = False + self.path = "/workspaces/review/workspace-services/guac/user-resources/vm" + self.enterContext(patch.object(helpers.config, "TRE_URL", "https://tre.example.test")) + self.enterContext(patch.object(airlock, "post_resource", AsyncMock(return_value=(self.path, "vm")))) + self.enterContext( + patch.object(airlock, "delete_owned_resource_if_present", resource.delete_owned_resource_if_present) + ) + self.enterContext( + patch.object( + resource, "AsyncClient", side_effect=lambda **_: AsyncClient(transport=MockTransport(self.handle)) + ) + ) + + async def handle(self, request): + self.requests.append((request.method, request.url.path)) + self.assertEqual(request.headers["Authorization"], "Bearer review-token") + if request.method == "GET" and request.url.path == f"/api{self.path}": + if self.state == "absent": + return Response(404) + return Response(200, json={"userResource": {"deploymentStatus": self.state}}) + if request.url.path.endswith("/operations"): + return Response(200, json={"operations": [{"status": "deleting" if self.active_operation else "deployed"}]}) + if request.method == "PATCH": + phase, code, state = "disable", 202, "updating" + elif request.method == "DELETE": + phase, code, state = "delete", 200, "deleting" + else: + phase, code = request.url.path.rsplit("/", 1)[-1], 200 + state = {"disable": "updated", "delete": "deleted"}[phase] + return Response( + code, + headers={"Location": f"/api/operations/{phase}"}, + json={ + "operation": { + "resourceId": "vm", + "status": state, + "message": "test", + "steps": [], + } + }, + ) + + def owner(self): + return airlock.managed_review_vm("/workspaces/research", "request", "research-token", "review-token", True) + + async def test_successful_creation_is_cleaned_after_approval_failure(self): + error = RuntimeError("approval failed") + approve = AsyncMock(side_effect=error) + with self.assertRaises(RuntimeError) as caught: + async with self.owner() as owned: + self.assertEqual(owned, self.path) + await approve() + self.assertIs(caught.exception, error) + self.assertIn(("DELETE", f"/api{self.path}"), self.requests) + + async def test_body_cancellation_waits_for_child_cleanup(self): + with self.assertRaises(TimeoutError): + async with asyncio.timeout(0.01): + async with self.owner(): + await asyncio.sleep(1) + self.assertEqual(self.requests[-1], ("GET", "/api/operations/delete")) + + async def test_normal_airlock_deletion_does_not_send_a_second_delete(self): + self.state = "absent" + async with self.owner(): + pass + self.assertEqual(self.requests, [("GET", f"/api{self.path}")]) + + async def test_active_approval_deletion_finishes_before_cleanup_returns(self): + self.state, self.active_operation = "updating", True + + async def finish_operation(seconds): + self.assertFalse(any(method in ("PATCH", "DELETE") for method, _ in self.requests)) + self.state = "absent" + + with patch.object(resource.asyncio, "sleep", finish_operation): + async with self.owner(): + pass + self.assertEqual([method for method, _ in self.requests], ["GET", "GET", "GET"]) + + async def test_cleanup_failure_preserves_the_original_body_failure(self): + original = RuntimeError("approval failed") + approve = AsyncMock(side_effect=original) + with patch.object(airlock, "delete_owned_resource_if_present", AsyncMock(side_effect=TimeoutError("stalled"))): + with self.assertRaises(RuntimeError) as caught: + async with self.owner(): + await approve() + self.assertIs(caught.exception, original) + self.assertIn(self.path, str(caught.exception.__notes__)) + + async def test_outer_timeout_during_recovery_preserves_body_failure(self): + started, finished = asyncio.Event(), asyncio.Event() + original = TimeoutError("automatic deletion stalled") + + async def cleanup(*args): + started.set() + await asyncio.sleep(0.03) + finished.set() + + @airlock.async_test_timeout(0.01) + async def run(): + async with self.owner(): + raise original + + with patch.object(airlock, "delete_owned_resource_if_present", cleanup): + with self.assertRaises(TimeoutError) as caught: + await run() + self.assertTrue(started.is_set()) + self.assertTrue(finished.is_set()) + self.assertIs(caught.exception, original) + + async def test_repeated_cancellation_during_successful_exit_waits_for_cleanup(self): + started, release, finished = asyncio.Event(), asyncio.Event(), asyncio.Event() + + async def cleanup(*args): + started.set() + await release.wait() + finished.set() + + async def run(): + async with self.owner(): + pass + + with patch.object(airlock, "delete_owned_resource_if_present", cleanup): + task = asyncio.create_task(run()) + try: + await asyncio.wait_for(started.wait(), 1) + for _ in range(2): + task.cancel() + await asyncio.sleep(0) + self.assertFalse(task.done()) + release.set() + with self.assertRaises(asyncio.CancelledError): + await asyncio.wait_for(task, 1) + self.assertTrue(finished.is_set()) + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + + async def test_cleanup_deadline_after_outer_timeout_preserves_body_failure(self): + original = RuntimeError("approval failed") + stopped = asyncio.Event() + + async def cleanup(*args): + try: + await asyncio.Event().wait() + finally: + stopped.set() + + @airlock.async_test_timeout(0.01) + async def run(): + async with self.owner(): + raise original + + with ( + patch.object(airlock, "delete_owned_resource_if_present", cleanup), + patch.object(airlock, "cleanup_deadline", side_effect=lambda _: resource.cleanup_deadline(0.03)), + ): + with self.assertRaises(RuntimeError) as caught: + await run() + self.assertIs(caught.exception, original) + self.assertTrue(stopped.is_set()) + self.assertIn("TimeoutError", str(original.__notes__)) + + async def test_cleanup_deadline_failure_is_reported_after_successful_body(self): + with ( + patch.object(airlock, "delete_owned_resource_if_present", AsyncMock(side_effect=TimeoutError("stalled"))), + self.assertRaisesRegex(TimeoutError, "stalled"), + ): + async with self.owner(): + pass + + async def test_cancelled_recovery_preserves_the_original_failure(self): + original = RuntimeError("approval failed") + with patch.object(airlock, "delete_owned_resource_if_present", AsyncMock(side_effect=asyncio.CancelledError)): + with self.assertRaises(RuntimeError) as caught: + async with self.owner(): + raise original + self.assertIs(caught.exception, original) + self.assertIn("CancelledError", str(original.__notes__)) + + +class ReviewVmDeletionTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.enterContext(patch.object(helpers.config, "TRE_URL", "https://tre.example.test")) + self.requests = [] + + async def run_states(self, states, **kwargs): + remaining = iter(states) + + def respond(request): + self.requests.append(request.method) + self.assertEqual(request.headers["Authorization"], "Bearer review-token") + state = next(remaining) + if isinstance(state, int): + return Response(state) + return Response(200, json={"userResource": {"deploymentStatus": state}}) + + with ( + patch.object(airlock, "AsyncClient", side_effect=lambda **_: AsyncClient(transport=MockTransport(respond))), + patch.object(airlock.asyncio, "sleep", AsyncMock()), + ): + await airlock.wait_for_review_vm_deletion( + "/workspaces/review/user-resources/vm", "review-token", True, **kwargs + ) + + async def test_fast_deletion_does_not_require_observing_updating(self): + for states in ([404], ["awaiting_deletion", "deleting", 404], ["deleted"]): + with self.subTest(states=states): + self.requests.clear() + await self.run_states(states) + self.assertEqual(self.requests, ["GET"] * len(states)) + + async def test_waits_through_disable_and_delete_progress(self): + states = ["deployed", "awaiting_update", "updating", "updated", "awaiting_deletion", "deleting", 404] + await self.run_states(states) + self.assertEqual(len(self.requests), len(states)) + + async def test_failed_deletion_is_not_reported_as_success(self): + for state in ("updating_failed", "deleting_failed", "pipeline_failed"): + with self.subTest(state=state): + with self.assertRaisesRegex(AssertionError, state): + await self.run_states([state]) + + async def test_server_error_is_not_treated_as_deleted(self): + with self.assertRaises(AssertionError): + await self.run_states([500]) + + async def test_stalled_deletion_has_a_finite_deadline(self): + def respond(request): + return Response(200, json={"userResource": {"deploymentStatus": "deleting"}}) + + with patch.object( + airlock, "AsyncClient", side_effect=lambda **_: AsyncClient(transport=MockTransport(respond)) + ): + with self.assertRaisesRegex(TimeoutError, "Last state: deleting"): + await airlock.wait_for_review_vm_deletion("/vm", "review-token", True, timeout_seconds=0.01) diff --git a/e2e_tests/unit_tests/test_review_workspace_ownership.py b/e2e_tests/unit_tests/test_review_workspace_ownership.py new file mode 100644 index 000000000..fe14eb957 --- /dev/null +++ b/e2e_tests/unit_tests/test_review_workspace_ownership.py @@ -0,0 +1,76 @@ +"""Check ownership when an Airlock review workspace is reused.""" + +import unittest +from unittest.mock import AsyncMock, patch + +from e2e_tests import conftest as fixtures + + +class ReviewWorkspaceOwnershipTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.enterContext(patch.object(fixtures.config, "TEST_AIRLOCK_IMPORT_REVIEW_WORKSPACE_ID", "existing")) + self.enterContext(patch.object(fixtures.config, "TEST_AIRLOCK_IMPORT_REVIEW_WORKSPACE_SERVICE_ID", "")) + self.enterContext( + patch.object( + fixtures, "create_or_get_test_workspace", AsyncMock(return_value=("/workspaces/existing", "existing")) + ) + ) + self.enterContext(patch.object(fixtures, "get_admin_token", AsyncMock(return_value="admin"))) + self.enterContext(patch.object(fixtures, "get_workspace_auth_details", AsyncMock(return_value=("owner", {})))) + self.create_service = self.enterContext( + patch.object( + fixtures, + "create_or_get_test_workpace_service", + AsyncMock(return_value=("/workspaces/existing/workspace-services/new", "new")), + ) + ) + self.delete_service = self.enterContext(patch.object(fixtures, "disable_and_delete_ws_resource", AsyncMock())) + self.delete_workspace = self.enterContext( + patch.object(fixtures, "disable_and_delete_tre_resource", AsyncMock()) + ) + + def fixture(self): + return fixtures.setup_test_airlock_import_review_workspace_and_guacamole_service.__wrapped__(None, True, None) + + async def finish(self, fixture): + with self.assertRaises(StopAsyncIteration): + await anext(fixture) + + async def test_created_child_is_removed_and_reused_parent_is_preserved(self): + fixture = self.fixture() + await anext(fixture) + await self.finish(fixture) + self.delete_service.assert_awaited_once_with("/workspaces/existing/workspace-services/new", "existing", True) + self.delete_workspace.assert_not_awaited() + + async def test_reused_child_is_preserved(self): + with patch.object(fixtures.config, "TEST_AIRLOCK_IMPORT_REVIEW_WORKSPACE_SERVICE_ID", "existing-child"): + fixture = self.fixture() + await anext(fixture) + await self.finish(fixture) + self.delete_service.assert_not_awaited() + self.delete_workspace.assert_not_awaited() + + async def test_owned_parent_uses_cascading_cleanup(self): + with patch.object(fixtures.config, "TEST_AIRLOCK_IMPORT_REVIEW_WORKSPACE_ID", ""): + fixture = self.fixture() + await anext(fixture) + await self.finish(fixture) + self.delete_service.assert_not_awaited() + self.delete_workspace.assert_awaited_once_with("/workspaces/existing", True) + + async def test_failed_child_cleanup_still_calls_parent_cleanup(self): + self.delete_service.side_effect = RuntimeError("child delete failed") + with patch.object(fixtures, "clean_up_test_workspace", AsyncMock()) as parent: + fixture = self.fixture() + await anext(fixture) + with self.assertRaisesRegex(RuntimeError, "child delete failed"): + await anext(fixture) + parent.assert_awaited_once() + + async def test_failed_child_setup_does_not_delete_reused_workspace(self): + self.create_service.side_effect = RuntimeError("service setup failed") + with self.assertRaisesRegex(RuntimeError, "service setup failed"): + await anext(self.fixture()) + self.delete_service.assert_not_awaited() + self.delete_workspace.assert_not_awaited()