Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions samples/_TEMPLATE/create.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,8 @@
"\n",
"# 2) Service-defined parameters (please do not change these)\n",
"rg_name = utils.get_infra_rg_name(deployment, index)\n",
"supported_infrastructures = [] # ENTER SUPPORTED INFRASTRUCTURES HERE, e.g., [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.AFD_APIM_FE]\n",
"utils.validate_infrastructure(deployment, supported_infrastructures)\n",
"\n",
"# 3) Define the APIs and their operations and policies\n",
"\n",
Expand Down
2 changes: 2 additions & 0 deletions samples/general/create.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@
"\n",
"# 2) Service-defined parameters (please do not change these)\n",
"rg_name = utils.get_infra_rg_name(deployment, index)\n",
"supported_infrastructures = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n",
"utils.validate_infrastructure(deployment, supported_infrastructures)\n",
"\n",
"# 3) Define the APIs and their operations and policies\n",
"\n",
Expand Down
2 changes: 2 additions & 0 deletions samples/load-balancing/create.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,8 @@
"\n",
"# 2) Service-defined parameters (please do not change these)\n",
"rg_name = utils.get_infra_rg_name(deployment, index)\n",
"supported_infrastructures = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA]\n",
"utils.validate_infrastructure(deployment, supported_infrastructures)\n",
"\n",
"# 3) Define the APIs and their operations and policies\n",
"\n",
Expand Down
10 changes: 5 additions & 5 deletions shared/python/apimtypes.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
Types and constants for Azure API Management automation and deployment.
"""

from enum import Enum
from enum import StrEnum
from dataclasses import dataclass
from typing import List, Optional

Expand Down Expand Up @@ -51,7 +51,7 @@ def _read_policy_xml(policy_xml_filepath: str) -> str:
# CLASSES
# ------------------------------

class APIMNetworkMode(str, Enum):
class APIMNetworkMode(StrEnum):
"""
Networking configuration modes for Azure API Management (APIM).
"""
Expand All @@ -62,7 +62,7 @@ class APIMNetworkMode(str, Enum):
NONE = "None" # No explicit network configuration (legacy or default)


class APIM_SKU(str, Enum):
class APIM_SKU(StrEnum):
"""
APIM SKU types.
"""
Expand All @@ -76,7 +76,7 @@ class APIM_SKU(str, Enum):
PREMIUMV2 = "Premiumv2"


class HTTP_VERB(str, Enum):
class HTTP_VERB(StrEnum):
"""
HTTP verbs that can be used for API operations.
"""
Expand All @@ -90,7 +90,7 @@ class HTTP_VERB(str, Enum):
HEAD = "HEAD"


class INFRASTRUCTURE(str, Enum):
class INFRASTRUCTURE(StrEnum):
"""
Infrastructure types for APIM automation scenarios.
"""
Expand Down
168 changes: 87 additions & 81 deletions shared/python/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,73 @@ def get(self, key: str, label: str = '', secure: bool = False) -> str | None:
# PRIVATE METHODS
# ------------------------------

def _cleanup_resources(deployment_name: str, rg_name: str) -> None:
Copy link

Copilot AI May 28, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The function docstring for _cleanup_resources indicates that it raises an Exception on error, but the implementation catches and logs exceptions instead. Update the docstring to reflect the actual behavior.

Copilot uses AI. Check for mistakes.
"""
Clean up resources associated with a deployment in a resource group.
Deletes and purges Cognitive Services, API Management, and Key Vault resources, then deletes the resource group itself.

Args:
deployment_name (str): The deployment name (string).
rg_name (str): The resource group name.

Returns:
None

