From 0497ec755b9e37f6084fd7af88362c9030409261 Mon Sep 17 00:00:00 2001 From: Cyclam <95434717+Cyclam@users.noreply.github.com> Date: Tue, 4 Jul 2023 16:45:51 +0100 Subject: [PATCH] Improve path handling, other small refactorings (#39) # Main changes ## Make methods relying on self._tmp_dir private - `self._tmp_dir` is only available in the context of calling `generate_nfd()`, so methods relying on `self._tmp_dir` should be private ## Use pathlib.Path rather than os file operations - Provides clearer and stronger typing than passing `str`s around - Adds some handy utility functions ## Variable renaming for clarity - E.g. consistently use 'directory' / 'dir' (rather than mix with 'folder') - Obvs somewhat subjective, but as someone new to most of this code, the changes made sense to me ## Add nfd_bicep_path as abstract property on NFDGenerator - We rely on it when calling the concrete implementations - Also use ABC rather than raise NotImplementedError ## Miscellaneous style updates to keep `azdev style aosm` happy - isort - black --- src/aosm/azext_aosm/_configuration.py | 20 +- src/aosm/azext_aosm/custom.py | 17 +- src/aosm/azext_aosm/deploy/artifact.py | 8 +- src/aosm/azext_aosm/deploy/deploy_with_arm.py | 4 +- .../generate_nfd/cnf_nfd_generator.py | 288 ++++++++---------- .../generate_nfd/nfd_generator_base.py | 15 +- .../generate_nfd/vnf_nfd_generator.py | 124 ++++---- .../azext_aosm/generate_nsd/nsd_generator.py | 2 +- src/aosm/azext_aosm/util/constants.py | 3 +- 9 files changed, 230 insertions(+), 251 deletions(-) diff --git a/src/aosm/azext_aosm/_configuration.py b/src/aosm/azext_aosm/_configuration.py index f933e3bd524..a36cca1bba5 100644 --- a/src/aosm/azext_aosm/_configuration.py +++ b/src/aosm/azext_aosm/_configuration.py @@ -13,7 +13,7 @@ NSD, NSD_OUTPUT_BICEP_PREFIX, VNF, - SOURCE_ACR_REGEX + SOURCE_ACR_REGEX, ) DESCRIPTION_MAP: Dict[str, str] = { @@ -229,7 +229,7 @@ def network_function_name(self) -> str: @property def acr_manifest_name(self) -> str: """Return the ACR manifest name from the NFD name.""" - sanitised_nf_name = self.network_function_name.lower().replace('_', '-') + sanitised_nf_name = self.network_function_name.lower().replace("_", "-") return ( f"{sanitised_nf_name}-nsd-acr-manifest-{self.nsd_version.replace('.', '-')}" ) @@ -329,10 +329,10 @@ def sa_manifest_name(self) -> str: return f"{sanitized_nf_name}-sa-manifest-{self.version.replace('.', '-')}" @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.""" - arm_template_path = self.arm_template.file_path - return f"{NF_DEFINITION_OUTPUT_BICEP_PREFIX}{Path(str(arm_template_path)).stem}" + arm_template_name = Path(self.arm_template.file_path).stem + return Path(f"{NF_DEFINITION_OUTPUT_BICEP_PREFIX}{arm_template_name}") @dataclass @@ -362,9 +362,9 @@ def __post_init__(self): self.helm_packages[package_index] = HelmPackageConfig(**dict(package)) @property - def build_output_folder_name(self) -> str: - """Return the local folder for generating the bicep template to.""" - return f"{NF_DEFINITION_OUTPUT_BICEP_PREFIX}{self.nf_name}" + def output_directory_for_build(self) -> Path: + """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 @@ -379,7 +379,9 @@ def validate(self): if not source_registry_match or len(source_registry_match.groups()) < 2: raise ValidationError( "CNF config has an invalid source registry ID. Please run `az aosm " - "nfd generate-config` to see the valid formats.") + "nfd generate-config` to see the valid formats." + ) + def get_configuration( configuration_type: str, config_as_dict: Optional[Dict[Any, Any]] = None diff --git a/src/aosm/azext_aosm/custom.py b/src/aosm/azext_aosm/custom.py index bbe7ed00abd..e69d35e8780 100644 --- a/src/aosm/azext_aosm/custom.py +++ b/src/aosm/azext_aosm/custom.py @@ -7,6 +7,7 @@ import os import shutil from dataclasses import asdict +from pathlib import Path from typing import Optional from azure.cli.core.azclierror import ( @@ -116,15 +117,15 @@ def _generate_nfd( "Generate NFD called for unrecognised definition_type. Only VNF and CNF" " have been implemented." ) - if nfd_generator.bicep_path: + if nfd_generator.nfd_bicep_path: carry_on = input( - f"The folder {os.path.dirname(nfd_generator.bicep_path)} already exists -" + f"The {nfd_generator.nfd_bicep_path.parent} directory already exists -" " delete it and continue? (y/n)" ) if carry_on != "y": - raise UnclassifiedUserFault("User aborted! ") + raise UnclassifiedUserFault("User aborted!") - shutil.rmtree(os.path.dirname(nfd_generator.bicep_path)) + shutil.rmtree(nfd_generator.nfd_bicep_path.parent) nfd_generator.generate_nfd() @@ -177,7 +178,7 @@ def publish_definition( parameters_json_file=parameters_json_file, manifest_bicep_path=manifest_file, manifest_parameters_json_file=manifest_parameters_json_file, - skip=skip + skip=skip, ) elif definition_type == CNF: deployer = DeployerViaArm(api_clients, config=config) @@ -187,7 +188,7 @@ def publish_definition( parameters_json_file=parameters_json_file, manifest_bicep_path=manifest_file, manifest_parameters_json_file=manifest_parameters_json_file, - skip=skip + skip=skip, ) else: raise ValueError( @@ -253,7 +254,7 @@ def _generate_config(configuration_type: str, output_file: str = "input.json"): config = get_configuration(configuration_type) config_as_dict = json.dumps(asdict(config), indent=4) - if os.path.exists(output_file): + if Path(output_file).exists(): carry_on = input( f"The file {output_file} already exists - do you want to overwrite it?" " (y/n)" @@ -369,7 +370,7 @@ def publish_design( parameters_json_file=parameters_json_file, manifest_bicep_path=manifest_file, manifest_parameters_json_file=manifest_parameters_json_file, - skip=skip + skip=skip, ) diff --git a/src/aosm/azext_aosm/deploy/artifact.py b/src/aosm/azext_aosm/deploy/artifact.py index 7e5d03bd532..614b38d5784 100644 --- a/src/aosm/azext_aosm/deploy/artifact.py +++ b/src/aosm/azext_aosm/deploy/artifact.py @@ -5,10 +5,12 @@ """A module to handle interacting with artifacts.""" import subprocess from dataclasses import dataclass -from typing import Union, List +from typing import List, Union from azure.cli.core.commands import LongRunningOperation -from azure.mgmt.containerregistry.models import ImportImageParameters, ImportSource +from azure.mgmt.containerregistry import ContainerRegistryManagementClient +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 @@ -157,7 +159,7 @@ def copy_image( :param source_image: source image :param target_registry_resource_group_name: target registry resource group name :param target_registry_name: target registry name - :param target_tags: the list of tags to be applied to the imported image + :param target_tags: the list of tags to be applied to the imported image should be of form: namepace/name:tag or name:tag :param mode: mode for import """ diff --git a/src/aosm/azext_aosm/deploy/deploy_with_arm.py b/src/aosm/azext_aosm/deploy/deploy_with_arm.py index e0251be0351..a0746a2045e 100644 --- a/src/aosm/azext_aosm/deploy/deploy_with_arm.py +++ b/src/aosm/azext_aosm/deploy/deploy_with_arm.py @@ -5,7 +5,6 @@ """Contains class for deploying generated definitions using ARM.""" import json import os -import re import shutil import subprocess # noqa import tempfile @@ -37,7 +36,6 @@ VNF, VNF_DEFINITION_BICEP_TEMPLATE_FILENAME, VNF_MANIFEST_BICEP_TEMPLATE_FILENAME, - SOURCE_ACR_REGEX, ) from azext_aosm.util.management_clients import ApiClients @@ -120,7 +118,7 @@ def deploy_vnfd_from_bicep( # User has not passed in a bicep template, so we are deploying the default # one produced from building the NFDV using this CLI bicep_path = os.path.join( - self.config.build_output_folder_name, + self.config.output_directory_for_build, VNF_DEFINITION_BICEP_TEMPLATE_FILENAME, ) diff --git a/src/aosm/azext_aosm/generate_nfd/cnf_nfd_generator.py b/src/aosm/azext_aosm/generate_nfd/cnf_nfd_generator.py index 9eef0c0db51..deec27f7306 100644 --- a/src/aosm/azext_aosm/generate_nfd/cnf_nfd_generator.py +++ b/src/aosm/azext_aosm/generate_nfd/cnf_nfd_generator.py @@ -4,11 +4,11 @@ # -------------------------------------------------------------------------------------- """Contains a class for generating CNF NFDs and associated resources.""" import json -import os import re import shutil import tarfile import tempfile +from pathlib import Path from typing import Any, Dict, Iterator, List, Optional, Tuple import yaml @@ -23,6 +23,7 @@ CNF_DEFINITION_JINJA2_SOURCE_TEMPLATE_FILENAME, CNF_MANIFEST_BICEP_TEMPLATE_FILENAME, CNF_MANIFEST_JINJA2_SOURCE_TEMPLATE_FILENAME, + CNF_VALUES_SCHEMA_FILENAME, CONFIG_MAPPINGS_DIR_NAME, DEPLOYMENT_PARAMETER_MAPPING_REGEX, DEPLOYMENT_PARAMETERS_FILENAME, @@ -59,38 +60,37 @@ def __init__(self, config: CNFConfiguration, interactive: bool = False): mapping file in config to be blank. """ self.config = config - self.nfd_jinja2_template_path = os.path.join( - os.path.dirname(__file__), - "templates", - CNF_DEFINITION_JINJA2_SOURCE_TEMPLATE_FILENAME, + self.nfd_jinja2_template_path = ( + Path(__file__).parent + / "templates" + / CNF_DEFINITION_JINJA2_SOURCE_TEMPLATE_FILENAME ) - self.manifest_jinja2_template_path = os.path.join( - os.path.dirname(__file__), - "templates", - CNF_MANIFEST_JINJA2_SOURCE_TEMPLATE_FILENAME, + self.manifest_jinja2_template_path = ( + Path(__file__).parent + / "templates" + / CNF_MANIFEST_JINJA2_SOURCE_TEMPLATE_FILENAME ) - self.output_folder_name = self.config.build_output_folder_name + self.output_directory: Path = self.config.output_directory_for_build + self._cnfd_bicep_path = ( + self.output_directory / CNF_DEFINITION_BICEP_TEMPLATE_FILENAME + ) + self._tmp_dir: Optional[Path] = None self.artifacts = [] self.nf_application_configurations = [] self.deployment_parameter_schema = SCHEMA_PREFIX - - self._bicep_path = os.path.join( - self.output_folder_name, CNF_DEFINITION_BICEP_TEMPLATE_FILENAME - ) self.interactive = interactive - self._tmp_folder_name = "" def generate_nfd(self) -> None: """Generate a CNF NFD which comprises a group, an Artifact Manifest and an NFDV.""" - # Create temporary folder. + # Create temporary directory. with tempfile.TemporaryDirectory() as tmpdirname: - self._tmp_folder_name = tmpdirname + self._tmp_dir = Path(tmpdirname) try: for helm_package in self.config.helm_packages: - # Unpack the chart into the tmp folder - self._extract_chart(helm_package.path_to_chart) + # Unpack the chart into the tmp directory + self._extract_chart(Path(helm_package.path_to_chart)) # TODO: Validate charts @@ -99,15 +99,15 @@ def generate_nfd(self) -> None: self._generate_chart_value_mappings(helm_package) # Get schema for each chart - # (extract mappings and take the schema bits we need from values.schema.json) + # (extract mappings and relevant parts of the schema) # + Add that schema to the big schema. self.deployment_parameter_schema["properties"].update( - self.get_chart_mapping_schema(helm_package) + self._get_chart_mapping_schema(helm_package) ) # Get all image line matches for files in the chart. # Do this here so we don't have to do it multiple times. - image_line_matches = self.find_pattern_matches_in_chart( + image_line_matches = self._find_pattern_matches_in_chart( helm_package, IMAGE_START_STRING ) # Creates a flattened list of image registry paths to prevent set error @@ -118,28 +118,28 @@ def generate_nfd(self) -> None: # Generate the NF application configuration for the chart # passed to jinja2 renderer to render bicep template self.nf_application_configurations.append( - self.generate_nf_application_config( + self._generate_nf_application_config( helm_package, image_registry_paths, - self.find_pattern_matches_in_chart( + self._find_pattern_matches_in_chart( helm_package, IMAGE_PULL_SECRETS_START_STRING ), ) ) # Workout the list of artifacts for the chart and # update the list for the NFD with any unique artifacts. - chart_artifacts = self.get_artifact_list( + chart_artifacts = self._get_artifact_list( helm_package, image_line_matches ) self.artifacts += [ a for a in chart_artifacts if a not in self.artifacts ] - self.write_nfd_bicep_file() - self.write_schema_to_file() - self.write_manifest_bicep_file() - self.copy_to_output_folder() + self._write_nfd_bicep_file() + self._write_schema_to_file() + self._write_manifest_bicep_file() + self._copy_to_output_directory() print( - f"Generated NFD bicep template created in {self.output_folder_name}" + f"Generated NFD bicep template created in {self.output_directory}" ) print( "Please review these templates." @@ -151,30 +151,29 @@ def generate_nfd(self) -> None: raise e @property - def bicep_path(self) -> Optional[str]: + def nfd_bicep_path(self) -> Optional[Path]: """Returns the path to the bicep file for the NFD if it has been created.""" - if os.path.exists(self._bicep_path): - return self._bicep_path - + if self._cnfd_bicep_path.exists(): + return self._cnfd_bicep_path return None - def _extract_chart(self, path: str) -> None: + def _extract_chart(self, path: Path) -> None: """ - Extract the chart into the tmp folder. + Extract the chart into the tmp directory. :param path: The path to helm package """ logger.debug("Extracting helm package %s", path) - (_, ext) = os.path.splitext(path) - if ext in (".gz", ".tgz"): + file_extension = path.suffix + if file_extension in (".gz", ".tgz"): with tarfile.open(path, "r:gz") as tar: - tar.extractall(path=self._tmp_folder_name) + tar.extractall(path=self._tmp_dir) - elif ext == ".tar": + elif file_extension == ".tar": with tarfile.open(path, "r:") as tar: - tar.extractall(path=self._tmp_folder_name) + tar.extractall(path=self._tmp_dir) else: raise InvalidTemplateError( @@ -202,12 +201,10 @@ def _generate_chart_value_mappings(self, helm_package: HelmPackageConfig) -> Non ) # Write the mapping to a file - folder_name = os.path.join(self._tmp_folder_name, GENERATED_VALUES_MAPPINGS_DIR_NAME) - os.makedirs(folder_name, exist_ok=True) - mapping_filepath = os.path.join( - self._tmp_folder_name, - GENERATED_VALUES_MAPPINGS_DIR_NAME, - f"{helm_package.name}-generated-mapping.yaml", + mapping_directory: Path = self._tmp_dir / GENERATED_VALUES_MAPPINGS_DIR_NAME + mapping_directory.mkdir(exist_ok=True) + mapping_filepath = ( + mapping_directory / f"{helm_package.name}-generated-mapping.yaml" ) with open(mapping_filepath, "w", encoding="UTF-8") as mapping_file: yaml.dump(mapping_to_write, mapping_file) @@ -227,13 +224,9 @@ def _read_top_level_values_yaml( :return: A dictionary of the yaml read from the file :rtype: Dict[str, Any] """ - for file in os.listdir(os.path.join(self._tmp_folder_name, helm_package.name)): - if file in ("values.yaml", "values.yml"): - with open( - os.path.join(self._tmp_folder_name, helm_package.name, file), - "r", - encoding="UTF-8", - ) as values_file: + for file in Path(self._tmp_dir / helm_package.name).iterdir(): + if file.name in ("values.yaml", "values.yml"): + with file.open(encoding="UTF-8") as values_file: values_yaml = yaml.safe_load(values_file) return values_yaml @@ -241,8 +234,8 @@ def _read_top_level_values_yaml( "Cannot find top level values.yaml/.yml file in Helm package." ) - def write_manifest_bicep_file(self) -> None: - """Write the bicep file for the Artifact Manifest.""" + def _write_manifest_bicep_file(self) -> None: + """Write the bicep file for the Artifact Manifest to the temp directory.""" with open(self.manifest_jinja2_template_path, "r", encoding="UTF-8") as f: template: Template = Template( f.read(), @@ -253,14 +246,14 @@ def write_manifest_bicep_file(self) -> None: artifacts=self.artifacts, ) - path = os.path.join(self._tmp_folder_name, CNF_MANIFEST_BICEP_TEMPLATE_FILENAME) + path = self._tmp_dir / CNF_MANIFEST_BICEP_TEMPLATE_FILENAME with open(path, "w", encoding="utf-8") as f: f.write(bicep_contents) logger.info("Created artifact manifest bicep template: %s", path) - def write_nfd_bicep_file(self) -> None: - """Write the bicep file for the NFD.""" + def _write_nfd_bicep_file(self) -> None: + """Write the bicep file for the NFD to the temp directory.""" with open(self.nfd_jinja2_template_path, "r", encoding="UTF-8") as f: template: Template = Template( f.read(), @@ -268,91 +261,76 @@ def write_nfd_bicep_file(self) -> None: ) bicep_contents: str = template.render( - deployParametersPath=os.path.join(SCHEMAS_DIR_NAME, DEPLOYMENT_PARAMETERS_FILENAME), + deployParametersPath=Path(SCHEMAS_DIR_NAME, DEPLOYMENT_PARAMETERS_FILENAME), nf_application_configurations=self.nf_application_configurations, ) - path = os.path.join(self._tmp_folder_name, CNF_DEFINITION_BICEP_TEMPLATE_FILENAME) + path = self._tmp_dir / CNF_DEFINITION_BICEP_TEMPLATE_FILENAME with open(path, "w", encoding="utf-8") as f: f.write(bicep_contents) logger.info("Created NFD bicep template: %s", path) - def write_schema_to_file(self) -> None: - """Write the schema to file deploymentParameters.json.""" + def _write_schema_to_file(self) -> None: + """Write the schema to file deploymentParameters.json to the temp directory.""" logger.debug("Create deploymentParameters.json") - full_schema = os.path.join(self._tmp_folder_name, DEPLOYMENT_PARAMETERS_FILENAME) + full_schema = self._tmp_dir / DEPLOYMENT_PARAMETERS_FILENAME with open(full_schema, "w", encoding="UTF-8") as f: json.dump(self.deployment_parameter_schema, f, indent=4) logger.debug("%s created", full_schema) - def copy_to_output_folder(self) -> None: - """Copy the config mappings, schema and bicep templates (artifact manifest and NFDV) to the output folder.""" - - logger.info("Create NFD bicep %s", self.output_folder_name) + def _copy_to_output_directory(self) -> None: + """Copy the config mappings, schema and bicep templates (artifact manifest and NFDV) from the temp directory to the output directory.""" - os.mkdir(self.output_folder_name) - os.mkdir(os.path.join(self.output_folder_name, SCHEMAS_DIR_NAME)) + logger.info("Create NFD bicep %s", self.output_directory) - # Copy the nfd and the manifest bicep files to the output folder - tmp_nfd_bicep_path = os.path.join( - self._tmp_folder_name, CNF_DEFINITION_BICEP_TEMPLATE_FILENAME + Path(self.output_directory / SCHEMAS_DIR_NAME).mkdir( + parents=True, exist_ok=True ) - shutil.copy(tmp_nfd_bicep_path, self.output_folder_name) - tmp_manifest_bicep_path = os.path.join( - self._tmp_folder_name, CNF_MANIFEST_BICEP_TEMPLATE_FILENAME + # Copy the nfd and the manifest bicep files to the output directory + shutil.copy( + self._tmp_dir / CNF_DEFINITION_BICEP_TEMPLATE_FILENAME, + self.output_directory, + ) + shutil.copy( + self._tmp_dir / CNF_MANIFEST_BICEP_TEMPLATE_FILENAME, self.output_directory ) - shutil.copy(tmp_manifest_bicep_path, self.output_folder_name) - # Copy any generated values mappings YAML files to the corresponding folder in + # Copy any generated values mappings YAML files to the corresponding directory in # the output directory so that the user can edit them and re-run the build if # required - if os.path.exists( - os.path.join(self._tmp_folder_name, GENERATED_VALUES_MAPPINGS_DIR_NAME) - ): - generated_mappings_path = os.path.join( - self.output_folder_name, GENERATED_VALUES_MAPPINGS_DIR_NAME - ) + if Path(self._tmp_dir / GENERATED_VALUES_MAPPINGS_DIR_NAME).exists(): shutil.copytree( - os.path.join(self._tmp_folder_name, GENERATED_VALUES_MAPPINGS_DIR_NAME), - generated_mappings_path, + self._tmp_dir / GENERATED_VALUES_MAPPINGS_DIR_NAME, + self.output_directory / GENERATED_VALUES_MAPPINGS_DIR_NAME, ) # Copy the JSON config mappings and deploymentParameters schema that are used - # for the NFD to the output folder - tmp_config_mappings_path = os.path.join(self._tmp_folder_name, CONFIG_MAPPINGS_DIR_NAME) - output_config_mappings_path = os.path.join( - self.output_folder_name, CONFIG_MAPPINGS_DIR_NAME - ) + # for the NFD to the output directory shutil.copytree( - tmp_config_mappings_path, - output_config_mappings_path, + self._tmp_dir / CONFIG_MAPPINGS_DIR_NAME, + self.output_directory / CONFIG_MAPPINGS_DIR_NAME, dirs_exist_ok=True, ) - - tmp_schema_path = os.path.join(self._tmp_folder_name, DEPLOYMENT_PARAMETERS_FILENAME) - output_schema_path = os.path.join( - self.output_folder_name, SCHEMAS_DIR_NAME, DEPLOYMENT_PARAMETERS_FILENAME - ) shutil.copy( - tmp_schema_path, - output_schema_path, + self._tmp_dir / DEPLOYMENT_PARAMETERS_FILENAME, + self.output_directory / SCHEMAS_DIR_NAME / DEPLOYMENT_PARAMETERS_FILENAME, ) - logger.info("Copied files to %s", self.output_folder_name) + logger.info("Copied files to %s", self.output_directory) - def generate_nf_application_config( + def _generate_nf_application_config( self, helm_package: HelmPackageConfig, image_registry_path: List[str], image_pull_secret_line_matches: List[Tuple[str, ...]], ) -> Dict[str, Any]: """Generate NF application config.""" - (name, version) = self.get_chart_name_and_version(helm_package) + (name, version) = self._get_chart_name_and_version(helm_package) registry_values_paths = set(image_registry_path) image_pull_secrets_values_paths = set(image_pull_secret_line_matches) @@ -364,22 +342,20 @@ def generate_nf_application_config( "dependsOnProfile": helm_package.depends_on, "registryValuesPaths": list(registry_values_paths), "imagePullSecretsValuesPaths": list(image_pull_secrets_values_paths), - "valueMappingsPath": self.jsonify_value_mappings(helm_package), + "valueMappingsPath": self._jsonify_value_mappings(helm_package), } @staticmethod - def _find_yaml_files(directory) -> Iterator[str]: + def _find_yaml_files(directory: Path) -> Iterator[str]: """ - Find all yaml files in given directory. + Find all yaml files recursively in given directory. :param directory: The directory to search. """ - for root, _, files in os.walk(directory): - for file in files: - if file.endswith(".yaml") or file.endswith(".yml"): - yield os.path.join(root, file) + yield from directory.glob("**/*.yaml") + yield from directory.glob("**/*.yml") - def find_pattern_matches_in_chart( + def _find_pattern_matches_in_chart( self, helm_package: HelmPackageConfig, start_string: str ) -> List[Tuple[str, ...]]: """ @@ -395,7 +371,7 @@ def find_pattern_matches_in_chart( paths and the name and version of the image. e.g. (Values.foo.bar.repoPath, foo, 1.2.3) """ - chart_dir = os.path.join(self._tmp_folder_name, helm_package.name) + chart_dir = self._tmp_dir / helm_package.name matches = [] path = [] @@ -413,20 +389,20 @@ def find_pattern_matches_in_chart( ) logger.debug( "Regex match for name and version is %s", - name_and_version - ) + name_and_version, + ) if name_and_version and len(name_and_version.groups()) == 2: logger.debug( "Found image name and version %s %s", - name_and_version.group('name'), - name_and_version.group('version') + name_and_version.group("name"), + name_and_version.group("version"), ) matches.append( ( path, - name_and_version.group('name'), - name_and_version.group('version'), + name_and_version.group("name"), + name_and_version.group("version"), ) ) else: @@ -435,7 +411,7 @@ def find_pattern_matches_in_chart( matches += path return matches - def get_artifact_list( + def _get_artifact_list( self, helm_package: HelmPackageConfig, image_line_matches: List[Tuple[str, ...]], @@ -447,7 +423,7 @@ def get_artifact_list( :param image_line_matches: The list of image line matches. """ artifact_list = [] - (chart_name, chart_version) = self.get_chart_name_and_version(helm_package) + (chart_name, chart_version) = self._get_chart_name_and_version(helm_package) helm_artifact = { "name": chart_name, "version": chart_version, @@ -463,12 +439,12 @@ def get_artifact_list( return artifact_list - def get_chart_mapping_schema( + def _get_chart_mapping_schema( self, helm_package: HelmPackageConfig ) -> Dict[Any, Any]: """ Get the schema for the non default values (those with {deploymentParameter...}). - Based on user provided values.schema.json. + Based on the user provided values schema. param helm_package: The helm package config. """ @@ -476,18 +452,16 @@ def get_chart_mapping_schema( logger.debug("Get chart mapping schema for %s", helm_package.name) mappings_path = helm_package.path_to_mappings - values_schema = os.path.join( - self._tmp_folder_name, helm_package.name, "values.schema.json" - ) - if not os.path.exists(mappings_path): + values_schema = self._tmp_dir / helm_package.name / CNF_VALUES_SCHEMA_FILENAME + if not Path(mappings_path).exists(): raise InvalidTemplateError( f"ERROR: The helm package '{helm_package.name}' does not have a valid values" " mappings file. The file at '{helm_package.path_to_mappings}' does not exist." "\nPlease fix this and run the command again." ) - if not os.path.exists(values_schema): + if not values_schema.exists(): raise InvalidTemplateError( - f"ERROR: The helm package '{helm_package.name}' is missing values.schema.json." + f"ERROR: The helm package '{helm_package.name}' is missing {CNF_VALUES_SCHEMA_FILENAME}." "\nPlease fix this and run the command again." ) @@ -515,14 +489,11 @@ def get_chart_mapping_schema( @staticmethod def traverse_dict( - dict_to_search: Dict[Any, Any], - target_regex: str - ) -> Dict[str, List[str]]: + dict_to_search: Dict[Any, Any], target_regex: str + ) -> Dict[str, List[str]]: """ - Traverse the dictionary that is loaded from the file provided by path_to_mappings in the input.json. - - Returns a dictionary of all the values that match the target regex, - with the key being the deploy parameter and the value being the path to the value. + Traverse the dictionary provided and return a dictionary of all the values that match the target regex, + with the key being the deploy parameter and the value being the path (as a list) to the value. e.g. {"foo": ["global", "foo", "bar"]} :param d: The dictionary to traverse. @@ -563,17 +534,16 @@ def traverse_dict( "at path %s, which this tool cannot parse. " "Please check the output configMappings and schemas " "files and check that they are as required.", - path + [k] + path + [k], ) return result @staticmethod def search_schema( - deployParams_paths: Dict[str, List[str]], - full_schema - ) -> Dict[str, Dict[str, str]]: + deployParams_paths: Dict[str, List[str]], full_schema + ) -> Dict[str, Dict[str, str]]: """ - Search through provided schema for the types of the deployment parameters. + Search through the provided schema for the types of the deployment parameters. This assumes that the type of the key will be the type of the deployment parameter. e.g. if foo: {deployParameter.bar} and foo is type string, then bar is type string. @@ -581,7 +551,7 @@ def search_schema( {"foo": {"type": "string"}, "bar": {"type": "string"}} param deployParams_paths: a dictionary of all the deploy parameters to search for, - with the key being the deploy parameter and the value being the + with the key being the deploy parameter and the value being the path to the value. e.g. {"foo": ["global", "foo", "bar"]} param full_schema: The schema to search through. @@ -591,13 +561,14 @@ def search_schema( for deploy_param, path_list in deployParams_paths.items(): logger.debug( "Searching for %s in schema at path %s", deploy_param, path_list - ) + ) node = full_schema for path in path_list: if "properties" in node.keys(): logger.debug( "Searching properties for %s in schema at path %s", - deploy_param, path + deploy_param, + path, ) node = node["properties"][path] else: @@ -613,7 +584,8 @@ def search_schema( logger.warning( "We default these parameters to type string. " "Please edit schemas/%s in the output before publishing " - "if this is wrong", DEPLOYMENT_PARAMETERS_FILENAME + "if this is wrong", + DEPLOYMENT_PARAMETERS_FILENAME, ) return new_schema @@ -693,19 +665,19 @@ def _replace_values_with_deploy_params( return final_values_mapping_dict - def get_chart_name_and_version( + def _get_chart_name_and_version( self, helm_package: HelmPackageConfig ) -> Tuple[str, str]: """Get the name and version of the chart.""" - chart = os.path.join(self._tmp_folder_name, helm_package.name, "Chart.yaml") + chart_path = self._tmp_dir / helm_package.name / "Chart.yaml" - if not os.path.exists(chart): + if not chart_path.exists(): raise InvalidTemplateError( f"There is no Chart.yaml file in the helm package '{helm_package.name}'. " "\nPlease fix this and run the command again." ) - with open(chart, "r", encoding="utf-8") as f: + with open(chart_path, "r", encoding="utf-8") as f: data = yaml.load(f, Loader=yaml.FullLoader) if "name" in data and "version" in data: chart_name = data["name"] @@ -719,23 +691,19 @@ def get_chart_name_and_version( return (chart_name, chart_version) - def jsonify_value_mappings(self, helm_package: HelmPackageConfig) -> str: + def _jsonify_value_mappings(self, helm_package: HelmPackageConfig) -> Path: """Yaml->JSON values mapping file, then return path to it.""" - mappings_yaml = helm_package.path_to_mappings - - mappings_folder_path = os.path.join(self._tmp_folder_name, CONFIG_MAPPINGS_DIR_NAME) - mappings_filename = f"{helm_package.name}-mappings.json" - - if not os.path.exists(mappings_folder_path): - os.mkdir(mappings_folder_path) + mappings_yaml_file = helm_package.path_to_mappings + mappings_dir = self._tmp_dir / CONFIG_MAPPINGS_DIR_NAME + mappings_output_file = mappings_dir / f"{helm_package.name}-mappings.json" - mapping_file_path = os.path.join(mappings_folder_path, mappings_filename) + mappings_dir.mkdir(exist_ok=True) - with open(mappings_yaml, "r", encoding="utf-8") as f: + with open(mappings_yaml_file, "r", encoding="utf-8") as f: data = yaml.load(f, Loader=yaml.FullLoader) - with open(mapping_file_path, "w", encoding="utf-8") as file: + with open(mappings_output_file, "w", encoding="utf-8") as file: json.dump(data, file, indent=4) logger.debug("Generated parameter mappings for %s", helm_package.name) - return os.path.join(CONFIG_MAPPINGS_DIR_NAME, mappings_filename) + return Path(CONFIG_MAPPINGS_DIR_NAME, f"{helm_package.name}-mappings.json") diff --git a/src/aosm/azext_aosm/generate_nfd/nfd_generator_base.py b/src/aosm/azext_aosm/generate_nfd/nfd_generator_base.py index a57604f3009..e155950c03d 100644 --- a/src/aosm/azext_aosm/generate_nfd/nfd_generator_base.py +++ b/src/aosm/azext_aosm/generate_nfd/nfd_generator_base.py @@ -3,7 +3,10 @@ # License. See License.txt in the project root for license information. # -------------------------------------------------------------------------------------- """Contains a base class for generating NFDs.""" -from abc import ABC +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Optional + from knack.log import get_logger logger = get_logger(__name__) @@ -12,7 +15,11 @@ class NFDGenerator(ABC): """A class for generating an NFD from a config file.""" - # pylint: disable=too-few-public-methods + @abstractmethod def generate_nfd(self) -> None: - """No-op on base class.""" - raise NotImplementedError + ... + + @property + @abstractmethod + def nfd_bicep_path(self) -> Optional[Path]: + ... diff --git a/src/aosm/azext_aosm/generate_nfd/vnf_nfd_generator.py b/src/aosm/azext_aosm/generate_nfd/vnf_nfd_generator.py index ec2d44a1214..98dd6a587e0 100644 --- a/src/aosm/azext_aosm/generate_nfd/vnf_nfd_generator.py +++ b/src/aosm/azext_aosm/generate_nfd/vnf_nfd_generator.py @@ -5,10 +5,10 @@ """Contains a class for generating VNF NFDs and associated resources.""" import json -import os import shutil import tempfile from functools import cached_property +from pathlib import Path from typing import Any, Dict, Optional from knack.log import get_logger @@ -62,21 +62,19 @@ class VnfNfdGenerator(NFDGenerator): def __init__(self, config: VNFConfiguration, order_params: bool, interactive: bool): self.config = config - self.bicep_template_name = VNF_DEFINITION_BICEP_TEMPLATE_FILENAME - self.manifest_template_name = VNF_MANIFEST_BICEP_TEMPLATE_FILENAME - self.arm_template_path = self.config.arm_template.file_path - self.output_folder_name = self.config.build_output_folder_name + self.arm_template_path = Path(self.config.arm_template.file_path) + self.output_directory: Path = self.config.output_directory_for_build - self._bicep_path = os.path.join( - self.output_folder_name, self.bicep_template_name + self._vnfd_bicep_path = Path( + self.output_directory, VNF_DEFINITION_BICEP_TEMPLATE_FILENAME ) - self._manifest_path = os.path.join( - self.output_folder_name, self.manifest_template_name + self._manifest_bicep_path = Path( + self.output_directory, VNF_MANIFEST_BICEP_TEMPLATE_FILENAME ) self.order_params = order_params self.interactive = interactive - self.tmp_folder_name = "" + self._tmp_dir: Optional[Path] = None self.image_name = f"{self.config.nf_name}Image" def generate_nfd(self) -> None: @@ -85,32 +83,30 @@ def generate_nfd(self) -> None: Create a bicep template for an NFD from the ARM template for the VNF. """ - # Create temporary folder. + # Create temporary directory. with tempfile.TemporaryDirectory() as tmpdirname: - self.tmp_folder_name = tmpdirname + self._tmp_dir = Path(tmpdirname) - self.create_parameter_files() - self.copy_to_output_folder() - print(f"Generated NFD bicep templates created in {self.output_folder_name}") + self._create_parameter_files() + self._copy_to_output_directory() + print(f"Generated NFD bicep templates created in {self.output_directory}") print( "Please review these templates. When you are happy with them run " "`az aosm nfd publish` with the same arguments." ) @property - def bicep_path(self) -> Optional[str]: + def nfd_bicep_path(self) -> Optional[Path]: """Returns the path to the bicep file for the NFD if it has been created.""" - if os.path.exists(self._bicep_path): - return self._bicep_path - + if self._vnfd_bicep_path.exists(): + return self._vnfd_bicep_path return None @property - def manifest_path(self) -> Optional[str]: + def manifest_bicep_path(self) -> Optional[str]: """Returns the path to the bicep file for the NFD if it has been created.""" - if os.path.exists(self._manifest_path): - return self._manifest_path - + if self._manifest_bicep_path.exists(): + return self._manifest_bicep_path return None @cached_property @@ -151,22 +147,22 @@ def vm_parameters_ordered(self) -> Dict[str, Any]: return {**vm_parameters_no_default, **vm_parameters_with_default} - def create_parameter_files(self) -> None: - """Create the Deployment and Template json parameter files.""" - schemas_folder_path = os.path.join(self.tmp_folder_name, SCHEMAS_DIR_NAME) - os.mkdir(schemas_folder_path) - self.write_deployment_parameters(schemas_folder_path) + def _create_parameter_files(self) -> None: + """Create the deployment, template and VHD parameter files.""" + tmp_schemas_directory: Path = self._tmp_dir / SCHEMAS_DIR_NAME + tmp_schemas_directory.mkdir() + self.write_deployment_parameters(tmp_schemas_directory) - mappings_folder_path = os.path.join(self.tmp_folder_name, CONFIG_MAPPINGS_DIR_NAME) - os.mkdir(mappings_folder_path) - self.write_template_parameters(mappings_folder_path) - self.write_vhd_parameters(mappings_folder_path) + tmp_mappings_directory: Path = self._tmp_dir / CONFIG_MAPPINGS_DIR_NAME + tmp_mappings_directory.mkdir() + self.write_template_parameters(tmp_mappings_directory) + self.write_vhd_parameters(tmp_mappings_directory) - def write_deployment_parameters(self, folder_path: str) -> None: + def write_deployment_parameters(self, directory: Path) -> None: """ - Write out the NFD deploymentParameters.json file. + Write out the NFD deploymentParameters.json file to `directory` - :param folder_path: The folder to put this file in. + :param directory: The directory to put this file in. """ logger.debug("Create deploymentParameters.json") @@ -180,10 +176,10 @@ def write_deployment_parameters(self, folder_path: str) -> None: for key in vm_parameters: if key == self.config.image_name_parameter: - # There is only one correct answer for the image name, so don't ask the + # There is only one correct answer for the image name, so don't ask the # user, instead it is hardcoded in config mappings. continue - + # Order parameters into those without and then with defaults has_default_field = "defaultValue" in self.vm_parameters[key] has_default = ( @@ -213,7 +209,7 @@ def write_deployment_parameters(self, folder_path: str) -> None: for key in vm_parameters_to_exclude: self.vm_parameters.pop(key, None) - deployment_parameters_path = os.path.join(folder_path, DEPLOYMENT_PARAMETERS_FILENAME) + deployment_parameters_path = directory / DEPLOYMENT_PARAMETERS_FILENAME # Heading for the deployParameters schema deploy_parameters_full: Dict[str, Any] = SCHEMA_PREFIX @@ -232,8 +228,8 @@ def write_deployment_parameters(self, folder_path: str) -> None: # Extra output file to help the user know which parameters are optional if not self.interactive: if nfd_parameters_with_default: - optional_deployment_parameters_path = os.path.join( - folder_path, OPTIONAL_DEPLOYMENT_PARAMETERS_FILENAME + optional_deployment_parameters_path = ( + directory / OPTIONAL_DEPLOYMENT_PARAMETERS_FILENAME ) with open( optional_deployment_parameters_path, "w", encoding="utf-8" @@ -245,12 +241,12 @@ def write_deployment_parameters(self, folder_path: str) -> None: f"{OPTIONAL_DEPLOYMENT_PARAMETERS_FILENAME} to help you choose which " "to expose." ) - - def write_template_parameters(self, folder_path: str) -> None: + + def write_template_parameters(self, directory: Path) -> None: """ - Write out the NFD templateParameters.json file. + Write out the NFD templateParameters.json file to `directory`. - :param folder_path: The folder to put this file in. + :param directory: The directory to put this file in. """ logger.debug("Create %s", TEMPLATE_PARAMETERS_FILENAME) vm_parameters = ( @@ -266,18 +262,18 @@ def write_template_parameters(self, folder_path: str) -> None: template_parameters[key] = f"{{deployParameters.{key}}}" - template_parameters_path = os.path.join(folder_path, TEMPLATE_PARAMETERS_FILENAME) + template_parameters_path = directory / TEMPLATE_PARAMETERS_FILENAME with open(template_parameters_path, "w", encoding="utf-8") as _file: _file.write(json.dumps(template_parameters, indent=4)) logger.debug("%s created", template_parameters_path) - def write_vhd_parameters(self, folder_path: str) -> None: + def write_vhd_parameters(self, directory: Path) -> None: """ - Write out the NFD vhdParameters.json file. + Write out the NFD vhdParameters.json file to `directory`. - :param folder_path: The folder to put this file in. + :param directory: The directory to put this file in. """ azureDeployLocation: str if self.vm_parameters.get("location"): @@ -293,29 +289,33 @@ def write_vhd_parameters(self, folder_path: str) -> None: "azureDeployLocation": azureDeployLocation, } - vhd_parameters_path = os.path.join(folder_path, VHD_PARAMETERS_FILENAME) + vhd_parameters_path = directory / VHD_PARAMETERS_FILENAME with open(vhd_parameters_path, "w", encoding="utf-8") as _file: _file.write(json.dumps(vhd_parameters, indent=4)) logger.debug("%s created", vhd_parameters_path) - def copy_to_output_folder(self) -> None: - """Copy the bicep templates, config mappings and schema into the build output folder.""" - code_dir = os.path.dirname(__file__) + def _copy_to_output_directory(self) -> None: + """Copy the static bicep templates and generated config mappings and schema into the build output directory.""" + logger.info("Create NFD bicep %s", self.output_directory) + Path(self.output_directory).mkdir(exist_ok=True) - logger.info("Create NFD bicep %s", self.output_folder_name) - os.mkdir(self.output_folder_name) + static_bicep_templates_dir = Path(__file__).parent / "templates" - bicep_path = os.path.join(code_dir, "templates", self.bicep_template_name) - shutil.copy(bicep_path, self.output_folder_name) + static_vnfd_bicep_path = ( + static_bicep_templates_dir / VNF_DEFINITION_BICEP_TEMPLATE_FILENAME + ) + shutil.copy(static_vnfd_bicep_path, self.output_directory) - manifest_path = os.path.join(code_dir, "templates", self.manifest_template_name) - shutil.copy(manifest_path, self.output_folder_name) - # Copy everything in the temp folder to the output folder + static_manifest_bicep_path = ( + static_bicep_templates_dir / VNF_MANIFEST_BICEP_TEMPLATE_FILENAME + ) + shutil.copy(static_manifest_bicep_path, self.output_directory) + # Copy everything in the temp directory to the output directory shutil.copytree( - self.tmp_folder_name, - self.output_folder_name, + self._tmp_dir, + self.output_directory, dirs_exist_ok=True, ) - logger.info("Copied files to %s", self.output_folder_name) + logger.info("Copied files to %s", self.output_directory) diff --git a/src/aosm/azext_aosm/generate_nsd/nsd_generator.py b/src/aosm/azext_aosm/generate_nsd/nsd_generator.py index 33dd6596bb0..836da37dc1d 100644 --- a/src/aosm/azext_aosm/generate_nsd/nsd_generator.py +++ b/src/aosm/azext_aosm/generate_nsd/nsd_generator.py @@ -298,7 +298,7 @@ def write_nsd_manifest(self) -> None: {}, ) - def generate_bicep(self, + def generate_bicep(self, template_name: str, output_file_name: str, params: Dict[Any,Any]) -> None: diff --git a/src/aosm/azext_aosm/util/constants.py b/src/aosm/azext_aosm/util/constants.py index cf76d00f895..d2f0f529f7c 100644 --- a/src/aosm/azext_aosm/util/constants.py +++ b/src/aosm/azext_aosm/util/constants.py @@ -35,6 +35,7 @@ CNF_MANIFEST_JINJA2_SOURCE_TEMPLATE_FILENAME = "cnfartifactmanifest.bicep.j2" CNF_DEFINITION_BICEP_TEMPLATE_FILENAME = "cnfdefinition.bicep" CNF_MANIFEST_BICEP_TEMPLATE_FILENAME = "cnfartifactmanifest.bicep" +CNF_VALUES_SCHEMA_FILENAME = "values.schema.json" # Names of directories used in the repo @@ -84,4 +85,4 @@ SOURCE_ACR_REGEX = ( r".*\/resourceGroups\/(?P[^\/]*)\/providers\/Microsoft." r"ContainerRegistry\/registries\/(?P[^\/]*)" - ) +)