Bring over bicep conversion and deploy functions

src/aosm/azext_aosm/cli_handlers/onboarding_base_handler.py

@@ -149,20 +149,18 @@ def build(self):
         self.build_artifact_list()
         self.build_resource_bicep()
 
-    def publish(self, input_json_path: str):
+    def publish(self, resource_client):
         """Publish the definition."""
         # Takes folder, deploys to Azure
         #  - Read folder/ create folder object
-        #  - For each step (element):
-        #    - Do element.deploy()
+        #  - Do folder.deploy(resource_client)
         # TODO: Implement
 
-    def delete(self, input_json_path: str):
+    def delete(self):
         """Delete the definition."""
         # Takes folder, deletes to Azure
         #  - Read folder/ create folder object
-        #  - For each element (reversed):
-        #    - Do element.delete()
+        #  - Do folder.delete(resource_client)
         # TODO: Implement
 
     @abstractmethod

src/aosm/azext_aosm/common/artifact.py

@@ -71,6 +71,8 @@ def __init__(self, artifact_manifest: ManifestArtifactFormat, file_path: Path):
         self.file_path = file_path
 
     # TODO: Implement
+    # Note: check if the artifact name ends in .bicep and if so use utils.convert_bicep_to_arm()
+    # This way we can support in-place Bicep artifacts in the folder.
     def upload(self):
         """Upload the artifact."""
         raise NotImplementedError

src/aosm/azext_aosm/common/utils.py

