Sunny/fix artifact upload windows (Azure#41)

* Fix Artifact upload on Windows

* mypy fixups

* mypy fixes

* linting

* mypy

* mypy for _configuration.py

* mypy for vnf_nfd_generator.py

* mypy appeasement

* python-static-checks fmt

* az style happy

* lint

* mypy cnf_nfd_generator

* copyright

* more lint

* Remove windows oras workaround now 0.0.18 oras out

* history

---------

Co-authored-by: Jamie Parsons <jamie.parsons@metaswitch.com>

src/aosm/HISTORY.rst

@@ -17,6 +17,7 @@ unreleased
 * Fix Manifest name for NSDs so it isn't the same as that for NFDs
 * Add validation of source_registry_id format for CNF configuration
 * Workaround Oras client bug (#90) on Windows for Artifact upload to ACR
+* Take Oras 0.1.18 so above Workaround could be removed
 
 0.2.0
 ++++++

src/aosm/azext_aosm/_configuration.py

@@ -1,3 +1,10 @@
+# --------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT
+# License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------
+"""Configuration class for input config file parsing,"""
+import abc
+import logging
 import json
 import os
 import re
@@ -6,7 +13,6 @@
 from typing import Any, Dict, List, Optional
 
 from azure.cli.core.azclierror import InvalidArgumentValueError, ValidationError
-
 from azext_aosm.util.constants import (
     CNF,
     NF_DEFINITION_OUTPUT_BICEP_PREFIX,
@@ -17,6 +23,8 @@
     SOURCE_ACR_REGEX,
 )
 
+logger = logging.getLogger(__name__)
+
 DESCRIPTION_MAP: Dict[str, str] = {
     "publisher_resource_group_name": (
         "Resource group for the Publisher resource. "
@@ -118,8 +126,14 @@ class ArtifactConfig:
 
 
 @dataclass
-class Configuration:
+class Configuration(abc.ABC):
     config_file: Optional[str] = None
+    publisher_name: str = DESCRIPTION_MAP["publisher_name"]
+    publisher_resource_group_name: str = DESCRIPTION_MAP[
+        "publisher_resource_group_name"
+    ]
+    acr_artifact_store_name: str = DESCRIPTION_MAP["acr_artifact_store_name"]
+    location: str = DESCRIPTION_MAP["location"]
 
     def path_from_cli_dir(self, path: str) -> str:
         """
@@ -131,6 +145,8 @@ def path_from_cli_dir(self, path: str) -> str:
 
         :param path: The path relative to the config file.
         """
+        assert self.config_file
+
         # If no path has been supplied we shouldn't try to update it.
         if path == "":
             return ""
@@ -139,11 +155,29 @@ def path_from_cli_dir(self, path: str) -> str:
         if os.path.isabs(path):
             return path
 
-        return os.path.join(os.path.dirname(self.config_file), path)
+        config_file_dir = Path(self.config_file).parent
+
+        updated_path = str(config_file_dir / path)
+
+        logger.debug("Updated path: %s", updated_path)
+
+        return updated_path
+
+    @property
+    def output_directory_for_build(self) -> Path:
+        """Base class method to ensure subclasses implement this function."""
+        raise NotImplementedError("Subclass must define property")
+
+    @property
+    def acr_manifest_name(self) -> str:
+        """Base class method to ensure subclasses implement this function."""
+        raise NotImplementedError("Subclass must define property")
 
 
 @dataclass
 class NFConfiguration(Configuration):
+    """Network Function configuration."""
+
     publisher_name: str = DESCRIPTION_MAP["publisher_name"]
     publisher_resource_group_name: str = DESCRIPTION_MAP[
         "publisher_resource_group_name"
@@ -237,10 +271,10 @@ def validate(self):
             raise ValueError("NSD Version must be set")
 
     @property
-    def build_output_folder_name(self) -> str:
+    def output_directory_for_build(self) -> Path:
         """Return the local folder for generating the bicep template to."""
         current_working_directory = os.getcwd()
-        return f"{current_working_directory}/{NSD_OUTPUT_BICEP_PREFIX}"
+        return Path(f"{current_working_directory}/{NSD_OUTPUT_BICEP_PREFIX}")
 
     @property
     def resource_element_name(self) -> str:
@@ -276,7 +310,7 @@ def arm_template(self) -> ArtifactConfig:
         artifact = ArtifactConfig()
         artifact.version = self.nsd_version
         artifact.file_path = os.path.join(
-            self.build_output_folder_name, NF_DEFINITION_JSON_FILENAME
+            self.output_directory_for_build, NF_DEFINITION_JSON_FILENAME
         )
         return artifact
 
@@ -389,19 +423,22 @@ def __post_init__(self):
         """
         for package_index, package in enumerate(self.helm_packages):
             if isinstance(package, dict):
-                package["path_to_chart"] = self.path_from_cli_dir(package["path_to_chart"])
+                package["path_to_chart"] = self.path_from_cli_dir(
+                    package["path_to_chart"]
+                )
                 package["path_to_mappings"] = self.path_from_cli_dir(
                     package["path_to_mappings"]
                 )
                 self.helm_packages[package_index] = HelmPackageConfig(**dict(package))
 
     @property
     def output_directory_for_build(self) -> Path:
-        """Return the directory the build command will writes its output to"""
+        """Return the directory the build command will writes its output to."""
         return Path(f"{NF_DEFINITION_OUTPUT_BICEP_PREFIX}{self.nf_name}")
 
     def validate(self):
-        """Validate the CNF config
+        """
+        Validate the CNF config.
 
         :raises ValidationError: If source registry ID doesn't match the regex
         """
@@ -418,8 +455,8 @@ def validate(self):
 
 
 def get_configuration(
-    configuration_type: str, config_file: str = None
-) -> NFConfiguration or NSConfiguration:
+    configuration_type: str, config_file: Optional[str] = None
+) -> Configuration:
     """
     Return the correct configuration object based on the type.
 
@@ -433,6 +470,8 @@ def get_configuration(
     else:
         config_as_dict = {}
 
+    config: Configuration
+
     if configuration_type == VNF:
         config = VNFConfiguration(config_file=config_file, **config_as_dict)
     elif configuration_type == CNF:

src/aosm/azext_aosm/custom.py

@@ -20,6 +20,7 @@
 from azext_aosm._client_factory import cf_acr_registries, cf_resources
 from azext_aosm._configuration import (
     CNFConfiguration,
+    Configuration,
     NFConfiguration,
     NSConfiguration,
     VNFConfiguration,
@@ -57,6 +58,7 @@ def build_definition(
     config = _get_config_from_file(
         config_file=config_file, configuration_type=definition_type
     )
+    assert isinstance(config, NFConfiguration)
 
     # Generate the NFD and the artifact manifest.
     _generate_nfd(
@@ -78,9 +80,7 @@ def generate_definition_config(definition_type: str, output_file: str = "input.j
     _generate_config(configuration_type=definition_type, output_file=output_file)
 
 
-def _get_config_from_file(
-    config_file: str, configuration_type: str
-) -> NFConfiguration or NSConfiguration:
+def _get_config_from_file(config_file: str, configuration_type: str) -> Configuration:
     """
     Read input config file JSON and turn it into a Configuration object.
 
@@ -363,7 +363,7 @@ def publish_design(
     )
 
     config = _get_config_from_file(config_file=config_file, configuration_type=NSD)
-
+    assert isinstance(config, NSConfiguration)
     config.validate()
 
     deployer = DeployerViaArm(api_clients, config=config)
@@ -379,14 +379,14 @@ def publish_design(
 
 def _generate_nsd(config: NSConfiguration, api_clients: ApiClients):
     """Generate a Network Service Design for the given config."""
-    if os.path.exists(config.build_output_folder_name):
+    if os.path.exists(config.output_directory_for_build):
         carry_on = input(
-            f"The folder {config.build_output_folder_name} already exists - delete it"
+            f"The folder {config.output_directory_for_build} already exists - delete it"
             " and continue? (y/n)"
         )
         if carry_on != "y":
             raise UnclassifiedUserFault("User aborted! ")
 
-        shutil.rmtree(config.build_output_folder_name)
+        shutil.rmtree(config.output_directory_for_build)
     nsd_generator = NSDGenerator(api_clients, config)
     nsd_generator.generate_nsd()

src/aosm/azext_aosm/delete/delete.py

@@ -5,7 +5,12 @@
 """Contains class for deploying generated definitions using the Python SDK."""
 from knack.log import get_logger
 
-from azext_aosm._configuration import NFConfiguration, NSConfiguration, VNFConfiguration
+from azext_aosm._configuration import (
+    Configuration,
+    NFConfiguration,
+    NSConfiguration,
+    VNFConfiguration,
+)
 from azext_aosm.util.management_clients import ApiClients
 from azext_aosm.util.utils import input_ack
 
@@ -16,7 +21,7 @@ class ResourceDeleter:
     def __init__(
         self,
         api_clients: ApiClients,
-        config: NFConfiguration or NSConfiguration,
+        config: Configuration,
     ) -> None:
         """
         Initializes a new instance of the Deployer class.
@@ -32,12 +37,12 @@ def __init__(
 
     def delete_nfd(self, clean: bool = False):
         """
-        Delete the NFDV and manifests.  If they don't exist it still reports them as
-        deleted.
+        Delete the NFDV and manifests.  If they don't exist it still reports them as deleted.
 
         :param clean: Delete the NFDG, artifact stores and publisher too. Defaults to False.
         Use with care.
         """
+        assert isinstance(self.config, NFConfiguration)
 
         if clean:
             print(
@@ -105,6 +110,7 @@ def delete_nsd(self):
         self.delete_config_group_schema()
 
     def delete_nfdv(self):
+        assert isinstance(self.config, NFConfiguration)
         message = (
             f"Delete NFDV {self.config.version} from group {self.config.nfdg_name} and"
             f" publisher {self.config.publisher_name}"
@@ -199,6 +205,7 @@ def delete_artifact_manifest(self, store_type: str) -> None:
 
     def delete_nsdg(self) -> None:
         """Delete the NSDG."""
+        assert isinstance(self.config, NSConfiguration)
         message = f"Delete NSD Group {self.config.nsdg_name}"
         logger.debug(message)
         print(message)
@@ -218,6 +225,7 @@ def delete_nsdg(self) -> None:
 
     def delete_nfdg(self) -> None:
         """Delete the NFDG."""
+        assert isinstance(self.config, NFConfiguration)
         message = f"Delete NFD Group {self.config.nfdg_name}"
         logger.debug(message)
         print(message)
@@ -287,6 +295,7 @@ def delete_publisher(self) -> None:
 
     def delete_config_group_schema(self) -> None:
         """Delete the Configuration Group Schema."""
+        assert isinstance(self.config, NSConfiguration)
         message = f"Delete Configuration Group Schema {self.config.cg_schema_name}"
         logger.debug(message)
         print(message)

src/aosm/azext_aosm/deploy/artifact.py

@@ -3,21 +3,17 @@
 
 # pylint: disable=unidiomatic-typecheck
 """A module to handle interacting with artifacts."""
-import os
 import subprocess
 from dataclasses import dataclass
 from typing import List, Union
 
 from azure.cli.core.commands import LongRunningOperation
 from azure.mgmt.containerregistry import ContainerRegistryManagementClient
-from azure.mgmt.containerregistry.models import (ImportImageParameters,
-                                                 ImportSource)
+from azure.mgmt.containerregistry.models import ImportImageParameters, ImportSource
 from azure.storage.blob import BlobClient, BlobType
 from knack.log import get_logger
 from knack.util import CLIError
 from oras.client import OrasClient
-from azure.cli.core.commands import LongRunningOperation
-from azure.mgmt.containerregistry import ContainerRegistryManagementClient
 
 from azext_aosm._configuration import ArtifactConfig, HelmPackageConfig
 
@@ -33,7 +29,7 @@ class Artifact:
     artifact_version: str
     artifact_client: Union[BlobClient, OrasClient]
 
-    def upload(self, artifact_config: ArtifactConfig or HelmPackageConfig) -> None:
+    def upload(self, artifact_config: Union[ArtifactConfig, HelmPackageConfig]) -> None:
         """
         Upload aritfact.
 
@@ -47,6 +43,7 @@ def upload(self, artifact_config: ArtifactConfig or HelmPackageConfig) -> None:
             else:
                 raise ValueError(f"Unsupported artifact type: {type(artifact_config)}.")
         else:
+            assert isinstance(artifact_config, ArtifactConfig)
             self._upload_to_storage_account(artifact_config)
 
     def _upload_arm_to_acr(self, artifact_config: ArtifactConfig) -> None:
@@ -58,30 +55,17 @@ def _upload_arm_to_acr(self, artifact_config: ArtifactConfig) -> None:
         assert type(self.artifact_client) == OrasClient
 
         if artifact_config.file_path:
-            try:
-                # OrasClient 0.1.17 has a bug
-                # https://github.com/oras-project/oras-py/issues/90 which means on
-                # Windows we need a real blank file on disk, without a colon in the
-                # filepath (so tempfile can't be used and we just put it in the working
-                # directory), that can act as the manifest config file. So create one
-                # and then delete it after the upload.
-                with open("dummyManifestConfig.json", "w", encoding='utf-8') as f:
-                    target = (
-                        f"{self.artifact_client.remote.hostname.replace('https://', '')}"
-                        f"/{self.artifact_name}:{self.artifact_version}"
-                    )
-                    logger.debug("Uploading %s to %s", artifact_config.file_path, target)
-                    self.artifact_client.push(
-                        files=[artifact_config.file_path],
-                        target=target,
-                        manifest_config=f.name,
-                    )
-            finally:
-                # Delete the dummy file
-                try:
-                    os.remove("dummyManifestConfig.json")
-                except FileNotFoundError:
-                    pass
+            if not self.artifact_client.remote.hostname:
+                raise ValueError(
+                    "Cannot upload ARM template as OrasClient has no remote hostname."
+                    " Please check your ACR config."
+                )
+            target = (
+                f"{self.artifact_client.remote.hostname.replace('https://', '')}"
+                f"/{self.artifact_name}:{self.artifact_version}"
+            )
+            logger.debug("Uploading %s to %s", artifact_config.file_path, target)
+            self.artifact_client.push(files=[artifact_config.file_path], target=target)
         else:
             raise NotImplementedError(
                 "Copying artifacts is not implemented for ACR artifacts stores."
@@ -93,7 +77,12 @@ def _upload_helm_to_acr(self, artifact_config: HelmPackageConfig) -> None:
 
         :param artifact_config: configuration for the artifact being uploaded
         """
+        assert isinstance(self.artifact_client, OrasClient)
         chart_path = artifact_config.path_to_chart
+        if not self.artifact_client.remote.hostname:
+            raise ValueError(
+                "Cannot upload artifact. Oras client has no remote hostname."
+            )
         registry = self.artifact_client.remote.hostname.replace("https://", "")
         target_registry = f"oci://{registry}"
         registry_name = registry.replace(".azurecr.io", "")
@@ -137,6 +126,8 @@ def _upload_to_storage_account(self, artifact_config: ArtifactConfig) -> None:
                 self.artifact_client.account_name,
             )
         else:
+            # Config Validation will raise error if not true
+            assert artifact_config.blob_sas_url
             logger.info("Copy from SAS URL to blob store")
             source_blob = BlobClient.from_blob_url(artifact_config.blob_sas_url)