diff --git a/eng/tools/azure-sdk-tools/devtools_testutils/proxy_startup.py b/eng/tools/azure-sdk-tools/devtools_testutils/proxy_startup.py index a5f560a4714d..0967e36ade6c 100644 --- a/eng/tools/azure-sdk-tools/devtools_testutils/proxy_startup.py +++ b/eng/tools/azure-sdk-tools/devtools_testutils/proxy_startup.py @@ -344,7 +344,11 @@ def set_common_sanitizers() -> None: # General regex sanitizers for sensitive patterns throughout interactions batch_sanitizers[Sanitizer.GENERAL_REGEX] = [ - {"regex": '(?:[\\?&](sig|se|st|sv)=)(?[^&\\"\\s]*)', "group_for_replace": "secret", "value": SANITIZED}, + { + "regex": r'(?:[?&](sig|se|st|sv)=)(?(?:[^&"\\\s]|\\+[^"\\\s&])*)', + "group_for_replace": "secret", + "value": SANITIZED, + }, ] # Header regex sanitizers for sensitive patterns in request/response headers diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/assets.json b/sdk/platformvalidation/azure-mgmt-platformvalidation/assets.json new file mode 100644 index 000000000000..69d58aeff4eb --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/assets.json @@ -0,0 +1,6 @@ +{ + "AssetsRepo": "Azure/azure-sdk-assets", + "AssetsRepoPrefixPath": "python", + "TagPrefix": "python/platformvalidation/azure-mgmt-platformvalidation", + "Tag": "python/platformvalidation/azure-mgmt-platformvalidation_71c39c2214" +} diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/dev_requirements.txt b/sdk/platformvalidation/azure-mgmt-platformvalidation/dev_requirements.txt index ece056fe0984..9aaf96bebf46 100644 --- a/sdk/platformvalidation/azure-mgmt-platformvalidation/dev_requirements.txt +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/dev_requirements.txt @@ -2,4 +2,5 @@ ../../core/azure-core ../../identity/azure-identity ../../core/azure-mgmt-core +../../resources/azure-mgmt-resource aiohttp \ No newline at end of file diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/README.md b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/README.md new file mode 100644 index 000000000000..ec1ee2ae8eac --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/README.md @@ -0,0 +1,212 @@ +# PlatformValidation recorded integration tests + +These handwritten tests follow the [Python SDK testing guide](https://github.com/Azure/azure-sdk-for-python/blob/main/doc/dev/tests.md) +and [management-plane testing guide](https://github.com/Azure/azure-sdk-for-python/blob/main/doc/dev/mgmt/tests.md). +Do not edit regeneration-owned files in `generated_tests/`. + +## Verified coverage + +Twenty recorded scenarios, ten per client, exercise all 25 SDK methods in both +sync and async clients. Mocked regressions separately exercise ownership, cleanup, +timeouts, serialization and sanitization. Method coverage does not mean every +input, authorization boundary or deployment has been tested. + +| Operation group | Methods exercised per client | Main assertions | +| --- | --- | --- | +| `cloud_validations` | `begin_create_or_update`, `get`, `begin_update`, `begin_delete`, `list_by_resource_group`, `list_by_subscription` | Identity, location, state, tags, description, existing-resource PUT, missing GET and post-delete 404 | +| `validation_execution_plans` | `begin_create_or_update`, `get`, `begin_update`, `begin_delete`, `list_by_resource_group` | Tags/description/configuration PATCH, omitted-field preservation, existing-resource PUT, lists and post-delete 404 | +| `execution_plan_runs` | `begin_create_or_update`, `get`, `begin_delete`, `list_by_execution_plan` | New run completes with one passing boot test; deletion removes the run and its test results | +| `validation_test_runs` | `get`, `list_by_execution_plan_run` | Exact parent identity, completed state, pass counts, timestamps and absence of failure details | +| `validation_tests` | `get`, `list_by_subscription` | Public entry discovery and exact subscription-scoped identity | +| `validation_test_versions` | `get`, `list` | Public version discovery and exact parent/version identity | +| `validation_test_categories` | `get`, `list_by_subscription` | Public category discovery and exact identity | +| `operation_status` | `get` | Direct polling of the supported route returned by a plan PATCH | +| `operations` | `list` | Expected permission names, display metadata and data-action flags | + +Existing-resource PUT is covered for CloudValidation and execution plans. +`execution_plan_runs.begin_create_or_update` is covered by creation of a new run, +not replay of an existing completed run. + +Catalog tests, categories and child versions follow the subscription-scoped ARM +contract. Expected IDs use the framework's configured subscription: +`/subscriptions/{subscriptionId}/providers/Microsoft.PlatformValidation/...`. +The current deployment and published recordings instead return provider-global +IDs beginning `/providers/Microsoft.PlatformValidation/`. Both recorded catalog +scenarios therefore fail strict identity assertions, reproducing the service +contract mismatch observed in Go and JavaScript. No fallback or expected-failure +masking is applied. Refresh and publish recordings after the service is corrected; +do not rewrite recorded responses to make the assertions pass. +Description-only PATCH tests verify that omitted tags and configuration are preserved. + +Run serially; do not re-submit an existing execution merely to test its +update-shaped SDK entrypoint. + +## Setup and authentication + +From this package directory, use an isolated virtual environment and install +the repository tooling, test dependencies and local SDK: + +```powershell +python -m pip install -r dev_requirements.txt -e . +``` + +Follow the repository's [feed configuration](https://github.com/Azure/azure-sdk-for-python/blob/main/CONTRIBUTING.md#package-index-configuration) +and [test authentication guide](https://github.com/Azure/azure-sdk-for-python/blob/main/doc/dev/tests.md#configure-test-variables). +Use an approved subscription with `Microsoft.PlatformValidation` registered, +a supported location, and permission to create/delete test resource groups and +exercise the listed operations. The resource-management SDK is a test dependency, +not a runtime dependency of PlatformValidation. + +For Azure CLI authentication, sign in with `az login`, then configure: + +```powershell +$env:AZURE_SUBSCRIPTION_ID = "" +$env:PLATFORMVALIDATION_LOCATION = "" +$env:AZURE_TEST_USE_CLI_AUTH = "true" +$env:AZURE_TEST_USE_PWSH_AUTH = "false" +$env:AZURE_TEST_USE_AZD_AUTH = "false" +Remove-Item Env:\AZURE_RESOURCEGROUP_NAME -ErrorAction SilentlyContinue +``` + +Other documented authentication modes remain available. With the environment +loader, service-principal inputs are `PLATFORMVALIDATION_TENANT_ID`, +`PLATFORMVALIDATION_CLIENT_ID` and `PLATFORMVALIDATION_CLIENT_SECRET`. +Never commit credentials or local target configuration. + +`AzureMgmtRecordedTestCase` creates framework-managed clients and selects +live/fake credentials. Async tests use `get_credential(..., is_async=True)` and +`create_client_from_credential`, and close their credentials. + +## Execution prerequisites and cleanup + +The full workload lifecycle uses this preparer order: + +```python +@ExecutionEnvironmentPreparer() +@ExecutionResourceGroupPreparer( + name_prefix="platformvalidationexec", random_name_enabled=True, location="eastus" +) +@ExecutionPrerequisitesPreparer() +@recorded_by_proxy +def test_execution_lifecycle(self, platformvalidation_location, resource_group, cloud_validation_name): + ... +``` + +The async scenario uses the same preparers and `recorded_by_proxy_async`. +Preparers create a fresh parent RG and CloudValidation outside recordings. +The service must create `-mrg`; setup verifies its `managedBy` association. +There is no manual MRG creation, role assignment or fallback. +Playback supplies fake prerequisites and creates no Azure resources. + +Cleanup order is EPR and cascaded VTRs, VEP, CV, service-managed MRG disappearance, +then the owned parent RG. Exact IDs, ownership tags, run descriptions and terminal +states are required. Unexpected siblings, active resources and orphan results +block deletion. The parent is retained if prerequisite cleanup is unconfirmed, +even when ARM's generic resource list is empty. + +CV deletion retries only HTTP 409 `CannotDeleteResource`, at most 30 attempts +ten seconds apart, checking that the plan list is empty before each attempt. +Other errors still fail. Managed-group absence requires three consecutive checks, +with at most 60 checks ten seconds apart; playback skips sleeps. +Managed groups are never deleted directly. + +Plan-only/update scenarios borrow an approved, successfully provisioned CV. +Read-only result scenarios additionally need a completed, one-test boot run: + +```powershell +$env:PLATFORMVALIDATION_EXECUTION_RESOURCE_GROUP = "" +$env:PLATFORMVALIDATION_EXECUTION_CLOUD_VALIDATION = "" +$env:PLATFORMVALIDATION_EXECUTION_OWNERSHIP = "borrowed-parent" +$env:PLATFORMVALIDATION_EXECUTION_PLAN = "" +$env:PLATFORMVALIDATION_EXECUTION_RUN = "" +``` + +Borrowed-parent tests never create/delete the CV, MRG or parent group. Update +tests create and delete only their own new plans and submit no executions. +Read-only result tests create or delete nothing. + +Creation tests also require `PLATFORMVALIDATION_SOURCE_VHD_URI`: supply an +approved HTTPS Linux Gen1 X64 image through a secure process environment. +Do not place a signed URI in source, shell history, `.env`, logs or recordings. +Playback uses `https://example.invalid/image.vhd` and needs no image access. +The approved fixture uses `Standard_D4s_v3` and one +`compute-vmimagevalidation-vm-boot-test` step. + +The embedded configuration uses `microsoft.validate/executionPlan.v0` and +`ExecutionPlan`, separately from ARM API version `2026-08-01-preview`. + +## Record, replay and publish + +The following maps the official guide to this suite: + +| Guide step | Implementation or command | +| --- | --- | +| Install tooling and local library | `python -m pip install -r dev_requirements.txt -e .` | +| Select framework credentials | Base test case and environment loaders; no handwritten credential construction | +| Prepare auxiliary resources | Framework preparers, excluded from recordings | +| Record actual service traffic | `recorded_by_proxy` / `recorded_by_proxy_async` and the `test_proxy` fixture | +| Sanitize secrets | Package-scoped fixtures plus image-input and list-response sanitizers | +| Replay the same tests | Set `AZURE_TEST_RUN_LIVE=false`; run `python -m pytest tests -q` | +| Publish updates | Supported `manage_recordings.py push`; include its generated `assets.json` | +| Verify contribution | Replay the full included suite from a clean checkout using published assets; inspect PR CI | + +Start with a single provisioning scenario: + +```powershell +$env:AZURE_TEST_RUN_LIVE = "true" +$env:AZURE_SKIP_LIVE_RECORDING = "false" +python -m pytest tests\test_cloud_validations.py::TestCloudValidations::test_cloud_validation_lifecycle -v --maxfail=1 +``` + +Then record its async equivalent, fresh execution lifecycles and other configured +scenarios serially. Catalog live tests additionally accept +`PLATFORMVALIDATION_CATALOG_ENDPOINT` and `PLATFORMVALIDATION_CATALOG_TEST`. +The approved retest used regional Canary ARM and the boot catalog entry; +these are environment settings, not public SDK endpoint defaults. + +Stop and investigate live failures. Before recording, locate the cache and +preserve unpublished changes: + +```powershell +python ..\..\..\scripts\manage_recordings.py locate +``` + +A failed live test can replace its cached recording. Retain failure evidence +privately and restore that scenario's last-known-good asset, if any. +Do not fabricate recordings, hand-edit failures into successes or publish +failed recordings as passing coverage. + +After successful recording: + +```powershell +$env:AZURE_TEST_RUN_LIVE = "false" +python -m pytest tests -q +``` + +The included suite needs no Azure credentials, image URI or real prerequisites +in playback. Inspect recordings for secrets and valid nested JSON before +publication. Sanitizers remove cookies/private headers, redact signed query +values and identity fields, replace image URLs, and redact unrelated resources +in subscription lists while preserving the target identity. The package replaces +the generic SAS sanitizer with an escape-preserving expression. The shared +SDK-tools SAS expression also preserves closing JSON escapes; otherwise it can +corrupt embedded documents before package sanitizers run. Mocked regressions +cover both expressions with nested JSON, escaped slashes and Unicode escapes. + +This package is already migrated to external recording assets. Use the +[recording update workflow](https://github.com/Azure/azure-sdk-for-python/blob/main/doc/dev/tests.md#update-test-recordings), +not initial migration: + +```powershell +python ..\..\..\scripts\manage_recordings.py push +``` + +This requires asset-repository write access and generates a new tag pointer in +`assets.json`. Verify the tag remotely and replay from a clean checkout without +the old cache or local shared-tool modifications. Commit the generated pointer +with the tests, not recordings, credentials or private diagnostics. + +ResourceGroupPreparer can request deletion without waiting. Independently +confirm live parent-group cleanup before declaring a run complete. +Passing playback is evidence for recorded behavior, not a repeatability or +concurrent-live-run claim. diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_cloud_validation_testcase.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_cloud_validation_testcase.py new file mode 100644 index 000000000000..4f6d9ac19ccb --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_cloud_validation_testcase.py @@ -0,0 +1,133 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from functools import partial +import asyncio +import json +import logging +import time + +from azure.mgmt.platformvalidation.models import CloudValidation +from azure.mgmt.resource.resources import ResourceManagementClient +from azure.mgmt.resource.resources.aio import ResourceManagementClient as AsyncResourceManagementClient +from devtools_testutils import ( + AzureMgmtRecordedTestCase, + EnvironmentVariableLoader, + add_body_key_sanitizer, + add_body_regex_sanitizer, +) + +PlatformValidationPreparer = partial( + EnvironmentVariableLoader, + "platformvalidation", + platformvalidation_location="eastus", +) + +OPERATION_TIMEOUT = 600 +MANAGED_GROUP_CHECKS = 60 +MANAGED_GROUP_CHECK_INTERVAL = 10 +MANAGED_GROUP_ABSENCE_CONFIRMATIONS = 3 +_LOGGER = logging.getLogger(__name__) + + +def wait_for_managed_group_deletion(client: ResourceManagementClient, name: str) -> None: + absent = 0 + for attempt in range(MANAGED_GROUP_CHECKS): + absent = 0 if client.resource_groups.check_existence(name) else absent + 1 + if absent == MANAGED_GROUP_ABSENCE_CONFIRMATIONS: + return + if attempt + 1 < MANAGED_GROUP_CHECKS: + time.sleep(MANAGED_GROUP_CHECK_INTERVAL) + raise AssertionError( + f"Managed resource group {name} cleanup was not confirmed. No explicit RG deletion was attempted." + ) + + +async def wait_for_managed_group_deletion_async(client: AsyncResourceManagementClient, name: str) -> None: + absent = 0 + for attempt in range(MANAGED_GROUP_CHECKS): + absent = 0 if await client.resource_groups.check_existence(name) else absent + 1 + if absent == MANAGED_GROUP_ABSENCE_CONFIRMATIONS: + return + if attempt + 1 < MANAGED_GROUP_CHECKS: + await asyncio.sleep(MANAGED_GROUP_CHECK_INTERVAL) + raise AssertionError( + f"Managed resource group {name} cleanup was not confirmed. No explicit RG deletion was attempted." + ) + + +class CloudValidationTestCase(AzureMgmtRecordedTestCase): + def sanitize_subscription_list(self, name: str) -> None: + condition = json.dumps( + {"uriRegex": r"(?i)/subscriptions/[^/]+/providers/Microsoft.PlatformValidation/cloudValidations(?:\?|$)"} + ) + other = f"$.value[?(@.name != '{name}')]" + prefix = "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/other-group" + for path, value in ( + (".id", f"{prefix}/providers/Microsoft.PlatformValidation/cloudValidations/other-cloud-validation"), + (".location", "eastus"), + (".properties.description", "Sanitized"), + ( + ".properties.managedOnBehalfOfConfiguration.moboBrokerResources[*].id", + f"{prefix}/providers/Microsoft.Resources/deployments/other", + ), + (".name", "other-cloud-validation"), + ): + add_body_key_sanitizer(json_path=other + path, value=value, condition=condition, function_scoped=True) + # Tag dictionaries can contain tenant-specific keys as well as values. + add_body_regex_sanitizer( + regex=r'"tags"\s*:\s*\{(?:\s*"(?:\\.|[^"\\])*"\s*:\s*"(?:\\.|[^"\\])*"\s*,?)*\s*\}', + value='"tags":{}', + condition=condition, + function_scoped=True, + ) + + def log_live_target(self, resource_group: str, name: str, location: str) -> None: + if self.is_live: + _LOGGER.info( + "CloudValidation live test: subscription=%s, resource_group=%s, name=%s, " + "location=%s, managed_resource_group=%s-mrg", + self.get_settings_value("SUBSCRIPTION_ID"), + resource_group, + name, + location, + name, + ) + + def resource_id(self, resource_group: str, name: str) -> str: + subscription = self.get_settings_value("SUBSCRIPTION_ID") + return ( + f"/subscriptions/{subscription}/resourceGroups/{resource_group}" + f"/providers/Microsoft.PlatformValidation/cloudValidations/{name}" + ) + + def assert_identity(self, resource: CloudValidation, resource_group: str, name: str) -> None: + assert resource.id is not None + assert resource.id.lower() == self.resource_id(resource_group, name).lower() + assert resource.name == name + assert resource.type is not None + assert resource.type.lower() == "microsoft.platformvalidation/cloudvalidations" + + def assert_resource( + self, + resource: CloudValidation, + resource_group: str, + name: str, + location: str, + description: str, + tags: dict[str, str], + ) -> None: + self.assert_identity(resource, resource_group, name) + assert resource.location == location + assert resource.tags == tags + assert resource.properties is not None + assert resource.properties.description == description + assert resource.properties.provisioning_state == "Succeeded" + + def assert_list_contains_resource(self, resources: list[CloudValidation], resource_group: str, name: str) -> None: + resource_id = self.resource_id(resource_group, name).lower() + matches = [resource for resource in resources if resource.id and resource.id.lower() == resource_id] + assert len(matches) == 1 + self.assert_identity(matches[0], resource_group, name) diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_execution_prerequisites.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_execution_prerequisites.py new file mode 100644 index 000000000000..14da79e98fca --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_execution_prerequisites.py @@ -0,0 +1,218 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from contextlib import ExitStack +from functools import partial +import logging +import time +from typing import TypeVar + +from azure.core.exceptions import AzureError, ResourceExistsError, ResourceNotFoundError +from azure.core.polling import LROPoller +from azure.mgmt.platformvalidation import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.models import CloudValidation, CloudValidationProperties +from azure.mgmt.resource.resources import ResourceManagementClient +from devtools_testutils import AzureMgmtPreparer, EnvironmentVariableLoader, ResourceGroupPreparer +from devtools_testutils import add_general_string_sanitizer + +from _cloud_validation_testcase import OPERATION_TIMEOUT, wait_for_managed_group_deletion + +_LOGGER = logging.getLogger(__name__) +_ResultT = TypeVar("_ResultT") +CV_DELETE_ATTEMPTS = 30 +CV_DELETE_INTERVAL = 10 + +ExecutionEnvironmentPreparer = partial( + EnvironmentVariableLoader, + "platformvalidation", + platformvalidation_location="eastus", +) + + +def complete_prerequisite_operation(poller: LROPoller[_ResultT]) -> _ResultT: + result = poller.result(timeout=OPERATION_TIMEOUT) + if not poller.done(): + raise TimeoutError("Execution prerequisite operation did not finish within the timeout.") + return result + + +class ExecutionResourceGroupPreparer(ResourceGroupPreparer): + """Keep the parent group when child cleanup fails; never cascade-delete a plan.""" + + def remove_resource(self, name, **kwargs): + if not self.is_live or not self._need_creation: + return + group = self._prefix_name(name) + try: + cleaned_name = getattr(self.test_class_instance, "_cleaned_execution_cv", None) + attempted = getattr(self.test_class_instance, "_execution_cv_attempted", False) + assert ( + not attempted or cleaned_name + ), f"Refusing to delete {group}: CloudValidation prerequisite cleanup was not confirmed." + remaining = list(self.client.resources.list_by_resource_group(group)) + for resource in remaining: + expected_id = ( + f"/subscriptions/{self.test_class_instance.get_settings_value('SUBSCRIPTION_ID')}" + f"/resourceGroups/{group}/providers/Microsoft.PlatformValidation/cloudValidations/{cleaned_name}" + ) + assert ( + cleaned_name and resource.id and resource.id.lower() == expected_id.lower() + ), f"Refusing to delete {group}: resources remain after execution prerequisite cleanup." + # ARM's generic listing may lag a completed provider DELETE. + # Recheck only this test's verified-deleted CV, never arbitrary children. + with self.create_mgmt_client(PlatformValidationMgmtClient) as cv: + try: + cv.cloud_validations.get(group, cleaned_name) + except ResourceNotFoundError as error: + if error.status_code != 404: + raise + else: + raise AssertionError(f"Refusing to delete {group}: CloudValidation still exists.") + complete_prerequisite_operation(self.client.resource_groups.begin_delete(group)) + wait_for_managed_group_deletion(self.client, group) + finally: + self.client.close() + + +class ExecutionPrerequisitesPreparer(AzureMgmtPreparer): + """Create an isolated CV and verify its service-managed MRG without permission changes.""" + + def __init__(self): + super().__init__("pyexec", 24, disable_recording=True, random_name_enabled=True) + self.resource_moniker = "executioncv" + self._clients = ExitStack() + self._cv_attempted = False + + def create_resource(self, name, **kwargs): + if not self.is_live: + return {"cloud_validation_name": name} + # Reset per invocation; decorator instances can be reused by the test runner. + self._clients = ExitStack() + self._cv_attempted = False + self.test_class_instance._execution_cv_attempted = False + self.test_class_instance._cleaned_execution_cv = None + self.name = name + self.group = kwargs["resource_group"].name + self.location = kwargs["platformvalidation_location"] + self.subscription = self.test_class_instance.get_settings_value("SUBSCRIPTION_ID") + self.managed_group = f"{name}-mrg" + self.tags = {"sdk-test": name} + add_general_string_sanitizer(target=name, value=self.moniker) + _LOGGER.warning( + "Execution prerequisites: subscription=%s, group=%s, cv=%s, managed_group=%s (service-managed)", + self.subscription, + self.group, + name, + self.managed_group, + ) + try: + self.resources = self._clients.enter_context(self.create_mgmt_client(ResourceManagementClient)) + self.cv = self._clients.enter_context(self.create_mgmt_client(PlatformValidationMgmtClient)) + if self.resources.resource_groups.check_existence(self.managed_group): + raise AssertionError("Refusing to adopt an existing managed group.") + try: + self.cv.cloud_validations.get(self.group, name) + except ResourceNotFoundError as error: + if error.status_code != 404: + raise + else: + raise AssertionError("Refusing to overwrite an existing CloudValidation.") + self._cv_attempted = True + self.test_class_instance._execution_cv_attempted = True + result = complete_prerequisite_operation( + self.cv.cloud_validations.begin_create_or_update( + self.group, + name, + CloudValidation( + location=self.location, + tags=self.tags, + properties=CloudValidationProperties(description="SDK execution prerequisites"), + ), + ) + ) + assert result.properties.provisioning_state == "Succeeded" + assert self.resources.resource_groups.check_existence( + self.managed_group + ), "Service-managed MRG is absent; no fallback to manual provisioning." + managed = self.resources.resource_groups.get(self.managed_group) + expected_id = ( + f"/subscriptions/{self.subscription}/resourceGroups/{self.group}" + f"/providers/Microsoft.PlatformValidation/cloudValidations/{name}" + ) + assert ( + managed.managed_by and managed.managed_by.lower() == expected_id.lower() + ), "Service-managed MRG is not associated with this test's CloudValidation." + return {"cloud_validation_name": name} + except (AzureError, AssertionError, TimeoutError, ValueError): + # Framework preparers do not run remove_resource when create_resource fails. + try: + if self._cv_attempted: + self._cleanup() + except (AzureError, AssertionError, TimeoutError): + _LOGGER.exception("Prerequisite cleanup failed; retain the logged resource names for recovery.") + finally: + self._clients.close() + raise + + def _cleanup(self) -> None: + try: + parent = self.cv.cloud_validations.get(self.group, self.name) + except ResourceNotFoundError as error: + if error.status_code != 404: + raise + parent = None + if parent is not None: + expected = ( + f"/subscriptions/{self.subscription}/resourceGroups/{self.group}" + f"/providers/Microsoft.PlatformValidation/cloudValidations/{self.name}" + ) + assert parent.id and parent.id.lower() == expected.lower() + assert ( + parent.tags and parent.tags.get("sdk-test") == self.name + ), "Refusing to delete a CloudValidation without this test's ownership tag." + assert parent.properties.provisioning_state in ( + "Succeeded", + "Failed", + "Canceled", + ), "Refusing cleanup while CloudValidation provisioning is nonterminal." + if parent is not None: + self._delete_cloud_validation() + try: + self.cv.cloud_validations.get(self.group, self.name) + except ResourceNotFoundError as error: + if error.status_code != 404: + raise + else: + raise AssertionError("CloudValidation still exists after cleanup.") + wait_for_managed_group_deletion(self.resources, self.managed_group) + if parent is not None: + self.test_class_instance._cleaned_execution_cv = self.name + + def _delete_cloud_validation(self) -> None: + for attempt in range(CV_DELETE_ATTEMPTS): + assert not list( + self.cv.validation_execution_plans.list_by_resource_group(self.group, self.name) + ), "Execution plans remain: delete EPRs and VEPs before removing prerequisites." + try: + complete_prerequisite_operation(self.cv.cloud_validations.begin_delete(self.group, self.name)) + return + except ResourceExistsError as error: + code = getattr(getattr(error, "error", None), "code", None) + if error.status_code != 409 or code != "CannotDeleteResource" or attempt + 1 == CV_DELETE_ATTEMPTS: + raise + _LOGGER.warning( + "CV %s deletion still reports nested resources after an empty plan list; retrying (%s/%s).", + self.name, + attempt + 1, + CV_DELETE_ATTEMPTS, + ) + time.sleep(CV_DELETE_INTERVAL) + + def remove_resource(self, name, **kwargs): + if self.is_live: + try: + self._cleanup() + finally: + self._clients.close() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_execution_testcase.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_execution_testcase.py new file mode 100644 index 000000000000..9848e2ed5bf7 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_execution_testcase.py @@ -0,0 +1,315 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import asyncio +from functools import partial, wraps +import inspect +import json +import os +import time +from urllib.parse import urlsplit + +from azure.core.exceptions import AzureError, ResourceNotFoundError +from devtools_testutils import EnvironmentVariableLoader, add_general_string_sanitizer + +from _cloud_validation_testcase import CloudValidationTestCase + + +ExistingExecutionPreparer = partial( + EnvironmentVariableLoader, + "platformvalidation", + platformvalidation_location="eastus", + platformvalidation_execution_resource_group="execution-parent", + platformvalidation_execution_cloud_validation="executioncv", + platformvalidation_execution_ownership="borrowed-parent", +) +ExistingRunPreparer = partial( + EnvironmentVariableLoader, + "platformvalidation", + platformvalidation_execution_resource_group="execution-parent", + platformvalidation_execution_cloud_validation="executioncv", + platformvalidation_execution_plan="existing-plan", + platformvalidation_execution_run="existing-run", + platformvalidation_execution_ownership="borrowed-parent", +) +IMAGE_PLACEHOLDER = "https://example.invalid/image.vhd" +EXECUTION_TIMEOUT = 1800 +TERMINAL_RUN_STATES = { + "Succeeded", + "Completed", + "Failed", + "Canceled", + "Cancelled", + "TimedOut", +} +TERMINAL_PROVISIONING_STATES = {"Succeeded", "Failed", "Canceled"} + + +def protect_execution_errors(function): + """Do not let service errors echo credential-bearing embedded documents.""" + if inspect.iscoroutinefunction(function): + + @wraps(function) + async def async_wrapper(*args, **kwargs): + try: + return await function(*args, **kwargs) + except AzureError as error: + raise RuntimeError( + f"Execution service request failed: HTTP {getattr(error, 'status_code', None)} " + f"({type(error).__name__})." + ) from None + except Exception as error: + # A cleanup failure must not expose an earlier service-error context. + raise error from None + + # Preparers use getfullargspec to remove their internal keyword arguments. + async_wrapper.__signature__ = inspect.signature(function) + return async_wrapper + + @wraps(function) + def wrapper(*args, **kwargs): + try: + return function(*args, **kwargs) + except AzureError as error: + raise RuntimeError( + f"Execution service request failed: HTTP {getattr(error, 'status_code', None)} " + f"({type(error).__name__})." + ) from None + except Exception as error: + raise error from None + + wrapper.__signature__ = inspect.signature(function) + return wrapper + + +def plan_configuration(name, source): + return json.dumps( + { + "apiVersion": "microsoft.validate/executionPlan.v0", + "kind": "ExecutionPlan", + "metadata": {"name": name}, + "parameters": { + "certificationPackageReference": { + "osType": "Linux", + "vmGenerationType": "V1", + "architectureType": "X64", + "recommendedVMSizes": ["Standard_D4s_v3"], + "storageProfile": { + "osDiskImage": {"sourceVhdUri": source}, + "dataDiskImages": [], + }, + "additionalProperties": {}, + } + }, + "authoring": { + "steps": [ + { + "name": "vm-boot-test", + "type": "test", + "testRef": "/providers/Microsoft.PlatformValidation/validationTests/" + "compute-vmimagevalidation-vm-boot-test", + } + ] + }, + }, + separators=(",", ":"), + ) + + +def image_source(is_live): + if not is_live: + return IMAGE_PLACEHOLDER + source = os.environ.get("PLATFORMVALIDATION_SOURCE_VHD_URI") + if not source or urlsplit(source).scheme != "https": + raise ValueError("An approved HTTPS PLATFORMVALIDATION_SOURCE_VHD_URI is required.") + add_general_string_sanitizer(target=source, value=IMAGE_PLACEHOLDER) + return source + + +def operation_status_target(response, target): + headers = response.http_response.headers + url = headers.get("Azure-AsyncOperation") or headers.get("Operation-Location") + if not url or target: + return + # Never retain signed query parameters, or attempt another endpoint's route. + parts = urlsplit(url).path.strip("/").split("/") + lower = [part.lower() for part in parts] + if "locations" in lower and len(parts) > 2 and lower[-2] == "operationstatuses": + target.extend((parts[lower.index("locations") + 1], parts[-1])) + + +def complete(poller): + result = poller.result(timeout=EXECUTION_TIMEOUT) + if not poller.done(): + raise TimeoutError("Execution operation did not finish; do not delete active resources.") + return result + + +async def complete_async(poller): + result = await asyncio.wait_for(poller.result(), timeout=EXECUTION_TIMEOUT) + if not poller.done(): + raise TimeoutError("Execution operation did not finish; do not delete active resources.") + return result + + +def assert_missing(operation, *args): + try: + operation(*args) + except ResourceNotFoundError as error: + assert error.status_code == 404 + else: + raise AssertionError("Expected HTTP 404.") + + +async def assert_missing_async(operation, *args): + try: + await operation(*args) + except ResourceNotFoundError as error: + assert error.status_code == 404 + else: + raise AssertionError("Expected HTTP 404.") + + +def assert_owned_plan(plan, expected_id, name): + assert plan.id and plan.id.lower() == expected_id.lower() + assert plan.tags and plan.tags.get("sdk-test") == name, "Refusing to delete an unowned plan." + assert plan.properties.provisioning_state in TERMINAL_PROVISIONING_STATES + + +def assert_deletable_run(run, expected_id, description): + assert run.id and run.id.lower() == expected_id.lower() + assert run.properties.description == description, "Refusing to delete an unowned run." + assert run.properties.provisioning_state in TERMINAL_PROVISIONING_STATES + assert run.properties.status in TERMINAL_RUN_STATES, "Execution is still active; retain its parent." + + +def cleanup_execution(client, args, plan_id, run_name, description): + try: + plan = client.validation_execution_plans.get(*args) + except ResourceNotFoundError as error: + assert error.status_code == 404 + return + assert_owned_plan(plan, plan_id, args[-1]) + runs = list(client.execution_plan_runs.list_by_execution_plan(*args)) + for run in runs: + assert run.name == run_name, "Unexpected run; refusing parent cascade." + assert_deletable_run(run, f"{plan_id}/executionPlanRuns/{run_name}", description) + for run in runs: + run_args = (*args, run.name) + tests = list(client.validation_test_runs.list_by_execution_plan_run(*run_args)) + complete(client.execution_plan_runs.begin_delete(*run_args)) + assert_missing(client.execution_plan_runs.get, *run_args) + for test in tests: + assert_missing(client.validation_test_runs.get, *run_args, test.name) + if not runs: + try: + orphans = list(client.validation_test_runs.list_by_execution_plan_run(*args, run_name)) + except ResourceNotFoundError as error: + assert error.status_code == 404 + orphans = [] + assert not orphans, "Orphan test runs remain; refusing parent cascade." + assert not list(client.execution_plan_runs.list_by_execution_plan(*args)) + complete(client.validation_execution_plans.begin_delete(*args)) + assert_missing(client.validation_execution_plans.get, *args) + + +async def cleanup_execution_async(client, args, plan_id, run_name, description): + try: + plan = await client.validation_execution_plans.get(*args) + except ResourceNotFoundError as error: + assert error.status_code == 404 + return + assert_owned_plan(plan, plan_id, args[-1]) + runs = [run async for run in client.execution_plan_runs.list_by_execution_plan(*args)] + for run in runs: + assert run.name == run_name, "Unexpected run; refusing parent cascade." + assert_deletable_run(run, f"{plan_id}/executionPlanRuns/{run_name}", description) + for run in runs: + run_args = (*args, run.name) + tests = [test async for test in client.validation_test_runs.list_by_execution_plan_run(*run_args)] + await complete_async(await client.execution_plan_runs.begin_delete(*run_args)) + await assert_missing_async(client.execution_plan_runs.get, *run_args) + for test in tests: + await assert_missing_async(client.validation_test_runs.get, *run_args, test.name) + if not runs: + try: + orphans = [test async for test in client.validation_test_runs.list_by_execution_plan_run(*args, run_name)] + except ResourceNotFoundError as error: + assert error.status_code == 404 + orphans = [] + assert not orphans, "Orphan test runs remain; refusing parent cascade." + assert not [run async for run in client.execution_plan_runs.list_by_execution_plan(*args)] + await complete_async(await client.validation_execution_plans.begin_delete(*args)) + await assert_missing_async(client.validation_execution_plans.get, *args) + + +class ExecutionTestCase(CloudValidationTestCase): + def assert_parent(self, parent, group, name, ownership): + assert ownership in ("borrowed-parent", "owned-parent"), "Explicit parent ownership is required." + self.assert_identity(parent, group, name) + assert parent.properties.provisioning_state == "Succeeded" + + def assert_plan(self, plan, plan_id, name, tags, description, configuration): + assert plan.id and plan.id.lower() == plan_id.lower() + assert plan.name == name + assert plan.tags == tags + assert plan.properties.provisioning_state == "Succeeded" + assert plan.properties.description == description + # Avoid pytest assertion introspection printing credential-bearing documents. + configuration_matches = json.loads(plan.properties.plan_configuration_json) == json.loads(configuration) + assert configuration_matches, "Plan configuration changed." + + def assert_workload(self, run, run_id, description): + assert run.id and run.id.lower() == run_id.lower() + assert run.properties.description == description + assert run.properties.provisioning_state == "Succeeded" + assert run.properties.status == "Succeeded" + assert run.properties.error is None + summary = run.properties.test_run_summary + assert summary is not None + assert summary.overall_result == "Passed" + assert summary.total_tests == 1 + assert summary.passed_tests == 1 + assert summary.failed_tests == 0 + assert summary.skipped_tests == 0 + assert run.properties.started_at is not None + assert run.properties.completed_at is not None + assert run.properties.completed_at >= run.properties.started_at + + def assert_test_run(self, test, run_id): + assert test.name in ("vm-boot-test", "compute-vmimagevalidation-vm-boot-test") + assert test.id.lower() == f"{run_id}/validationTestRuns/{test.name}".lower() + assert test.properties.status == "Completed" + assert test.properties.provisioning_state == "Succeeded" + assert test.properties.test_id == ( + "/providers/Microsoft.PlatformValidation/validationTests/compute-vmimagevalidation-vm-boot-test" + ) + assert test.properties.error is None + assert test.properties.pass_details + assert not test.properties.failure_details + + +def wait_for_run(client, args): + deadline = time.monotonic() + EXECUTION_TIMEOUT + while time.monotonic() < deadline: + run = client.execution_plan_runs.get(*args) + if run.properties.status in TERMINAL_RUN_STATES: + return run + if run.properties.provisioning_state in {"Failed", "Canceled"}: + raise AssertionError("Run provisioning failed before workload completion.") + time.sleep(10) + raise TimeoutError("Workload is not terminal; retain the run and plan for investigation.") + + +async def wait_for_run_async(client, args): + deadline = time.monotonic() + EXECUTION_TIMEOUT + while time.monotonic() < deadline: + run = await client.execution_plan_runs.get(*args) + if run.properties.status in TERMINAL_RUN_STATES: + return run + if run.properties.provisioning_state in {"Failed", "Canceled"}: + raise AssertionError("Run provisioning failed before workload completion.") + await asyncio.sleep(10) + raise TimeoutError("Workload is not terminal; retain the run and plan for investigation.") diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_sanitizers.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_sanitizers.py new file mode 100644 index 000000000000..6df84ffec59b --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_sanitizers.py @@ -0,0 +1,66 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import os +import re +from urllib.parse import quote + +from devtools_testutils import ( + add_body_key_sanitizer, + add_general_regex_sanitizer, + add_general_string_sanitizer, + add_header_regex_sanitizer, + add_remove_header_sanitizer, + remove_batch_sanitizers, +) + + +def configure_sanitizers() -> None: + # Preserve resource identity assertions and LRO polling URLs, not credentials. + # Replace the generic SAS expression: it consumes JSON escape characters in + # planConfigurationJson, making credential-bearing embedded documents invalid. + remove_batch_sanitizers(["AZSDK3430", "AZSDK3493", "AZSDK2003", "AZSDK1007"]) + for variable in ( + "AZURE_SUBSCRIPTION_ID", + "AZURE_TENANT_ID", + "AZURE_CLIENT_ID", + "AZURE_CLIENT_SECRET", + "PLATFORMVALIDATION_TENANT_ID", + "PLATFORMVALIDATION_CLIENT_ID", + "PLATFORMVALIDATION_CLIENT_SECRET", + ): + value = os.environ.get(variable) + if value: + for target in {value, quote(value, safe="")}: + add_general_string_sanitizer(target=target, value="00000000-0000-0000-0000-000000000000") + # A replacement Set-Cookie can introduce a cookie that was never sent live. + add_remove_header_sanitizer( + headers="Set-Cookie,Cookie,Request-Context,x-ms-resource-provider-hint,x-ms-providerhub-traffic" + ) + # This diagnostic header can contain a tenant GUID even with CLI authentication. + add_header_regex_sanitizer(key="x-ms-operation-identifier", value="Sanitized") + location = os.environ.get("PLATFORMVALIDATION_LOCATION") + if location: + add_general_regex_sanitizer(regex=f"(?i){re.escape(location)}", value="eastus") + add_general_regex_sanitizer( + regex=r"""(?i)https://[a-z0-9.-]+\.blob\.core\.windows\.net/[^"\s<>\\]*""", + value="https://example.invalid/image.vhd", + ) + # The service returns this embedded JSON document verbatim, including image SAS. + add_body_key_sanitizer( + json_path="$..planConfigurationJson", + regex=r'("sourceVhdUri"\s*:\s*")([^"]*)', + group_for_replace="2", + value="https://example.invalid/image.vhd", + ) + # Cover SAS and signed ARM polling parameters without consuming JSON escapes. + add_general_regex_sanitizer( + regex=r"""(?i)([?&](?:c|h|s|t|sig|se|st|sv)=)((?:[^&\s"<>\\]|\\+[^"\\\s&])+)""", + group_for_replace="2", + value="Sanitized", + ) + add_body_key_sanitizer(json_path="$..access_token", value="Sanitized") + add_body_key_sanitizer(json_path="$..systemData.createdBy", value="Sanitized") + add_body_key_sanitizer(json_path="$..systemData.lastModifiedBy", value="Sanitized") diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_update_testcase.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_update_testcase.py new file mode 100644 index 000000000000..69b32573f668 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/_update_testcase.py @@ -0,0 +1,47 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import json +from urllib.parse import urlsplit + +from azure.mgmt.platformvalidation.models import ( + ValidationExecutionPlan, + ValidationExecutionPlanProperties, + ValidationExecutionPlanUpdate, + ValidationExecutionPlanUpdateProperties, +) + + +def replacement_plan(location, tags, description, configuration): + return ValidationExecutionPlan( + location=location, + tags=tags, + properties=ValidationExecutionPlanProperties(description=description, plan_configuration_json=configuration), + ) + + +def plan_update(*, tags=None, **properties): + return ValidationExecutionPlanUpdate( + **({"tags": tags} if tags is not None else {}), + properties=ValidationExecutionPlanUpdateProperties(**properties), + ) + + +def updated_configuration(configuration): + document = json.loads(configuration) + document["metadata"]["description"] = "Updated SDK recorded plan metadata" + return json.dumps(document, separators=(",", ":")) + + +def response_observation(response): + """Retain correlation evidence, never signed queries or response documents.""" + headers = response.http_response.headers + return { + "method": response.http_request.method, + "path": urlsplit(response.http_request.url).path, + "status": response.http_response.status_code, + "request_id": headers.get("x-ms-request-id"), + "correlation_id": headers.get("x-ms-correlation-request-id"), + } diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/conftest.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/conftest.py new file mode 100644 index 000000000000..5ff919da8b66 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/conftest.py @@ -0,0 +1,16 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import pytest +from dotenv import load_dotenv + +from _sanitizers import configure_sanitizers + +load_dotenv() + + +@pytest.fixture(scope="session", autouse=True) +def configure_test_proxy(test_proxy, patch_sleep, patch_async_sleep) -> None: + configure_sanitizers() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_catalog.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_catalog.py new file mode 100644 index 000000000000..269388e02039 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_catalog.py @@ -0,0 +1,115 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from functools import partial + +from azure.mgmt.platformvalidation import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.aio import PlatformValidationMgmtClient as AsyncPlatformValidationMgmtClient +from devtools_testutils import ( + EnvironmentVariableLoader, + add_body_key_sanitizer, + add_general_regex_sanitizer, + recorded_by_proxy, +) +from devtools_testutils.aio import recorded_by_proxy_async + +from _cloud_validation_testcase import CloudValidationTestCase + + +CatalogPreparer = partial( + EnvironmentVariableLoader, + "platformvalidation", + platformvalidation_catalog_endpoint="https://management.azure.com", + platformvalidation_catalog_test="compute-vmimagevalidation-vm-boot-test", +) + + +def sanitize_catalog(): + # Canary uses a regional ARM host; playback uses the public ARM placeholder. + add_general_regex_sanitizer( + regex=r"(?i)https://[a-z0-9-]+\.management\.azure\.com", + value="https://management.azure.com", + function_scoped=True, + ) + add_body_key_sanitizer(json_path="$..owners[*]", value="Sanitized", function_scoped=True) + + +class TestCatalog(CloudValidationTestCase): + def public_entries(self, entries): + public = [entry for entry in entries if entry.properties and entry.properties.audience == "Public"] + assert public, "The approved public catalog must not be empty." + return public + + def assert_entry(self, entry, resource_id, name): + assert entry.name == name + actual_id = entry.id + assert ( + actual_id and actual_id.lower() == resource_id.lower() + ), "Catalog ID differs from the canonical subscription-scoped identity." + is_public = bool(entry.properties and entry.properties.audience == "Public") + assert is_public, "The selected catalog entry must have Public audience." + + def catalog_id(self, name): + subscription_id = self.get_settings_value("SUBSCRIPTION_ID") + return f"/subscriptions/{subscription_id}/providers/Microsoft.PlatformValidation/{name}" + + @CatalogPreparer() + @recorded_by_proxy + def test_public_catalog(self, platformvalidation_catalog_endpoint, platformvalidation_catalog_test): + sanitize_catalog() + name = platformvalidation_catalog_test + test_id = self.catalog_id(f"validationTests/{name}") + with self.create_mgmt_client( + PlatformValidationMgmtClient, base_url=platformvalidation_catalog_endpoint + ) as client: + test = client.validation_tests.get(name) + tests = self.public_entries(list(client.validation_tests.list_by_subscription())) + assert sum(test.name == name for test in tests) == 1 + + versions = self.public_entries(list(client.validation_test_versions.list(name))) + version = versions[0].name + test_version = client.validation_test_versions.get(name, version) + + categories = self.public_entries(list(client.validation_test_categories.list_by_subscription())) + category = categories[0].name + test_category = client.validation_test_categories.get(category) + self.assert_entry(test, test_id, name) + self.assert_entry(test_version, f"{test_id}/versions/{version}", version) + self.assert_entry(test_category, self.catalog_id(f"validationTestCategories/{category}"), category) + + @CatalogPreparer() + @recorded_by_proxy_async + async def test_public_catalog_async(self, platformvalidation_catalog_endpoint, platformvalidation_catalog_test): + sanitize_catalog() + name = platformvalidation_catalog_test + test_id = self.catalog_id(f"validationTests/{name}") + credential = self.get_credential(AsyncPlatformValidationMgmtClient, is_async=True) + try: + async with self.create_client_from_credential( + AsyncPlatformValidationMgmtClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + base_url=platformvalidation_catalog_endpoint, + ) as client: + test = await client.validation_tests.get(name) + tests = self.public_entries([test async for test in client.validation_tests.list_by_subscription()]) + assert sum(test.name == name for test in tests) == 1 + + versions = self.public_entries( + [version async for version in client.validation_test_versions.list(name)] + ) + version = versions[0].name + test_version = await client.validation_test_versions.get(name, version) + + categories = self.public_entries( + [category async for category in client.validation_test_categories.list_by_subscription()] + ) + category = categories[0].name + test_category = await client.validation_test_categories.get(category) + self.assert_entry(test, test_id, name) + self.assert_entry(test_version, f"{test_id}/versions/{version}", version) + self.assert_entry(test_category, self.catalog_id(f"validationTestCategories/{category}"), category) + finally: + await credential.close() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_catalog_helpers.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_catalog_helpers.py new file mode 100644 index 000000000000..3cff2f12f43e --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_catalog_helpers.py @@ -0,0 +1,40 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from types import SimpleNamespace + +import pytest + +from test_catalog import TestCatalog as CatalogTestCase + + +@pytest.mark.parametrize( + "subscription_id", + ["11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222"], +) +@pytest.mark.parametrize( + "resource", + [ + "validationTests/boot", + "validationTests/boot/versions/1.0.0", + "validationTestCategories/image-certification", + ], +) +def test_catalog_identity_is_subscription_scoped(monkeypatch, resource, subscription_id): + case = CatalogTestCase() + monkeypatch.setattr(case, "get_settings_value", lambda _: subscription_id) + provider_id = f"/providers/Microsoft.PlatformValidation/{resource}" + expected = f"/subscriptions/{subscription_id}{provider_id}" + assert case.catalog_id(resource) == expected + entry = SimpleNamespace( + name=resource.rsplit("/", 1)[-1], + id=expected, + properties=SimpleNamespace(audience="Public"), + ) + case.assert_entry(entry, expected, entry.name) + for invalid_id in (provider_id, f"/subscriptions/unexpected{provider_id}"): + entry.id = invalid_id + with pytest.raises(AssertionError, match="subscription-scoped"): + case.assert_entry(entry, expected, entry.name) diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_cloud_validations.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_cloud_validations.py new file mode 100644 index 000000000000..60b8b26ec68c --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_cloud_validations.py @@ -0,0 +1,185 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import logging +from typing import TypeVar + +import pytest + +from azure.core.exceptions import ResourceNotFoundError +from azure.core.polling import LROPoller +from azure.mgmt.platformvalidation import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.models import ( + CloudValidation, + CloudValidationProperties, + CloudValidationUpdate, + CloudValidationUpdateProperties, +) +from azure.mgmt.resource.resources import ResourceManagementClient +from devtools_testutils import ResourceGroupPreparer, recorded_by_proxy + +from _cloud_validation_testcase import ( + CloudValidationTestCase, + OPERATION_TIMEOUT, + PlatformValidationPreparer, + wait_for_managed_group_deletion, +) + +_LOGGER = logging.getLogger(__name__) +_ResultT = TypeVar("_ResultT") + + +def complete_operation(poller: LROPoller[_ResultT]) -> _ResultT: + result = poller.result(timeout=OPERATION_TIMEOUT) + assert poller.done(), "CloudValidation operation did not finish within the timeout." + return result + + +class TestCloudValidations(CloudValidationTestCase): + @PlatformValidationPreparer() + @ResourceGroupPreparer(name_prefix="platformvalidation", random_name_enabled=True, location="eastus") + @recorded_by_proxy + def test_cloud_validation_put_existing(self, resource_group, platformvalidation_location: str) -> None: + from _update_testcase import response_observation + + group = resource_group.name + name = self.get_resource_name("pyputcv") + location = platformvalidation_location + managed_group = f"{name}-mrg" + tags = {"sdk-test": name} + description = "SDK isolated PUT CloudValidation" + self.update_observations = [] + + def observe(response): + self.update_observations.append(response_observation(response)) + + with ( + self.create_mgmt_client(PlatformValidationMgmtClient) as client, + self.create_mgmt_client(ResourceManagementClient) as resources, + ): + operations = client.cloud_validations + with pytest.raises(ResourceNotFoundError) as missing: + operations.get(group, name) + assert missing.value.status_code == 404 + assert not resources.resource_groups.check_existence(managed_group) + try: + for phase in ("create", "idempotent", "metadata-replacement"): + if phase == "metadata-replacement": + tags = {**tags, "phase": phase} + description = "PUT existing CloudValidation metadata" + result = complete_operation( + operations.begin_create_or_update( + group, + name, + CloudValidation( + location=location, + tags=tags, + properties=CloudValidationProperties(description=description), + ), + raw_response_hook=observe, + ) + ) + self.assert_resource(result, group, name, location, description, tags) + self.assert_resource(operations.get(group, name), group, name, location, description, tags) + finally: + try: + self.delete_owned_resource(client, group, name, allow_missing=True) + finally: + wait_for_managed_group_deletion(resources, managed_group) + + @PlatformValidationPreparer() + @ResourceGroupPreparer(name_prefix="platformvalidation", random_name_enabled=True, location="eastus") + @recorded_by_proxy + def test_cloud_validation_lifecycle(self, resource_group, platformvalidation_location: str) -> None: + group = resource_group.name + location = platformvalidation_location + name = self.get_resource_name("pysdkrec") + tags = {"sdk-test": name, "phase": "created"} + description = "Python SDK recorded CloudValidation" + + managed_group = f"{name}-mrg" + self.log_live_target(group, name, location) + with ( + self.create_mgmt_client(PlatformValidationMgmtClient) as client, + self.create_mgmt_client(ResourceManagementClient) as resources, + ): + operations = client.cloud_validations + # Refuse to overwrite a resource left behind by an earlier live run. + with pytest.raises(ResourceNotFoundError) as missing: + operations.get(group, name) + assert missing.value.status_code == 404 + assert not resources.resource_groups.check_existence( + managed_group + ), "Managed resource group already exists." + + deleted = False + try: + created = complete_operation( + operations.begin_create_or_update( + group, + name, + CloudValidation( + location=location, + tags=tags, + properties=CloudValidationProperties(description=description), + ), + ) + ) + self.assert_resource(created, group, name, location, description, tags) + self.assert_resource(operations.get(group, name), group, name, location, description, tags) + self.assert_list_contains_resource(list(operations.list_by_resource_group(group)), group, name) + self.sanitize_subscription_list(name) + self.assert_list_contains_resource(list(operations.list_by_subscription()), group, name) + + tags = {"sdk-test": name, "phase": "updated"} + description = "Updated Python SDK recorded CloudValidation" + updated = complete_operation( + operations.begin_update( + group, + name, + CloudValidationUpdate( + tags=tags, properties=CloudValidationUpdateProperties(description=description) + ), + ) + ) + self.assert_resource(updated, group, name, location, description, tags) + self.assert_resource(operations.get(group, name), group, name, location, description, tags) + + self.delete_owned_resource(client, group, name) + deleted = True + finally: + try: + if not deleted: + self.delete_owned_resource(client, group, name, allow_missing=True) + finally: + wait_for_managed_group_deletion(resources, managed_group) + + def delete_owned_resource( + self, client: PlatformValidationMgmtClient, group: str, name: str, *, allow_missing: bool = False + ) -> None: + try: + resource = client.cloud_validations.get(group, name) + except ResourceNotFoundError as error: + if not allow_missing or error.status_code != 404: + raise + _LOGGER.warning( + "Cleanup found no CloudValidation. If creation failed or timed out, check for late provisioning." + ) + return + self.assert_identity(resource, group, name) + assert resource.tags and resource.tags.get("sdk-test") == name, "Refusing to delete an unowned resource." + complete_operation(client.cloud_validations.begin_delete(group, name)) + with pytest.raises(ResourceNotFoundError) as missing: + client.cloud_validations.get(group, name) + assert missing.value.status_code == 404 + + @PlatformValidationPreparer() + @ResourceGroupPreparer(name_prefix="platformvalidation", random_name_enabled=True, location="eastus") + @recorded_by_proxy + def test_get_missing_cloud_validation_raises_not_found(self, resource_group) -> None: + with self.create_mgmt_client(PlatformValidationMgmtClient) as client: + with pytest.raises(ResourceNotFoundError) as missing: + client.cloud_validations.get(resource_group.name, self.get_resource_name("pysdkmissing")) + assert missing.value.status_code == 404 diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_cloud_validations_async.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_cloud_validations_async.py new file mode 100644 index 000000000000..6737b419e534 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_cloud_validations_async.py @@ -0,0 +1,236 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import asyncio +import logging +from contextlib import asynccontextmanager +from typing import AsyncIterator, TypeVar + +import pytest + +from azure.core.exceptions import ResourceNotFoundError +from azure.core.polling import AsyncLROPoller +from azure.mgmt.platformvalidation.aio import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.models import ( + CloudValidation, + CloudValidationProperties, + CloudValidationUpdate, + CloudValidationUpdateProperties, +) +from azure.mgmt.resource.resources.aio import ResourceManagementClient +from devtools_testutils import ResourceGroupPreparer +from devtools_testutils.aio import recorded_by_proxy_async + +from _cloud_validation_testcase import ( + CloudValidationTestCase, + OPERATION_TIMEOUT, + PlatformValidationPreparer, + wait_for_managed_group_deletion_async, +) + +_LOGGER = logging.getLogger(__name__) +_ResultT = TypeVar("_ResultT") + + +async def complete_operation(poller: AsyncLROPoller[_ResultT]) -> _ResultT: + result = await asyncio.wait_for(poller.result(), timeout=OPERATION_TIMEOUT) + assert poller.done(), "CloudValidation operation did not finish within the timeout." + return result + + +class TestCloudValidationsAsync(CloudValidationTestCase): + @PlatformValidationPreparer() + @ResourceGroupPreparer(name_prefix="platformvalidation", random_name_enabled=True, location="eastus") + @recorded_by_proxy_async + async def test_cloud_validation_put_existing(self, resource_group, platformvalidation_location: str) -> None: + from _update_testcase import response_observation + + group = resource_group.name + name = self.get_resource_name("pyaputcv") + location = platformvalidation_location + managed_group = f"{name}-mrg" + tags = {"sdk-test": name} + description = "SDK isolated PUT CloudValidation" + self.update_observations = [] + + def observe(response): + self.update_observations.append(response_observation(response)) + + credential = self.get_credential(ResourceManagementClient, is_async=True) + try: + async with ( + self.create_client() as client, + self.create_client_from_credential( + ResourceManagementClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + ) as resources, + ): + operations = client.cloud_validations + with pytest.raises(ResourceNotFoundError) as missing: + await operations.get(group, name) + assert missing.value.status_code == 404 + assert not await resources.resource_groups.check_existence(managed_group) + try: + for phase in ("create", "idempotent", "metadata-replacement"): + if phase == "metadata-replacement": + tags = {**tags, "phase": phase} + description = "PUT existing CloudValidation metadata" + result = await complete_operation( + await operations.begin_create_or_update( + group, + name, + CloudValidation( + location=location, + tags=tags, + properties=CloudValidationProperties(description=description), + ), + raw_response_hook=observe, + ) + ) + self.assert_resource(result, group, name, location, description, tags) + self.assert_resource( + await operations.get(group, name), group, name, location, description, tags + ) + finally: + try: + await self.delete_owned_resource(client, group, name, allow_missing=True) + finally: + await wait_for_managed_group_deletion_async(resources, managed_group) + finally: + await credential.close() + + @asynccontextmanager + async def create_client(self) -> AsyncIterator[PlatformValidationMgmtClient]: + credential = self.get_credential(PlatformValidationMgmtClient, is_async=True) + try: + async with self.create_client_from_credential( + PlatformValidationMgmtClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + ) as client: + yield client + finally: + await credential.close() + + @PlatformValidationPreparer() + @ResourceGroupPreparer(name_prefix="platformvalidation", random_name_enabled=True, location="eastus") + @recorded_by_proxy_async + async def test_cloud_validation_lifecycle(self, resource_group, platformvalidation_location: str) -> None: + group = resource_group.name + location = platformvalidation_location + name = self.get_resource_name("pyasdkrec") + tags = {"sdk-test": name, "phase": "created"} + description = "Async Python SDK recorded CloudValidation" + managed_group = f"{name}-mrg" + self.log_live_target(group, name, location) + credential = self.get_credential(ResourceManagementClient, is_async=True) + try: + async with ( + self.create_client() as client, + self.create_client_from_credential( + ResourceManagementClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + ) as resources, + ): + await self.exercise_lifecycle( + client, resources, group, name, location, description, tags, managed_group + ) + finally: + await credential.close() + + async def exercise_lifecycle( + self, + client: PlatformValidationMgmtClient, + resources: ResourceManagementClient, + group: str, + name: str, + location: str, + description: str, + tags: dict[str, str], + managed_group: str, + ) -> None: + operations = client.cloud_validations + with pytest.raises(ResourceNotFoundError) as missing: + await operations.get(group, name) + assert missing.value.status_code == 404 + assert not await resources.resource_groups.check_existence( + managed_group + ), "Managed resource group already exists." + + deleted = False + try: + created = await complete_operation( + await operations.begin_create_or_update( + group, + name, + CloudValidation( + location=location, + tags=tags, + properties=CloudValidationProperties(description=description), + ), + ) + ) + self.assert_resource(created, group, name, location, description, tags) + self.assert_resource(await operations.get(group, name), group, name, location, description, tags) + self.assert_list_contains_resource( + [resource async for resource in operations.list_by_resource_group(group)], group, name + ) + self.sanitize_subscription_list(name) + self.assert_list_contains_resource( + [resource async for resource in operations.list_by_subscription()], group, name + ) + + tags = {"sdk-test": name, "phase": "updated"} + description = "Updated async Python SDK recorded CloudValidation" + updated = await complete_operation( + await operations.begin_update( + group, + name, + CloudValidationUpdate( + tags=tags, properties=CloudValidationUpdateProperties(description=description) + ), + ) + ) + self.assert_resource(updated, group, name, location, description, tags) + self.assert_resource(await operations.get(group, name), group, name, location, description, tags) + + await self.delete_owned_resource(client, group, name) + deleted = True + finally: + try: + if not deleted: + await self.delete_owned_resource(client, group, name, allow_missing=True) + finally: + await wait_for_managed_group_deletion_async(resources, managed_group) + + async def delete_owned_resource( + self, client: PlatformValidationMgmtClient, group: str, name: str, *, allow_missing: bool = False + ) -> None: + try: + resource = await client.cloud_validations.get(group, name) + except ResourceNotFoundError as error: + if not allow_missing or error.status_code != 404: + raise + _LOGGER.warning( + "Cleanup found no CloudValidation. If creation failed or timed out, check for late provisioning." + ) + return + self.assert_identity(resource, group, name) + assert resource.tags and resource.tags.get("sdk-test") == name, "Refusing to delete an unowned resource." + await complete_operation(await client.cloud_validations.begin_delete(group, name)) + with pytest.raises(ResourceNotFoundError) as missing: + await client.cloud_validations.get(group, name) + assert missing.value.status_code == 404 + + @PlatformValidationPreparer() + @ResourceGroupPreparer(name_prefix="platformvalidation", random_name_enabled=True, location="eastus") + @recorded_by_proxy_async + async def test_get_missing_cloud_validation_raises_not_found(self, resource_group) -> None: + async with self.create_client() as client: + with pytest.raises(ResourceNotFoundError) as missing: + await client.cloud_validations.get(resource_group.name, self.get_resource_name("pyasdkmissing")) + assert missing.value.status_code == 404 diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_helpers.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_helpers.py new file mode 100644 index 000000000000..cd171391c95c --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_helpers.py @@ -0,0 +1,341 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import json +import inspect +import re +import traceback +from types import SimpleNamespace as NS +from unittest.mock import AsyncMock, Mock + +import pytest +from azure.core.exceptions import HttpResponseError, ResourceNotFoundError + +import _execution_testcase as helpers + + +ARGS = ("group", "cv", "plan") +PLAN_ID = "/subscriptions/test/resourceGroups/group/providers/Microsoft.PlatformValidation/cloudValidations/cv/validationExecutionPlans/plan" +RUN_ID = f"{PLAN_ID}/executionPlanRuns/run" + + +def missing(): + error = ResourceNotFoundError("Missing") + error.status_code = 404 + return error + + +def plan(): + return NS( + id=PLAN_ID, + tags={"sdk-test": "plan"}, + properties=NS(provisioning_state="Succeeded"), + ) + + +def run(status="Succeeded"): + return NS( + name="run", + id=RUN_ID, + properties=NS(description="owned", provisioning_state="Succeeded", status=status), + ) + + +def poller(): + result = Mock() + result.done.return_value = True + return result + + +async def items(values): + for value in values: + yield value + + +def test_configuration_uses_public_boot_contract(): + value = json.loads(helpers.plan_configuration("plan", helpers.IMAGE_PLACEHOLDER)) + assert value["apiVersion"] == "microsoft.validate/executionPlan.v0" + assert value["kind"] == "ExecutionPlan" + assert value["parameters"]["certificationPackageReference"]["vmGenerationType"] == "V1" + assert value["authoring"]["steps"][0]["testRef"].endswith("/compute-vmimagevalidation-vm-boot-test") + + +def test_playback_never_uses_live_image(monkeypatch): + monkeypatch.setenv("PLATFORMVALIDATION_SOURCE_VHD_URI", "private") + assert helpers.image_source(False) == helpers.IMAGE_PLACEHOLDER + + +def test_live_image_requires_https(monkeypatch): + monkeypatch.setenv("PLATFORMVALIDATION_SOURCE_VHD_URI", "http://example.invalid/image") + with pytest.raises(ValueError, match="approved HTTPS"): + helpers.image_source(True) + + +def test_live_image_registers_sanitizer(monkeypatch): + sanitizer = Mock() + monkeypatch.setattr(helpers, "add_general_string_sanitizer", sanitizer) + monkeypatch.setenv("PLATFORMVALIDATION_SOURCE_VHD_URI", helpers.IMAGE_PLACEHOLDER) + assert helpers.image_source(True) == helpers.IMAGE_PLACEHOLDER + sanitizer.assert_called_once_with(target=helpers.IMAGE_PLACEHOLDER, value=helpers.IMAGE_PLACEHOLDER) + + +@pytest.mark.parametrize( + "url,expected", + [ + ( + "https://example.invalid/providers/x/locations/eastus/operationStatuses/id?sig=private", + ["eastus", "id"], + ), + ("https://example.invalid/operationResults/id?sig=private", []), + (None, []), + ], +) +def test_operation_target_discards_signed_query(url, expected): + target = [] + helpers.operation_status_target(NS(http_response=NS(headers={"Azure-AsyncOperation": url})), target) + assert target == expected + + +def test_incomplete_sync_poller_is_not_success(): + operation = poller() + operation.done.return_value = False + with pytest.raises(TimeoutError): + helpers.complete(operation) + + +@pytest.mark.asyncio +async def test_incomplete_async_poller_is_not_success(): + operation = poller() + operation.result = AsyncMock(return_value=None) + operation.done.return_value = False + with pytest.raises(TimeoutError): + await helpers.complete_async(operation) + + +@pytest.mark.parametrize("status", ["Accepted", "Running", None]) +def test_active_run_cannot_be_deleted(status): + with pytest.raises(AssertionError, match="still active"): + helpers.assert_deletable_run(run(status), RUN_ID, "owned") + + +def test_wrong_run_owner_cannot_be_deleted(): + with pytest.raises(AssertionError, match="unowned"): + helpers.assert_deletable_run(run(), RUN_ID, "someone else") + + +def test_wrong_plan_owner_cannot_be_deleted(): + with pytest.raises(AssertionError, match="unowned"): + helpers.assert_owned_plan(plan(), PLAN_ID, "someone else") + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_cleanup_child_first_and_checks_vtr_cascade(async_mode): + client = Mock() + order = [] + if async_mode: + client.validation_execution_plans.get = AsyncMock(side_effect=[plan(), missing()]) + client.execution_plan_runs.list_by_execution_plan.side_effect = lambda *a: items([run()] if not order else []) + client.validation_test_runs.list_by_execution_plan_run.side_effect = lambda *a: items([NS(name="boot")]) + for operations, label in ( + (client.execution_plan_runs, "run"), + (client.validation_execution_plans, "plan"), + ): + + async def delete(*args, label=label): + order.append(label) + operation = poller() + operation.result = AsyncMock(return_value=None) + return operation + + operations.begin_delete = AsyncMock(side_effect=delete) + client.execution_plan_runs.get = AsyncMock(side_effect=missing()) + client.validation_test_runs.get = AsyncMock(side_effect=missing()) + await helpers.cleanup_execution_async(client, ARGS, PLAN_ID, "run", "owned") + client.validation_test_runs.get.assert_awaited_once_with(*ARGS, "run", "boot") + else: + client.validation_execution_plans.get.side_effect = [plan(), missing()] + client.execution_plan_runs.list_by_execution_plan.side_effect = [[run()], []] + client.validation_test_runs.list_by_execution_plan_run.return_value = [NS(name="boot")] + client.execution_plan_runs.begin_delete.side_effect = lambda *a: order.append("run") or poller() + client.validation_execution_plans.begin_delete.side_effect = lambda *a: order.append("plan") or poller() + client.execution_plan_runs.get.side_effect = missing() + client.validation_test_runs.get.side_effect = missing() + helpers.cleanup_execution(client, ARGS, PLAN_ID, "run", "owned") + client.validation_test_runs.get.assert_called_once_with(*ARGS, "run", "boot") + assert order == ["run", "plan"] + client.cloud_validations.begin_delete.assert_not_called() + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_cleanup_refuses_unexpected_run_before_deleting_anything(async_mode): + client = Mock() + if async_mode: + client.validation_execution_plans.get = AsyncMock(return_value=plan()) + client.execution_plan_runs.list_by_execution_plan.side_effect = lambda *a: items([run()]) + with pytest.raises(AssertionError, match="Unexpected run"): + await helpers.cleanup_execution_async(client, ARGS, PLAN_ID, "other", "owned") + else: + client.validation_execution_plans.get.return_value = plan() + client.execution_plan_runs.list_by_execution_plan.return_value = [run()] + with pytest.raises(AssertionError, match="Unexpected run"): + helpers.cleanup_execution(client, ARGS, PLAN_ID, "other", "owned") + client.execution_plan_runs.begin_delete.assert_not_called() + client.validation_execution_plans.begin_delete.assert_not_called() + + +def test_workload_rejects_partial_success(): + result = run() + result.properties.error = None + result.properties.test_run_summary = NS( + overall_result="Passed", + total_tests=1, + passed_tests=0, + failed_tests=1, + skipped_tests=0, + ) + with pytest.raises(AssertionError): + helpers.ExecutionTestCase().assert_workload(result, RUN_ID, "owned") + + +def test_proxy_sanitizer_registration(monkeypatch): + import _sanitizers as sanitizers + + header, body = Mock(), Mock() + for name in ( + "remove_batch_sanitizers", + "add_general_string_sanitizer", + "add_remove_header_sanitizer", + "add_general_regex_sanitizer", + ): + monkeypatch.setattr(sanitizers, name, Mock()) + monkeypatch.setattr(sanitizers, "add_header_regex_sanitizer", header) + monkeypatch.setattr(sanitizers, "add_body_key_sanitizer", body) + sanitizers.configure_sanitizers() + header.assert_called_once_with(key="x-ms-operation-identifier", value="Sanitized") + assert any(call.kwargs.get("json_path") == "$..planConfigurationJson" for call in body.call_args_list) + + +def test_subscription_list_sanitizers_preserve_target_identity(monkeypatch): + import _cloud_validation_testcase as cloud + + body, tags = Mock(), Mock() + monkeypatch.setattr(cloud, "add_body_key_sanitizer", body) + monkeypatch.setattr(cloud, "add_body_regex_sanitizer", tags) + cloud.CloudValidationTestCase().sanitize_subscription_list("owned") + assert body.call_count == 5 + for call in body.call_args_list: + assert call.kwargs["json_path"].startswith("$.value[?(@.name != 'owned')]") + assert call.kwargs["function_scoped"] + condition = json.loads(call.kwargs["condition"])["uriRegex"] + assert re.search( + condition, + "https://example.invalid/subscriptions/test/providers/Microsoft.PlatformValidation/cloudValidations?api-version=test", + ) + assert not re.search( + condition, + "https://example.invalid/subscriptions/test/resourceGroups/group/providers/Microsoft.PlatformValidation/cloudValidations?api-version=test", + ) + tags.assert_called_once() + tagged = json.dumps({"tags": {"private-key": 'value with {"nested"} braces and \\ escapes'}}) + settings = tags.call_args.kwargs + assert json.loads(re.sub(settings["regex"], settings["value"], tagged)) == {"tags": {}} + + +@pytest.mark.parametrize("key", ["sig", "se", "st", "sv"]) +@pytest.mark.parametrize("depth", [0, 1, 2]) +@pytest.mark.parametrize("value", ["private", r"\/private", r"private\/tail", r"\u0070rivate"]) +def test_common_sas_sanitizer_preserves_embedded_json_escapes(monkeypatch, key, depth, value): + import devtools_testutils.proxy_startup as startup + from devtools_testutils.sanitizers import Sanitizer + + batch = Mock() + monkeypatch.setattr(startup, "add_batch_sanitizers", batch) + monkeypatch.setattr(startup, "set_custom_default_matcher", Mock()) + monkeypatch.setattr(startup, "remove_batch_sanitizers", Mock()) + startup.set_common_sanitizers() + settings = batch.call_args.kwargs["sanitizers"][Sanitizer.GENERAL_REGEX][0] + regex = settings["regex"].replace("(?", "(?P") + document = '{"sourceVhdUri": "https://example.invalid/?' + key + "=" + value + '"}' + for _ in range(depth): + document = json.dumps({"planConfigurationJson": document}) + sanitized = re.sub(regex, lambda match: match[0].replace(match["secret"], "Sanitized"), document) + assert "private" not in sanitized + for _ in range(depth): + sanitized = json.loads(sanitized)["planConfigurationJson"] + assert json.loads(sanitized)["sourceVhdUri"].endswith("=Sanitized") + + +@pytest.mark.parametrize("key", ["sig", "se", "st", "sv", "s", "c", "h", "t"]) +@pytest.mark.parametrize("depth", [0, 1, 2]) +@pytest.mark.parametrize("value", ["private", r"\/private", r"private\/tail", r"\u0070rivate"]) +def test_signed_url_sanitizer_removes_escaped_values(monkeypatch, key, depth, value): + import _sanitizers as sanitizers + + general = Mock() + for name in ( + "remove_batch_sanitizers", + "add_general_string_sanitizer", + "add_remove_header_sanitizer", + "add_header_regex_sanitizer", + "add_body_key_sanitizer", + ): + monkeypatch.setattr(sanitizers, name, Mock()) + monkeypatch.setattr(sanitizers, "add_general_regex_sanitizer", general) + sanitizers.configure_sanitizers() + settings = next(call.kwargs for call in general.call_args_list if call.kwargs.get("group_for_replace") == "2") + document = '{"url": "https://example.invalid/?' + key + "=" + value + '"}' + for _ in range(depth): + document = json.dumps({"planConfigurationJson": document}) + sanitized = re.sub(settings["regex"], lambda match: match[1] + settings["value"], document) + assert "private" not in sanitized + for _ in range(depth): + sanitized = json.loads(sanitized)["planConfigurationJson"] + assert json.loads(sanitized)["url"].endswith("=Sanitized") + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_cleanup_refuses_orphans_after_prior_run_deletion(async_mode): + client = Mock() + if async_mode: + client.validation_execution_plans.get = AsyncMock(return_value=plan()) + client.execution_plan_runs.list_by_execution_plan.side_effect = lambda *a: items([]) + client.validation_test_runs.list_by_execution_plan_run.side_effect = lambda *a: items([NS(name="orphan")]) + with pytest.raises(AssertionError, match="Orphan"): + await helpers.cleanup_execution_async(client, ARGS, PLAN_ID, "run", "owned") + else: + client.validation_execution_plans.get.return_value = plan() + client.execution_plan_runs.list_by_execution_plan.return_value = [] + client.validation_test_runs.list_by_execution_plan_run.return_value = [NS(name="orphan")] + with pytest.raises(AssertionError, match="Orphan"): + helpers.cleanup_execution(client, ARGS, PLAN_ID, "run", "owned") + client.validation_execution_plans.begin_delete.assert_not_called() + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_service_error_does_not_echo_embedded_credentials(async_mode): + message = "sensitive-" + "image-signature" + if async_mode: + + @helpers.protect_execution_errors + async def operation(): + raise HttpResponseError(message) + + else: + + @helpers.protect_execution_errors + def operation(): + raise HttpResponseError(message) + + with pytest.raises(RuntimeError) as failure: + if async_mode: + await operation() + else: + operation() + assert inspect.getfullargspec(operation).varkw is None + assert message not in "".join(traceback.format_exception(failure.value)) diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_lifecycle.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_lifecycle.py new file mode 100644 index 000000000000..da496b2c3a23 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_lifecycle.py @@ -0,0 +1,169 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from functools import partial + +from azure.mgmt.platformvalidation import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.models import ( + ExecutionPlanRun, + ExecutionPlanRunProperties, + ValidationExecutionPlan, + ValidationExecutionPlanProperties, + ValidationExecutionPlanUpdate, +) +from devtools_testutils import recorded_by_proxy + +from _execution_prerequisites import ( + ExecutionEnvironmentPreparer, + ExecutionPrerequisitesPreparer, + ExecutionResourceGroupPreparer, +) +from _execution_testcase import ( + ExecutionTestCase, + ExistingExecutionPreparer, + assert_missing, + cleanup_execution, + complete, + image_source, + operation_status_target, + plan_configuration, + protect_execution_errors, + wait_for_run, +) + + +class TestExecutionLifecycle(ExecutionTestCase): + @ExistingExecutionPreparer() + @recorded_by_proxy + def test_execution_plan_lifecycle( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ): + self.exercise_lifecycle( + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + execute=False, + ) + + @ExecutionEnvironmentPreparer() + @ExecutionResourceGroupPreparer(name_prefix="platformvalidationexec", random_name_enabled=True, location="eastus") + @ExecutionPrerequisitesPreparer() + @recorded_by_proxy + def test_execution_lifecycle( + self, + platformvalidation_location, + resource_group, + cloud_validation_name, + ): + self.exercise_lifecycle( + platformvalidation_location, + resource_group.name, + cloud_validation_name, + "owned-parent", + execute=True, + ) + + @protect_execution_errors + def exercise_lifecycle( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + *, + execute, + ): + group = platformvalidation_execution_resource_group + cv = platformvalidation_execution_cloud_validation + name = self.get_resource_name("pyexec") + run_name = self.get_resource_name("pyrun") + args = (group, cv, name) + run_args = (*args, run_name) + plan_id = f"{self.resource_id(group, cv)}/validationExecutionPlans/{name}" + run_id = f"{plan_id}/executionPlanRuns/{run_name}" + description = f"SDK owned execution {run_name}" + configuration = plan_configuration(name, image_source(self.is_live)) + tags = {"sdk-test": name} + with self.create_mgmt_client(PlatformValidationMgmtClient) as client: + self.assert_parent( + client.cloud_validations.get(group, cv), + group, + cv, + platformvalidation_execution_ownership, + ) + assert_missing(client.validation_execution_plans.get, *args) + try: + plan = complete( + client.validation_execution_plans.begin_create_or_update( + *args, + ValidationExecutionPlan( + location=platformvalidation_location, + tags=tags, + properties=ValidationExecutionPlanProperties( + description="SDK boot plan", + plan_configuration_json=configuration, + ), + ), + ) + ) + self.assert_plan(plan, plan_id, name, tags, "SDK boot plan", configuration) + self.assert_plan( + client.validation_execution_plans.get(*args), + plan_id, + name, + tags, + "SDK boot plan", + configuration, + ) + matches = [ + p + for p in client.validation_execution_plans.list_by_resource_group(group, cv) + if p.id.lower() == plan_id.lower() + ] + assert len(matches) == 1 + tags = {**tags, "phase": "updated"} + target = [] + updated = complete( + client.validation_execution_plans.begin_update( + *args, + ValidationExecutionPlanUpdate(tags=tags), + raw_response_hook=partial(operation_status_target, target=target), + ) + ) + self.assert_plan(updated, plan_id, name, tags, "SDK boot plan", configuration) + self.assert_plan( + client.validation_execution_plans.get(*args), + plan_id, + name, + tags, + "SDK boot plan", + configuration, + ) + assert len(target) == 2, "PATCH must expose a supported operation-status route." + assert client.operation_status.get(*target).status == "Succeeded" + if not execute: + return + assert_missing(client.execution_plan_runs.get, *run_args) + complete( + client.execution_plan_runs.begin_create_or_update( + *run_args, + ExecutionPlanRun(properties=ExecutionPlanRunProperties(description=description)), + ) + ) + self.assert_workload(wait_for_run(client, run_args), run_id, description) + runs = list(client.execution_plan_runs.list_by_execution_plan(*args)) + assert len(runs) == 1 + self.assert_workload(runs[0], run_id, description) + tests = list(client.validation_test_runs.list_by_execution_plan_run(*run_args)) + assert len(tests) == 1 + self.assert_test_run(tests[0], run_id) + self.assert_test_run(client.validation_test_runs.get(*run_args, tests[0].name), run_id) + finally: + cleanup_execution(client, args, plan_id, run_name, description) diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_lifecycle_async.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_lifecycle_async.py new file mode 100644 index 000000000000..22cecd5b7f33 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_lifecycle_async.py @@ -0,0 +1,180 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from functools import partial + +from azure.mgmt.platformvalidation.aio import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.models import ( + ExecutionPlanRun, + ExecutionPlanRunProperties, + ValidationExecutionPlan, + ValidationExecutionPlanProperties, + ValidationExecutionPlanUpdate, +) +from devtools_testutils.aio import recorded_by_proxy_async + +from _execution_prerequisites import ( + ExecutionEnvironmentPreparer, + ExecutionPrerequisitesPreparer, + ExecutionResourceGroupPreparer, +) +from _execution_testcase import ( + ExecutionTestCase, + ExistingExecutionPreparer, + assert_missing_async, + cleanup_execution_async, + complete_async, + image_source, + operation_status_target, + plan_configuration, + protect_execution_errors, + wait_for_run_async, +) + + +class TestExecutionLifecycleAsync(ExecutionTestCase): + @ExistingExecutionPreparer() + @recorded_by_proxy_async + async def test_execution_plan_lifecycle( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ): + await self.exercise_lifecycle( + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + execute=False, + ) + + @ExecutionEnvironmentPreparer() + @ExecutionResourceGroupPreparer(name_prefix="platformvalidationexec", random_name_enabled=True, location="eastus") + @ExecutionPrerequisitesPreparer() + @recorded_by_proxy_async + async def test_execution_lifecycle( + self, + platformvalidation_location, + resource_group, + cloud_validation_name, + ): + await self.exercise_lifecycle( + platformvalidation_location, + resource_group.name, + cloud_validation_name, + "owned-parent", + execute=True, + ) + + @protect_execution_errors + async def exercise_lifecycle( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + *, + execute, + ): + group = platformvalidation_execution_resource_group + cv = platformvalidation_execution_cloud_validation + name = self.get_resource_name("pyaexec") + run_name = self.get_resource_name("pyarun") + args = (group, cv, name) + run_args = (*args, run_name) + plan_id = f"{self.resource_id(group, cv)}/validationExecutionPlans/{name}" + run_id = f"{plan_id}/executionPlanRuns/{run_name}" + description = f"SDK owned execution {run_name}" + configuration = plan_configuration(name, image_source(self.is_live)) + tags = {"sdk-test": name} + credential = self.get_credential(PlatformValidationMgmtClient, is_async=True) + try: + async with self.create_client_from_credential( + PlatformValidationMgmtClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + ) as client: + self.assert_parent( + await client.cloud_validations.get(group, cv), + group, + cv, + platformvalidation_execution_ownership, + ) + await assert_missing_async(client.validation_execution_plans.get, *args) + try: + plan = await complete_async( + await client.validation_execution_plans.begin_create_or_update( + *args, + ValidationExecutionPlan( + location=platformvalidation_location, + tags=tags, + properties=ValidationExecutionPlanProperties( + description="SDK boot plan", + plan_configuration_json=configuration, + ), + ), + ) + ) + self.assert_plan(plan, plan_id, name, tags, "SDK boot plan", configuration) + self.assert_plan( + await client.validation_execution_plans.get(*args), + plan_id, + name, + tags, + "SDK boot plan", + configuration, + ) + matches = [ + p + async for p in client.validation_execution_plans.list_by_resource_group(group, cv) + if p.id.lower() == plan_id.lower() + ] + assert len(matches) == 1 + tags = {**tags, "phase": "updated"} + target = [] + updated = await complete_async( + await client.validation_execution_plans.begin_update( + *args, + ValidationExecutionPlanUpdate(tags=tags), + raw_response_hook=partial(operation_status_target, target=target), + ) + ) + self.assert_plan(updated, plan_id, name, tags, "SDK boot plan", configuration) + self.assert_plan( + await client.validation_execution_plans.get(*args), + plan_id, + name, + tags, + "SDK boot plan", + configuration, + ) + assert len(target) == 2, "PATCH must expose a supported operation-status route." + assert (await client.operation_status.get(*target)).status == "Succeeded" + if not execute: + return + await assert_missing_async(client.execution_plan_runs.get, *run_args) + await complete_async( + await client.execution_plan_runs.begin_create_or_update( + *run_args, + ExecutionPlanRun(properties=ExecutionPlanRunProperties(description=description)), + ) + ) + self.assert_workload(await wait_for_run_async(client, run_args), run_id, description) + runs = [run async for run in client.execution_plan_runs.list_by_execution_plan(*args)] + assert len(runs) == 1 + self.assert_workload(runs[0], run_id, description) + tests = [test async for test in client.validation_test_runs.list_by_execution_plan_run(*run_args)] + assert len(tests) == 1 + self.assert_test_run(tests[0], run_id) + self.assert_test_run( + await client.validation_test_runs.get(*run_args, tests[0].name), + run_id, + ) + finally: + await cleanup_execution_async(client, args, plan_id, run_name, description) + finally: + await credential.close() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_prerequisite_cleanup.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_prerequisite_cleanup.py new file mode 100644 index 000000000000..70e9709b0c12 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_prerequisite_cleanup.py @@ -0,0 +1,341 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from types import SimpleNamespace +from unittest.mock import MagicMock, Mock + +import pytest +from azure.core.exceptions import HttpResponseError, ResourceExistsError, ResourceNotFoundError + +import _execution_prerequisites as setup + +SUBSCRIPTION = "00000000-0000-0000-0000-000000000000" +NAME = "pyexecowned" +GROUP = "owned-parent" +CV_ID = ( + f"/subscriptions/{SUBSCRIPTION}/resourceGroups/{GROUP}" + f"/providers/Microsoft.PlatformValidation/cloudValidations/{NAME}" +) + + +def not_found(status=404): + error = ResourceNotFoundError(message="Not found") + error.status_code = status + return error + + +def poller(result=None): + operation = Mock() + operation.result.return_value = result + operation.done.return_value = True + return operation + + +@pytest.fixture +def prerequisites(monkeypatch): + monkeypatch.setattr(setup, "add_general_string_sanitizer", Mock()) + monkeypatch.setattr(setup, "wait_for_managed_group_deletion", Mock()) + monkeypatch.setattr(setup.time, "sleep", Mock()) + preparer = setup.ExecutionPrerequisitesPreparer() + preparer.test_class_instance = Mock(is_live=True) + preparer.test_class_instance.get_settings_value.return_value = SUBSCRIPTION + resources, cv = MagicMock(), MagicMock() + clients = { + setup.ResourceManagementClient: resources, + setup.PlatformValidationMgmtClient: cv, + } + for client in clients.values(): + client.__enter__.return_value = client + preparer.create_mgmt_client = Mock(side_effect=lambda client_type: clients[client_type]) + resources.resource_groups.check_existence.side_effect = [False, True] + resources.resource_groups.get.return_value = SimpleNamespace(managed_by=CV_ID) + cv.cloud_validations.get.side_effect = not_found() + cv.cloud_validations.begin_create_or_update.return_value = poller( + SimpleNamespace(properties=SimpleNamespace(provisioning_state="Succeeded")) + ) + arguments = { + "resource_group": SimpleNamespace(name=GROUP), + "platformvalidation_location": "eastus", + } + return preparer, resources, cv, arguments + + +def test_setup_verifies_service_managed_group_without_creating_it(prerequisites): + preparer, resources, cv, arguments = prerequisites + assert preparer.create_resource(NAME, **arguments) == {"cloud_validation_name": NAME} + cv.cloud_validations.begin_create_or_update.assert_called_once() + resources.resource_groups.create_or_update.assert_not_called() + resources.resource_groups.get.assert_called_once_with(f"{NAME}-mrg") + assert {call.args[0] for call in preparer.create_mgmt_client.call_args_list} == { + setup.ResourceManagementClient, + setup.PlatformValidationMgmtClient, + } + + +def test_service_mode_never_falls_back_to_manual(prerequisites): + preparer, resources, _, arguments = prerequisites + resources.resource_groups.check_existence.side_effect = [False, False] + preparer._cleanup = Mock() + with pytest.raises(AssertionError, match="no fallback"): + preparer.create_resource(NAME, **arguments) + resources.resource_groups.create_or_update.assert_not_called() + preparer._cleanup.assert_called_once() + + +def test_playback_performs_no_prerequisite_azure_calls(prerequisites): + preparer, _, _, _ = prerequisites + preparer.test_class_instance.is_live = False + assert preparer.create_resource("executioncv") == {"cloud_validation_name": "executioncv"} + preparer.remove_resource("executioncv") + preparer.create_mgmt_client.assert_not_called() + + +def test_existing_group_is_never_adopted(prerequisites): + preparer, resources, cv, arguments = prerequisites + resources.resource_groups.check_existence.side_effect = None + resources.resource_groups.check_existence.return_value = True + with pytest.raises(AssertionError, match="existing managed group"): + preparer.create_resource(NAME, **arguments) + cv.cloud_validations.begin_create_or_update.assert_not_called() + resources.resource_groups.create_or_update.assert_not_called() + resources.resource_groups.begin_delete.assert_not_called() + + +@pytest.mark.parametrize("managed_by", [None, "", f"{CV_ID}-other"]) +def test_managed_group_must_belong_to_created_cv(prerequisites, managed_by): + preparer, resources, _, arguments = prerequisites + resources.resource_groups.get.return_value.managed_by = managed_by + preparer._cleanup = Mock() + with pytest.raises(AssertionError, match="not associated"): + preparer.create_resource(NAME, **arguments) + resources.resource_groups.create_or_update.assert_not_called() + preparer._cleanup.assert_called_once() + + +def test_managed_group_read_failure_is_propagated_and_cleanup_attempted(prerequisites): + preparer, resources, _, arguments = prerequisites + error = HttpResponseError(message="Access denied") + resources.resource_groups.get.side_effect = error + preparer._cleanup = Mock() + with pytest.raises(HttpResponseError) as raised: + preparer.create_resource(NAME, **arguments) + assert raised.value is error + preparer._cleanup.assert_called_once() + + +def test_failed_cv_creation_is_tracked_for_partial_cleanup(prerequisites): + preparer, resources, cv, arguments = prerequisites + error = HttpResponseError(message="CV creation response failed") + cv.cloud_validations.begin_create_or_update.side_effect = error + preparer._cleanup = Mock() + with pytest.raises(HttpResponseError) as raised: + preparer.create_resource(NAME, **arguments) + assert raised.value is error + assert preparer._cv_attempted + preparer._cleanup.assert_called_once() + resources.resource_groups.create_or_update.assert_not_called() + + +def test_cleanup_error_is_logged_without_replacing_setup_error(prerequisites, caplog): + preparer, resources, _, arguments = prerequisites + error = HttpResponseError(message="AuthorizationFailed") + resources.resource_groups.get.side_effect = error + preparer._cleanup = Mock(side_effect=AssertionError("Cleanup blocked")) + with pytest.raises(HttpResponseError) as raised: + preparer.create_resource(NAME, **arguments) + assert raised.value is error + assert "Prerequisite cleanup failed" in caplog.text + + +@pytest.fixture +def cleanup_ready(prerequisites): + preparer, resources, cv, arguments = prerequisites + preparer.create_resource(NAME, **arguments) + parent = SimpleNamespace( + id=f"/subscriptions/{SUBSCRIPTION}/resourceGroups/{GROUP}/providers/Microsoft.PlatformValidation/cloudValidations/{NAME}", + tags={"sdk-test": NAME}, + properties=SimpleNamespace(provisioning_state="Succeeded"), + ) + cv.cloud_validations.get.side_effect = [parent, not_found()] + cv.validation_execution_plans.list_by_resource_group.return_value = [] + resources.resource_groups.check_existence.side_effect = None + resources.resource_groups.check_existence.return_value = True + resources.resource_groups.get.return_value = SimpleNamespace(tags={"sdk-test": NAME}) + resources.resource_groups.begin_delete.return_value = poller() + cv.cloud_validations.begin_delete.return_value = poller() + return preparer, resources, cv, parent + + +def test_cleanup_deletes_cv_and_waits_for_service_managed_group(cleanup_ready): + preparer, resources, cv, _ = cleanup_ready + preparer.remove_resource(NAME) + cv.cloud_validations.begin_delete.assert_called_once_with(GROUP, NAME) + resources.resource_groups.begin_delete.assert_not_called() + setup.wait_for_managed_group_deletion.assert_called_once_with(resources, f"{NAME}-mrg") + + +def test_remaining_plan_blocks_all_parent_cleanup(cleanup_ready): + preparer, resources, cv, _ = cleanup_ready + cv.validation_execution_plans.list_by_resource_group.return_value = [Mock()] + with pytest.raises(AssertionError, match="delete EPRs and VEPs"): + preparer.remove_resource(NAME) + resources.resource_groups.begin_delete.assert_not_called() + cv.cloud_validations.begin_delete.assert_not_called() + + +def deletion_conflict(code="CannotDeleteResource", status=409): + error = ResourceExistsError(message="Nested resource index has not converged") + error.status_code = status + error.error = SimpleNamespace(code=code) + return error + + +def test_cv_cleanup_retries_only_known_nested_resource_conflict(cleanup_ready): + preparer, resources, cv, _ = cleanup_ready + cv.cloud_validations.begin_delete.side_effect = [deletion_conflict(), poller()] + preparer.remove_resource(NAME) + assert cv.cloud_validations.begin_delete.call_count == 2 + assert cv.validation_execution_plans.list_by_resource_group.call_count == 2 + setup.time.sleep.assert_called_once_with(setup.CV_DELETE_INTERVAL) + resources.resource_groups.begin_delete.assert_not_called() + + +def test_cv_cleanup_stops_if_a_plan_appears_during_retry(cleanup_ready): + preparer, resources, cv, _ = cleanup_ready + cv.cloud_validations.begin_delete.side_effect = deletion_conflict() + cv.validation_execution_plans.list_by_resource_group.side_effect = [[], [Mock()]] + with pytest.raises(AssertionError, match="Execution plans remain"): + preparer.remove_resource(NAME) + cv.cloud_validations.begin_delete.assert_called_once() + resources.resource_groups.begin_delete.assert_not_called() + + +@pytest.mark.parametrize("code,status", [("Conflict", 409), ("CannotDeleteResource", 400)]) +def test_cv_cleanup_does_not_retry_other_errors(cleanup_ready, code, status): + preparer, _, cv, _ = cleanup_ready + cv.cloud_validations.begin_delete.side_effect = deletion_conflict(code, status) + with pytest.raises(ResourceExistsError): + preparer.remove_resource(NAME) + cv.cloud_validations.begin_delete.assert_called_once() + setup.time.sleep.assert_not_called() + + +def test_cv_cleanup_reports_persistent_nested_resource_conflict(cleanup_ready, monkeypatch): + preparer, _, cv, _ = cleanup_ready + monkeypatch.setattr(setup, "CV_DELETE_ATTEMPTS", 2) + cv.cloud_validations.begin_delete.side_effect = deletion_conflict() + with pytest.raises(ResourceExistsError): + preparer.remove_resource(NAME) + assert cv.cloud_validations.begin_delete.call_count == 2 + setup.wait_for_managed_group_deletion.assert_not_called() + + +def test_unowned_cv_blocks_cleanup(cleanup_ready): + preparer, resources, cv, parent = cleanup_ready + parent.tags = {"sdk-test": "someone-else"} + with pytest.raises(AssertionError, match="ownership tag"): + preparer.remove_resource(NAME) + resources.resource_groups.begin_delete.assert_not_called() + cv.cloud_validations.begin_delete.assert_not_called() + + +def test_service_managed_group_cleanup_failure_is_reported(cleanup_ready): + preparer, resources, cv, _ = cleanup_ready + setup.wait_for_managed_group_deletion.side_effect = AssertionError("Managed group remains") + with pytest.raises(AssertionError, match="Managed group remains"): + preparer.remove_resource(NAME) + cv.cloud_validations.begin_delete.assert_called_once() + resources.resource_groups.begin_delete.assert_not_called() + + +def test_nonterminal_cv_blocks_cleanup(cleanup_ready): + preparer, resources, cv, parent = cleanup_ready + parent.properties.provisioning_state = "Accepted" + with pytest.raises(AssertionError, match="nonterminal"): + preparer.remove_resource(NAME) + resources.resource_groups.begin_delete.assert_not_called() + cv.cloud_validations.begin_delete.assert_not_called() + + +def test_incomplete_operation_is_not_success(): + operation = poller() + operation.done.return_value = False + with pytest.raises(TimeoutError, match="did not finish"): + setup.complete_prerequisite_operation(operation) + + +@pytest.mark.parametrize("remaining", [[], [Mock()]]) +def test_outer_group_cleanup_does_not_cascade_into_children(remaining, monkeypatch): + monkeypatch.setattr(setup, "wait_for_managed_group_deletion", Mock()) + preparer = setup.ExecutionResourceGroupPreparer() + preparer.test_class_instance = Mock(is_live=True) + preparer._need_creation = True + preparer.client = Mock() + preparer.test_class_instance._execution_cv_attempted = False + preparer.client.resources.list_by_resource_group.return_value = remaining + preparer.client.resource_groups.begin_delete.return_value = poller() + preparer.client.resource_groups.check_existence.return_value = False + if remaining: + with pytest.raises(AssertionError, match="resources remain"): + preparer.remove_resource("owned") + preparer.client.resource_groups.begin_delete.assert_not_called() + else: + preparer.remove_resource("owned") + preparer.client.resource_groups.begin_delete.assert_called_once_with("rgpy-owned") + + +@pytest.mark.parametrize("remaining", [[], [Mock()]]) +def test_outer_group_retained_when_cv_cleanup_failed_even_if_arm_listing_is_empty(remaining): + preparer = setup.ExecutionResourceGroupPreparer() + preparer.test_class_instance = Mock(is_live=True) + preparer.test_class_instance._execution_cv_attempted = True + preparer.test_class_instance._cleaned_execution_cv = None + preparer._need_creation = True + preparer.client = Mock() + preparer.client.resources.list_by_resource_group.return_value = remaining + with pytest.raises(AssertionError, match="prerequisite cleanup was not confirmed"): + preparer.remove_resource("owned") + preparer.client.resource_groups.begin_delete.assert_not_called() + + +@pytest.mark.parametrize( + "provider_state,cleanup_confirmed", + [("deleted", True), ("present", True), ("forbidden", True), ("deleted", False)], +) +def test_stale_arm_listing_requires_confirmed_cleanup_and_fresh_provider_404( + provider_state, cleanup_confirmed, monkeypatch +): + monkeypatch.setattr(setup, "wait_for_managed_group_deletion", Mock()) + preparer = setup.ExecutionResourceGroupPreparer() + preparer.test_class_instance = Mock(is_live=True) + preparer.test_class_instance.get_settings_value.return_value = SUBSCRIPTION + preparer.test_class_instance._execution_cv_attempted = True + preparer.test_class_instance._cleaned_execution_cv = NAME if cleanup_confirmed else None + preparer._need_creation = True + preparer.client = Mock() + preparer.client.resources.list_by_resource_group.return_value = [ + SimpleNamespace( + id=f"/subscriptions/{SUBSCRIPTION}/resourceGroups/rgpy-owned" + f"/providers/Microsoft.PlatformValidation/cloudValidations/{NAME}" + ) + ] + preparer.client.resource_groups.begin_delete.return_value = poller() + cv = MagicMock() + cv.__enter__.return_value = cv + preparer.create_mgmt_client = Mock(return_value=cv) + if provider_state == "deleted": + cv.cloud_validations.get.side_effect = not_found() + elif provider_state == "forbidden": + cv.cloud_validations.get.side_effect = HttpResponseError(message="Access denied") + if provider_state == "deleted" and cleanup_confirmed: + preparer.remove_resource("owned") + cv.cloud_validations.get.assert_called_once_with("rgpy-owned", NAME) + preparer.client.resource_groups.begin_delete.assert_called_once_with("rgpy-owned") + else: + with pytest.raises((AssertionError, HttpResponseError)): + preparer.remove_resource("owned") + preparer.client.resource_groups.begin_delete.assert_not_called() + if not cleanup_confirmed: + preparer.create_mgmt_client.assert_not_called() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_results.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_results.py new file mode 100644 index 000000000000..ec0e051e529e --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_results.py @@ -0,0 +1,87 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from azure.mgmt.platformvalidation import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.aio import PlatformValidationMgmtClient as AsyncPlatformValidationMgmtClient +from devtools_testutils import recorded_by_proxy +from devtools_testutils.aio import recorded_by_proxy_async + +from _execution_testcase import ExecutionTestCase, ExistingRunPreparer, protect_execution_errors + + +class TestExecutionResults(ExecutionTestCase): + @ExistingRunPreparer() + @recorded_by_proxy + @protect_execution_errors + def test_existing_execution_results( + self, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_plan, + platformvalidation_execution_run, + platformvalidation_execution_ownership, + ): + group = platformvalidation_execution_resource_group + cv = platformvalidation_execution_cloud_validation + args = (group, cv, platformvalidation_execution_plan, platformvalidation_execution_run) + run_id = f"{self.resource_id(group, cv)}/validationExecutionPlans/{args[2]}" f"/executionPlanRuns/{args[3]}" + with self.create_mgmt_client(PlatformValidationMgmtClient) as client: + self.assert_parent( + client.cloud_validations.get(group, cv), group, cv, platformvalidation_execution_ownership + ) + run = client.execution_plan_runs.get(*args) + self.assert_workload(run, run_id, run.properties.description) + matches = [ + item + for item in client.execution_plan_runs.list_by_execution_plan(*args[:3]) + if item.id.lower() == run_id.lower() + ] + assert len(matches) == 1 + self.assert_workload(matches[0], run_id, run.properties.description) + tests = list(client.validation_test_runs.list_by_execution_plan_run(*args)) + assert len(tests) == 1 + self.assert_test_run(tests[0], run_id) + self.assert_test_run(client.validation_test_runs.get(*args, tests[0].name), run_id) + + @ExistingRunPreparer() + @recorded_by_proxy_async + @protect_execution_errors + async def test_existing_execution_results_async( + self, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_plan, + platformvalidation_execution_run, + platformvalidation_execution_ownership, + ): + group = platformvalidation_execution_resource_group + cv = platformvalidation_execution_cloud_validation + args = (group, cv, platformvalidation_execution_plan, platformvalidation_execution_run) + run_id = f"{self.resource_id(group, cv)}/validationExecutionPlans/{args[2]}" f"/executionPlanRuns/{args[3]}" + credential = self.get_credential(AsyncPlatformValidationMgmtClient, is_async=True) + try: + async with self.create_client_from_credential( + AsyncPlatformValidationMgmtClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + ) as client: + self.assert_parent( + await client.cloud_validations.get(group, cv), group, cv, platformvalidation_execution_ownership + ) + run = await client.execution_plan_runs.get(*args) + self.assert_workload(run, run_id, run.properties.description) + matches = [ + item + async for item in client.execution_plan_runs.list_by_execution_plan(*args[:3]) + if item.id.lower() == run_id.lower() + ] + assert len(matches) == 1 + self.assert_workload(matches[0], run_id, run.properties.description) + tests = [test async for test in client.validation_test_runs.list_by_execution_plan_run(*args)] + assert len(tests) == 1 + self.assert_test_run(tests[0], run_id) + self.assert_test_run(await client.validation_test_runs.get(*args, tests[0].name), run_id) + finally: + await credential.close() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_updates.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_updates.py new file mode 100644 index 000000000000..ea9624c36d94 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_updates.py @@ -0,0 +1,120 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from azure.mgmt.platformvalidation import PlatformValidationMgmtClient +from devtools_testutils import recorded_by_proxy + +from _execution_testcase import ( + ExecutionTestCase, + ExistingExecutionPreparer, + assert_missing, + cleanup_execution, + complete, + image_source, + plan_configuration, + protect_execution_errors, +) +from _update_testcase import ( + plan_update, + replacement_plan, + response_observation, + updated_configuration, +) + + +class TestExecutionUpdates(ExecutionTestCase): + @ExistingExecutionPreparer() + @recorded_by_proxy + def test_configuration_and_put_updates( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ): + self.exercise_updates( + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + include_description=False, + ) + + @ExistingExecutionPreparer() + @recorded_by_proxy + def test_plan_updates( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ): + self.exercise_updates( + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ) + + @protect_execution_errors + def exercise_updates(self, location, group, cv, ownership, *, include_description=True): + name = self.get_resource_name("pyupd") + args = (group, cv, name) + plan_id = f"{self.resource_id(group, cv)}/validationExecutionPlans/{name}" + tags = {"sdk-test": name} + description = "SDK isolated update plan" + configuration = plan_configuration(name, image_source(self.is_live)) + self.update_observations = [] + + def observe(response): + self.update_observations.append(response_observation(response)) + + with self.create_mgmt_client(PlatformValidationMgmtClient) as client: + operations = client.validation_execution_plans + self.assert_parent(client.cloud_validations.get(group, cv), group, cv, ownership) + assert_missing(operations.get, *args) + try: + created = complete( + operations.begin_create_or_update( + *args, + replacement_plan(location, tags, description, configuration), + raw_response_hook=observe, + ) + ) + self.assert_plan(created, plan_id, name, tags, description, configuration) + if include_description: + description = "Description-only PATCH" + patched = complete( + operations.begin_update(*args, plan_update(description=description), raw_response_hook=observe) + ) + self.assert_plan(patched, plan_id, name, tags, description, configuration) + self.assert_plan(operations.get(*args), plan_id, name, tags, description, configuration) + configuration = updated_configuration(configuration) + patched = complete( + operations.begin_update( + *args, + plan_update(tags=tags, plan_configuration_json=configuration), + raw_response_hook=observe, + ) + ) + self.assert_plan(patched, plan_id, name, tags, description, configuration) + self.assert_plan(operations.get(*args), plan_id, name, tags, description, configuration) + for phase in ("idempotent", "metadata-replacement"): + if phase == "metadata-replacement": + tags = {**tags, "phase": phase} + description = "PUT existing plan metadata" + replaced = complete( + operations.begin_create_or_update( + *args, + replacement_plan(location, tags, description, configuration), + raw_response_hook=observe, + ) + ) + self.assert_plan(replaced, plan_id, name, tags, description, configuration) + self.assert_plan(operations.get(*args), plan_id, name, tags, description, configuration) + self.assert_plan(operations.get(*args), plan_id, name, tags, description, configuration) + assert not list(client.execution_plan_runs.list_by_execution_plan(*args)) + finally: + cleanup_execution(client, args, plan_id, "never-submitted", "No execution was submitted") diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_updates_async.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_updates_async.py new file mode 100644 index 000000000000..d0e8dc95cab2 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_execution_updates_async.py @@ -0,0 +1,132 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from azure.mgmt.platformvalidation.aio import PlatformValidationMgmtClient +from devtools_testutils.aio import recorded_by_proxy_async + +from _execution_testcase import ( + ExecutionTestCase, + ExistingExecutionPreparer, + assert_missing_async, + cleanup_execution_async, + complete_async, + image_source, + plan_configuration, + protect_execution_errors, +) +from _update_testcase import ( + plan_update, + replacement_plan, + response_observation, + updated_configuration, +) + + +class TestExecutionUpdatesAsync(ExecutionTestCase): + @ExistingExecutionPreparer() + @recorded_by_proxy_async + async def test_configuration_and_put_updates( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ): + await self.exercise_updates( + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + include_description=False, + ) + + @ExistingExecutionPreparer() + @recorded_by_proxy_async + async def test_plan_updates( + self, + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ): + await self.exercise_updates( + platformvalidation_location, + platformvalidation_execution_resource_group, + platformvalidation_execution_cloud_validation, + platformvalidation_execution_ownership, + ) + + @protect_execution_errors + async def exercise_updates(self, location, group, cv, ownership, *, include_description=True): + name = self.get_resource_name("pyaupd") + args = (group, cv, name) + plan_id = f"{self.resource_id(group, cv)}/validationExecutionPlans/{name}" + tags = {"sdk-test": name} + description = "SDK isolated update plan" + configuration = plan_configuration(name, image_source(self.is_live)) + self.update_observations = [] + + def observe(response): + self.update_observations.append(response_observation(response)) + + credential = self.get_credential(PlatformValidationMgmtClient, is_async=True) + try: + async with self.create_client_from_credential( + PlatformValidationMgmtClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + ) as client: + operations = client.validation_execution_plans + self.assert_parent(await client.cloud_validations.get(group, cv), group, cv, ownership) + await assert_missing_async(operations.get, *args) + try: + created = await complete_async( + await operations.begin_create_or_update( + *args, + replacement_plan(location, tags, description, configuration), + raw_response_hook=observe, + ) + ) + self.assert_plan(created, plan_id, name, tags, description, configuration) + if include_description: + description = "Description-only PATCH" + patched = await complete_async( + await operations.begin_update( + *args, plan_update(description=description), raw_response_hook=observe + ) + ) + self.assert_plan(patched, plan_id, name, tags, description, configuration) + self.assert_plan(await operations.get(*args), plan_id, name, tags, description, configuration) + configuration = updated_configuration(configuration) + patched = await complete_async( + await operations.begin_update( + *args, + plan_update(tags=tags, plan_configuration_json=configuration), + raw_response_hook=observe, + ) + ) + self.assert_plan(patched, plan_id, name, tags, description, configuration) + self.assert_plan(await operations.get(*args), plan_id, name, tags, description, configuration) + for phase in ("idempotent", "metadata-replacement"): + if phase == "metadata-replacement": + tags = {**tags, "phase": phase} + description = "PUT existing plan metadata" + replaced = await complete_async( + await operations.begin_create_or_update( + *args, + replacement_plan(location, tags, description, configuration), + raw_response_hook=observe, + ) + ) + self.assert_plan(replaced, plan_id, name, tags, description, configuration) + self.assert_plan(await operations.get(*args), plan_id, name, tags, description, configuration) + self.assert_plan(await operations.get(*args), plan_id, name, tags, description, configuration) + assert not [run async for run in client.execution_plan_runs.list_by_execution_plan(*args)] + finally: + await cleanup_execution_async( + client, args, plan_id, "never-submitted", "No execution was submitted" + ) + finally: + await credential.close() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_managed_group_cleanup.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_managed_group_cleanup.py new file mode 100644 index 000000000000..3f5f9e30f1ca --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_managed_group_cleanup.py @@ -0,0 +1,96 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +# Generated by AI (UnitTest MCP) +from unittest.mock import AsyncMock, Mock + +import pytest + +from azure.core.exceptions import HttpResponseError + +import _cloud_validation_testcase as cleanup + + +@pytest.fixture +def no_cleanup_delay(monkeypatch) -> None: + monkeypatch.setattr(cleanup.time, "sleep", Mock()) + monkeypatch.setattr(cleanup.asyncio, "sleep", AsyncMock()) + + +@pytest.mark.parametrize( + "existence", + [ + [False, False, False], + [True, False, False, False], + [False, False, True, False, False, False], + ], +) +def test_cleanup_requires_consecutive_absence(existence, no_cleanup_delay) -> None: + client = Mock() + client.resource_groups.check_existence.side_effect = existence + cleanup.wait_for_managed_group_deletion(client, "owned-mrg") + assert client.resource_groups.check_existence.call_count == len(existence) + client.resource_groups.check_existence.assert_called_with("owned-mrg") + client.resource_groups.begin_delete.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "existence", + [ + [False, False, False], + [True, False, False, False], + [False, False, True, False, False, False], + ], +) +async def test_async_cleanup_requires_consecutive_absence(existence, no_cleanup_delay) -> None: + client = Mock() + client.resource_groups.check_existence = AsyncMock(side_effect=existence) + await cleanup.wait_for_managed_group_deletion_async(client, "owned-mrg") + assert client.resource_groups.check_existence.await_count == len(existence) + client.resource_groups.check_existence.assert_awaited_with("owned-mrg") + client.resource_groups.begin_delete.assert_not_called() + + +def test_cleanup_fails_when_managed_group_remains(no_cleanup_delay) -> None: + client = Mock() + client.resource_groups.check_existence.return_value = True + with pytest.raises(AssertionError, match="cleanup was not confirmed"): + cleanup.wait_for_managed_group_deletion(client, "owned-mrg") + assert client.resource_groups.check_existence.call_count == cleanup.MANAGED_GROUP_CHECKS + client.resource_groups.begin_delete.assert_not_called() + + +@pytest.mark.asyncio +async def test_async_cleanup_fails_when_managed_group_remains(no_cleanup_delay) -> None: + client = Mock() + client.resource_groups.check_existence = AsyncMock(return_value=True) + with pytest.raises(AssertionError, match="cleanup was not confirmed"): + await cleanup.wait_for_managed_group_deletion_async(client, "owned-mrg") + assert client.resource_groups.check_existence.await_count == cleanup.MANAGED_GROUP_CHECKS + client.resource_groups.begin_delete.assert_not_called() + + +def test_cleanup_does_not_treat_request_failure_as_absence(no_cleanup_delay) -> None: + client = Mock() + error = HttpResponseError(message="Authorization failed") + client.resource_groups.check_existence.side_effect = error + with pytest.raises(HttpResponseError) as raised: + cleanup.wait_for_managed_group_deletion(client, "owned-mrg") + assert raised.value is error + client.resource_groups.check_existence.assert_called_once_with("owned-mrg") + client.resource_groups.begin_delete.assert_not_called() + + +@pytest.mark.asyncio +async def test_async_cleanup_does_not_treat_request_failure_as_absence(no_cleanup_delay) -> None: + client = Mock() + error = HttpResponseError(message="Authorization failed") + client.resource_groups.check_existence = AsyncMock(side_effect=error) + with pytest.raises(HttpResponseError) as raised: + await cleanup.wait_for_managed_group_deletion_async(client, "owned-mrg") + assert raised.value is error + client.resource_groups.check_existence.assert_awaited_once_with("owned-mrg") + client.resource_groups.begin_delete.assert_not_called() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_operations.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_operations.py new file mode 100644 index 000000000000..3654a71ff98f --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_operations.py @@ -0,0 +1,45 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +from azure.mgmt.platformvalidation import PlatformValidationMgmtClient +from azure.mgmt.platformvalidation.aio import PlatformValidationMgmtClient as AsyncPlatformValidationMgmtClient +from azure.mgmt.platformvalidation.models import Operation +from devtools_testutils import AzureMgmtRecordedTestCase, recorded_by_proxy +from devtools_testutils.aio import recorded_by_proxy_async + + +def assert_operations(operations: list[Operation]) -> None: + names = {operation.name.lower() for operation in operations if operation.name} + for action in ("read", "write", "delete"): + assert f"microsoft.platformvalidation/cloudvalidations/{action}" in names + assert f"microsoft.platformvalidation/cloudvalidations/validationexecutionplans/{action}" in names + for operation in operations: + # PUT and PATCH may advertise the same ARM permission name. + assert operation.name and operation.name.lower().startswith("microsoft.platformvalidation/") + assert operation.display is not None + assert operation.display.operation + assert operation.display.provider + assert isinstance(operation.is_data_action, bool) + + +class TestOperations(AzureMgmtRecordedTestCase): + @recorded_by_proxy + def test_list_management_operations(self) -> None: + with self.create_mgmt_client(PlatformValidationMgmtClient) as client: + assert_operations(list(client.operations.list())) + + @AzureMgmtRecordedTestCase.await_prepared_test + @recorded_by_proxy_async + async def test_list_management_operations_async(self) -> None: + credential = self.get_credential(AsyncPlatformValidationMgmtClient, is_async=True) + try: + async with self.create_client_from_credential( + AsyncPlatformValidationMgmtClient, + credential, + subscription_id=self.get_settings_value("SUBSCRIPTION_ID"), + ) as client: + assert_operations([operation async for operation in client.operations.list()]) + finally: + await credential.close() diff --git a/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_update_helpers.py b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_update_helpers.py new file mode 100644 index 000000000000..a02807e32e73 --- /dev/null +++ b/sdk/platformvalidation/azure-mgmt-platformvalidation/tests/test_update_helpers.py @@ -0,0 +1,51 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import json +from types import SimpleNamespace as NS +from _execution_testcase import IMAGE_PLACEHOLDER, plan_configuration +from _update_testcase import ( + plan_update, + replacement_plan, + response_observation, + updated_configuration, +) + + +def test_configuration_update_changes_only_metadata_description(): + original = json.loads(plan_configuration("plan", IMAGE_PLACEHOLDER)) + updated = json.loads(updated_configuration(json.dumps(original))) + assert updated["metadata"].pop("description") == "Updated SDK recorded plan metadata" + assert updated == original + + +def test_patch_omits_configuration_when_updating_description(): + assert plan_update(description="updated").as_dict() == {"properties": {"description": "updated"}} + + +def test_replacement_preserves_explicit_configuration(): + result = replacement_plan("eastus", {"sdk-test": "plan"}, "description", "{}").as_dict() + assert result["properties"] == {"description": "description", "planConfigurationJson": "{}"} + assert result["tags"] == {"sdk-test": "plan"} + assert result["location"] == "eastus" + + +def test_observation_discards_credentials_and_payloads(): + result = response_observation( + NS( + http_request=NS(method="PATCH", url="https://management.azure.com/resource?sig=secret"), + http_response=NS( + status_code=202, + headers={"x-ms-request-id": "request", "x-ms-operation-identifier": "private"}, + ), + ) + ) + assert result == { + "method": "PATCH", + "path": "/resource", + "status": 202, + "request_id": "request", + "correlation_id": None, + }