Pk5/add image parsing

src/aosm/azext_aosm/build_processors/helm_chart_processor.py

@@ -10,16 +10,26 @@
 from knack.log import get_logger
 
 from azext_aosm.build_processors.base_processor import BaseInputProcessor
-from azext_aosm.common.artifact import (BaseArtifact, LocalFileACRArtifact,
-                                        RemoteACRArtifact)
+from azext_aosm.common.artifact import (
+    BaseArtifact,
+    LocalFileACRArtifact,
+    RemoteACRArtifact,
+)
 from azext_aosm.common.local_file_builder import LocalFileBuilder
 from azext_aosm.inputs.helm_chart_input import HelmChartInput
 from azext_aosm.vendored_sdks.models import (
-    ApplicationEnablement, ArtifactType, AzureArcKubernetesArtifactProfile,
+    ApplicationEnablement,
+    ArtifactType,
+    AzureArcKubernetesArtifactProfile,
     AzureArcKubernetesDeployMappingRuleProfile,
-    AzureArcKubernetesHelmApplication, DependsOnProfile, HelmArtifactProfile,
-    HelmMappingRuleProfile, ManifestArtifactFormat, ReferencedResource,
-    ResourceElementTemplate)
+    AzureArcKubernetesHelmApplication,
+    DependsOnProfile,
+    HelmArtifactProfile,
+    HelmMappingRuleProfile,
+    ManifestArtifactFormat,
+    ReferencedResource,
+    ResourceElementTemplate,
+)
 
 logger = get_logger(__name__)
 
@@ -150,24 +160,25 @@ def generate_nf_application(self) -> AzureArcKubernetesHelmApplication:
         return AzureArcKubernetesHelmApplication(
             name=self.name,
             # Current implementation is set all depends on profiles to empty lists
-            depends_on_profile=DependsOnProfile(install_depends_on=[],
-                                                uninstall_depends_on=[], update_depends_on=[]),
+            depends_on_profile=DependsOnProfile(
+                install_depends_on=[], uninstall_depends_on=[], update_depends_on=[]
+            ),
             artifact_profile=artifact_profile,
             deploy_parameters_mapping_rule_profile=mapping_rule_profile,
         )
 
-    def _find_chart_images(self) -> List[Tuple[str, str]]:
+    def _find_chart_images(self) -> Set[Tuple[str, str]]:
         """
         Find the images used by the Helm chart.
 
         :return: A list of tuples containing the image name and version.
-        :rtype: List[Tuple[str, str]]
+        :rtype: Set[Tuple[str, str]]
         """
         logger.debug("Finding images used by Helm chart %s", self.name)
         image_lines: Set[str] = set()
         self._find_image_lines(self.input_artifact, image_lines)
 
-        images: List[Tuple[str, str]] = []
+        images: Set[Tuple[str, str]] = set()
         for line in image_lines:
             name_and_tag = re.search(IMAGE_NAME_AND_VERSION_REGEX, line)
             if name_and_tag and len(name_and_tag.groups()) == 2:
@@ -179,7 +190,7 @@ def _find_chart_images(self) -> List[Tuple[str, str]]:
                     image_tag,
                     self.name,
                 )