@@ -3,9 +3,46 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-def convert_bicep_to_arm():
-    # Need this in bicep element and artifacts
-    return NotImplementedError
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+
+
+def convert_bicep_to_arm(bicep_template_path: Path) -> dict:
+    """
+    Convert a bicep template into an ARM template.
+
+    :param bicep_template_path: The path to the bicep template to be converted
+    :return: Output dictionary representation of the ARM template JSON.
+    """
+    with tempfile.TemporaryDirectory() as tmpdir:
+        bicep_filename = bicep_template_path.name
+        arm_template_name = bicep_filename.replace(".bicep", ".json")
+        arm_path = Path(tmpdir) / arm_template_name
+
+        try:
+            subprocess.run(  # noqa
+                [
+                    str(shutil.which("az")),
+                    "bicep",
+                    "build",
+                    "--file",
+                    bicep_template_path,
+                    "--outfile",
+                    str(arm_path),
+                ],
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
+        except subprocess.CalledProcessError as err:
+            raise RuntimeError("Bicep to ARM template compilation failed")
+
+        arm_json = json.loads(arm_path.read_text())
+
+    return arm_json
 
 
 def create_bicep_from_template():

src/aosm/azext_aosm/custom.py

@@ -3,7 +3,10 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+from azure.cli.core.commands import AzCliCommand
+
 # from azext_aosm.cli_handlers.onboarding_nfd_base_handler import OnboardingNFDBaseCLIHandler
+from azext_aosm._client_factory import cf_resources
 from azext_aosm.cli_handlers.onboarding_cnf_handler import OnboardingCNFCLIHandler
 from azext_aosm.cli_handlers.onboarding_vnf_handler import OnboardingVNFCLIHandler
 from azext_aosm.cli_handlers.onboarding_nsd_handler import OnboardingNSDCLIHandler
@@ -33,25 +36,27 @@ def onboard_nfd_build(definition_type: str, config_file: str):
         raise Exception("Invalid definition type")
 
 
-def onboard_nfd_publish(definition_type: str, config_file: str):
+def onboard_nfd_publish(cmd: AzCliCommand, definition_type: str, config_file: str):
+    client = cf_resources(cmd.cli_ctx)
     if definition_type == "cnf":
         handler = OnboardingCNFCLIHandler(config_file)
-        handler.publish()
+        handler.publish(client)
     elif definition_type == "vnf":
         handler = OnboardingVNFCLIHandler(config_file)
-        handler.publish()
+        handler.publish(client)
     else:
         # TODO: better error
         raise Exception("Invalid definition type")
 
 
-def onboard_nfd_delete(definition_type: str, config_file: str):
+def onboard_nfd_delete(cmd: AzCliCommand, definition_type: str, config_file: str):
+    client = cf_resources(cmd.cli_ctx)
     if definition_type == "cnf":
         handler = OnboardingCNFCLIHandler(config_file)
-        handler.delete()
+        handler.delete(client)
     elif definition_type == "vnf":
         handler = OnboardingVNFCLIHandler(config_file)
-        handler.delete()
+        handler.delete(client)
     else:
         # TODO: better error
         raise Exception("Invalid definition type")
@@ -67,11 +72,13 @@ def onboard_nsd_build(config_file: str):
     handler.build()
 
 
-def onboard_nsd_publish(config_file: str):
+def onboard_nsd_publish(cmd: AzCliCommand, config_file: str):
+    client = cf_resources(cmd.cli_ctx)
     handler = OnboardingNSDCLIHandler(config_file)
-    handler.publish()
+    handler.publish(client)
 
 
-def onboard_nsd_delete(config_file: str):
+def onboard_nsd_delete(cmd: AzCliCommand, config_file: str):
+    client = cf_resources(cmd.cli_ctx)
     handler = OnboardingNSDCLIHandler(config_file)
-    handler.delete()
+    handler.delete(client)

src/aosm/azext_aosm/definition_folder/reader/artifact_definition.py

@@ -24,12 +24,12 @@ def __init__(self, path: Path, only_delete_on_clean: bool):
                 raise ValueError("Artifact type is missing or invalid")
             self.artifacts.append(ARTIFACT_TYPE_TO_CLASS[artifact["type"]].from_dict(artifact))
 
-    def deploy(self):
+    def deploy(self, resource_client):
         """Deploy the element."""
         for artifact in self.artifacts:
-            artifact.upload()
+            artifact.upload(resource_client)
 
-    def delete(self):
+    def delete(self, resource_client):
         """Delete the element."""
         # TODO: Implement?
         pass

src/aosm/azext_aosm/definition_folder/reader/bicep_definition.py

@@ -3,8 +3,13 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import time
+from azure.cli.core.commands import LongRunningOperation
+from azure.mgmt.resource import ResourceManagementClient
+from azure.mgmt.resource.resources.models import DeploymentExtended
 from pathlib import Path
 
+from azext_aosm.common.utils import convert_bicep_to_arm
 from azext_aosm.definition_folder.reader.base_definition import BaseDefinitionElement
 
 
@@ -14,12 +19,105 @@ class BicepDefinitionElement(BaseDefinitionElement):
     def __init__(self, path: Path, only_delete_on_clean: bool):
         super().__init__(path, only_delete_on_clean)
 
-    def deploy(self):
+    def deploy(self, resource_client: ResourceManagementClient):
         """Deploy the element."""
-        # TODO: Implement.
-        pass
+        arm_json = convert_bicep_to_arm(self.path / "deploy.bicep")
+        self._validate_and_deploy_arm_template(
+            template=arm_json,
+            # TODO: Create/store parameters somewhere? also pass in resource group
+            parameters={},
+            resource_group="",
+            resource_client=resource_client,
+        )
 
-    def delete(self):
+    def delete(self, resource_client: ResourceManagementClient):
         """Delete the element."""
-        # TODO: Implement.
+        # TODO: Delete resources by Bicep or ARM template.
+        # The ResourceManagementClient doesn't just do it, so we may need to parse the
+        # template for resources and delete them individually.
         pass
+
+    def _validate_and_deploy_arm_template(
+        self, template: any, parameters: dict[any, any], resource_group: str, resource_client: ResourceManagementClient
+    ) -> any:
+        """
+        Validate and deploy an individual ARM template.
+
+        This ARM template will be created in the resource group passed in.
+
+        :param template: The JSON contents of the template to deploy
+        :param parameters: The JSON contents of the parameters file
+        :param resource_group: The name of the resource group that has been deployed
+
+        :return: Output dictionary from the bicep template.
+        :raise RuntimeError if validation or deploy fails
+        """
+        # Get current time from the time module and remove all digits after the decimal
+        # point
+        current_time = str(time.time()).split(".", maxsplit=1)[0]
+
+        # Add a timestamp to the deployment name to ensure it is unique
+        deployment_name = f"AOSM_CLI_deployment_{current_time}"
+
+        # Validation is automatically re-attempted in live runs, but not in test
+        # playback, causing them to fail. This explicitly re-attempts validation to
+        # ensure the tests pass
+        validation_res = None
+        for validation_attempt in range(2):
+            try:
+                validation = (
+                    resource_client.deployments.begin_validate(
+                        resource_group_name=resource_group,
+                        deployment_name=deployment_name,
+                        parameters={
+                            "properties": {
+                                "mode": "Incremental",
+                                "template": template,
+                                "parameters": parameters,
+                            }
+                        },
+                    )
+                )
+                validation_res = LongRunningOperation(
+                    self.cli_ctx, "Validating ARM template..."
+                )(validation)
+                break
+            except Exception:  # pylint: disable=broad-except
+                if validation_attempt == 1:
+                    raise
+
+        if not validation_res or validation_res.error:
+            raise RuntimeError(f"Validation of template {template} failed.")
+
+        # Validation succeeded so proceed with deployment
+        poller = resource_client.deployments.begin_create_or_update(
+            resource_group_name=resource_group,
+            deployment_name=deployment_name,
+            parameters={
+                "properties": {
+                    "mode": "Incremental",
+                    "template": template,
+                    "parameters": parameters,
+                }
+            },
+        )
+
+        # Wait for the deployment to complete and get the outputs
+        deployment: DeploymentExtended = LongRunningOperation(
+            self.cli_ctx, "Deploying ARM template"
+        )(poller)
+
+        if deployment.properties is not None:
+            depl_props = deployment.properties
+        else:
+            raise RuntimeError("The deployment has no properties.\nAborting")
+
+        if depl_props.provisioning_state != "Succeeded":
+            raise RuntimeError(
+                "Deploy of template to resource group"
+                f" {resource_group} proceeded but the provisioning"
+                f" state returned is {depl_props.provisioning_state}."
+                "\nAborting"
+            )
+
+        return depl_props.outputs

src/aosm/azext_aosm/definition_folder/reader/definition_folder.py

@@ -49,13 +49,13 @@ def _parse_index_file(self, file_content: str) -> list[dict[str, any]]:
             })
         return parsed_elements
 
-    def deploy(self):
+    def deploy(self, resource_client):
         """Deploy the resources defined in the folder."""
         for element in self.elements:
-            element.deploy()
+            element.deploy(resource_client)
 
-    def delete(self, clean: bool = False):
+    def delete(self, resource_client, clean: bool = False):
         """Delete the definition folder."""
         for element in reversed(self.elements):
             if clean or not element.only_delete_on_clean:
-                element.delete()
+                element.delete(resource_client)