Raises:
Exception: If an error occurs during cleanup.
"""
if not deployment_name:
print_error("Missing deployment name parameter.")
return

if not rg_name:
print_error("Missing resource group name parameter.")
return

try:
print_info(f"🧹 Cleaning up resource group '{rg_name}'...")

# Show the deployment details
output = run(f"az deployment group show --name {deployment_name} -g {rg_name} -o json", "Deployment retrieved", "Failed to retrieve the deployment")

if output.success and output.json_data:
provisioning_state = output.json_data.get("properties").get("provisioningState")
print_info(f"Deployment provisioning state: {provisioning_state}")

# Delete and purge CognitiveService accounts
output = run(f" az cognitiveservices account list -g {rg_name}", f"Listed CognitiveService accounts", f"Failed to list CognitiveService accounts")
if output.success and output.json_data:
for resource in output.json_data:
print_info(f"Deleting and purging Cognitive Service Account '{resource['name']}' in resource group '{rg_name}'...")
output = run(f"az cognitiveservices account delete -g {rg_name} -n {resource['name']}", f"Cognitive Services '{resource['name']}' deleted", f"Failed to delete Cognitive Services '{resource['name']}'")
output = run(f"az cognitiveservices account purge -g {rg_name} -n {resource['name']} -l \"{resource['location']}\"", f"Cognitive Services '{resource['name']}' purged", f"Failed to purge Cognitive Services '{resource['name']}'")

# Delete and purge APIM resources
output = run(f" az apim list -g {rg_name}", f"Listed APIM resources", f"Failed to list APIM resources")
if output.success and output.json_data:
for resource in output.json_data:
print_info(f"Deleting and purging API Management '{resource['name']}' in resource group '{rg_name}'...")
output = run(f"az apim delete -n {resource['name']} -g {rg_name} -y", f"API Management '{resource['name']}' deleted", f"Failed to delete API Management '{resource['name']}'")
output = run(f"az apim deletedservice purge --service-name {resource['name']} --location \"{resource['location']}\"", f"API Management '{resource['name']}' purged", f"Failed to purge API Management '{resource['name']}'")

# Delete and purge Key Vault resources
output = run(f" az keyvault list -g {rg_name}", f"Listed Key Vault resources", f"Failed to list Key Vault resources")
if output.success and output.json_data:
for resource in output.json_data:
print_info(f"Deleting and purging Key Vault '{resource['name']}' in resource group '{rg_name}'...")
output = run(f"az keyvault delete -n {resource['name']} -g {rg_name}", f"Key Vault '{resource['name']}' deleted", f"Failed to delete Key Vault '{resource['name']}'")
output = run(f"az keyvault purge -n {resource['name']} --location \"{resource['location']}\"", f"Key Vault '{resource['name']}' purged", f"Failed to purge Key Vault '{resource['name']}'")

# Delete the resource group last
print_message(f"🧹 Deleting resource group '{rg_name}'...")
output = run(f"az group delete --name {rg_name} -y", f"Resource group '{rg_name}' deleted", f"Failed to delete resource group '{rg_name}'")

print_message("🧹 Cleanup completed.")

except Exception as e:
print(f"An error occurred during cleanup: {e}")
traceback.print_exc()

def _print_log(message: str, prefix: str = '', color: str = '', output: str = '', duration: str = '', show_time: bool = False, blank_above: bool = False, blank_below: bool = False) -> None:
"""
Print a formatted log message with optional prefix, color, output, duration, and time.
Expand Down Expand Up @@ -163,12 +230,6 @@ def _print_log(message: str, prefix: str = '', color: str = '', output: str = ''
print_warning = lambda msg, output = '', duration = '' : _print_log(msg, '⚠️ ', BOLD_Y, output, duration, True)
print_val = lambda name, value, val_below = False : _print_log(f"{name:<25}:{'\n' if val_below else ' '}{value}", '👉🏽 ', BOLD_B)

# Validation functions will raise ValueError if the value is not valid

validate_http_verb = lambda val: HTTP_VERB(val)
validate_infrastructure = lambda val: INFRASTRUCTURE(val)
validate_sku = lambda val: APIM_SKU(val)

def create_bicep_deployment_group(rg_name: str, rg_location: str, deployment: str | INFRASTRUCTURE, bicep_parameters: dict, bicep_parameters_file: str = 'params.json') -> Output:
"""
Create a Bicep deployment in a resource group, writing parameters to a file and running the deployment.
Expand Down Expand Up @@ -279,78 +340,6 @@ def read_policy_xml(policy_xml_filepath: str) -> str:

return policy_template_xml

def _cleanup_resources(deployment_name: str, rg_name: str) -> None:
"""
Clean up resources associated with a deployment in a resource group.
Deletes and purges Cognitive Services, API Management, and Key Vault resources, then deletes the resource group itself.

Args:
deployment_name (str): The deployment name (string).
rg_name (str): The resource group name.

Returns:
None

Raises:
Exception: If an error occurs during cleanup.
"""
if not deployment_name:
print_error("Missing deployment name parameter.")
return

if not rg_name:
print_error("Missing resource group name parameter.")
return

try:
print_info(f"🧹 Cleaning up resource group '{rg_name}'...")

# Show the deployment details
output = run(f"az deployment group show --name {deployment_name} -g {rg_name} -o json", "Deployment retrieved", "Failed to retrieve the deployment")

if output.success and output.json_data:
provisioning_state = output.json_data.get("properties").get("provisioningState")
print_info(f"Deployment provisioning state: {provisioning_state}")

# Delete and purge CognitiveService accounts
output = run(f" az cognitiveservices account list -g {rg_name}", f"Listed CognitiveService accounts", f"Failed to list CognitiveService accounts")
if output.success and output.json_data:
for resource in output.json_data:
print_info(f"Deleting and purging Cognitive Service Account '{resource['name']}' in resource group '{rg_name}'...")
output = run(f"az cognitiveservices account delete -g {rg_name} -n {resource['name']}", f"Cognitive Services '{resource['name']}' deleted", f"Failed to delete Cognitive Services '{resource['name']}'")
output = run(f"az cognitiveservices account purge -g {rg_name} -n {resource['name']} -l \"{resource['location']}\"", f"Cognitive Services '{resource['name']}' purged", f"Failed to purge Cognitive Services '{resource['name']}'")

# Delete and purge APIM resources
output = run(f" az apim list -g {rg_name}", f"Listed APIM resources", f"Failed to list APIM resources")
if output.success and output.json_data:
for resource in output.json_data:
print_info(f"Deleting and purging API Management '{resource['name']}' in resource group '{rg_name}'...")
output = run(f"az apim delete -n {resource['name']} -g {rg_name} -y", f"API Management '{resource['name']}' deleted", f"Failed to delete API Management '{resource['name']}'")
output = run(f"az apim deletedservice purge --service-name {resource['name']} --location \"{resource['location']}\"", f"API Management '{resource['name']}' purged", f"Failed to purge API Management '{resource['name']}'")

# Delete and purge Key Vault resources
output = run(f" az keyvault list -g {rg_name}", f"Listed Key Vault resources", f"Failed to list Key Vault resources")
if output.success and output.json_data:
for resource in output.json_data:
print_info(f"Deleting and purging Key Vault '{resource['name']}' in resource group '{rg_name}'...")
output = run(f"az keyvault delete -n {resource['name']} -g {rg_name}", f"Key Vault '{resource['name']}' deleted", f"Failed to delete Key Vault '{resource['name']}'")
output = run(f"az keyvault purge -n {resource['name']} --location \"{resource['location']}\"", f"Key Vault '{resource['name']}' purged", f"Failed to purge Key Vault '{resource['name']}'")

# Delete the resource group last
print_message(f"🧹 Deleting resource group '{rg_name}'...")
output = run(f"az group delete --name {rg_name} -y", f"Resource group '{rg_name}' deleted", f"Failed to delete resource group '{rg_name}'")

print_message("🧹 Cleanup completed.")

except Exception as e:
print(f"An error occurred during cleanup: {e}")
traceback.print_exc()


# ------------------------------
# PUBLIC METHODS
# ------------------------------

def cleanup_infra_deployments(deployment: INFRASTRUCTURE, indexes: int | list[int] | None = None) -> None:
"""
Clean up infrastructure deployments by deployment enum and index/indexes.
Expand All @@ -360,7 +349,6 @@ def cleanup_infra_deployments(deployment: INFRASTRUCTURE, indexes: int | list[in
deployment (INFRASTRUCTURE): The infrastructure deployment enum value.
indexes (int | list[int] | None): A single index, a list of indexes, or None for no index.
"""
validate_infrastructure(deployment)

if indexes is None:
indexes_list = [None]
Expand Down Expand Up @@ -545,8 +533,6 @@ def get_infra_rg_name(deployment_name: INFRASTRUCTURE, index: int | None = None)
str: The generated resource group name.
"""

validate_infrastructure(deployment_name)

rg_name = f"apim-infra-{deployment_name.value}"

if index is not None:
Expand Down Expand Up @@ -638,3 +624,23 @@ def run(command: str, ok_message: str = '', error_message: str = '', print_outpu
print_message(ok_message if success else error_message, output_text if not success or print_output else "", f"[{int(minutes)}m:{int(seconds)}s]")

return Output(success, output_text)

# Validation functions will raise ValueError if the value is not valid

validate_http_verb = lambda val: HTTP_VERB(val)
validate_sku = lambda val: APIM_SKU(val)

def validate_infrastructure(infra: INFRASTRUCTURE, supported_infras: list[INFRASTRUCTURE]) -> None:
"""
Validate that the provided infrastructure is a supported infrastructure.

Args:
infra (INFRASTRUCTURE): The infrastructure deployment enum value.
supported_infras (list[INFRASTRUCTURE]): List of supported infrastructures.

Raises:
ValueError: If the infrastructure is not supported.
"""

if infra not in supported_infras:
raise ValueError(f"Unsupported infrastructure: {infra}. Supported infrastructures are: {', '.join([i.value for i in supported_infras])}")
22 changes: 21 additions & 1 deletion tests/python/test_utils.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import pytest
from apimtypes import INFRASTRUCTURE
import os
import builtins
import pytest
from io import StringIO
from unittest.mock import patch, MagicMock, mock_open
from shared.python import utils
Expand Down Expand Up @@ -252,3 +253,22 @@ def test_extract_json_multiple_json_types():
assert utils.extract_json(s) == [1, 2, 3]
s2 = '{"a": 1}[1,2,3]'
assert utils.extract_json(s2) == {"a": 1}

# ------------------------------
# validate_infrastructure
# ------------------------------

def test_validate_infrastructure_supported():
# Should return None for supported infra
assert utils.validate_infrastructure(INFRASTRUCTURE.SIMPLE_APIM, [INFRASTRUCTURE.SIMPLE_APIM]) is None

def test_validate_infrastructure_unsupported():
# Should raise ValueError for unsupported infra
with pytest.raises(ValueError) as exc:
utils.validate_infrastructure(INFRASTRUCTURE.SIMPLE_APIM, [INFRASTRUCTURE.APIM_ACA])
assert "Unsupported infrastructure" in str(exc.value)

def test_validate_infrastructure_multiple_supported():
# Should return True if infra is in the supported list
supported = [INFRASTRUCTURE.SIMPLE_APIM, INFRASTRUCTURE.APIM_ACA]
assert utils.validate_infrastructure(INFRASTRUCTURE.APIM_ACA, supported) is None