-                images.append((image_name, image_tag))
+                images.add((image_name, image_tag))
             else:
                 logger.warning(
                     "Could not parse image name and tag in line %s in Helm chart %s",
@@ -198,19 +209,20 @@ def _find_image_lines(self, chart: HelmChartInput, image_lines: Set[str]) -> Non
         """
         logger.debug("Finding image lines in Helm chart %s", chart.artifact_name)
         # Find the image lines in the current chart
-        for template in chart.get_templates():
-            for line in template.data:
-                if "image:" in line:
-                    logger.debug(
-                        "Found image line %s in Helm chart %s",
-                        line,
-                        chart.artifact_name,
-                    )
-                    image_lines.add(line.replace("image:", "").strip())
 
-        # Recursively search the dependency charts for image lines
-        for dep in chart.get_dependencies():
-            self._find_image_lines(dep, image_lines)
+        template_lines = []
+
+        if chart.helm_template is not None:
+            template_lines = chart.helm_template.split("\n")
+
+        for line in template_lines:
+            if "image:" in line:
+                logger.debug(
+                    "Found image line %s in Helm chart %s",
+                    line,
+                    chart.artifact_name,
+                )
+                image_lines.add(line.replace("image:", "").strip())
 
     def _generate_artifact_profile(self) -> AzureArcKubernetesArtifactProfile:
         """

src/aosm/azext_aosm/cli_handlers/onboarding_cnf_handler.py

@@ -115,7 +115,7 @@ def _get_processor_list(self) -> [HelmChartProcessor]:
 
     def _validate_helm_template(self):
         """Validate the helm packages."""
-        helm_chart_processors = self._get_processor_list()
+        helm_chart_processors = self.processors
 
         validation_errors = {}
 

src/aosm/azext_aosm/common/artifact.py

@@ -159,16 +159,18 @@ def upload(self, config: BaseCommonParametersConfig, command_context: CommandCon
         )
         logger.debug("Uploading %s to %s", self.file_path, target)
         retries = 0
-        while True:
+        while retries < 20:
             try:
                 oras_client.push(files=[self.file_path], target=target)
                 break
             except ValueError:
-                if retries < 20:
-                    logger.info("Retrying pushing local artifact to ACR. Retries so far: %s", retries)
-                    retries += 1
-                    sleep(3)
-                    continue
+                logger.info(
+                    "Retrying pushing local artifact to ACR. Retries so far: %s",
+                    retries,
+                )
+                retries += 1
+                sleep(3)
+                continue
         logger.info("LocalFileACRArtifact uploaded %s to %s", self.file_path, target)
 
 
@@ -543,10 +545,9 @@ def upload(self, config: VNFCommonParametersConfig, command_context: CommandCont
                 blob_type=BlobType.PAGEBLOB,
                 progress_hook=self._vhd_upload_progress_callback,
             )
+
         logger.info(
-            "Successfully uploaded %s to %s",
-            artifact_config.file_path,
-            self.artifact_client.account_name,
+            "Successfully uploaded %s to %s", self.file_path, blob_client.container_name
         )
 
     def _vhd_upload_progress_callback(

src/aosm/azext_aosm/inputs/helm_chart_input.py

@@ -104,7 +104,7 @@ def __init__(
     @staticmethod
     def from_chart_path(
         chart_path: Path,
-        default_config: Optional[Dict[str, Any]],
+        default_config: Optional[Dict[str, Any]] = None,
         default_config_path: Optional[str] = None,
     ) -> "HelmChartInput":
         """
@@ -173,7 +173,7 @@ def validate_template(self) -> None:
         try:
             result = subprocess.run(cmd, capture_output=True, check=True)
             helm_template_output = result.stdout
-            self.helm_template = helm_template_output
+            self.helm_template = helm_template_output.decode()
 
             logger.debug(
                 "Helm template output for Helm chart %s:\n%s",

src/aosm/azext_aosm/tests/latest/input_file_templates/input-nf-agent-cnf-template-invalid-chart.jsonc

@@ -0,0 +1,19 @@
+{
+  "location": "uksouth",
+  "publisher_name": "sunnyclipub",
+  "publisher_resource_group_name": "sunny-uksouth",
+  "nf_name": "nf-agent-cnf",
+  "version": "0.1.0",
+  "acr_artifact_store_name": "sunny-nfagent-acr-2",
+  "images": {
+    "source_registry": "--this was copied here and renamed from https://ms.portal.azure.com/#@microsoft.onmicrosoft.com/resource/subscriptions/4a0479c0-b795-4d0f-96fd-c7edd2a2928f/resourceGroups/pez-nfagent-pipelines/providers/Microsoft.ContainerRegistry/registries/peznfagenttemp/overview new one was /subscriptions/c7bd9d96-70dd-4f61-af56-6e0abd8d80b5/resourceGroups/sunny-nfagent-acr-HostedResources-4CDE264A/providers/Microsoft.ContainerRegistry/registries/SunnyclipubSunnyNfagentAcre00abc1832"
+  },
+  "helm_packages": [
+    {
+      "name": "nf-agent-cnf",
+      "path_to_chart": "{{tests_directory}}/latest/mock_cnf/helm-charts/nf-agent-cnf-invalid",
+      "default_values": "",
+      "depends_on": []
+    }
+  ]
+}

src/aosm/azext_aosm/tests/latest/input_file_templates/input-nf-agent-cnf-template.jsonc

@@ -0,0 +1,19 @@
+{
+  "location": "uksouth",
+  "publisher_name": "sunnyclipub",
+  "publisher_resource_group_name": "sunny-uksouth",
+  "nf_name": "nf-agent-cnf",
+  "version": "0.1.0",
+  "acr_artifact_store_name": "sunny-nfagent-acr-2",
+  "images": {
+    "source_registry": "--this was copied here and renamed from https://ms.portal.azure.com/#@microsoft.onmicrosoft.com/resource/subscriptions/4a0479c0-b795-4d0f-96fd-c7edd2a2928f/resourceGroups/pez-nfagent-pipelines/providers/Microsoft.ContainerRegistry/registries/peznfagenttemp/overview new one was /subscriptions/c7bd9d96-70dd-4f61-af56-6e0abd8d80b5/resourceGroups/sunny-nfagent-acr-HostedResources-4CDE264A/providers/Microsoft.ContainerRegistry/registries/SunnyclipubSunnyNfagentAcre00abc1832"
+  },
+  "helm_packages": [
+    {
+      "name": "nf-agent-cnf",
+      "path_to_chart": "{{tests_directory}}/latest/mock_cnf/helm-charts/nf-agent-cnf",
+      "default_values": "",
+      "depends_on": []
+    }
+  ]
+}

src/aosm/azext_aosm/tests/latest/tests_utils.py

@@ -0,0 +1,42 @@
+# # --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# # --------------------------------------------------------------------------------------------
+
+import os
+from typing import Dict
+
+from jinja2 import Template
+
+
+def get_tests_path():
+    code_directory = os.path.dirname(__file__)
+    tests_directory = os.path.join(code_directory, "../")
+    return tests_directory
+
+
+def update_input_file(input_template_name, output_file_name, params: Dict[str, str]):
+    tests_directory = get_tests_path()
+    templates_directory = os.path.join(
+        tests_directory, "latest", "input_file_templates"
+    )
+    output_files_directory = os.path.join(
+        tests_directory, "latest", "autogenerated_test_files"
+    )
+    os.makedirs(output_files_directory, exist_ok=True)
+
+    input_template_path = os.path.join(templates_directory, input_template_name)
+
+    with open(input_template_path, "r", encoding="utf-8") as file:
+        contents = file.read()
+
+    jinja_template = Template(contents)
+
+    rendered_template = jinja_template.render(**params)
+
+    output_path = os.path.join(output_files_directory, output_file_name)
+
+    with open(output_path, "w", encoding="utf-8") as file:
+        file.write(rendered_template)
+
+    return output_path

src/aosm/azext_aosm/tests/latest/unit_test/test_helm_chart_processor.py

@@ -0,0 +1,35 @@
+from unittest import TestCase
+from azext_aosm.tests.latest.tests_utils import update_input_file, get_tests_path
+from azext_aosm.cli_handlers.onboarding_cnf_handler import (
+    OnboardingCNFCLIHandler,
+)
+from pathlib import Path
+
+CNF_NF_AGENT_INPUT_TEMPLATE_NAME = "input-nf-agent-cnf-template.jsonc"
+CNF_NF_AGENT_INPUT_FILE_NAME = "test_helm_chart_processor_input-nf-agent-cnf.jsonc"
+
+
+class TestHelmChartProcessor(TestCase):
+    def setUp(self):
+        config_file = update_input_file(
+            CNF_NF_AGENT_INPUT_TEMPLATE_NAME,
+            CNF_NF_AGENT_INPUT_FILE_NAME,
+            params={
+                "tests_directory": get_tests_path(),
+            },
+        )
+
+        handler = OnboardingCNFCLIHandler(Path(config_file))
+        # We want to test a specific private method so disable the pylint warning
+        # pylint: disable=protected-access
+        handler._validate_helm_template()
+        self.helm_chart_processor = handler.processors[0]
+
+    def test_find_chart_images(self):
+        # We want to test a specific private method so disable the pylint warning
+        # pylint: disable=protected-access
+        collected_images = self.helm_chart_processor._find_chart_images()
+
+        # Assert the expected images are returned
+        expected_images = {("pez-nfagent", "879624")}
+        self.assertEqual(collected_images, expected_images)

src/aosm/azext_aosm/tests/latest/unit_test/test_pre_validate_build.py

@@ -3,58 +3,55 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 from unittest import TestCase
-import os
+from pathlib import Path
 
 from azure.cli.core.azclierror import ValidationError
 
 from azext_aosm.cli_handlers.onboarding_cnf_handler import (
     OnboardingCNFCLIHandler,
 )
-from azext_aosm.configuration_models.onboarding_cnf_input_config import (
-    HelmPackageConfig,
-)
+from azext_aosm.tests.latest.tests_utils import update_input_file, get_tests_path
 
-current_file_path = os.path.abspath(__file__)
-current_directory = os.path.dirname(current_file_path)
-test_charts_directory = os.path.join(current_directory, "../mock_cnf/helm-charts")
+CNF_NF_AGENT_INPUT_TEMPLATE_NAME = "input-nf-agent-cnf-template.jsonc"
+CNF_NF_AGENT_INPUT_FILE_NAME = "test_pre_validate_build_input-nf-agent-cnf.jsonc"
+CNF_NF_AGENT_INVALID_INPUT_TEMPLATE_NAME = (
+    "input-nf-agent-cnf-template-invalid-chart.jsonc"
+)
+CNF_NF_AGENT_INVALID_INPUT_FILE_NAME = (
+    "test_pre_validate_build_input-nf-agent-cnf-invalid.jsonc"
+)
 
 
 class TestOnboardingCNFCLIHandler(TestCase):
     """Test the OnboardingCNFCLIHandler class."""
 
     def test_validate_helm_template_valid_chart(self):
         """Test validating a valid Helm chart using helm template."""
-
-        handler = OnboardingCNFCLIHandler()
-        handler.config.helm_packages = [
-            HelmPackageConfig(
-                name="nf-agent-cnf-UNIT-TEST",
-                path_to_chart=os.path.join(test_charts_directory, "nf-agent-cnf"),
-                default_values="",
-                depends_on=[],
-            )
-        ]
-        handler.config.images_source_registry = ""
-        handler.config.source_registry_username = ""
+        config_file = update_input_file(
+            CNF_NF_AGENT_INPUT_TEMPLATE_NAME,
+            CNF_NF_AGENT_INPUT_FILE_NAME,
+            params={
+                "tests_directory": get_tests_path(),
+            },
+        )
+
+        handler = OnboardingCNFCLIHandler(Path(config_file))
         # We want to test a specific private method so disable the pylint warning
         # pylint: disable=protected-access
         handler._validate_helm_template()
 
     def test_validate_helm_template_invalid_chart(self):
         """Test validating an invalid Helm chart using helm template."""
-        handler = OnboardingCNFCLIHandler()
-        handler.config.helm_packages = [
-            HelmPackageConfig(
-                name="nf-agent-cnf-invalid-UNIT-TEST",
-                path_to_chart=os.path.join(
-                    test_charts_directory, "nf-agent-cnf-invalid"
-                ),
-                default_values="",
-                depends_on=[],
-            )
-        ]
-        handler.config.images_source_registry = ""
-        handler.config.source_registry_username = ""
+        config_file = update_input_file(
+            CNF_NF_AGENT_INVALID_INPUT_TEMPLATE_NAME,
+            CNF_NF_AGENT_INVALID_INPUT_FILE_NAME,
+            params={
+                "tests_directory": get_tests_path(),
+            },
+        )
+
+        handler = OnboardingCNFCLIHandler(Path(config_file))
+
         with self.assertRaises(ValidationError):
             # We want to test a specific private method so disable the pylint warning
             # pylint: disable=protected-access