From 577633e95624c7517553932d159c538a7a96ae74 Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 07:43:56 -0400 Subject: [PATCH 1/9] Add initial infrastructure classes and tests --- shared/python/infrastructures.py | 213 +++++++++ tests/python/test_infrastructures.py | 633 +++++++++++++++++++++++++++ 2 files changed, 846 insertions(+) create mode 100644 shared/python/infrastructures.py create mode 100644 tests/python/test_infrastructures.py diff --git a/shared/python/infrastructures.py b/shared/python/infrastructures.py new file mode 100644 index 0000000..affafb6 --- /dev/null +++ b/shared/python/infrastructures.py @@ -0,0 +1,213 @@ +""" +Infrastructure Types +""" + +import json +import os +from pathlib import Path +from apimtypes import * +import utils +# from abc import ABC, abstractmethod + + +# ------------------------------ +# INFRASTRUCTURE CLASSES +# ------------------------------ + +class Infrastructure: + """ + Represents the base Infrastructure class + """ + + # ------------------------------ + # CONSTRUCTOR + # ------------------------------ + + def __init__(self, infra: INFRASTRUCTURE, index: int, location: str, apim_sku: APIM_SKU = APIM_SKU.BASICV2, networkMode: APIMNetworkMode = APIMNetworkMode.PUBLIC, + infra_pfs: List[PolicyFragment] | None = None, infra_apis: List[API] | None = None): + self.infra = infra + self.index = index + self.rg_location = location + self.apim_sku = apim_sku + self.networkMode = networkMode + self.infra_apis = infra_apis + self.infra_pfs = infra_pfs + + self.rg_name = utils.get_infra_rg_name(infra, index) + self.rg_tags = utils.build_infrastructure_tags(infra) + + print(f'\nšŸš€ Initializing infrastructure...\n') + print(f' Infrastructure : {self.infra.value}') + print(f' Index : {self.index}') + print(f' Resource group : {self.rg_name}') + print(f' Location : {self.rg_location}') + print(f' APIM SKU : {self.apim_sku.value}\n') + + self._define_policy_fragments(self.infra_pfs) + self._define_apis(self.infra_apis) + + # ------------------------------ + # PUBLIC METHODS + # ------------------------------ + + def deploy_infrastructure(self) -> None: + """ + Deploy the infrastructure using the defined Bicep parameters. + This method should be implemented in subclasses to handle specific deployment logic. + """ + + self._define_bicep_parameters() + + # Change to the infrastructure directory to ensure bicep files are found + original_cwd = os.getcwd() + infra_dir = Path(__file__).parent + + try: + os.chdir(infra_dir) + print(f'šŸ“ Changed working directory to: {infra_dir}') + + # Prepare deployment parameters and run directly to avoid path detection issues + bicep_parameters_format = { + '$schema': 'https://schema.management.azure.com/schemas/2019-04-01/deploymentParameters.json#', + 'contentVersion': '1.0.0.0', + 'parameters': self.bicep_parameters + } + + # Write the parameters file + params_file_path = infra_dir / 'params.json' + + with open(params_file_path, 'w') as file: + file.write(json.dumps(bicep_parameters_format)) + + print(f"šŸ“ Updated the policy XML in the bicep parameters file 'params.json'") + + # ------------------------------ + # EXECUTE DEPLOYMENT + # ------------------------------ + + # Create the resource group if it doesn't exist + utils.create_resource_group(self.rg_name, self.rg_location, self.rg_tags) + + # Run the deployment directly + main_bicep_path = infra_dir / 'main.bicep' + output = utils.run( + f'az deployment group create --name {self.infra.value} --resource-group {self.rg_name} --template-file "{main_bicep_path}" --parameters "{params_file_path}" --query "properties.outputs"', + f"Deployment '{self.infra.value}' succeeded", + f"Deployment '{self.infra.value}' failed.", + print_command_to_run = False + ) + + # ------------------------------ + # VERIFY DEPLOYMENT RESULTS + # ------------------------------ + + if output.success: + print('\nāœ… Infrastructure creation completed successfully!') + if output.json_data: + apim_gateway_url = output.get('apimResourceGatewayURL', 'APIM API Gateway URL', suppress_logging = True) + apim_apis = output.getJson('apiOutputs', 'APIs', suppress_logging = True) + + print(f'\nšŸ“‹ Infrastructure Details:') + print(f' Resource Group : {self.rg_name}') + print(f' Location : {self.rg_location}') + print(f' APIM SKU : {self.apim_sku.value}') + print(f' Gateway URL : {apim_gateway_url}') + print(f' APIs Created : {len(apim_apis)}') + + # TODO: Perform basic verification + # utils.verify_infrastructure(self.rg_name) + else: + print('āŒ Infrastructure creation failed!') + + return output + + finally: + # Always restore the original working directory + os.chdir(original_cwd) + print(f'šŸ“ Restored working directory to: {original_cwd}') + + # @abstractmethod + # def verify_infrastructure(self) -> bool: + # """ + # Verify the infrastructure deployment. + # This method should be implemented in subclasses to handle specific verification logic. + # """ + # pass + + # ------------------------------ + # PRIVATE METHODS + # ------------------------------ + + def _define_bicep_parameters(self) -> dict: + # Define the Bicep parameters with serialized APIs + self.bicep_parameters = { + 'apimSku' : {'value': self.apim_sku.value}, + 'apis' : {'value': [api.to_dict() for api in self.apis]}, + 'policyFragments' : {'value': [pf.to_dict() for pf in self.pfs]} + } + + return self.bicep_parameters + + + def _define_policy_fragments(self, infra_pfs: List[PolicyFragment] | None) -> List[PolicyFragment]: + """ + Define policy fragments for the infrastructure. + """ + + # The base policy fragments common to all infrastructures + self.base_pfs = [ + PolicyFragment('AuthZ-Match-All', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-authz-match-all.xml')), 'Authorizes if all of the specified roles match the JWT role claims.'), + PolicyFragment('AuthZ-Match-Any', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-authz-match-any.xml')), 'Authorizes if any of the specified roles match the JWT role claims.'), + PolicyFragment('Http-Response-200', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-http-response-200.xml')), 'Returns a 200 OK response for the current HTTP method.'), + PolicyFragment('Product-Match-Any', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-product-match-any.xml')), 'Proceeds if any of the specified products match the context product name.'), + PolicyFragment('Remove-Request-Headers', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-remove-request-headers.xml')), 'Removes request headers from the incoming request.') + ] + + # Combine base policy fragments with infrastructure-specific ones + self.pfs = self.base_pfs + infra_pfs if infra_pfs else self.base_pfs + + return self.pfs + + def _define_apis(self, infra_apis: List[API] | None) -> List[API]: + """ + Define APIs for the infrastructure. + """ + + # The base APIs common to all infrastructures + # Hello World API + pol_hello_world = utils.read_policy_xmll(HELLO_WORLD_XML_POLICY_PATH) + api_hwroot_get = GET_APIOperation('Gets a Hello World message', pol_hello_world) + api_hwroot = API('hello-world', 'Hello World', '', 'This is the root API for Hello World', operations = [api_hwroot_get]) + self.base_apis = [api_hwroot] + + # Combine base APIs with infrastructure-specific ones + self.apis = self.base_apis + infra_apis if infra_apis else self.base_apis + + return self.apis + + +class SimpleApimInfrastructure(Infrastructure): + """ + Represents a simple API Management infrastructure. + """ + + def __init__(self, location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2): + super().__init__(INFRASTRUCTURE.SIMPLE_APIM, index, location, apim_sku, APIMNetworkMode.PUBLIC) + + +class ApimAcaInfrastructure(Infrastructure): + """ + Represents an API Management with Azure Container Apps infrastructure. + """ + + def __init__(self, location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2): + super().__init__(INFRASTRUCTURE.APIM_ACA, index, location, apim_sku, APIMNetworkMode.PUBLIC) + + +class AfdApimAcaInfrastructure(Infrastructure): + """ + Represents an Azure Front Door with API Management and Azure Container Apps infrastructure. + """ + + def __init__(self, location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2): + super().__init__(INFRASTRUCTURE.AFD_APIM_PE, index, location, apim_sku, APIMNetworkMode.PUBLIC) diff --git a/tests/python/test_infrastructures.py b/tests/python/test_infrastructures.py new file mode 100644 index 0000000..e4d9ba1 --- /dev/null +++ b/tests/python/test_infrastructures.py @@ -0,0 +1,633 @@ +""" +Unit tests for infrastructures.py. +""" + +import pytest +from unittest.mock import Mock, patch, call, MagicMock +import json +import os +from pathlib import Path + +import infrastructures +from apimtypes import INFRASTRUCTURE, APIM_SKU, APIMNetworkMode, API, PolicyFragment, HTTP_VERB, GET_APIOperation + + +# ------------------------------ +# CONSTANTS +# ------------------------------ + +TEST_LOCATION = 'eastus2' +TEST_INDEX = 1 +TEST_APIM_SKU = APIM_SKU.BASICV2 +TEST_NETWORK_MODE = APIMNetworkMode.PUBLIC + + +# ------------------------------ +# FIXTURES +# ------------------------------ + +@pytest.fixture +def mock_utils(): + """Mock the utils module to avoid external dependencies.""" + with patch('infrastructures.utils') as mock_utils: + mock_utils.get_infra_rg_name.return_value = 'rg-test-infrastructure-01' + mock_utils.build_infrastructure_tags.return_value = {'environment': 'test', 'project': 'apim-samples'} + mock_utils.read_policy_xmll.return_value = '' + mock_utils.determine_shared_policy_path.return_value = '/mock/path/policy.xml' + mock_utils.create_resource_group.return_value = None + mock_utils.verify_infrastructure.return_value = True + + # Mock the run command with proper return object + mock_output = Mock() + mock_output.success = True + mock_output.json_data = {'outputs': 'test'} + mock_output.get.return_value = 'https://test-apim.azure-api.net' + mock_output.getJson.return_value = ['api1', 'api2'] + mock_utils.run.return_value = mock_output + + yield mock_utils + +@pytest.fixture +def mock_policy_fragments(): + """Provide mock policy fragments for testing.""" + return [ + PolicyFragment('Test-Fragment-1', 'test1', 'Test fragment 1'), + PolicyFragment('Test-Fragment-2', 'test2', 'Test fragment 2') + ] + +@pytest.fixture +def mock_apis(): + """Provide mock APIs for testing.""" + return [ + API('test-api-1', 'Test API 1', '/test1', 'Test API 1 description', 'api1'), + API('test-api-2', 'Test API 2', '/test2', 'Test API 2 description', 'api2') + ] + + +# ------------------------------ +# BASE INFRASTRUCTURE CLASS TESTS +# ------------------------------ + +@pytest.mark.unit +def test_infrastructure_creation_basic(mock_utils): + """Test basic Infrastructure creation with default values.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + assert infra.infra == INFRASTRUCTURE.SIMPLE_APIM + assert infra.index == TEST_INDEX + assert infra.rg_location == TEST_LOCATION + assert infra.apim_sku == APIM_SKU.BASICV2 # default value + assert infra.networkMode == APIMNetworkMode.PUBLIC # default value + assert infra.rg_name == 'rg-test-infrastructure-01' + assert infra.rg_tags == {'environment': 'test', 'project': 'apim-samples'} + +@pytest.mark.unit +def test_infrastructure_creation_with_custom_values(mock_utils): + """Test Infrastructure creation with custom values.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.APIM_ACA, + index=2, + location='westus2', + apim_sku=APIM_SKU.PREMIUM, + networkMode=APIMNetworkMode.EXTERNAL_VNET + ) + + assert infra.infra == INFRASTRUCTURE.APIM_ACA + assert infra.index == 2 + assert infra.rg_location == 'westus2' + assert infra.apim_sku == APIM_SKU.PREMIUM + assert infra.networkMode == APIMNetworkMode.EXTERNAL_VNET + +@pytest.mark.unit +def test_infrastructure_creation_with_custom_policy_fragments(mock_utils, mock_policy_fragments): + """Test Infrastructure creation with custom policy fragments.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION, + infra_pfs=mock_policy_fragments + ) + + # Should have base policy fragments + custom ones + assert len(infra.pfs) == 7 # 5 base + 2 custom + assert any(pf.name == 'Test-Fragment-1' for pf in infra.pfs) + assert any(pf.name == 'Test-Fragment-2' for pf in infra.pfs) + assert any(pf.name == 'AuthZ-Match-All' for pf in infra.pfs) + +@pytest.mark.unit +def test_infrastructure_creation_with_custom_apis(mock_utils, mock_apis): + """Test Infrastructure creation with custom APIs.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION, + infra_apis=mock_apis + ) + + # Should have base APIs + custom ones + assert len(infra.apis) == 3 # 1 base (hello-world) + 2 custom + assert any(api.name == 'test-api-1' for api in infra.apis) + assert any(api.name == 'test-api-2' for api in infra.apis) + assert any(api.name == 'hello-world' for api in infra.apis) + +@pytest.mark.unit +def test_infrastructure_creation_calls_utils_functions(mock_utils): + """Test that Infrastructure creation calls expected utility functions.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + mock_utils.get_infra_rg_name.assert_called_once_with(INFRASTRUCTURE.SIMPLE_APIM, TEST_INDEX) + mock_utils.build_infrastructure_tags.assert_called_once_with(INFRASTRUCTURE.SIMPLE_APIM) + + # Should call read_policy_xmll for base policy fragments + assert mock_utils.read_policy_xmll.call_count >= 5 # At least 5 base policy fragments + assert mock_utils.determine_shared_policy_path.call_count >= 5 + +@pytest.mark.unit +def test_infrastructure_base_policy_fragments_creation(mock_utils): + """Test that base policy fragments are created correctly.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + # Check that all base policy fragments are created + expected_fragment_names = [ + 'AuthZ-Match-All', + 'AuthZ-Match-Any', + 'Http-Response-200', + 'Product-Match-Any', + 'Remove-Request-Headers' + ] + + base_fragment_names = [pf.name for pf in infra.base_pfs] + for expected_name in expected_fragment_names: + assert expected_name in base_fragment_names + +@pytest.mark.unit +def test_infrastructure_base_apis_creation(mock_utils): + """Test that base APIs are created correctly.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + # Check that hello-world API is created + assert len(infra.base_apis) == 1 + hello_world_api = infra.base_apis[0] + assert hello_world_api.name == 'hello-world' + assert hello_world_api.displayName == 'Hello World' + assert hello_world_api.path == '' + assert len(hello_world_api.operations) == 1 + assert hello_world_api.operations[0].method == HTTP_VERB.GET + + +# ------------------------------ +# POLICY FRAGMENT TESTS +# ------------------------------ + +@pytest.mark.unit +def test_define_policy_fragments_with_none_input(mock_utils): + """Test _define_policy_fragments with None input.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION, + infra_pfs=None + ) + + # Should only have base policy fragments + assert len(infra.pfs) == 5 + assert all(pf.name in ['AuthZ-Match-All', 'AuthZ-Match-Any', 'Http-Response-200', 'Product-Match-Any', 'Remove-Request-Headers'] for pf in infra.pfs) + +@pytest.mark.unit +def test_define_policy_fragments_with_custom_input(mock_utils, mock_policy_fragments): + """Test _define_policy_fragments with custom input.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION, + infra_pfs=mock_policy_fragments + ) + + # Should have base + custom policy fragments + assert len(infra.pfs) == 7 # 5 base + 2 custom + fragment_names = [pf.name for pf in infra.pfs] + assert 'Test-Fragment-1' in fragment_names + assert 'Test-Fragment-2' in fragment_names + assert 'AuthZ-Match-All' in fragment_names + + +# ------------------------------ +# API TESTS +# ------------------------------ + +@pytest.mark.unit +def test_define_apis_with_none_input(mock_utils): + """Test _define_apis with None input.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION, + infra_apis=None + ) + + # Should only have base APIs + assert len(infra.apis) == 1 + assert infra.apis[0].name == 'hello-world' + +@pytest.mark.unit +def test_define_apis_with_custom_input(mock_utils, mock_apis): + """Test _define_apis with custom input.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION, + infra_apis=mock_apis + ) + + # Should have base + custom APIs + assert len(infra.apis) == 3 # 1 base + 2 custom + api_names = [api.name for api in infra.apis] + assert 'test-api-1' in api_names + assert 'test-api-2' in api_names + assert 'hello-world' in api_names + + +# ------------------------------ +# BICEP PARAMETERS TESTS +# ------------------------------ + +@pytest.mark.unit +def test_define_bicep_parameters(mock_utils): + """Test _define_bicep_parameters method.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + bicep_params = infra._define_bicep_parameters() + + assert 'apimSku' in bicep_params + assert bicep_params['apimSku']['value'] == APIM_SKU.BASICV2.value + + assert 'apis' in bicep_params + assert isinstance(bicep_params['apis']['value'], list) + assert len(bicep_params['apis']['value']) == 1 # hello-world API + + assert 'policyFragments' in bicep_params + assert isinstance(bicep_params['policyFragments']['value'], list) + assert len(bicep_params['policyFragments']['value']) == 5 # base policy fragments + + +# ------------------------------ +# ABSTRACT METHOD TESTS +# ------------------------------ + +# ------------------------------ +# DEPLOYMENT TESTS +# ------------------------------ + +@pytest.mark.unit +@patch('os.getcwd') +@patch('os.chdir') +@patch('pathlib.Path') +def test_deploy_infrastructure_success(mock_path_class, mock_chdir, mock_getcwd, mock_utils): + """Test successful infrastructure deployment.""" + # Setup mocks + mock_getcwd.return_value = '/original/path' + mock_infra_dir = Mock() + mock_path_instance = Mock() + mock_path_instance.parent = mock_infra_dir + mock_path_class.return_value = mock_path_instance + + # Create a concrete subclass for testing + class TestInfrastructure(infrastructures.Infrastructure): + def verify_infrastructure(self) -> bool: + return True + + # Mock file writing and JSON dumps to avoid MagicMock serialization issues + mock_open = MagicMock() + + with patch('builtins.open', mock_open), \ + patch('json.dumps', return_value='{"mocked": "params"}') as mock_json_dumps: + + infra = TestInfrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + result = infra.deploy_infrastructure() + + # Verify the deployment process + mock_utils.create_resource_group.assert_called_once() + mock_utils.run.assert_called_once() + # Note: utils.verify_infrastructure is currently commented out in the actual code + # mock_utils.verify_infrastructure.assert_called_once() + + # Verify directory changes - just check that chdir was called twice (to infra dir and back) + assert mock_chdir.call_count == 2 + # Second call should restore original path + mock_chdir.assert_any_call('/original/path') + + # Verify file writing (open will be called multiple times - for reading policies and writing params) + assert mock_open.call_count >= 1 # At least called once for writing params.json + mock_json_dumps.assert_called_once() + + assert result.success is True + +@pytest.mark.unit +@patch('os.getcwd') +@patch('os.chdir') +@patch('pathlib.Path') +def test_deploy_infrastructure_failure(mock_path_class, mock_chdir, mock_getcwd, mock_utils): + """Test infrastructure deployment failure.""" + # Setup mocks for failure scenario + mock_getcwd.return_value = '/original/path' + mock_infra_dir = Mock() + mock_path_instance = Mock() + mock_path_instance.parent = mock_infra_dir + mock_path_class.return_value = mock_path_instance + + # Mock failed deployment + mock_output = Mock() + mock_output.success = False + mock_utils.run.return_value = mock_output + + # Create a concrete subclass for testing + class TestInfrastructure(infrastructures.Infrastructure): + def verify_infrastructure(self) -> bool: + return True + + # Mock file operations to prevent actual file writes and JSON serialization issues + with patch('builtins.open', MagicMock()), \ + patch('json.dumps', return_value='{"mocked": "params"}'): + + infra = TestInfrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + result = infra.deploy_infrastructure() + + # Verify the deployment process was attempted + mock_utils.create_resource_group.assert_called_once() + mock_utils.run.assert_called_once() + # Note: utils.verify_infrastructure is currently commented out in the actual code + # mock_utils.verify_infrastructure.assert_not_called() # Should not be called on failure + + # Verify directory changes (should restore even on failure) + assert mock_chdir.call_count == 2 + # Second call should restore original path + mock_chdir.assert_any_call('/original/path') + + assert result.success is False + + +# ------------------------------ +# CONCRETE INFRASTRUCTURE CLASSES TESTS +# ------------------------------ + +@pytest.mark.unit +def test_simple_apim_infrastructure_creation(mock_utils): + """Test SimpleApimInfrastructure creation.""" + infra = infrastructures.SimpleApimInfrastructure( + location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.DEVELOPER + ) + + assert infra.infra == INFRASTRUCTURE.SIMPLE_APIM + assert infra.index == TEST_INDEX + assert infra.rg_location == TEST_LOCATION + assert infra.apim_sku == APIM_SKU.DEVELOPER + assert infra.networkMode == APIMNetworkMode.PUBLIC + +@pytest.mark.unit +def test_simple_apim_infrastructure_defaults(mock_utils): + """Test SimpleApimInfrastructure with default values.""" + infra = infrastructures.SimpleApimInfrastructure( + location=TEST_LOCATION, + index=TEST_INDEX + ) + + assert infra.apim_sku == APIM_SKU.BASICV2 # default value + +@pytest.mark.unit +def test_apim_aca_infrastructure_creation(mock_utils): + """Test ApimAcaInfrastructure creation.""" + infra = infrastructures.ApimAcaInfrastructure( + location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.STANDARD + ) + + assert infra.infra == INFRASTRUCTURE.APIM_ACA + assert infra.index == TEST_INDEX + assert infra.rg_location == TEST_LOCATION + assert infra.apim_sku == APIM_SKU.STANDARD + assert infra.networkMode == APIMNetworkMode.PUBLIC + +@pytest.mark.unit +def test_afd_apim_aca_infrastructure_creation(mock_utils): + """Test AfdApimAcaInfrastructure creation.""" + infra = infrastructures.AfdApimAcaInfrastructure( + location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.PREMIUM + ) + + assert infra.infra == INFRASTRUCTURE.AFD_APIM_PE + assert infra.index == TEST_INDEX + assert infra.rg_location == TEST_LOCATION + assert infra.apim_sku == APIM_SKU.PREMIUM + assert infra.networkMode == APIMNetworkMode.PUBLIC + + +# ------------------------------ +# INTEGRATION TESTS +# ------------------------------ + +@pytest.mark.unit +def test_infrastructure_end_to_end_simple(mock_utils): + """Test end-to-end Infrastructure creation with SimpleApim.""" + infra = infrastructures.SimpleApimInfrastructure( + location='eastus', + index=1, + apim_sku=APIM_SKU.DEVELOPER + ) + + # Verify all components are created correctly + assert infra.infra == INFRASTRUCTURE.SIMPLE_APIM + assert len(infra.base_pfs) == 5 + assert len(infra.pfs) == 5 + assert len(infra.base_apis) == 1 + assert len(infra.apis) == 1 + + # Verify bicep parameters + bicep_params = infra._define_bicep_parameters() + assert bicep_params['apimSku']['value'] == 'Developer' + assert len(bicep_params['apis']['value']) == 1 + assert len(bicep_params['policyFragments']['value']) == 5 + +@pytest.mark.unit +def test_infrastructure_with_all_custom_components(mock_utils, mock_policy_fragments, mock_apis): + """Test Infrastructure creation with all custom components.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.APIM_ACA, + index=2, + location='westus2', + apim_sku=APIM_SKU.PREMIUM, + networkMode=APIMNetworkMode.EXTERNAL_VNET, + infra_pfs=mock_policy_fragments, + infra_apis=mock_apis + ) + + # Verify all components are combined correctly + assert len(infra.base_pfs) == 5 + assert len(infra.pfs) == 7 # 5 base + 2 custom + assert len(infra.base_apis) == 1 + assert len(infra.apis) == 3 # 1 base + 2 custom + + # Verify bicep parameters include all components + bicep_params = infra._define_bicep_parameters() + assert bicep_params['apimSku']['value'] == 'Premium' + assert len(bicep_params['apis']['value']) == 3 + assert len(bicep_params['policyFragments']['value']) == 7 + + +# ------------------------------ +# ERROR HANDLING TESTS +# ------------------------------ + +@pytest.mark.unit +def test_infrastructure_missing_required_params(): + """Test Infrastructure creation with missing required parameters.""" + with pytest.raises(TypeError): + infrastructures.Infrastructure() + + with pytest.raises(TypeError): + infrastructures.Infrastructure(infra=INFRASTRUCTURE.SIMPLE_APIM) + +@pytest.mark.unit +def test_concrete_infrastructure_missing_params(): + """Test concrete infrastructure classes with missing parameters.""" + with pytest.raises(TypeError): + infrastructures.SimpleApimInfrastructure() + + with pytest.raises(TypeError): + infrastructures.SimpleApimInfrastructure(location=TEST_LOCATION) + + +# ------------------------------ +# EDGE CASES AND COVERAGE TESTS +# ------------------------------ + +@pytest.mark.unit +def test_infrastructure_empty_custom_lists(mock_utils): + """Test Infrastructure with empty custom lists.""" + empty_pfs = [] + empty_apis = [] + + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION, + infra_pfs=empty_pfs, + infra_apis=empty_apis + ) + + # Empty lists should behave the same as None + assert len(infra.pfs) == 5 # Only base policy fragments + assert len(infra.apis) == 1 # Only base APIs + +@pytest.mark.unit +def test_infrastructure_attribute_access(mock_utils): + """Test that all Infrastructure attributes are accessible.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + # Test all attributes are accessible + assert hasattr(infra, 'infra') + assert hasattr(infra, 'index') + assert hasattr(infra, 'rg_location') + assert hasattr(infra, 'apim_sku') + assert hasattr(infra, 'networkMode') + assert hasattr(infra, 'rg_name') + assert hasattr(infra, 'rg_tags') + assert hasattr(infra, 'base_pfs') + assert hasattr(infra, 'pfs') + assert hasattr(infra, 'base_apis') + assert hasattr(infra, 'apis') + # bicep_parameters is only created during deployment via _define_bicep_parameters() + infra._define_bicep_parameters() + assert hasattr(infra, 'bicep_parameters') + +@pytest.mark.unit +def test_infrastructure_string_representation(mock_utils): + """Test Infrastructure string representation.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + # Test that the object can be converted to string without error + str_repr = str(infra) + assert isinstance(str_repr, str) + assert 'Infrastructure' in str_repr + +@pytest.mark.unit +def test_all_infrastructure_types_coverage(mock_utils): + """Test that all infrastructure types can be instantiated.""" + # Test all concrete infrastructure classes + simple_infra = infrastructures.SimpleApimInfrastructure(TEST_LOCATION, TEST_INDEX) + assert simple_infra.infra == INFRASTRUCTURE.SIMPLE_APIM + + aca_infra = infrastructures.ApimAcaInfrastructure(TEST_LOCATION, TEST_INDEX) + assert aca_infra.infra == INFRASTRUCTURE.APIM_ACA + + afd_infra = infrastructures.AfdApimAcaInfrastructure(TEST_LOCATION, TEST_INDEX) + assert afd_infra.infra == INFRASTRUCTURE.AFD_APIM_PE + +@pytest.mark.unit +def test_policy_fragment_creation_robustness(mock_utils): + """Test that policy fragment creation is robust.""" + # Test with various mock return values + mock_utils.read_policy_xmll.side_effect = [ + '', + '', + '', + '', + '', + '' + ] + + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + location=TEST_LOCATION + ) + + # Verify all policy fragments were created with different XML + policy_xmls = [pf.policyXml for pf in infra.base_pfs] + assert '' in policy_xmls + assert '' in policy_xmls + assert '' in policy_xmls + assert '' in policy_xmls + assert '' in policy_xmls From 56a55070ba9c19da6038273b23dec90af36ab0ac Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 09:22:00 -0400 Subject: [PATCH 2/9] Further refinements to infrastructure and tests --- shared/python/infrastructures.py | 216 ++++++++++++++++++--------- tests/python/test_infrastructures.py | 145 ++++++++++++------ 2 files changed, 247 insertions(+), 114 deletions(-) diff --git a/shared/python/infrastructures.py b/shared/python/infrastructures.py index affafb6..d56aff9 100644 --- a/shared/python/infrastructures.py +++ b/shared/python/infrastructures.py @@ -23,11 +23,11 @@ class Infrastructure: # CONSTRUCTOR # ------------------------------ - def __init__(self, infra: INFRASTRUCTURE, index: int, location: str, apim_sku: APIM_SKU = APIM_SKU.BASICV2, networkMode: APIMNetworkMode = APIMNetworkMode.PUBLIC, + def __init__(self, infra: INFRASTRUCTURE, index: int, rg_location: str, apim_sku: APIM_SKU = APIM_SKU.BASICV2, networkMode: APIMNetworkMode = APIMNetworkMode.PUBLIC, infra_pfs: List[PolicyFragment] | None = None, infra_apis: List[API] | None = None): self.infra = infra self.index = index - self.rg_location = location + self.rg_location = rg_location self.apim_sku = apim_sku self.networkMode = networkMode self.infra_apis = infra_apis @@ -36,31 +36,158 @@ def __init__(self, infra: INFRASTRUCTURE, index: int, location: str, apim_sku: A self.rg_name = utils.get_infra_rg_name(infra, index) self.rg_tags = utils.build_infrastructure_tags(infra) - print(f'\nšŸš€ Initializing infrastructure...\n') - print(f' Infrastructure : {self.infra.value}') - print(f' Index : {self.index}') - print(f' Resource group : {self.rg_name}') - print(f' Location : {self.rg_location}') - print(f' APIM SKU : {self.apim_sku.value}\n') - self._define_policy_fragments(self.infra_pfs) - self._define_apis(self.infra_apis) + # ------------------------------ + # PRIVATE METHODS + # ------------------------------ + + def _define_bicep_parameters(self) -> dict: + # Define the Bicep parameters with serialized APIs + self.bicep_parameters = { + 'apimSku' : {'value': self.apim_sku.value}, + 'apis' : {'value': [api.to_dict() for api in self.apis]}, + 'policyFragments' : {'value': [pf.to_dict() for pf in self.pfs]} + } + + return self.bicep_parameters + + + def _define_policy_fragments(self) -> List[PolicyFragment]: + """ + Define policy fragments for the infrastructure. + """ + + # The base policy fragments common to all infrastructures + self.base_pfs = [ + PolicyFragment('AuthZ-Match-All', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-all.xml')), 'Authorizes if all of the specified roles match the JWT role claims.'), + PolicyFragment('AuthZ-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-any.xml')), 'Authorizes if any of the specified roles match the JWT role claims.'), + PolicyFragment('Http-Response-200', utils.read_policy_xml(utils.determine_shared_policy_path('pf-http-response-200.xml')), 'Returns a 200 OK response for the current HTTP method.'), + PolicyFragment('Product-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-product-match-any.xml')), 'Proceeds if any of the specified products match the context product name.'), + PolicyFragment('Remove-Request-Headers', utils.read_policy_xml(utils.determine_shared_policy_path('pf-remove-request-headers.xml')), 'Removes request headers from the incoming request.') + ] + + # Combine base policy fragments with infrastructure-specific ones + self.pfs = self.base_pfs + self.infra_pfs if self.infra_pfs else self.base_pfs + + return self.pfs + + def _define_apis(self) -> List[API]: + """ + Define APIs for the infrastructure. + """ + + # The base APIs common to all infrastructures + # Hello World API + pol_hello_world = utils.read_policy_xml(HELLO_WORLD_XML_POLICY_PATH) + api_hwroot_get = GET_APIOperation('Gets a Hello World message', pol_hello_world) + api_hwroot = API('hello-world', 'Hello World', '', 'This is the root API for Hello World', operations = [api_hwroot_get]) + self.base_apis = [api_hwroot] + + # Combine base APIs with infrastructure-specific ones + self.apis = self.base_apis + self.infra_apis if self.infra_apis else self.base_apis + + return self.apis + + def _verify_infrastructure(self, rg_name: str) -> bool: + """ + Verify that the infrastructure was created successfully. + + Args: + rg_name (str): Resource group name. + + Returns: + bool: True if verification passed, False otherwise. + """ + + print('\nšŸ” Verifying infrastructure...') + + try: + # Check if the resource group exists + if not utils.does_resource_group_exist(rg_name): + print('āŒ Resource group does not exist!') + return False + + print('āœ… Resource group verified') + + # Get APIM service details + output = utils.run(f'az apim list -g {rg_name} --query "[0]" -o json', print_command_to_run = False, print_errors = False) + + if output.success and output.json_data: + apim_name = output.json_data.get('name') + + print(f'āœ… APIM Service verified: {apim_name}') + + # Get API count + api_output = utils.run(f'az apim api list --service-name {apim_name} -g {rg_name} --query "length(@)"', + print_command_to_run = False, print_errors = False) + + if api_output.success: + api_count = int(api_output.text.strip()) + print(f'āœ… APIs verified: {api_count} API(s) created') + + # Test basic connectivity (optional) + if api_count > 0: + try: + # Get subscription key for testing + sub_output = utils.run(f'az apim subscription list --service-name {apim_name} -g {rg_name} --query "[0].primaryKey" -o tsv', + print_command_to_run = False, print_errors = False) + + if sub_output.success and sub_output.text.strip(): + print('āœ… Subscription key available for API testing') + except: + pass + + print('\nšŸŽ‰ Infrastructure verification completed successfully!') + return True + + else: + print('\nāŒ APIM service not found!') + return False + + except Exception as e: + print(f'\nāš ļø Verification failed with error: {str(e)}') + return False # ------------------------------ # PUBLIC METHODS # ------------------------------ - def deploy_infrastructure(self) -> None: + def deploy_infrastructure(self) -> 'utils.Output': """ Deploy the infrastructure using the defined Bicep parameters. This method should be implemented in subclasses to handle specific deployment logic. """ + print(f'\nšŸš€ Creating infrastructure...\n') + print(f' Infrastructure : {self.infra.value}') + print(f' Index : {self.index}') + print(f' Resource group : {self.rg_name}') + print(f' Location : {self.rg_location}') + print(f' APIM SKU : {self.apim_sku.value}\n') + + self._define_policy_fragments() + self._define_apis() self._define_bicep_parameters() - # Change to the infrastructure directory to ensure bicep files are found + # Determine the correct infrastructure directory based on the infrastructure type original_cwd = os.getcwd() - infra_dir = Path(__file__).parent + + # Map infrastructure types to their directory names + infra_dir_map = { + INFRASTRUCTURE.SIMPLE_APIM: 'simple-apim', + INFRASTRUCTURE.APIM_ACA: 'apim-aca', + INFRASTRUCTURE.AFD_APIM_PE: 'afd-apim-pe' + } + + # Get the infrastructure directory + infra_dir_name = infra_dir_map.get(self.infra) + if not infra_dir_name: + raise ValueError(f"Unknown infrastructure type: {self.infra}") + + # Navigate to the correct infrastructure directory + # From shared/python -> ../../infrastructure/{infra_type}/ + shared_dir = Path(__file__).parent + infra_dir = shared_dir.parent.parent / 'infrastructure' / infra_dir_name try: os.chdir(infra_dir) @@ -115,7 +242,7 @@ def deploy_infrastructure(self) -> None: print(f' APIs Created : {len(apim_apis)}') # TODO: Perform basic verification - # utils.verify_infrastructure(self.rg_name) + self._verify_infrastructure(self.rg_name) else: print('āŒ Infrastructure creation failed!') @@ -134,65 +261,16 @@ def deploy_infrastructure(self) -> None: # """ # pass - # ------------------------------ - # PRIVATE METHODS - # ------------------------------ - - def _define_bicep_parameters(self) -> dict: - # Define the Bicep parameters with serialized APIs - self.bicep_parameters = { - 'apimSku' : {'value': self.apim_sku.value}, - 'apis' : {'value': [api.to_dict() for api in self.apis]}, - 'policyFragments' : {'value': [pf.to_dict() for pf in self.pfs]} - } - - return self.bicep_parameters - def _define_policy_fragments(self, infra_pfs: List[PolicyFragment] | None) -> List[PolicyFragment]: - """ - Define policy fragments for the infrastructure. - """ - - # The base policy fragments common to all infrastructures - self.base_pfs = [ - PolicyFragment('AuthZ-Match-All', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-authz-match-all.xml')), 'Authorizes if all of the specified roles match the JWT role claims.'), - PolicyFragment('AuthZ-Match-Any', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-authz-match-any.xml')), 'Authorizes if any of the specified roles match the JWT role claims.'), - PolicyFragment('Http-Response-200', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-http-response-200.xml')), 'Returns a 200 OK response for the current HTTP method.'), - PolicyFragment('Product-Match-Any', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-product-match-any.xml')), 'Proceeds if any of the specified products match the context product name.'), - PolicyFragment('Remove-Request-Headers', utils.read_policy_xmll(utils.determine_shared_policy_path('pf-remove-request-headers.xml')), 'Removes request headers from the incoming request.') - ] - - # Combine base policy fragments with infrastructure-specific ones - self.pfs = self.base_pfs + infra_pfs if infra_pfs else self.base_pfs - - return self.pfs - - def _define_apis(self, infra_apis: List[API] | None) -> List[API]: - """ - Define APIs for the infrastructure. - """ - - # The base APIs common to all infrastructures - # Hello World API - pol_hello_world = utils.read_policy_xmll(HELLO_WORLD_XML_POLICY_PATH) - api_hwroot_get = GET_APIOperation('Gets a Hello World message', pol_hello_world) - api_hwroot = API('hello-world', 'Hello World', '', 'This is the root API for Hello World', operations = [api_hwroot_get]) - self.base_apis = [api_hwroot] - - # Combine base APIs with infrastructure-specific ones - self.apis = self.base_apis + infra_apis if infra_apis else self.base_apis - - return self.apis - class SimpleApimInfrastructure(Infrastructure): """ Represents a simple API Management infrastructure. """ - def __init__(self, location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2): - super().__init__(INFRASTRUCTURE.SIMPLE_APIM, index, location, apim_sku, APIMNetworkMode.PUBLIC) + def __init__(self, rg_location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2, infra_pfs: List[PolicyFragment] | None = None, infra_apis: List[API] | None = None): + super().__init__(INFRASTRUCTURE.SIMPLE_APIM, index, rg_location, apim_sku, APIMNetworkMode.PUBLIC, infra_pfs, infra_apis) class ApimAcaInfrastructure(Infrastructure): @@ -200,8 +278,8 @@ class ApimAcaInfrastructure(Infrastructure): Represents an API Management with Azure Container Apps infrastructure. """ - def __init__(self, location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2): - super().__init__(INFRASTRUCTURE.APIM_ACA, index, location, apim_sku, APIMNetworkMode.PUBLIC) + def __init__(self, rg_location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2, infra_pfs: List[PolicyFragment] | None = None, infra_apis: List[API] | None = None): + super().__init__(INFRASTRUCTURE.APIM_ACA, index, rg_location, apim_sku, APIMNetworkMode.PUBLIC, infra_pfs, infra_apis) class AfdApimAcaInfrastructure(Infrastructure): @@ -209,5 +287,5 @@ class AfdApimAcaInfrastructure(Infrastructure): Represents an Azure Front Door with API Management and Azure Container Apps infrastructure. """ - def __init__(self, location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2): - super().__init__(INFRASTRUCTURE.AFD_APIM_PE, index, location, apim_sku, APIMNetworkMode.PUBLIC) + def __init__(self, rg_location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2, infra_pfs: List[PolicyFragment] | None = None, infra_apis: List[API] | None = None): + super().__init__(INFRASTRUCTURE.AFD_APIM_PE, index, rg_location, apim_sku, APIMNetworkMode.PUBLIC, infra_pfs, infra_apis) diff --git a/tests/python/test_infrastructures.py b/tests/python/test_infrastructures.py index e4d9ba1..bd7cb2e 100644 --- a/tests/python/test_infrastructures.py +++ b/tests/python/test_infrastructures.py @@ -32,7 +32,7 @@ def mock_utils(): with patch('infrastructures.utils') as mock_utils: mock_utils.get_infra_rg_name.return_value = 'rg-test-infrastructure-01' mock_utils.build_infrastructure_tags.return_value = {'environment': 'test', 'project': 'apim-samples'} - mock_utils.read_policy_xmll.return_value = '' + mock_utils.read_policy_xml.return_value = '' mock_utils.determine_shared_policy_path.return_value = '/mock/path/policy.xml' mock_utils.create_resource_group.return_value = None mock_utils.verify_infrastructure.return_value = True @@ -74,7 +74,7 @@ def test_infrastructure_creation_basic(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) assert infra.infra == INFRASTRUCTURE.SIMPLE_APIM @@ -91,7 +91,7 @@ def test_infrastructure_creation_with_custom_values(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.APIM_ACA, index=2, - location='westus2', + rg_location='westus2', apim_sku=APIM_SKU.PREMIUM, networkMode=APIMNetworkMode.EXTERNAL_VNET ) @@ -108,15 +108,18 @@ def test_infrastructure_creation_with_custom_policy_fragments(mock_utils, mock_p infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION, + rg_location=TEST_LOCATION, infra_pfs=mock_policy_fragments ) + # Initialize policy fragments + pfs = infra._define_policy_fragments() + # Should have base policy fragments + custom ones - assert len(infra.pfs) == 7 # 5 base + 2 custom - assert any(pf.name == 'Test-Fragment-1' for pf in infra.pfs) - assert any(pf.name == 'Test-Fragment-2' for pf in infra.pfs) - assert any(pf.name == 'AuthZ-Match-All' for pf in infra.pfs) + assert len(pfs) == 7 # 5 base + 2 custom + assert any(pf.name == 'Test-Fragment-1' for pf in pfs) + assert any(pf.name == 'Test-Fragment-2' for pf in pfs) + assert any(pf.name == 'AuthZ-Match-All' for pf in pfs) @pytest.mark.unit def test_infrastructure_creation_with_custom_apis(mock_utils, mock_apis): @@ -124,15 +127,18 @@ def test_infrastructure_creation_with_custom_apis(mock_utils, mock_apis): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION, + rg_location=TEST_LOCATION, infra_apis=mock_apis ) + # Initialize APIs + apis = infra._define_apis() + # Should have base APIs + custom ones - assert len(infra.apis) == 3 # 1 base (hello-world) + 2 custom + assert len(apis) == 3 # 1 base (hello-world) + 2 custom assert any(api.name == 'test-api-1' for api in infra.apis) - assert any(api.name == 'test-api-2' for api in infra.apis) - assert any(api.name == 'hello-world' for api in infra.apis) + assert any(api.name == 'test-api-2' for api in apis) + assert any(api.name == 'hello-world' for api in apis) @pytest.mark.unit def test_infrastructure_creation_calls_utils_functions(mock_utils): @@ -140,14 +146,18 @@ def test_infrastructure_creation_calls_utils_functions(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) mock_utils.get_infra_rg_name.assert_called_once_with(INFRASTRUCTURE.SIMPLE_APIM, TEST_INDEX) mock_utils.build_infrastructure_tags.assert_called_once_with(INFRASTRUCTURE.SIMPLE_APIM) - # Should call read_policy_xmll for base policy fragments - assert mock_utils.read_policy_xmll.call_count >= 5 # At least 5 base policy fragments + # Initialize policy fragments to trigger utils calls + infra._define_policy_fragments() + infra._define_apis() + + # Should call read_policy_xml for base policy fragments and APIs + assert mock_utils.read_policy_xml.call_count >= 6 # 5 base policy fragments + 1 hello-world API assert mock_utils.determine_shared_policy_path.call_count >= 5 @pytest.mark.unit @@ -156,9 +166,12 @@ def test_infrastructure_base_policy_fragments_creation(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) + # Initialize policy fragments + pfs = infra._define_policy_fragments() + # Check that all base policy fragments are created expected_fragment_names = [ 'AuthZ-Match-All', @@ -178,9 +191,12 @@ def test_infrastructure_base_apis_creation(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) + # Initialize APIs + apis = infra._define_apis() + # Check that hello-world API is created assert len(infra.base_apis) == 1 hello_world_api = infra.base_apis[0] @@ -201,13 +217,16 @@ def test_define_policy_fragments_with_none_input(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION, + rg_location=TEST_LOCATION, infra_pfs=None ) + # Initialize policy fragments + pfs = infra._define_policy_fragments() + # Should only have base policy fragments - assert len(infra.pfs) == 5 - assert all(pf.name in ['AuthZ-Match-All', 'AuthZ-Match-Any', 'Http-Response-200', 'Product-Match-Any', 'Remove-Request-Headers'] for pf in infra.pfs) + assert len(pfs) == 5 + assert all(pf.name in ['AuthZ-Match-All', 'AuthZ-Match-Any', 'Http-Response-200', 'Product-Match-Any', 'Remove-Request-Headers'] for pf in pfs) @pytest.mark.unit def test_define_policy_fragments_with_custom_input(mock_utils, mock_policy_fragments): @@ -215,12 +234,15 @@ def test_define_policy_fragments_with_custom_input(mock_utils, mock_policy_fragm infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION, + rg_location=TEST_LOCATION, infra_pfs=mock_policy_fragments ) + # Initialize policy fragments + pfs = infra._define_policy_fragments() + # Should have base + custom policy fragments - assert len(infra.pfs) == 7 # 5 base + 2 custom + assert len(pfs) == 7 # 5 base + 2 custom fragment_names = [pf.name for pf in infra.pfs] assert 'Test-Fragment-1' in fragment_names assert 'Test-Fragment-2' in fragment_names @@ -237,13 +259,16 @@ def test_define_apis_with_none_input(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION, + rg_location=TEST_LOCATION, infra_apis=None ) + # Initialize APIs + apis = infra._define_apis() + # Should only have base APIs - assert len(infra.apis) == 1 - assert infra.apis[0].name == 'hello-world' + assert len(apis) == 1 + assert apis[0].name == 'hello-world' @pytest.mark.unit def test_define_apis_with_custom_input(mock_utils, mock_apis): @@ -251,13 +276,16 @@ def test_define_apis_with_custom_input(mock_utils, mock_apis): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION, + rg_location=TEST_LOCATION, infra_apis=mock_apis ) + # Initialize APIs + apis = infra._define_apis() + # Should have base + custom APIs - assert len(infra.apis) == 3 # 1 base + 2 custom - api_names = [api.name for api in infra.apis] + assert len(apis) == 3 # 1 base + 2 custom + api_names = [api.name for api in apis] assert 'test-api-1' in api_names assert 'test-api-2' in api_names assert 'hello-world' in api_names @@ -273,9 +301,13 @@ def test_define_bicep_parameters(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) + # Initialize APIs and policy fragments first + infra._define_policy_fragments() + infra._define_apis() + bicep_params = infra._define_bicep_parameters() assert 'apimSku' in bicep_params @@ -325,14 +357,15 @@ def verify_infrastructure(self) -> bool: infra = TestInfrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) result = infra.deploy_infrastructure() # Verify the deployment process mock_utils.create_resource_group.assert_called_once() - mock_utils.run.assert_called_once() + # The utils.run method is now called multiple times (deployment + verification steps) + assert mock_utils.run.call_count >= 1 # At least one call for deployment # Note: utils.verify_infrastructure is currently commented out in the actual code # mock_utils.verify_infrastructure.assert_called_once() @@ -377,7 +410,7 @@ def verify_infrastructure(self) -> bool: infra = TestInfrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) result = infra.deploy_infrastructure() @@ -404,7 +437,7 @@ def verify_infrastructure(self) -> bool: def test_simple_apim_infrastructure_creation(mock_utils): """Test SimpleApimInfrastructure creation.""" infra = infrastructures.SimpleApimInfrastructure( - location=TEST_LOCATION, + rg_location=TEST_LOCATION, index=TEST_INDEX, apim_sku=APIM_SKU.DEVELOPER ) @@ -419,7 +452,7 @@ def test_simple_apim_infrastructure_creation(mock_utils): def test_simple_apim_infrastructure_defaults(mock_utils): """Test SimpleApimInfrastructure with default values.""" infra = infrastructures.SimpleApimInfrastructure( - location=TEST_LOCATION, + rg_location=TEST_LOCATION, index=TEST_INDEX ) @@ -429,7 +462,7 @@ def test_simple_apim_infrastructure_defaults(mock_utils): def test_apim_aca_infrastructure_creation(mock_utils): """Test ApimAcaInfrastructure creation.""" infra = infrastructures.ApimAcaInfrastructure( - location=TEST_LOCATION, + rg_location=TEST_LOCATION, index=TEST_INDEX, apim_sku=APIM_SKU.STANDARD ) @@ -444,7 +477,7 @@ def test_apim_aca_infrastructure_creation(mock_utils): def test_afd_apim_aca_infrastructure_creation(mock_utils): """Test AfdApimAcaInfrastructure creation.""" infra = infrastructures.AfdApimAcaInfrastructure( - location=TEST_LOCATION, + rg_location=TEST_LOCATION, index=TEST_INDEX, apim_sku=APIM_SKU.PREMIUM ) @@ -464,11 +497,15 @@ def test_afd_apim_aca_infrastructure_creation(mock_utils): def test_infrastructure_end_to_end_simple(mock_utils): """Test end-to-end Infrastructure creation with SimpleApim.""" infra = infrastructures.SimpleApimInfrastructure( - location='eastus', + rg_location='eastus', index=1, apim_sku=APIM_SKU.DEVELOPER ) + # Initialize components + infra._define_policy_fragments() + infra._define_apis() + # Verify all components are created correctly assert infra.infra == INFRASTRUCTURE.SIMPLE_APIM assert len(infra.base_pfs) == 5 @@ -488,13 +525,17 @@ def test_infrastructure_with_all_custom_components(mock_utils, mock_policy_fragm infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.APIM_ACA, index=2, - location='westus2', + rg_location='westus2', apim_sku=APIM_SKU.PREMIUM, networkMode=APIMNetworkMode.EXTERNAL_VNET, infra_pfs=mock_policy_fragments, infra_apis=mock_apis ) + # Initialize components + infra._define_policy_fragments() + infra._define_apis() + # Verify all components are combined correctly assert len(infra.base_pfs) == 5 assert len(infra.pfs) == 7 # 5 base + 2 custom @@ -528,7 +569,7 @@ def test_concrete_infrastructure_missing_params(): infrastructures.SimpleApimInfrastructure() with pytest.raises(TypeError): - infrastructures.SimpleApimInfrastructure(location=TEST_LOCATION) + infrastructures.SimpleApimInfrastructure(rg_location=TEST_LOCATION) # ------------------------------ @@ -544,11 +585,15 @@ def test_infrastructure_empty_custom_lists(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION, + rg_location=TEST_LOCATION, infra_pfs=empty_pfs, infra_apis=empty_apis ) + # Initialize components + infra._define_policy_fragments() + infra._define_apis() + # Empty lists should behave the same as None assert len(infra.pfs) == 5 # Only base policy fragments assert len(infra.apis) == 1 # Only base APIs @@ -559,10 +604,10 @@ def test_infrastructure_attribute_access(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) - # Test all attributes are accessible + # Test constructor attributes are accessible assert hasattr(infra, 'infra') assert hasattr(infra, 'index') assert hasattr(infra, 'rg_location') @@ -570,6 +615,12 @@ def test_infrastructure_attribute_access(mock_utils): assert hasattr(infra, 'networkMode') assert hasattr(infra, 'rg_name') assert hasattr(infra, 'rg_tags') + + # Initialize components to create the lazily-loaded attributes + infra._define_policy_fragments() + infra._define_apis() + + # Test that lazy-loaded attributes are now accessible assert hasattr(infra, 'base_pfs') assert hasattr(infra, 'pfs') assert hasattr(infra, 'base_apis') @@ -584,7 +635,7 @@ def test_infrastructure_string_representation(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) # Test that the object can be converted to string without error @@ -609,7 +660,7 @@ def test_all_infrastructure_types_coverage(mock_utils): def test_policy_fragment_creation_robustness(mock_utils): """Test that policy fragment creation is robust.""" # Test with various mock return values - mock_utils.read_policy_xmll.side_effect = [ + mock_utils.read_policy_xml.side_effect = [ '', '', '', @@ -621,9 +672,13 @@ def test_policy_fragment_creation_robustness(mock_utils): infra = infrastructures.Infrastructure( infra=INFRASTRUCTURE.SIMPLE_APIM, index=TEST_INDEX, - location=TEST_LOCATION + rg_location=TEST_LOCATION ) + # Initialize policy fragments + infra._define_policy_fragments() + infra._define_apis() + # Verify all policy fragments were created with different XML policy_xmls = [pf.policyXml for pf in infra.base_pfs] assert '' in policy_xmls From 035b67b22d719f7396550c013071e60c49b3b7ff Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 09:34:14 -0400 Subject: [PATCH 3/9] Reduce Simple APIM infrastructure creation complexity --- .../simple-apim/create_infrastructure.py | 232 ++---------------- 1 file changed, 16 insertions(+), 216 deletions(-) diff --git a/infrastructure/simple-apim/create_infrastructure.py b/infrastructure/simple-apim/create_infrastructure.py index 7b1cec9..562f079 100644 --- a/infrastructure/simple-apim/create_infrastructure.py +++ b/infrastructure/simple-apim/create_infrastructure.py @@ -1,242 +1,42 @@ """ -Infrastructure creation module for Simple APIM. - -This module provides a reusable way to create Simple APIM infrastructure -that can be called from notebooks or other scripts. +This module provides a reusable way to create Simple APIM infrastructure that can be called from notebooks or other scripts. """ import sys -import os import argparse -from pathlib import Path -import utils -from apimtypes import * -import json - -def _create_simple_apim_infrastructure( - rg_location: str = 'eastus2', - index: int | None = None, - apim_sku: APIM_SKU = APIM_SKU.BASICV2, - custom_apis: list[API] | None = None, - custom_policy_fragments: list[PolicyFragment] | None = None -) -> utils.Output: - """ - Create Simple APIM infrastructure with the specified parameters. - - Args: - rg_location (str): Azure region for deployment. Defaults to 'eastus2'. - index (int | None): Index for the infrastructure. Defaults to None (no index). - apim_sku (APIM_SKU): SKU for API Management. Defaults to BASICV2. - custom_apis (list[API] | None): Custom APIs to deploy. If None, uses default Hello World API. - custom_policy_fragments (list[PolicyFragment] | None): Custom policy fragments. If None, uses defaults. - - Returns: - utils.Output: The deployment result. - """ - - # 1) Setup deployment parameters - deployment = INFRASTRUCTURE.SIMPLE_APIM - rg_name = utils.get_infra_rg_name(deployment, index) - rg_tags = utils.build_infrastructure_tags(deployment) - - print(f'\nšŸš€ Creating Simple APIM infrastructure...\n') - print(f' Infrastructure : {deployment.value}') - print(f' Index : {index}') - print(f' Resource group : {rg_name}') - print(f' Location : {rg_location}') - print(f' APIM SKU : {apim_sku.value}\n') - - # 2) Set up the policy fragments - if custom_policy_fragments is None: - pfs: List[PolicyFragment] = [ - PolicyFragment('AuthZ-Match-All', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-all.xml')), 'Authorizes if all of the specified roles match the JWT role claims.'), - PolicyFragment('AuthZ-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-any.xml')), 'Authorizes if any of the specified roles match the JWT role claims.'), - PolicyFragment('Http-Response-200', utils.read_policy_xml(utils.determine_shared_policy_path('pf-http-response-200.xml')), 'Returns a 200 OK response for the current HTTP method.'), - PolicyFragment('Product-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-product-match-any.xml')), 'Proceeds if any of the specified products match the context product name.'), - PolicyFragment('Remove-Request-Headers', utils.read_policy_xml(utils.determine_shared_policy_path('pf-remove-request-headers.xml')), 'Removes request headers from the incoming request.') - ] - else: - pfs = custom_policy_fragments - - # 3) Define the APIs - if custom_apis is None: - # Default Hello World API - pol_hello_world = utils.read_policy_xml(HELLO_WORLD_XML_POLICY_PATH) - api_hwroot_get = GET_APIOperation('This is a GET for API 1', pol_hello_world) - api_hwroot = API('hello-world', 'Hello World', '', 'This is the root API for Hello World', operations = [api_hwroot_get]) - apis: List[API] = [api_hwroot] - else: - apis = custom_apis - - # 4) Define the Bicep parameters with serialized APIs - # Define the Bicep parameters with serialized APIs - bicep_parameters = { - 'apimSku' : {'value': apim_sku.value}, - 'apis' : {'value': [api.to_dict() for api in apis]}, - 'policyFragments' : {'value': [pf.to_dict() for pf in pfs]} - } - - # Change to the infrastructure directory to ensure bicep files are found - original_cwd = os.getcwd() - infra_dir = Path(__file__).parent - - try: - os.chdir(infra_dir) - print(f'šŸ“ Changed working directory to: {infra_dir}') - - # Prepare deployment parameters and run directly to avoid path detection issues - bicep_parameters_format = { - '$schema': 'https://schema.management.azure.com/schemas/2019-04-01/deploymentParameters.json#', - 'contentVersion': '1.0.0.0', - 'parameters': bicep_parameters - } - - # Write the parameters file - params_file_path = infra_dir / 'params.json' - - with open(params_file_path, 'w') as file: - file.write(json.dumps(bicep_parameters_format)) - - print(f"šŸ“ Updated the policy XML in the bicep parameters file 'params.json'") - - # ------------------------------ - # EXECUTE DEPLOYMENT - # ------------------------------ - - # Create the resource group if it doesn't exist - utils.create_resource_group(rg_name, rg_location, rg_tags) - - # Run the deployment directly - main_bicep_path = infra_dir / 'main.bicep' - output = utils.run( - f'az deployment group create --name {deployment.value} --resource-group {rg_name} --template-file "{main_bicep_path}" --parameters "{params_file_path}" --query "properties.outputs"', - f"Deployment '{deployment.value}' succeeded", - f"Deployment '{deployment.value}' failed.", - print_command_to_run = False - ) - - # ------------------------------ - # VERIFY DEPLOYMENT RESULTS - # ------------------------------ - - if output.success: - print('\nāœ… Infrastructure creation completed successfully!') - if output.json_data: - apim_gateway_url = output.get('apimResourceGatewayURL', 'APIM API Gateway URL', suppress_logging = True) - apim_apis = output.getJson('apiOutputs', 'APIs', suppress_logging = True) - - print(f'\nšŸ“‹ Infrastructure Details:') - print(f' Resource Group : {rg_name}') - print(f' Location : {rg_location}') - print(f' APIM SKU : {apim_sku.value}') - print(f' Gateway URL : {apim_gateway_url}') - print(f' APIs Created : {len(apim_apis)}') - - # Perform basic verification - _verify_infrastructure(rg_name) - else: - print('āŒ Infrastructure creation failed!') - - return output - - finally: - # Always restore the original working directory - os.chdir(original_cwd) - print(f'šŸ“ Restored working directory to: {original_cwd}') - -def _verify_infrastructure(rg_name: str) -> bool: - """ - Verify that the infrastructure was created successfully. - - Args: - rg_name (str): Resource group name. - - Returns: - bool: True if verification passed, False otherwise. - """ - - print('\nšŸ” Verifying infrastructure...') - - try: - # Check if the resource group exists - if not utils.does_resource_group_exist(rg_name): - print('āŒ Resource group does not exist!') - return False - - print('āœ… Resource group verified') - - # Get APIM service details - output = utils.run(f'az apim list -g {rg_name} --query "[0]" -o json', print_command_to_run = False, print_errors = False) - - if output.success and output.json_data: - apim_name = output.json_data.get('name') - - print(f'āœ… APIM Service verified: {apim_name}') - - # Get API count - api_output = utils.run(f'az apim api list --service-name {apim_name} -g {rg_name} --query "length(@)"', - print_command_to_run = False, print_errors = False) - - if api_output.success: - api_count = int(api_output.text.strip()) - print(f'āœ… APIs verified: {api_count} API(s) created') - - # Test basic connectivity (optional) - if api_count > 0: - try: - # Get subscription key for testing - sub_output = utils.run(f'az apim subscription list --service-name {apim_name} -g {rg_name} --query "[0].primaryKey" -o tsv', - print_command_to_run = False, print_errors = False) - - if sub_output.success and sub_output.text.strip(): - print('āœ… Subscription key available for API testing') - except: - pass - - print('\nšŸŽ‰ Infrastructure verification completed successfully!') - return True - - else: - print('\nāŒ APIM service not found!') - return False - - except Exception as e: - print(f'\nāš ļø Verification failed with error: {str(e)}') - return False +from apimtypes import APIM_SKU +from infrastructures import SimpleApimInfrastructure def main(): """ Main entry point for command-line usage. """ - parser = argparse.ArgumentParser(description='Create Simple APIM infrastructure') - parser.add_argument('--location', default='eastus2', help='Azure region (default: eastus2)') - parser.add_argument('--index', type=int, help='Infrastructure index') - parser.add_argument('--sku', choices=['Basicv2', 'Standardv2', 'Premiumv2'], default='Basicv2', help='APIM SKU (default: Basicv2)') - + parser = argparse.ArgumentParser(description = 'Create Simple APIM infrastructure') + parser.add_argument('--location', default = 'eastus2', help = 'Azure region (default: eastus2)') + parser.add_argument('--index', type = int, help = 'Infrastructure index') + parser.add_argument('--sku', choices = ['Basicv2', 'Standardv2', 'Premiumv2'], default = 'Basicv2', help = 'APIM SKU (default: Basicv2)') args = parser.parse_args() - # Convert SKU string to enum - sku_map = { - 'Basicv2': APIM_SKU.BASICV2, - 'Standardv2': APIM_SKU.STANDARDV2, - 'Premiumv2': APIM_SKU.PREMIUMV2 - } + # Convert SKU string to enum using the enum's built-in functionality + try: + apim_sku = APIM_SKU(args.sku) + except ValueError: + print(f"Error: Invalid SKU '{args.sku}'. Valid options are: {', '.join([sku.value for sku in APIM_SKU])}") + sys.exit(1) try: - result = _create_simple_apim_infrastructure(rg_location = args.location, index = args.index, apim_sku = sku_map[args.sku]) + infra = SimpleApimInfrastructure(args.location, args.index, apim_sku) + result = infra.deploy_infrastructure() if result.success: - print('\nšŸŽ‰ Infrastructure creation completed successfully!') sys.exit(0) else: - print('\nšŸ’„ Infrastructure creation failed!') sys.exit(1) except Exception as e: print(f'\nšŸ’„ Error: {str(e)}') sys.exit(1) - if __name__ == '__main__': - main() + main() \ No newline at end of file From 9abe7432d53d186a1be0e36d90fe6541a902a7af Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 12:57:20 -0400 Subject: [PATCH 4/9] Homogenize infrastructure creation --- infrastructure/afd-apim-pe/create.ipynb | 13 +- .../afd-apim-pe/create_infrastructure.py | 314 +++--------------- infrastructure/apim-aca/create.ipynb | 19 +- .../apim-aca/create_infrastructure.py | 286 +++------------- infrastructure/simple-apim/create.ipynb | 12 +- .../simple-apim/create_infrastructure.py | 27 +- shared/python/infrastructures.py | 111 ++++++- tests/python/test_infrastructures.py | 257 +++++++++++++- 8 files changed, 493 insertions(+), 546 deletions(-) diff --git a/infrastructure/afd-apim-pe/create.ipynb b/infrastructure/afd-apim-pe/create.ipynb index 46ae39b..7fb88b7 100644 --- a/infrastructure/afd-apim-pe/create.ipynb +++ b/infrastructure/afd-apim-pe/create.ipynb @@ -22,13 +22,20 @@ "import utils\n", "from apimtypes import *\n", "\n", - "# User-defined parameters (change these as needed)\n", + "# ------------------------------\n", + "# USER CONFIGURATION\n", + "# ------------------------------\n", + "\n", "rg_location = 'eastus2' # Azure region for deployment\n", "index = 1 # Infrastructure index (use different numbers for multiple environments)\n", "apim_sku = APIM_SKU.STANDARDV2 # Options: 'STANDARDV2', 'PREMIUMV2' (Basic not supported for private endpoints)\n", - "use_aca = True # Include Azure Container Apps backends\n", "\n", - "# Create an instance of the desired infrastructure\n", + "\n", + "\n", + "# ------------------------------\n", + "# SYSTEM CONFIGURATION\n", + "# ------------------------------\n", + "\n", "inb_helper = utils.InfrastructureNotebookHelper(rg_location, INFRASTRUCTURE.AFD_APIM_PE, index, apim_sku) \n", "success = inb_helper.create_infrastructure()\n", "\n", diff --git a/infrastructure/afd-apim-pe/create_infrastructure.py b/infrastructure/afd-apim-pe/create_infrastructure.py index 96e1569..2004338 100644 --- a/infrastructure/afd-apim-pe/create_infrastructure.py +++ b/infrastructure/afd-apim-pe/create_infrastructure.py @@ -1,300 +1,82 @@ """ -Infrastructure creation module for AFD-APIM-PE. - -This module provides a reusable way to create Azure Front Door with API Management -(Private Endpoint) infrastructure that can be called from notebooks or other scripts. +This module provides a reusable way to create Azure Front Door with API Management (Private Endpoint) infrastructure that can be called from notebooks or other scripts. """ import sys -import os import argparse -from pathlib import Path +from apimtypes import APIM_SKU, API, GET_APIOperation, BACKEND_XML_POLICY_PATH +from infrastructures import AfdApimAcaInfrastructure import utils -from apimtypes import * -import json - -def _create_afd_apim_pe_infrastructure( - rg_location: str = 'eastus2', - index: int | None = None, - apim_sku: APIM_SKU = APIM_SKU.STANDARDV2, - use_aca: bool = True, - custom_apis: list[API] | None = None, - custom_policy_fragments: list[PolicyFragment] | None = None -) -> utils.Output: - """ - Create AFD-APIM-PE infrastructure with the specified parameters. - - Args: - rg_location (str): Azure region for deployment. Defaults to 'eastus2'. - index (int | None): Index for the infrastructure. Defaults to None (no index). - apim_sku (APIM_SKU): SKU for API Management. Defaults to STANDARDV2. - use_aca (bool): Whether to include Azure Container Apps. Defaults to True. - custom_apis (list[API] | None): Custom APIs to deploy. If None, uses default Hello World API. - custom_policy_fragments (list[PolicyFragment] | None): Custom policy fragments. If None, uses defaults. - - Returns: - utils.Output: The deployment result. - """ - - # 1) Setup deployment parameters - deployment = INFRASTRUCTURE.AFD_APIM_PE - rg_name = utils.get_infra_rg_name(deployment, index) - rg_tags = utils.build_infrastructure_tags(deployment) - apim_network_mode = APIMNetworkMode.EXTERNAL_VNET - - print(f'\nšŸš€ Creating AFD-APIM-PE infrastructure...\n') - print(f' Infrastructure : {deployment.value}') - print(f' Index : {index}') - print(f' Resource group : {rg_name}') - print(f' Location : {rg_location}') - print(f' APIM SKU : {apim_sku.value}\n') - - # 2) Set up the policy fragments - if custom_policy_fragments is None: - pfs: List[PolicyFragment] = [ - PolicyFragment('AuthZ-Match-All', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-all.xml')), 'Authorizes if all of the specified roles match the JWT role claims.'), - PolicyFragment('AuthZ-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-any.xml')), 'Authorizes if any of the specified roles match the JWT role claims.'), - PolicyFragment('Http-Response-200', utils.read_policy_xml(utils.determine_shared_policy_path('pf-http-response-200.xml')), 'Returns a 200 OK response for the current HTTP method.'), - PolicyFragment('Product-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-product-match-any.xml')), 'Proceeds if any of the specified products match the context product name.'), - PolicyFragment('Remove-Request-Headers', utils.read_policy_xml(utils.determine_shared_policy_path('pf-remove-request-headers.xml')), 'Removes request headers from the incoming request.') - ] - else: - pfs = custom_policy_fragments - - # 3) Define the APIs - if custom_apis is None: - # Default Hello World API - pol_hello_world = utils.read_policy_xml(HELLO_WORLD_XML_POLICY_PATH) - api_hwroot_get = GET_APIOperation('This is a GET for API 1', pol_hello_world) - api_hwroot = API('hello-world', 'Hello World', '', 'This is the root API for Hello World', operations = [api_hwroot_get]) - apis: List[API] = [api_hwroot] - - # If Container Apps is enabled, create the ACA APIs in APIM - if use_aca: - pol_backend = utils.read_policy_xml(BACKEND_XML_POLICY_PATH) - pol_aca_backend_1 = pol_backend.format(backend_id = 'aca-backend-1') - pol_aca_backend_2 = pol_backend.format(backend_id = 'aca-backend-2') - pol_aca_backend_pool = pol_backend.format(backend_id = 'aca-backend-pool') - - # API 1: Hello World (ACA Backend 1) - api_hwaca_1_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 1') - api_hwaca_1 = API('hello-world-aca-1', 'Hello World (ACA 1)', '/aca-1', 'This is the ACA API for Backend 1', pol_aca_backend_1, [api_hwaca_1_get]) - - # API 2: Hello World (ACA Backend 2) - api_hwaca_2_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 2') - api_hwaca_2 = API('hello-world-aca-2', 'Hello World (ACA 2)', '/aca-2', 'This is the ACA API for Backend 2', pol_aca_backend_2, [api_hwaca_2_get]) - # API 3: Hello World (ACA Backend Pool) - api_hwaca_pool_get = GET_APIOperation('This is a GET for Hello World on ACA Backend Pool') - api_hwaca_pool = API('hello-world-aca-pool', 'Hello World (ACA Pool)', '/aca-pool', 'This is the ACA API for Backend Pool', pol_aca_backend_pool, [api_hwaca_pool_get]) - # Add ACA APIs to the existing apis array - apis += [api_hwaca_1, api_hwaca_2, api_hwaca_pool] - else: - apis = custom_apis - - # 4) Define the Bicep parameters with serialized APIs - bicep_parameters = { - 'apimSku' : {'value': apim_sku.value}, - 'apis' : {'value': [api.to_dict() for api in apis]}, - 'policyFragments' : {'value': [pf.to_dict() for pf in pfs]}, - 'apimPublicAccess' : {'value': apim_network_mode in [APIMNetworkMode.PUBLIC, APIMNetworkMode.EXTERNAL_VNET]}, - 'useACA' : {'value': use_aca} - } - - # 5) Change to the infrastructure directory to ensure bicep files are found - original_cwd = os.getcwd() - infra_dir = Path(__file__).parent - +def create_infrastructure(location: str, index: int, apim_sku: APIM_SKU, no_aca: bool = False) -> None: try: - os.chdir(infra_dir) - print(f'šŸ“ Changed working directory to: {infra_dir}') - - # 6) Create the resource group if it doesn't exist - utils.create_resource_group(rg_name, rg_location, rg_tags) + # Create custom APIs for AFD-APIM-PE with optional Container Apps backends + custom_apis = _create_afd_specific_apis(not no_aca) - # 7) First deployment with public access enabled - print('\nšŸš€ Phase 1: Creating infrastructure with public access enabled...') - output = utils.create_bicep_deployment_group(rg_name, rg_location, deployment, bicep_parameters) + infra = AfdApimAcaInfrastructure(location, index, apim_sku, infra_apis = custom_apis) + result = infra.deploy_infrastructure() - if not output.success: - print('āŒ Phase 1 deployment failed!') - return output - - # Extract service details for private link approval - if output.json_data: - apim_service_id = output.get('apimServiceId', 'APIM Service Id', suppress_logging = True) - - print('āœ… Phase 1 deployment completed successfully!') - - # 8) Approve private link connections - print('\nšŸ”— Approving Front Door private link connections...') - _approve_private_link_connections(apim_service_id) + sys.exit(0 if result.success else 1) - # 9) Second deployment to disable public access - print('\nšŸ”’ Phase 2: Disabling APIM public access...') - bicep_parameters['apimPublicAccess']['value'] = False - - output = utils.create_bicep_deployment_group(rg_name, rg_location, deployment, bicep_parameters) - - if output.success: - print('\nāœ… Infrastructure creation completed successfully!') - if output.json_data: - apim_gateway_url = output.get('apimResourceGatewayURL', 'APIM API Gateway URL', suppress_logging = True) - afd_endpoint_url = output.get('fdeSecureUrl', 'Front Door Endpoint URL', suppress_logging = True) - apim_apis = output.getJson('apiOutputs', 'APIs', suppress_logging = True) - - print(f'\nšŸ“‹ Infrastructure Details:') - print(f' Resource Group : {rg_name}') - print(f' Location : {rg_location}') - print(f' APIM SKU : {apim_sku.value}') - print(f' Use ACA : {use_aca}') - print(f' Gateway URL : {apim_gateway_url}') - print(f' Front Door URL : {afd_endpoint_url}') - print(f' APIs Created : {len(apim_apis)}') - - # Perform basic verification - _verify_infrastructure(rg_name, use_aca) - else: - print('āŒ Phase 2 deployment failed!') - - return output - - finally: - # Always restore the original working directory - os.chdir(original_cwd) - print(f'šŸ“ Restored working directory to: {original_cwd}') - -def _approve_private_link_connections(apim_service_id: str) -> None: - """ - Approve pending private link connections for the APIM service. - - Args: - apim_service_id (str): The resource ID of the APIM service. - """ - - # Get all pending private endpoint connections as JSON - output = utils.run(f"az network private-endpoint-connection list --id {apim_service_id} --query \"[?contains(properties.privateLinkServiceConnectionState.status, 'Pending')]\" -o json", print_command_to_run = False) - - # Handle both a single object and a list of objects - pending_connections = output.json_data if output.success and output.is_json else [] - - if isinstance(pending_connections, dict): - pending_connections = [pending_connections] - - total = len(pending_connections) - print(f'Found {total} pending private link service connection(s).') - - if total > 0: - for i, conn in enumerate(pending_connections, 1): - conn_id = conn.get('id') - conn_name = conn.get('name', '') - print(f' {i}/{total}: Approving {conn_name}') - - approve_result = utils.run( - f"az network private-endpoint-connection approve --id {conn_id} --description 'Approved'", - f'Private Link Connection approved: {conn_name}', - f'Failed to approve Private Link Connection: {conn_name}', - print_command_to_run = False - ) + except Exception as e: + print(f'\nšŸ’„ Error: {str(e)}') + sys.exit(1) - print('āœ… Private link approvals completed') - else: - print('No pending private link service connections found. Nothing to approve.') -def _verify_infrastructure(rg_name: str, use_aca: bool) -> bool: +def _create_afd_specific_apis(use_aca: bool = True) -> list[API]: """ - Verify that the infrastructure was created successfully. + Create AFD-APIM-PE specific APIs with optional Container Apps backends. Args: - rg_name (str): Resource group name. - use_aca (bool): Whether Container Apps were included. + use_aca (bool): Whether to include Azure Container Apps backends. Defaults to true. Returns: - bool: True if verification passed, False otherwise. + list[API]: List of AFD-specific APIs. """ - print('\nšŸ” Verifying infrastructure...') + # If Container Apps is enabled, create the ACA APIs in APIM + if use_aca: + pol_backend = utils.read_policy_xml(BACKEND_XML_POLICY_PATH) + pol_aca_backend_1 = pol_backend.format(backend_id = 'aca-backend-1') + pol_aca_backend_2 = pol_backend.format(backend_id = 'aca-backend-2') + pol_aca_backend_pool = pol_backend.format(backend_id = 'aca-backend-pool') + + # API 1: Hello World (ACA Backend 1) + api_hwaca_1_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 1') + api_hwaca_1 = API('hello-world-aca-1', 'Hello World (ACA 1)', '/aca-1', 'This is the ACA API for Backend 1', pol_aca_backend_1, [api_hwaca_1_get]) + + # API 2: Hello World (ACA Backend 2) + api_hwaca_2_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 2') + api_hwaca_2 = API('hello-world-aca-2', 'Hello World (ACA 2)', '/aca-2', 'This is the ACA API for Backend 2', pol_aca_backend_2, [api_hwaca_2_get]) + + # API 3: Hello World (ACA Backend Pool) + api_hwaca_pool_get = GET_APIOperation('This is a GET for Hello World on ACA Backend Pool') + api_hwaca_pool = API('hello-world-aca-pool', 'Hello World (ACA Pool)', '/aca-pool', 'This is the ACA API for Backend Pool', pol_aca_backend_pool, [api_hwaca_pool_get]) + + return [api_hwaca_1, api_hwaca_2, api_hwaca_pool] - try: - # Check if the resource group exists - if not utils.does_resource_group_exist(rg_name): - print('āŒ Resource group does not exist!') - return False - - print('āœ… Resource group verified') - - # Get APIM service details - output = utils.run(f'az apim list -g {rg_name} --query "[0]" -o json', print_command_to_run = False, print_errors = False) - - if output.success and output.json_data: - apim_name = output.json_data.get('name') - print(f'āœ… APIM Service verified: {apim_name}') - - # Check Front Door - afd_output = utils.run(f'az afd profile list -g {rg_name} --query "[0]" -o json', print_command_to_run = False, print_errors = False) - - if afd_output.success and afd_output.json_data: - afd_name = afd_output.json_data.get('name') - print(f'āœ… Azure Front Door verified: {afd_name}') - - # Check Container Apps if enabled - if use_aca: - aca_output = utils.run(f'az containerapp list -g {rg_name} --query "length(@)"', print_command_to_run = False, print_errors = False) - - if aca_output.success: - aca_count = int(aca_output.text.strip()) - print(f'āœ… Container Apps verified: {aca_count} app(s) created') - - print('\nšŸŽ‰ Infrastructure verification completed successfully!') - return True - - else: - print('\nāŒ APIM service not found!') - return False - - except Exception as e: - print(f'\nāš ļø Verification failed with error: {str(e)}') - return False - + return [] def main(): """ Main entry point for command-line usage. """ - parser = argparse.ArgumentParser(description='Create AFD-APIM-PE infrastructure') - parser.add_argument('--location', default='eastus2', help='Azure region (default: eastus2)') - parser.add_argument('--index', type=int, help='Infrastructure index') - parser.add_argument('--sku', choices=['Standardv2', 'Premiumv2'], default='Standardv2', help='APIM SKU (default: Standardv2)') - parser.add_argument('--no-aca', action='store_true', help='Disable Azure Container Apps') - + parser = argparse.ArgumentParser(description = 'Create AFD-APIM-PE infrastructure') + parser.add_argument('--location', default = 'eastus2', help = 'Azure region (default: eastus2)') + parser.add_argument('--index', type = int, help = 'Infrastructure index') + parser.add_argument('--sku', choices = ['Standardv2', 'Premiumv2'], default = 'Standardv2', help = 'APIM SKU (default: Standardv2)') + parser.add_argument('--no-aca', action = 'store_true', help = 'Disable Azure Container Apps') args = parser.parse_args() - - # Convert SKU string to enum - sku_map = { - 'Standardv2': APIM_SKU.STANDARDV2, - 'Premiumv2': APIM_SKU.PREMIUMV2 - } - + + # Convert SKU string to enum using the enum's built-in functionality try: - result = _create_afd_apim_pe_infrastructure( - rg_location = args.location, - index = args.index, - apim_sku = sku_map[args.sku], - use_aca = not args.no_aca - ) - - if result.success: - print('\nšŸŽ‰ Infrastructure creation completed successfully!') - sys.exit(0) - else: - print('\nšŸ’„ Infrastructure creation failed!') - sys.exit(1) - - except Exception as e: - print(f'\nšŸ’„ Error: {str(e)}') + apim_sku = APIM_SKU(args.sku) + except ValueError: + print(f"Error: Invalid SKU '{args.sku}'. Valid options are: {', '.join([sku.value for sku in APIM_SKU])}") sys.exit(1) + create_infrastructure(args.location, args.index, apim_sku, args.no_aca) if __name__ == '__main__': main() diff --git a/infrastructure/apim-aca/create.ipynb b/infrastructure/apim-aca/create.ipynb index 22b6314..e4efc7d 100644 --- a/infrastructure/apim-aca/create.ipynb +++ b/infrastructure/apim-aca/create.ipynb @@ -20,13 +20,20 @@ "import utils\n", "from apimtypes import *\n", "\n", - "# User-defined parameters (change these as needed)\n", - "rg_location = 'eastus2' # Azure region for deployment\n", - "index = 1 # Infrastructure index (use different numbers for multiple environments)\n", - "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "reveal_backend = True # Set to True to reveal the backend details in the API operations\n", + "# ------------------------------\n", + "# USER CONFIGURATION\n", + "# ------------------------------\n", + "\n", + "rg_location = 'eastus2' # Azure region for deployment\n", + "index = 1 # Infrastructure index (use different numbers for multiple environments)\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", + "\n", + "\n", + "\n", + "# ------------------------------\n", + "# SYSTEM CONFIGURATION\n", + "# ------------------------------\n", "\n", - "# Create an instance of the desired infrastructure\n", "inb_helper = utils.InfrastructureNotebookHelper(rg_location, INFRASTRUCTURE.APIM_ACA, index, apim_sku) \n", "success = inb_helper.create_infrastructure()\n", "\n", diff --git a/infrastructure/apim-aca/create_infrastructure.py b/infrastructure/apim-aca/create_infrastructure.py index 691d0e0..0c6ea6e 100644 --- a/infrastructure/apim-aca/create_infrastructure.py +++ b/infrastructure/apim-aca/create_infrastructure.py @@ -1,274 +1,76 @@ """ -Infrastructure creation module for APIM-ACA. - -This module provides a reusable way to create API Management with Azure Container Apps -infrastructure that can be called from notebooks or other scripts. +This module provides a reusable way to create API Management with Azure Container Apps infrastructure that can be called from notebooks or other scripts. """ import sys -import os import argparse -from pathlib import Path +from apimtypes import APIM_SKU, API, GET_APIOperation, BACKEND_XML_POLICY_PATH +from infrastructures import ApimAcaInfrastructure import utils -from apimtypes import * -import json - -def _create_apim_aca_infrastructure( - rg_location: str = 'eastus2', - index: int | None = None, - apim_sku: APIM_SKU = APIM_SKU.BASICV2, - reveal_backend: bool = True, - custom_apis: list[API] | None = None, - custom_policy_fragments: list[PolicyFragment] | None = None -) -> utils.Output: - """ - Create APIM-ACA infrastructure with the specified parameters. - - Args: - rg_location (str): Azure region for deployment. Defaults to 'eastus2'. - index (int | None): Index for the infrastructure. Defaults to None (no index). - apim_sku (APIM_SKU): SKU for API Management. Defaults to BASICV2. - reveal_backend (bool): Whether to reveal backend details in API operations. Defaults to True. - custom_apis (list[API] | None): Custom APIs to deploy. If None, uses default Hello World API. - custom_policy_fragments (list[PolicyFragment] | None): Custom policy fragments. If None, uses defaults. - - Returns: - utils.Output: The deployment result. - """ - - # 1) Setup deployment parameters - deployment = INFRASTRUCTURE.APIM_ACA - rg_name = utils.get_infra_rg_name(deployment, index) - rg_tags = utils.build_infrastructure_tags(deployment) - - print(f'\nšŸš€ Creating APIM-ACA infrastructure...\n') - print(f' Infrastructure : {deployment.value}') - print(f' Index : {index}') - print(f' Resource group : {rg_name}') - print(f' Location : {rg_location}') - print(f' APIM SKU : {apim_sku.value}\n') - - # 2) Set up the policy fragments - if custom_policy_fragments is None: - pfs: List[PolicyFragment] = [ - PolicyFragment('AuthZ-Match-All', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-all.xml')), 'Authorizes if all of the specified roles match the JWT role claims.'), - PolicyFragment('AuthZ-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-authz-match-any.xml')), 'Authorizes if any of the specified roles match the JWT role claims.'), - PolicyFragment('Http-Response-200', utils.read_policy_xml(utils.determine_shared_policy_path('pf-http-response-200.xml')), 'Returns a 200 OK response for the current HTTP method.'), - PolicyFragment('Product-Match-Any', utils.read_policy_xml(utils.determine_shared_policy_path('pf-product-match-any.xml')), 'Proceeds if any of the specified products match the context product name.'), - PolicyFragment('Remove-Request-Headers', utils.read_policy_xml(utils.determine_shared_policy_path('pf-remove-request-headers.xml')), 'Removes request headers from the incoming request.') - ] - else: - pfs = custom_policy_fragments - - # 3) Define the APIs - if custom_apis is None: - # Default APIs with Container Apps backends - pol_hello_world = utils.read_policy_xml(HELLO_WORLD_XML_POLICY_PATH) - pol_backend = utils.read_policy_xml(BACKEND_XML_POLICY_PATH) - pol_aca_backend_1 = pol_backend.format(backend_id = 'aca-backend-1') - pol_aca_backend_2 = pol_backend.format(backend_id = 'aca-backend-2') - pol_aca_backend_pool = pol_backend.format(backend_id = 'aca-backend-pool') - - # Hello World (Root) - api_hwroot_get = GET_APIOperation('This is a GET for Hello World in the root', pol_hello_world) - api_hwroot = API('hello-world', 'Hello World', '', 'This is the root API for Hello World', operations = [api_hwroot_get]) - - # Hello World (ACA Backend 1) - api_hwaca_1_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 1') - api_hwaca_1 = API('hello-world-aca-1', 'Hello World (ACA 1)', '/aca-1', 'This is the ACA API for Backend 1', policyXml = pol_aca_backend_1, operations = [api_hwaca_1_get]) - # Hello World (ACA Backend 2) - api_hwaca_2_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 2') - api_hwaca_2 = API('hello-world-aca-2', 'Hello World (ACA 2)', '/aca-2', 'This is the ACA API for Backend 2', policyXml = pol_aca_backend_2, operations = [api_hwaca_2_get]) - # Hello World (ACA Backend Pool) - api_hwaca_pool_get = GET_APIOperation('This is a GET for Hello World on ACA Backend Pool') - api_hwaca_pool = API('hello-world-aca-pool', 'Hello World (ACA Pool)', '/aca-pool', 'This is the ACA API for Backend Pool', policyXml = pol_aca_backend_pool, operations = [api_hwaca_pool_get]) - - # APIs Array - apis: List[API] = [api_hwroot, api_hwaca_1, api_hwaca_2, api_hwaca_pool] - else: - apis = custom_apis - - # 4) Define the Bicep parameters with serialized APIs - bicep_parameters = { - 'apimSku' : {'value': apim_sku.value}, - 'apis' : {'value': [api.to_dict() for api in apis]}, - 'policyFragments' : {'value': [pf.to_dict() for pf in pfs]}, - 'revealBackendApiInfo' : {'value': reveal_backend} - } - - # 5) Change to the infrastructure directory to ensure bicep files are found - original_cwd = os.getcwd() - infra_dir = Path(__file__).parent - +def create_infrastructure(location: str, index: int, apim_sku: APIM_SKU) -> None: try: - os.chdir(infra_dir) - print(f'šŸ“ Changed working directory to: {infra_dir}') - - # 6) Prepare deployment parameters and run directly to avoid path detection issues - bicep_parameters_format = { - '$schema': 'https://schema.management.azure.com/schemas/2019-04-01/deploymentParameters.json#', - 'contentVersion': '1.0.0.0', - 'parameters': bicep_parameters - } - - # Write the parameters file - params_file_path = infra_dir / 'params.json' - - with open(params_file_path, 'w') as file: - file.write(json.dumps(bicep_parameters_format)) - - print(f"šŸ“ Updated the policy XML in the bicep parameters file 'params.json'") - - # Create the resource group if it doesn't exist - utils.create_resource_group(rg_name, rg_location, rg_tags) + # Create custom APIs for APIM-ACA with Container Apps backends + custom_apis = _create_aca_specific_apis() - # Run the deployment directly - main_bicep_path = infra_dir / 'main.bicep' - output = utils.run( - f'az deployment group create --name {deployment.value} --resource-group {rg_name} --template-file "{main_bicep_path}" --parameters "{params_file_path}" --query "properties.outputs"', - f"Deployment '{deployment.value}' succeeded", - f"Deployment '{deployment.value}' failed.", - print_command_to_run = False - ) + infra = ApimAcaInfrastructure(location, index, apim_sku, infra_apis = custom_apis) + result = infra.deploy_infrastructure() - # 7) Check the deployment results and perform verification - if output.success: - print('\nāœ… Infrastructure creation completed successfully!') - if output.json_data: - apim_gateway_url = output.get('apimResourceGatewayURL', 'APIM API Gateway URL', suppress_logging = True) - aca_url_1 = output.get('acaUrl1', 'ACA Backend 1 URL', suppress_logging = True) - aca_url_2 = output.get('acaUrl2', 'ACA Backend 2 URL', suppress_logging = True) - apim_apis = output.getJson('apiOutputs', 'APIs', suppress_logging = True) - - print(f'\nšŸ“‹ Infrastructure Details:') - print(f' Resource Group : {rg_name}') - print(f' Location : {rg_location}') - print(f' APIM SKU : {apim_sku.value}') - print(f' Reveal Backend : {reveal_backend}') - print(f' Gateway URL : {apim_gateway_url}') - print(f' ACA Backend 1 : {aca_url_1}') - print(f' ACA Backend 2 : {aca_url_2}') - print(f' APIs Created : {len(apim_apis)}') - - # Perform basic verification - _verify_infrastructure(rg_name) - else: - print('āŒ Infrastructure creation failed!') + sys.exit(0 if result.success else 1) - return output - - finally: - # Always restore the original working directory - os.chdir(original_cwd) - print(f'šŸ“ Restored working directory to: {original_cwd}') + except Exception as e: + print(f'\nšŸ’„ Error: {str(e)}') + sys.exit(1) + -def _verify_infrastructure(rg_name: str) -> bool: +def _create_aca_specific_apis() -> list[API]: """ - Verify that the infrastructure was created successfully. + Create APIM-ACA specific APIs with Container Apps backends. - Args: - rg_name (str): Resource group name. - Returns: - bool: True if verification passed, False otherwise. + list[API]: List of ACA-specific APIs. """ - print('\nšŸ” Verifying infrastructure...') + # Define the APIs with Container Apps backends + pol_backend = utils.read_policy_xml(BACKEND_XML_POLICY_PATH) + pol_aca_backend_1 = pol_backend.format(backend_id = 'aca-backend-1') + pol_aca_backend_2 = pol_backend.format(backend_id = 'aca-backend-2') + pol_aca_backend_pool = pol_backend.format(backend_id = 'aca-backend-pool') + + # API 1: Hello World (ACA Backend 1) + api_hwaca_1_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 1') + api_hwaca_1 = API('hello-world-aca-1', 'Hello World (ACA 1)', '/aca-1', 'This is the ACA API for Backend 1', pol_aca_backend_1, [api_hwaca_1_get]) + + # API 2: Hello World (ACA Backend 2) + api_hwaca_2_get = GET_APIOperation('This is a GET for Hello World on ACA Backend 2') + api_hwaca_2 = API('hello-world-aca-2', 'Hello World (ACA 2)', '/aca-2', 'This is the ACA API for Backend 2', pol_aca_backend_2, [api_hwaca_2_get]) + + # API 3: Hello World (ACA Backend Pool) + api_hwaca_pool_get = GET_APIOperation('This is a GET for Hello World on ACA Backend Pool') + api_hwaca_pool = API('hello-world-aca-pool', 'Hello World (ACA Pool)', '/aca-pool', 'This is the ACA API for Backend Pool', pol_aca_backend_pool, [api_hwaca_pool_get]) - try: - # Check if the resource group exists - if not utils.does_resource_group_exist(rg_name): - print('āŒ Resource group does not exist!') - return False - - print('āœ… Resource group verified') - - # Get APIM service details - output = utils.run(f'az apim list -g {rg_name} --query "[0]" -o json', print_command_to_run = False, print_errors = False) - - if output.success and output.json_data: - apim_name = output.json_data.get('name') - print(f'āœ… APIM Service verified: {apim_name}') - - # Get Container Apps count - aca_output = utils.run(f'az containerapp list -g {rg_name} --query "length(@)"', print_command_to_run = False, print_errors = False) - - if aca_output.success: - aca_count = int(aca_output.text.strip()) - print(f'āœ… Container Apps verified: {aca_count} app(s) created') - - # Get API count - api_output = utils.run(f'az apim api list --service-name {apim_name} -g {rg_name} --query "length(@)"', - print_command_to_run = False, print_errors = False) - - if api_output.success: - api_count = int(api_output.text.strip()) - print(f'āœ… APIs verified: {api_count} API(s) created') - - # Test basic connectivity (optional) - if api_count > 0: - try: - # Get subscription key for testing - sub_output = utils.run(f'az apim subscription list --service-name {apim_name} -g {rg_name} --query "[0].primaryKey" -o tsv', - print_command_to_run = False, print_errors = False) - - if sub_output.success and sub_output.text.strip(): - print('āœ… Subscription key available for API testing') - except: - pass - - print('\nšŸŽ‰ Infrastructure verification completed successfully!') - return True - - else: - print('\nāŒ APIM service not found!') - return False - - except Exception as e: - print(f'\nāš ļø Verification failed with error: {str(e)}') - return False + return [api_hwaca_1, api_hwaca_2, api_hwaca_pool] def main(): """ Main entry point for command-line usage. """ - parser = argparse.ArgumentParser(description='Create APIM-ACA infrastructure') - parser.add_argument('--location', default='eastus2', help='Azure region (default: eastus2)') - parser.add_argument('--index', type=int, help='Infrastructure index') - parser.add_argument('--sku', choices=['Basicv2', 'Standardv2', 'Premiumv2'], default='Basicv2', help='APIM SKU (default: Basicv2)') - parser.add_argument('--no-reveal-backend', action='store_true', help='Do not reveal backend details in API operations') - + parser = argparse.ArgumentParser(description = 'Create APIM-ACA infrastructure') + parser.add_argument('--location', default = 'eastus2', help = 'Azure region (default: eastus2)') + parser.add_argument('--index', type = int, help = 'Infrastructure index') + parser.add_argument('--sku', choices = ['Basicv2', 'Standardv2', 'Premiumv2'], default = 'Basicv2', help = 'APIM SKU (default: Basicv2)') args = parser.parse_args() - - # Convert SKU string to enum - sku_map = { - 'Basicv2': APIM_SKU.BASICV2, - 'Standardv2': APIM_SKU.STANDARDV2, - 'Premiumv2': APIM_SKU.PREMIUMV2 - } - + + # Convert SKU string to enum using the enum's built-in functionality try: - result = _create_apim_aca_infrastructure( - rg_location = args.location, - index = args.index, - apim_sku = sku_map[args.sku], - reveal_backend = not args.no_reveal_backend - ) - - if result.success: - print('\nšŸŽ‰ Infrastructure creation completed successfully!') - sys.exit(0) - else: - print('\nšŸ’„ Infrastructure creation failed!') - sys.exit(1) - - except Exception as e: - print(f'\nšŸ’„ Error: {str(e)}') + apim_sku = APIM_SKU(args.sku) + except ValueError: + print(f"Error: Invalid SKU '{args.sku}'. Valid options are: {', '.join([sku.value for sku in APIM_SKU])}") sys.exit(1) + create_infrastructure(args.location, args.index, apim_sku) if __name__ == '__main__': main() diff --git a/infrastructure/simple-apim/create.ipynb b/infrastructure/simple-apim/create.ipynb index 5ae77bb..1c9b7df 100644 --- a/infrastructure/simple-apim/create.ipynb +++ b/infrastructure/simple-apim/create.ipynb @@ -20,12 +20,20 @@ "import utils\n", "from apimtypes import *\n", "\n", - "# User-defined parameters (change these as needed)\n", + "# ------------------------------\n", + "# USER CONFIGURATION\n", + "# ------------------------------\n", + "\n", "rg_location = 'eastus2' # Azure region for deployment\n", "index = 1 # Infrastructure index (use different numbers for multiple environments)\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "\n", - "# Create an instance of the desired infrastructure\n", + "\n", + "\n", + "# ------------------------------\n", + "# SYSTEM CONFIGURATION\n", + "# ------------------------------\n", + "\n", "inb_helper = utils.InfrastructureNotebookHelper(rg_location, INFRASTRUCTURE.SIMPLE_APIM, index, apim_sku) \n", "success = inb_helper.create_infrastructure()\n", "\n", diff --git a/infrastructure/simple-apim/create_infrastructure.py b/infrastructure/simple-apim/create_infrastructure.py index 562f079..1fc0cd5 100644 --- a/infrastructure/simple-apim/create_infrastructure.py +++ b/infrastructure/simple-apim/create_infrastructure.py @@ -7,6 +7,16 @@ from apimtypes import APIM_SKU from infrastructures import SimpleApimInfrastructure + +def create_infrastructure(location: str, index: int, apim_sku: APIM_SKU) -> None: + try: + result = SimpleApimInfrastructure(location, index, apim_sku).deploy_infrastructure() + sys.exit(0 if result.success else 1) + + except Exception as e: + print(f'\nšŸ’„ Error: {str(e)}') + sys.exit(1) + def main(): """ Main entry point for command-line usage. @@ -17,26 +27,15 @@ def main(): parser.add_argument('--index', type = int, help = 'Infrastructure index') parser.add_argument('--sku', choices = ['Basicv2', 'Standardv2', 'Premiumv2'], default = 'Basicv2', help = 'APIM SKU (default: Basicv2)') args = parser.parse_args() - + # Convert SKU string to enum using the enum's built-in functionality try: apim_sku = APIM_SKU(args.sku) except ValueError: print(f"Error: Invalid SKU '{args.sku}'. Valid options are: {', '.join([sku.value for sku in APIM_SKU])}") sys.exit(1) - - try: - infra = SimpleApimInfrastructure(args.location, args.index, apim_sku) - result = infra.deploy_infrastructure() - - if result.success: - sys.exit(0) - else: - sys.exit(1) - - except Exception as e: - print(f'\nšŸ’„ Error: {str(e)}') - sys.exit(1) + + create_infrastructure(args.location, args.index, apim_sku) if __name__ == '__main__': main() \ No newline at end of file diff --git a/shared/python/infrastructures.py b/shared/python/infrastructures.py index d56aff9..764cf94 100644 --- a/shared/python/infrastructures.py +++ b/shared/python/infrastructures.py @@ -137,8 +137,13 @@ def _verify_infrastructure(self, rg_name: str) -> bool: except: pass - print('\nšŸŽ‰ Infrastructure verification completed successfully!') - return True + # Call infrastructure-specific verification + if self._verify_infrastructure_specific(rg_name): + print('\nšŸŽ‰ Infrastructure verification completed successfully!') + return True + else: + print('\nāŒ Infrastructure-specific verification failed!') + return False else: print('\nāŒ APIM service not found!') @@ -148,6 +153,20 @@ def _verify_infrastructure(self, rg_name: str) -> bool: print(f'\nāš ļø Verification failed with error: {str(e)}') return False + def _verify_infrastructure_specific(self, rg_name: str) -> bool: + """ + Verify infrastructure-specific components. + This is a virtual method that can be overridden by subclasses for specific verification logic. + + Args: + rg_name (str): Resource group name. + + Returns: + bool: True if verification passed, False otherwise. + """ + # Base implementation - no additional verification required + return True + # ------------------------------ # PUBLIC METHODS # ------------------------------ @@ -253,16 +272,6 @@ def deploy_infrastructure(self) -> 'utils.Output': os.chdir(original_cwd) print(f'šŸ“ Restored working directory to: {original_cwd}') - # @abstractmethod - # def verify_infrastructure(self) -> bool: - # """ - # Verify the infrastructure deployment. - # This method should be implemented in subclasses to handle specific verification logic. - # """ - # pass - - - class SimpleApimInfrastructure(Infrastructure): """ @@ -281,6 +290,32 @@ class ApimAcaInfrastructure(Infrastructure): def __init__(self, rg_location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2, infra_pfs: List[PolicyFragment] | None = None, infra_apis: List[API] | None = None): super().__init__(INFRASTRUCTURE.APIM_ACA, index, rg_location, apim_sku, APIMNetworkMode.PUBLIC, infra_pfs, infra_apis) + def _verify_infrastructure_specific(self, rg_name: str) -> bool: + """ + Verify APIM-ACA specific components. + + Args: + rg_name (str): Resource group name. + + Returns: + bool: True if verification passed, False otherwise. + """ + try: + # Get Container Apps count + aca_output = utils.run(f'az containerapp list -g {rg_name} --query "length(@)"', print_command_to_run = False, print_errors = False) + + if aca_output.success: + aca_count = int(aca_output.text.strip()) + print(f'āœ… Container Apps verified: {aca_count} app(s) created') + return True + else: + print('āŒ Container Apps verification failed!') + return False + + except Exception as e: + print(f'āš ļø Container Apps verification failed with error: {str(e)}') + return False + class AfdApimAcaInfrastructure(Infrastructure): """ @@ -289,3 +324,55 @@ class AfdApimAcaInfrastructure(Infrastructure): def __init__(self, rg_location: str, index: int, apim_sku: APIM_SKU = APIM_SKU.BASICV2, infra_pfs: List[PolicyFragment] | None = None, infra_apis: List[API] | None = None): super().__init__(INFRASTRUCTURE.AFD_APIM_PE, index, rg_location, apim_sku, APIMNetworkMode.PUBLIC, infra_pfs, infra_apis) + + def _define_bicep_parameters(self) -> dict: + """ + Define AFD-APIM-PE specific Bicep parameters. + """ + # Get base parameters + base_params = super()._define_bicep_parameters() + + # Add AFD-specific parameters + afd_params = { + 'apimPublicAccess': {'value': True}, # Initially true for private link approval + 'useACA': {'value': len(self.infra_apis) > 0 if self.infra_apis else False} # Enable ACA if custom APIs are provided + } + + # Merge with base parameters + base_params.update(afd_params) + return base_params + + def _verify_infrastructure_specific(self, rg_name: str) -> bool: + """ + Verify AFD-APIM-PE specific components. + + Args: + rg_name (str): Resource group name. + + Returns: + bool: True if verification passed, False otherwise. + """ + try: + # Check Front Door + afd_output = utils.run(f'az afd profile list -g {rg_name} --query "[0]" -o json', print_command_to_run = False, print_errors = False) + + if afd_output.success and afd_output.json_data: + afd_name = afd_output.json_data.get('name') + print(f'āœ… Azure Front Door verified: {afd_name}') + + # Check Container Apps if they exist (optional for this infrastructure) + aca_output = utils.run(f'az containerapp list -g {rg_name} --query "length(@)"', print_command_to_run = False, print_errors = False) + + if aca_output.success: + aca_count = int(aca_output.text.strip()) + if aca_count > 0: + print(f'āœ… Container Apps verified: {aca_count} app(s) created') + + return True + else: + print('āŒ Azure Front Door verification failed!') + return False + + except Exception as e: + print(f'āš ļø AFD-APIM-PE verification failed with error: {str(e)}') + return False diff --git a/tests/python/test_infrastructures.py b/tests/python/test_infrastructures.py index bd7cb2e..9164fbd 100644 --- a/tests/python/test_infrastructures.py +++ b/tests/python/test_infrastructures.py @@ -323,9 +323,264 @@ def test_define_bicep_parameters(mock_utils): # ------------------------------ -# ABSTRACT METHOD TESTS +# INFRASTRUCTURE VERIFICATION TESTS # ------------------------------ +@pytest.mark.unit +def test_base_infrastructure_verification_success(mock_utils): + """Test base infrastructure verification success.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + rg_location=TEST_LOCATION + ) + + # Mock successful resource group check + mock_utils.does_resource_group_exist.return_value = True + + # Mock successful APIM service check + mock_apim_output = Mock() + mock_apim_output.success = True + mock_apim_output.json_data = {'name': 'test-apim'} + + # Mock successful API count check + mock_api_output = Mock() + mock_api_output.success = True + mock_api_output.text = '5' # 5 APIs + + # Mock successful subscription check + mock_sub_output = Mock() + mock_sub_output.success = True + mock_sub_output.text = 'test-subscription-key' + + mock_utils.run.side_effect = [mock_apim_output, mock_api_output, mock_sub_output] + + result = infra._verify_infrastructure('test-rg') + + assert result is True + mock_utils.does_resource_group_exist.assert_called_once_with('test-rg') + assert mock_utils.run.call_count >= 2 # At least APIM list and API count + +@pytest.mark.unit +def test_base_infrastructure_verification_missing_rg(mock_utils): + """Test base infrastructure verification with missing resource group.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + rg_location=TEST_LOCATION + ) + + # Mock missing resource group + mock_utils.does_resource_group_exist.return_value = False + + result = infra._verify_infrastructure('test-rg') + + assert result is False + mock_utils.does_resource_group_exist.assert_called_once_with('test-rg') + +@pytest.mark.unit +def test_base_infrastructure_verification_missing_apim(mock_utils): + """Test base infrastructure verification with missing APIM service.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + rg_location=TEST_LOCATION + ) + + # Mock successful resource group check + mock_utils.does_resource_group_exist.return_value = True + + # Mock failed APIM service check + mock_apim_output = Mock() + mock_apim_output.success = False + mock_apim_output.json_data = None + + mock_utils.run.return_value = mock_apim_output + + result = infra._verify_infrastructure('test-rg') + + assert result is False + +@pytest.mark.unit +def test_infrastructure_specific_verification_base(mock_utils): + """Test the base infrastructure-specific verification method.""" + infra = infrastructures.Infrastructure( + infra=INFRASTRUCTURE.SIMPLE_APIM, + index=TEST_INDEX, + rg_location=TEST_LOCATION + ) + + # Base implementation should always return True + result = infra._verify_infrastructure_specific('test-rg') + + assert result is True + +# ------------------------------ +# APIM-ACA INFRASTRUCTURE SPECIFIC TESTS +# ------------------------------ + +@pytest.mark.unit +def test_apim_aca_infrastructure_verification_success(mock_utils): + """Test APIM-ACA infrastructure-specific verification success.""" + infra = infrastructures.ApimAcaInfrastructure( + rg_location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.BASICV2 + ) + + # Mock successful Container Apps check + mock_aca_output = Mock() + mock_aca_output.success = True + mock_aca_output.text = '3' # 3 Container Apps + + mock_utils.run.return_value = mock_aca_output + + result = infra._verify_infrastructure_specific('test-rg') + + assert result is True + mock_utils.run.assert_called_once_with( + 'az containerapp list -g test-rg --query "length(@)"', + print_command_to_run=False, + print_errors=False + ) + +@pytest.mark.unit +def test_apim_aca_infrastructure_verification_failure(mock_utils): + """Test APIM-ACA infrastructure-specific verification failure.""" + infra = infrastructures.ApimAcaInfrastructure( + rg_location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.BASICV2 + ) + + # Mock failed Container Apps check + mock_aca_output = Mock() + mock_aca_output.success = False + + mock_utils.run.return_value = mock_aca_output + + result = infra._verify_infrastructure_specific('test-rg') + + assert result is False + + +# ------------------------------ +# AFD-APIM-PE INFRASTRUCTURE SPECIFIC TESTS +# ------------------------------ + +@pytest.mark.unit +def test_afd_apim_infrastructure_verification_success(mock_utils): + """Test AFD-APIM-PE infrastructure-specific verification success.""" + infra = infrastructures.AfdApimAcaInfrastructure( + rg_location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.STANDARDV2 + ) + + # Mock successful Front Door check + mock_afd_output = Mock() + mock_afd_output.success = True + mock_afd_output.json_data = {'name': 'test-afd'} + + # Mock successful Container Apps check + mock_aca_output = Mock() + mock_aca_output.success = True + mock_aca_output.text = '2' # 2 Container Apps + + mock_utils.run.side_effect = [mock_afd_output, mock_aca_output] + + result = infra._verify_infrastructure_specific('test-rg') + + assert result is True + assert mock_utils.run.call_count == 2 + +@pytest.mark.unit +def test_afd_apim_infrastructure_verification_no_afd(mock_utils): + """Test AFD-APIM-PE infrastructure-specific verification with missing AFD.""" + infra = infrastructures.AfdApimAcaInfrastructure( + rg_location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.STANDARDV2 + ) + + # Mock failed Front Door check + mock_afd_output = Mock() + mock_afd_output.success = False + mock_afd_output.json_data = None + + mock_utils.run.return_value = mock_afd_output + + result = infra._verify_infrastructure_specific('test-rg') + + assert result is False + +@pytest.mark.unit +def test_afd_apim_infrastructure_bicep_parameters(mock_utils): + """Test AFD-APIM-PE specific Bicep parameters.""" + # Test with custom APIs (should enable ACA) + custom_apis = [ + API('test-api', 'Test API', '/test', 'Test API description') + ] + + infra = infrastructures.AfdApimAcaInfrastructure( + rg_location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.STANDARDV2, + infra_apis=custom_apis + ) + + # Initialize components + infra._define_policy_fragments() + infra._define_apis() + + bicep_params = infra._define_bicep_parameters() + + # Check AFD-specific parameters + assert 'apimPublicAccess' in bicep_params + assert bicep_params['apimPublicAccess']['value'] is True + assert 'useACA' in bicep_params + assert bicep_params['useACA']['value'] is True # Should be True due to custom APIs + + # Test without custom APIs (should disable ACA) + infra_no_apis = infrastructures.AfdApimAcaInfrastructure( + rg_location=TEST_LOCATION, + index=TEST_INDEX, + apim_sku=APIM_SKU.STANDARDV2 + ) + + # Initialize components + infra_no_apis._define_policy_fragments() + infra_no_apis._define_apis() + + bicep_params_no_apis = infra_no_apis._define_bicep_parameters() + + # Should disable ACA when no custom APIs + assert bicep_params_no_apis['useACA']['value'] is False + + +# ------------------------------ +# INFRASTRUCTURE CLASS CONSISTENCY TESTS +# ------------------------------ + +@pytest.mark.unit +def test_all_concrete_infrastructure_classes_have_verification(mock_utils): + """Test that all concrete infrastructure classes have verification methods.""" + # Test Simple APIM (uses base verification) + simple_infra = infrastructures.SimpleApimInfrastructure(TEST_LOCATION, TEST_INDEX) + assert hasattr(simple_infra, '_verify_infrastructure_specific') + assert callable(simple_infra._verify_infrastructure_specific) + + # Test APIM-ACA (has custom verification) + aca_infra = infrastructures.ApimAcaInfrastructure(TEST_LOCATION, TEST_INDEX) + assert hasattr(aca_infra, '_verify_infrastructure_specific') + assert callable(aca_infra._verify_infrastructure_specific) + + # Test AFD-APIM-PE (has custom verification) + afd_infra = infrastructures.AfdApimAcaInfrastructure(TEST_LOCATION, TEST_INDEX) + assert hasattr(afd_infra, '_verify_infrastructure_specific') + assert callable(afd_infra._verify_infrastructure_specific) + + # ------------------------------ # DEPLOYMENT TESTS # ------------------------------ From 57e1d668218917a6940d5fe3cee53fb466c10e0a Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 14:59:05 -0400 Subject: [PATCH 5/9] Add APIM SKU for Notebook Helper use --- samples/_TEMPLATE/create.ipynb | 7 ++++--- samples/authX-pro/create.ipynb | 3 ++- samples/authX/create.ipynb | 20 ++++++++++++++++---- samples/azure-maps/create.ipynb | 3 ++- samples/general/create.ipynb | 3 ++- samples/load-balancing/create.ipynb | 7 ++++--- samples/oauth-3rd-party/create.ipynb | 3 ++- samples/secure-blob-access/create.ipynb | 3 ++- shared/python/utils.py | 7 ++++--- 9 files changed, 38 insertions(+), 18 deletions(-) diff --git a/samples/_TEMPLATE/create.ipynb b/samples/_TEMPLATE/create.ipynb index 379fcb3..9609715 100644 --- a/samples/_TEMPLATE/create.ipynb +++ b/samples/_TEMPLATE/create.ipynb @@ -24,9 +24,10 @@ "\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.AFD_APIM_PE\n", - "api_prefix = 'template-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", - "tags = ['tag1', 'tag2'] # ENTER DESCRIPTIVE TAGS\n", + "api_prefix = 'template-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", + "tags = ['tag1', 'tag2'] # ENTER DESCRIPTIVE TAGS\n", "\n", "\n", "\n", @@ -37,7 +38,7 @@ "sample_folder = '_TEMPLATE'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index, apim_sku = apim_sku)\n", "\n", "# Define the APIs and their operations and policies\n", "\n", diff --git a/samples/authX-pro/create.ipynb b/samples/authX-pro/create.ipynb index 255b542..6afb1d7 100644 --- a/samples/authX-pro/create.ipynb +++ b/samples/authX-pro/create.ipynb @@ -24,6 +24,7 @@ "\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", "api_prefix = 'authX-pro-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", "tags = ['authX-pro', 'jwt', 'policy-fragment'] # ENTER DESCRIPTIVE TAG(S)\n", @@ -38,7 +39,7 @@ "sample_folder = 'authX-pro'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index, apim_sku = apim_sku)\n", "\n", "# Define the APIs and their operations and policies\n", "\n", diff --git a/samples/authX/create.ipynb b/samples/authX/create.ipynb index f4e07d7..8196d7a 100644 --- a/samples/authX/create.ipynb +++ b/samples/authX/create.ipynb @@ -13,7 +13,18 @@ "cell_type": "code", "execution_count": null, "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "šŸ‘‰šŸ½ \u001b[1;34mJWT key value : C6t1uTHcRRryHYODH4EWYmjlHmfDlQL2zq7DlfvzqfXzeJ6BZLTYZQUKCEfZ5IYo89jQGuLTgdb0jQW\u001b[0m \n", + "šŸ‘‰šŸ½ \u001b[1;34mJWT key value (base64) : QzZ0MXVUSGNSUnJ5SFlPREg0RVdZbWpsSG1mRGxRTDJ6cTdEbGZ2enFmWHplSjZCWkxUWVpRVUtDRWZaNUlZbzg5alFHdUxUZ2RiMGpRVw==\u001b[0m \n", + "\n", + "āœ… \u001b[1;32mNotebook initialized\u001b[0m āŒš 14:15:18.283524 \n" + ] + } + ], "source": [ "import utils\n", "from apimtypes import *\n", @@ -24,9 +35,10 @@ "\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", - "api_prefix = 'authX-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", - "tags = ['authX', 'jwt', 'hr'] # ENTER DESCRIPTIVE TAG(S)\n", + "api_prefix = 'authX-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", + "tags = ['authX', 'jwt', 'hr'] # ENTER DESCRIPTIVE TAG(S)\n", "\n", "\n", "\n", @@ -38,7 +50,7 @@ "sample_folder = 'authX'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index, apim_sku = apim_sku)\n", "\n", "# Define the APIs and their operations and policies\n", "\n", diff --git a/samples/azure-maps/create.ipynb b/samples/azure-maps/create.ipynb index 43b7c44..04e6f91 100644 --- a/samples/azure-maps/create.ipynb +++ b/samples/azure-maps/create.ipynb @@ -28,6 +28,7 @@ "\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", "api_prefix = '' # using a prefix results in some issues with at least the default API GET. It's fine to leave it off.\n", "tags = ['azure-maps']\n", @@ -41,7 +42,7 @@ "sample_folder = 'azure-maps'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index, apim_sku = apim_sku)\n", "azure_maps_url = 'https://atlas.microsoft.com'\n", "\n", "# Define the APIs and their operations and policies\n", diff --git a/samples/general/create.ipynb b/samples/general/create.ipynb index d70d9b8..26c666d 100644 --- a/samples/general/create.ipynb +++ b/samples/general/create.ipynb @@ -24,6 +24,7 @@ "\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", "api_prefix = '' # Not defining a prefix for general as these APIs will live off the root\n", "tags = ['general']\n", @@ -37,7 +38,7 @@ "sample_folder = 'general'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index, apim_sku = apim_sku)\n", "\n", "# Define the APIs and their operations and policies\n", "\n", diff --git a/samples/load-balancing/create.ipynb b/samples/load-balancing/create.ipynb index 0fab30a..b14c426 100644 --- a/samples/load-balancing/create.ipynb +++ b/samples/load-balancing/create.ipynb @@ -24,9 +24,10 @@ "\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.APIM_ACA\n", - "api_prefix = 'lb-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", - "tags = ['load-balancing'] # ENTER DESCRIPTIVE TAG(S)\n", + "api_prefix = 'lb-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", + "tags = ['load-balancing'] # ENTER DESCRIPTIVE TAG(S)\n", "\n", "\n", "\n", @@ -37,7 +38,7 @@ "sample_folder = 'load-balancing'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, index = index, apim_sku = apim_sku)\n", "\n", "# Define the APIs and their operations and policies\n", "\n", diff --git a/samples/oauth-3rd-party/create.ipynb b/samples/oauth-3rd-party/create.ipynb index 31d64b6..89f59d0 100644 --- a/samples/oauth-3rd-party/create.ipynb +++ b/samples/oauth-3rd-party/create.ipynb @@ -33,6 +33,7 @@ "# ------------------------------\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", "api_prefix = 'oauth-' # Prefix for API names # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", "tags = ['oauth-3rd-party', 'jwt', 'credential-manager', 'policy-fragment'] # ENTER DESCRIPTIVE TAG(S)\n", @@ -47,7 +48,7 @@ "sample_folder = 'oauth-3rd-party'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index, apim_sku = apim_sku)\n", "\n", "# OAuth credentials (required environment variables)\n", "client_id = os.getenv('SPOTIFY_CLIENT_ID')\n", diff --git a/samples/secure-blob-access/create.ipynb b/samples/secure-blob-access/create.ipynb index 4eebe40..0142512 100644 --- a/samples/secure-blob-access/create.ipynb +++ b/samples/secure-blob-access/create.ipynb @@ -26,6 +26,7 @@ "\n", "rg_location = 'eastus2'\n", "index = 1\n", + "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", "api_prefix = 'blob-'\n", "tags = ['secure-blob-access', 'valet-key', 'storage', 'jwt', 'authz']\n", @@ -40,7 +41,7 @@ "sample_folder = 'secure-blob-access'\n", "rg_name = utils.get_infra_rg_name(deployment, index)\n", "supported_infras = [INFRASTRUCTURE.AFD_APIM_PE, INFRASTRUCTURE.APIM_ACA, INFRASTRUCTURE.SIMPLE_APIM]\n", - "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index)\n", + "nb_helper = utils.NotebookHelper(sample_folder, rg_name, rg_location, deployment, supported_infras, True, index = index, apim_sku = apim_sku)\n", "\n", "# Blob storage configuration\n", "container_name = 'hr-assets'\n", diff --git a/shared/python/utils.py b/shared/python/utils.py index 60017bc..e7ebe00 100644 --- a/shared/python/utils.py +++ b/shared/python/utils.py @@ -317,7 +317,7 @@ class NotebookHelper: # CONSTRUCTOR # ------------------------------ - def __init__(self, sample_folder: str, rg_name: str, rg_location: str, deployment: INFRASTRUCTURE, supported_infrastructures = list[INFRASTRUCTURE], use_jwt: bool = False, index: int = 1, is_debug = False): + def __init__(self, sample_folder: str, rg_name: str, rg_location: str, deployment: INFRASTRUCTURE, supported_infrastructures = list[INFRASTRUCTURE], use_jwt: bool = False, index: int = 1, is_debug = False, apim_sku: APIM_SKU = APIM_SKU.BASICV2): """ Initialize the NotebookHelper with sample configuration and infrastructure details. @@ -340,6 +340,7 @@ def __init__(self, sample_folder: str, rg_name: str, rg_location: str, deploymen self.use_jwt = use_jwt self.index = index self.is_debug = is_debug + self.apim_sku = apim_sku validate_infrastructure(deployment, supported_infrastructures) @@ -479,7 +480,7 @@ def _query_and_select_infrastructure(self) -> tuple[INFRASTRUCTURE | None, int | print_info(f'Creating new infrastructure: {self.deployment.value}{' (index: ' + str(selected_index) + ')' if selected_index is not None else ''}') # Execute the infrastructure creation - inb_helper = InfrastructureNotebookHelper(self.rg_location, self.deployment, selected_index, APIM_SKU.BASICV2) + inb_helper = InfrastructureNotebookHelper(self.rg_location, self.deployment, selected_index, self.apim_sku) success = inb_helper.create_infrastructure(True) # Bypass infrastructure check to force creation if success: @@ -511,7 +512,7 @@ def _query_and_select_infrastructure(self) -> tuple[INFRASTRUCTURE | None, int | print_info(f'Creating new infrastructure: {selected_infra.value}{' (index: ' + str(selected_index) + ')' if selected_index is not None else ''}') # Execute the infrastructure creation - inb_helper = InfrastructureNotebookHelper(self.rg_location, self.deployment, selected_index, APIM_SKU.BASICV2) + inb_helper = InfrastructureNotebookHelper(self.rg_location, self.deployment, selected_index, self.apim_sku) success = inb_helper.create_infrastructure(True) # Bypass infrastructure check to force creation if success: From 31272cc1e8077580c59fe30defc66ebc3a0664a5 Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 19:13:26 -0400 Subject: [PATCH 6/9] Add comments for infra options --- samples/authX-pro/create.ipynb | 2 +- samples/authX/create.ipynb | 2 +- samples/azure-maps/create.ipynb | 2 +- samples/general/create.ipynb | 2 +- samples/load-balancing/create.ipynb | 2 +- samples/oauth-3rd-party/create.ipynb | 2 +- samples/secure-blob-access/create.ipynb | 2 +- 7 files changed, 7 insertions(+), 7 deletions(-) diff --git a/samples/authX-pro/create.ipynb b/samples/authX-pro/create.ipynb index 6afb1d7..af98b2b 100644 --- a/samples/authX-pro/create.ipynb +++ b/samples/authX-pro/create.ipynb @@ -25,7 +25,7 @@ "rg_location = 'eastus2'\n", "index = 1\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", + "deployment = INFRASTRUCTURE.SIMPLE_APIM # Options: 'AFD_APIM_PE', 'APIM_ACA', 'SIMPLE_APIM'\n", "api_prefix = 'authX-pro-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", "tags = ['authX-pro', 'jwt', 'policy-fragment'] # ENTER DESCRIPTIVE TAG(S)\n", "\n", diff --git a/samples/authX/create.ipynb b/samples/authX/create.ipynb index 8196d7a..83164f5 100644 --- a/samples/authX/create.ipynb +++ b/samples/authX/create.ipynb @@ -36,7 +36,7 @@ "rg_location = 'eastus2'\n", "index = 1\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", + "deployment = INFRASTRUCTURE.SIMPLE_APIM # Options: 'AFD_APIM_PE', 'APIM_ACA', 'SIMPLE_APIM'\n", "api_prefix = 'authX-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", "tags = ['authX', 'jwt', 'hr'] # ENTER DESCRIPTIVE TAG(S)\n", "\n", diff --git a/samples/azure-maps/create.ipynb b/samples/azure-maps/create.ipynb index 04e6f91..833ac10 100644 --- a/samples/azure-maps/create.ipynb +++ b/samples/azure-maps/create.ipynb @@ -29,7 +29,7 @@ "rg_location = 'eastus2'\n", "index = 1\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", + "deployment = INFRASTRUCTURE.SIMPLE_APIM # Options: 'AFD_APIM_PE', 'APIM_ACA', 'SIMPLE_APIM'\n", "api_prefix = '' # using a prefix results in some issues with at least the default API GET. It's fine to leave it off.\n", "tags = ['azure-maps']\n", "\n", diff --git a/samples/general/create.ipynb b/samples/general/create.ipynb index 26c666d..81493c6 100644 --- a/samples/general/create.ipynb +++ b/samples/general/create.ipynb @@ -25,7 +25,7 @@ "rg_location = 'eastus2'\n", "index = 1\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", + "deployment = INFRASTRUCTURE.SIMPLE_APIM # Options: 'AFD_APIM_PE', 'APIM_ACA', 'SIMPLE_APIM'\n", "api_prefix = '' # Not defining a prefix for general as these APIs will live off the root\n", "tags = ['general']\n", "\n", diff --git a/samples/load-balancing/create.ipynb b/samples/load-balancing/create.ipynb index b14c426..80f7289 100644 --- a/samples/load-balancing/create.ipynb +++ b/samples/load-balancing/create.ipynb @@ -25,7 +25,7 @@ "rg_location = 'eastus2'\n", "index = 1\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "deployment = INFRASTRUCTURE.APIM_ACA\n", + "deployment = INFRASTRUCTURE.APIM_ACA # Options: 'AFD_APIM_PE', 'APIM_ACA'\n", "api_prefix = 'lb-' # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", "tags = ['load-balancing'] # ENTER DESCRIPTIVE TAG(S)\n", "\n", diff --git a/samples/oauth-3rd-party/create.ipynb b/samples/oauth-3rd-party/create.ipynb index 89f59d0..ee6bb98 100644 --- a/samples/oauth-3rd-party/create.ipynb +++ b/samples/oauth-3rd-party/create.ipynb @@ -34,7 +34,7 @@ "rg_location = 'eastus2'\n", "index = 1\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", + "deployment = INFRASTRUCTURE.SIMPLE_APIM # Options: 'AFD_APIM_PE', 'APIM_ACA', 'SIMPLE_APIM'\n", "api_prefix = 'oauth-' # Prefix for API names # ENTER A PREFIX FOR THE APIS TO REDUCE COLLISION POTENTIAL WITH OTHER SAMPLES\n", "tags = ['oauth-3rd-party', 'jwt', 'credential-manager', 'policy-fragment'] # ENTER DESCRIPTIVE TAG(S)\n", "\n", diff --git a/samples/secure-blob-access/create.ipynb b/samples/secure-blob-access/create.ipynb index 0142512..4847f78 100644 --- a/samples/secure-blob-access/create.ipynb +++ b/samples/secure-blob-access/create.ipynb @@ -27,7 +27,7 @@ "rg_location = 'eastus2'\n", "index = 1\n", "apim_sku = APIM_SKU.BASICV2 # Options: 'BASICV2', 'STANDARDV2', 'PREMIUMV2'\n", - "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", + "deployment = INFRASTRUCTURE.SIMPLE_APIM # Options: 'AFD_APIM_PE', 'APIM_ACA', 'SIMPLE_APIM'\n", "api_prefix = 'blob-'\n", "tags = ['secure-blob-access', 'valet-key', 'storage', 'jwt', 'authz']\n", "\n", From b04a31b35d78e9659d3e15e14aa5c35c2c41fd19 Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 19:13:43 -0400 Subject: [PATCH 7/9] Order parameters --- shared/python/utils.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/shared/python/utils.py b/shared/python/utils.py index e7ebe00..8b21435 100644 --- a/shared/python/utils.py +++ b/shared/python/utils.py @@ -286,8 +286,8 @@ def create_infrastructure(self, bypass_infrastructure_check: bool = False) -> bo sys.executable, os.path.join(find_project_root(), 'infrastructure', infra_folder, 'create_infrastructure.py'), '--location', self.rg_location, - '--sku', str(self.apim_sku.value), - '--index', str(self.index) + '--index', str(self.index), + '--sku', str(self.apim_sku.value) ] # Execute the infrastructure creation script with real-time output streaming and UTF-8 encoding to handle Unicode characters properly From d8773933870847b9b43b1f2327714196809833dc Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 23:18:44 -0400 Subject: [PATCH 8/9] Fix the AFD infrastructure deployment --- shared/python/infrastructures.py | 226 ++++++++++++++++++++++++++- tests/python/test_infrastructures.py | 10 +- 2 files changed, 233 insertions(+), 3 deletions(-) diff --git a/shared/python/infrastructures.py b/shared/python/infrastructures.py index 764cf94..d76f25e 100644 --- a/shared/python/infrastructures.py +++ b/shared/python/infrastructures.py @@ -7,7 +7,7 @@ from pathlib import Path from apimtypes import * import utils -# from abc import ABC, abstractmethod +from utils import Output # ------------------------------ @@ -342,6 +342,217 @@ def _define_bicep_parameters(self) -> dict: base_params.update(afd_params) return base_params + def _approve_private_link_connections(self, apim_service_id: str) -> bool: + """ + Approve pending private link connections from AFD to APIM. + + Args: + apim_service_id (str): APIM service resource ID. + + Returns: + bool: True if all connections were approved successfully, False otherwise. + """ + print('\nšŸ”— Step 3: Approving Front Door private link connection to APIM...') + + try: + # Get all pending private endpoint connections + output = utils.run( + f'az network private-endpoint-connection list --id {apim_service_id} --query "[?contains(properties.privateLinkServiceConnectionState.status, \'Pending\')]" -o json', + print_command_to_run = False, + print_errors = False + ) + + if not output.success: + print('āŒ Failed to retrieve private endpoint connections') + return False + + pending_connections = output.json_data if output.is_json else [] + + # Handle both single object and list + if isinstance(pending_connections, dict): + pending_connections = [pending_connections] + + total = len(pending_connections) + print(f' Found {total} pending private link service connection(s)') + + if total == 0: + print(' āœ… No pending connections found - may already be approved') + return True + + # Approve each pending connection + for i, conn in enumerate(pending_connections, 1): + conn_id = conn.get('id') + conn_name = conn.get('name', '') + print(f' Approving {i}/{total}: {conn_name}') + + approve_result = utils.run( + f'az network private-endpoint-connection approve --id {conn_id} --description "Approved by infrastructure deployment"', + f'āœ… Private Link Connection approved: {conn_name}', + f'āŒ Failed to approve Private Link Connection: {conn_name}', + print_command_to_run = False + ) + + if not approve_result.success: + return False + + print(' āœ… All private link connections approved successfully') + return True + + except Exception as e: + print(f' āŒ Error during private link approval: {str(e)}') + return False + + def _disable_apim_public_access(self) -> bool: + """ + Disable public network access to APIM by redeploying with updated parameters. + + Returns: + bool: True if deployment succeeded, False otherwise. + """ + print('\nšŸ”’ Step 5: Disabling API Management public network access...') + + try: + # Update parameters to disable public access + self.bicep_parameters['apimPublicAccess']['value'] = False + + # Write updated parameters file + original_cwd = os.getcwd() + shared_dir = Path(__file__).parent + infra_dir = shared_dir.parent.parent / 'infrastructure' / 'afd-apim-pe' + + try: + os.chdir(infra_dir) + + bicep_parameters_format = { + '$schema': 'https://schema.management.azure.com/schemas/2019-04-01/deploymentParameters.json#', + 'contentVersion': '1.0.0.0', + 'parameters': self.bicep_parameters + } + + params_file_path = infra_dir / 'params.json' + with open(params_file_path, 'w') as file: + file.write(json.dumps(bicep_parameters_format)) + + print(' šŸ“ Updated parameters to disable public access') + + # Run the second deployment + main_bicep_path = infra_dir / 'main.bicep' + output = utils.run( + f'az deployment group create --name {self.infra.value}-lockdown --resource-group {self.rg_name} --template-file "{main_bicep_path}" --parameters "{params_file_path}" --query "properties.outputs"', + 'āœ… Public access disabled successfully', + 'āŒ Failed to disable public access', + print_command_to_run = False + ) + + return output.success + + finally: + os.chdir(original_cwd) + + except Exception as e: + print(f' āŒ Error during public access disable: {str(e)}') + return False + + def _verify_apim_connectivity(self, apim_gateway_url: str) -> bool: + """ + Verify APIM connectivity before disabling public access using the health check endpoint. + + Args: + apim_gateway_url (str): APIM gateway URL. + + Returns: + bool: True if connectivity test passed, False otherwise. + """ + print('\nāœ… Step 4: Verifying API request success via API Management...') + + try: + # Use the health check endpoint which doesn't require a subscription key + import requests + + healthcheck_url = f'{apim_gateway_url}/status-0123456789abcdef' + print(f' Testing connectivity to health check endpoint: {healthcheck_url}') + + response = requests.get(healthcheck_url, timeout=30) + + if response.status_code == 200: + print(' āœ… APIM connectivity verified - Health check returned 200') + return True + else: + print(f' āš ļø APIM health check returned status code {response.status_code} (expected 200)') + return True # Continue anyway as this might be expected during deployment + + except Exception as e: + print(f' āš ļø APIM connectivity test failed: {str(e)}') + print(' ā„¹ļø Continuing deployment - this may be expected during infrastructure setup') + return True # Continue anyway + + def deploy_infrastructure(self) -> Output: + """ + Deploy the AFD-APIM-PE infrastructure with the required multi-step process. + + Returns: + utils.Output: The deployment result. + """ + print('\nšŸš€ Starting AFD-APIM-PE infrastructure deployment...\n') + print(' This deployment requires multiple steps:\n') + print(' 1. Initial deployment with public access enabled') + print(' 2. Approve private link connections') + print(' 3. Verify connectivity') + print(' 4. Disable public access to APIM') + print(' 5. Final verification\n') + + # Step 1 & 2: Initial deployment using base class method + output = super().deploy_infrastructure() + + if not output.success: + print('āŒ Initial deployment failed!') + return output + + print('\nāœ… Step 1 & 2: Initial infrastructure deployment completed') + + # Extract required values from deployment output + if not output.json_data: + print('āŒ No deployment output data available') + return output + + apim_service_id = output.get('apimServiceId', 'APIM Service ID', suppress_logging = True) + apim_gateway_url = output.get('apimResourceGatewayURL', 'APIM Gateway URL', suppress_logging = True) + + if not apim_service_id or not apim_gateway_url: + print('āŒ Required APIM information not found in deployment output') + return output + + # Step 3: Approve private link connections + if not self._approve_private_link_connections(apim_service_id): + print('āŒ Private link approval failed!') + # Create a failed output object + failed_output = utils.Output() + failed_output.success = False + failed_output.text = 'Private link approval failed' + return failed_output + + # Step 4: Verify connectivity (optional - continues on failure) + self._verify_apim_connectivity(apim_gateway_url) + + # Step 5: Disable public access + if not self._disable_apim_public_access(): + print('āŒ Failed to disable public access!') + # Create a failed output object + failed_output = utils.Output() + failed_output.success = False + failed_output.text = 'Failed to disable public access' + return failed_output + + print('\nšŸŽ‰ AFD-APIM-PE infrastructure deployment completed successfully!\n') + print('\nšŸ“‹ Final Configuration:\n') + print(' āœ… Azure Front Door deployed') + print(' āœ… API Management deployed with private endpoints') + print(' āœ… Private link connections approved') + print(' āœ… Public access to APIM disabled') + print(' ā„¹ļø Traffic now flows: Internet ā†’ AFD ā†’ Private Endpoint ā†’ APIM') + + return output + def _verify_infrastructure_specific(self, rg_name: str) -> bool: """ Verify AFD-APIM-PE specific components. @@ -368,6 +579,19 @@ def _verify_infrastructure_specific(self, rg_name: str) -> bool: if aca_count > 0: print(f'āœ… Container Apps verified: {aca_count} app(s) created') + # Verify private endpoint connections (optional - don't fail if it errors) + try: + apim_output = utils.run(f'az apim list -g {rg_name} --query "[0].id" -o tsv', print_command_to_run = False, print_errors = False) + if apim_output.success and apim_output.text.strip(): + apim_id = apim_output.text.strip() + pe_output = utils.run(f'az network private-endpoint-connection list --id {apim_id} --query "length(@)"', print_command_to_run = False, print_errors = False) + if pe_output.success: + pe_count = int(pe_output.text.strip()) + print(f'āœ… Private endpoint connections: {pe_count}') + except: + # Don't fail verification if private endpoint check fails + pass + return True else: print('āŒ Azure Front Door verification failed!') diff --git a/tests/python/test_infrastructures.py b/tests/python/test_infrastructures.py index 9164fbd..0afcabf 100644 --- a/tests/python/test_infrastructures.py +++ b/tests/python/test_infrastructures.py @@ -487,12 +487,18 @@ def test_afd_apim_infrastructure_verification_success(mock_utils): mock_aca_output.success = True mock_aca_output.text = '2' # 2 Container Apps - mock_utils.run.side_effect = [mock_afd_output, mock_aca_output] + # Mock successful APIM check for private endpoints (optional third call) + mock_apim_output = Mock() + mock_apim_output.success = True + mock_apim_output.text = 'apim-resource-id' + + mock_utils.run.side_effect = [mock_afd_output, mock_aca_output, mock_apim_output] result = infra._verify_infrastructure_specific('test-rg') assert result is True - assert mock_utils.run.call_count == 2 + # Allow for 2-3 calls (3rd call is optional for private endpoint verification) + assert mock_utils.run.call_count >= 2 @pytest.mark.unit def test_afd_apim_infrastructure_verification_no_afd(mock_utils): From ab6461f774b81db8dbf4b11ca18a8de826947aab Mon Sep 17 00:00:00 2001 From: Simon Kurtz Date: Wed, 6 Aug 2025 23:19:18 -0400 Subject: [PATCH 9/9] Clean up --- shared/python/utils.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/shared/python/utils.py b/shared/python/utils.py index 8b21435..4ece227 100644 --- a/shared/python/utils.py +++ b/shared/python/utils.py @@ -619,6 +619,8 @@ def deploy_sample(self, bicep_parameters: dict) -> Output: print('āœ… Desired infrastructure already exists, proceeding with sample deployment') # Deploy the sample APIs to the selected infrastructure + print(f'\n------------------------------------------------') + print(f'\nSAMPLE DEPLOYMENT') print(f'\nDeploying sample to:\n') print(f' Infrastructure : {self.deployment.value}') print(f' Index : {self.index}') @@ -675,9 +677,6 @@ def _cleanup_resources(deployment_name: str, rg_name: str) -> None: output = run(f'az deployment group show --name {deployment_name} -g {rg_name} -o json', 'Deployment retrieved', 'Failed to retrieve the deployment', print_command_to_run = False) 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', print_command_to_run = False) @@ -1237,6 +1236,8 @@ def cleanup_infra_deployments(deployment: INFRASTRUCTURE, indexes: int | list[in _cleanup_resources(deployment.value, rg_name) i += 1 + print_ok('All done!') + def extract_json(text: str) -> Any: """ Extract the first valid JSON object or array from a string and return it as a Python object.