diff --git a/config/docker/files/default.json5 b/config/docker/files/default.json5 index f59571e7..92752e86 100644 --- a/config/docker/files/default.json5 +++ b/config/docker/files/default.json5 @@ -84,6 +84,7 @@ "image_manifest_files": [ "resources/image_manifests/stx-test-images.yaml", "resources/image_manifests/stx-test-images-invalid.yaml", + "resources/image_manifests/stx-third-party-images.yaml" // "resources/image_manifests/harbor-test-images.yaml", // "resources/stx-networking-images.yaml", ], @@ -98,6 +99,10 @@ "manifest_registry": "dockerhub", "override": false, }, + "resources/image_manifests/stx-third-party-images.yaml": { + "manifest_registry": "null", // No manifest fallback; each image uses its "source_registry" or "default_source_registry" + "override": false, + }, // // Use Harbor as the default for images in this manifest that do not specify "source_registry" // "resources/image_manifests/stx-sanity-images.yaml": { // "manifest_registry": "harbor", diff --git a/keywords/docker/images/docker_images_keywords.py b/keywords/docker/images/docker_images_keywords.py index 732334ea..4ae5a5a4 100644 --- a/keywords/docker/images/docker_images_keywords.py +++ b/keywords/docker/images/docker_images_keywords.py @@ -11,44 +11,46 @@ class DockerImagesKeywords(BaseKeyword): def __init__(self, ssh_connection: SSHConnection): """ Constructor + Args: - ssh_connection: + ssh_connection (SSHConnection): Active SSH connection to the system under test. """ self.ssh_connection = ssh_connection - def list_images(self) -> [DockerImagesOutput]: + def list_images(self) -> list[DockerImagesOutput]: """ - Lists docker images - Returns: list of docker images outputs + Lists docker images. + Returns: + list[DockerImagesOutput]: List of DockerImagesOutput objects representing the images """ - output = self.ssh_connection.send_as_sudo('docker images') + output = self.ssh_connection.send_as_sudo("docker images") docker_images = DockerImagesOutput(output).get_images() return docker_images - def remove_image(self, image): + def remove_image(self, image: str) -> None: """ - Removes the docker image + Removes the docker image. + Args: - image (): the image to remove - - Returns: - + image (str): the docker image to remove """ - output = self.ssh_connection.send_as_sudo(f'docker image rm {image}') + output = self.ssh_connection.send_as_sudo(f"docker image rm {image}") # both the Untagged Image and no such images messages are valid - assert ( - len(list(filter(lambda output_line: f'Untagged: {image}' in output_line, output))) > 0 or len(list(filter(lambda output_line: f'Error: No such image: {image}' in output_line, output))) > 0 - ) + assert len(list(filter(lambda output_line: f"Untagged: {image}" in output_line, output))) > 0 or len(list(filter(lambda output_line: f"Error: No such image: {image}" in output_line, output))) > 0 - def pull_image(self, image): + def pull_image(self, image: str) -> None: """ - Pulls the image + Pulls the image. + Args: - image (): the image - - Returns: - + image (str): the image to pull """ - self.ssh_connection.send_as_sudo(f'docker image pull {image}') + self.ssh_connection.send_as_sudo(f"docker image pull {image}") + + def prune_dangling_images(self) -> None: + """ + Prunes all dangling Docker images to free disk space. + """ + self.ssh_connection.send_as_sudo("docker image prune -f") diff --git a/keywords/docker/images/docker_sync_images_keywords.py b/keywords/docker/images/docker_sync_images_keywords.py index aa80931f..ec62141e 100644 --- a/keywords/docker/images/docker_sync_images_keywords.py +++ b/keywords/docker/images/docker_sync_images_keywords.py @@ -30,6 +30,141 @@ class DockerSyncImagesKeywords(BaseKeyword): self.docker_images_keywords = DockerImagesKeywords(ssh_connection) self.docker_load_keywords = DockerLoadImageKeywords(ssh_connection) + def _load_and_validate_manifest(self, manifest_path: str) -> dict: + """ + Load and validate manifest structure. + + Args: + manifest_path (str): Path to the manifest YAML file. + + Returns: + dict: Loaded manifest data. + + Raises: + KeywordException: If the manifest cannot be loaded or is missing required fields. + """ + try: + with open(manifest_path, "r") as f: + manifest = yaml.safe_load(f) + except Exception as e: + raise KeywordException(f"Failed to load manifest '{manifest_path}': {e}") + + if "images" not in manifest: + raise KeywordException(f"Manifest '{manifest_path}' missing required 'images' key") + + return manifest + + def _find_image_in_manifest(self, manifest: dict, image_name: str, image_tag: str, manifest_path: str) -> dict: + """ + Find and validate a specific image entry in manifest. + + Args: + manifest (dict): Loaded manifest data. + image_name (str): Name of the image to find. + image_tag (str): Tag of the image to find. + manifest_path (str): Path to manifest file (for error messages). + + Returns: + dict: The matching image entry. + + Raises: + KeywordException: If image is not found or duplicate entries exist. + """ + matches = [img for img in manifest["images"] if img["name"] == image_name and img["tag"] == image_tag] + + if not matches: + raise KeywordException(f"Image '{image_name}:{image_tag}' not found in manifest '{manifest_path}'") + + if len(matches) > 1: + base_message = f"Duplicate entries found for '{image_name}:{image_tag}' in manifest '{manifest_path}'. Each image:tag combination must be unique." + duplicate_entries_details = "\n".join([str(entry) for entry in matches]) + get_logger().log_error(f"{base_message}\n{duplicate_entries_details}") + raise KeywordException(base_message) + + return matches[0] + + def _sync_single_image_from_manifest(self, image: dict, manifest_path: str) -> None: + """ + Sync a single image using the pull/tag/push pattern. + + Args: + image (dict): Image entry from manifest with 'name' and 'tag' keys. + manifest_path (str): Path to manifest file (for registry resolution). + + Raises: + KeywordException: If no registry can be resolved or sync operations fail. + """ + docker_config = ConfigurationManager.get_docker_config() + local_registry = docker_config.get_registry("local_registry") + + name = image["name"] + tag = image["tag"] + + # Resolve the source registry using the config resolution precedence + source_registry_name = docker_config.get_effective_source_registry_name(image, manifest_path) + + if not source_registry_name: + raise KeywordException(f"Image '{name}:{tag}' has no registry resolved (manifest: {manifest_path}).") + + source_registry = docker_config.get_registry(source_registry_name) + + source_image = f"{source_registry.get_registry_url()}/{name}:{tag}" + target_image = f"{local_registry.get_registry_url()}/{name}:{tag}" + + get_logger().log_info(f"Pulling {source_image}") + self.docker_images_keywords.pull_image(source_image) + + get_logger().log_info(f"Tagging {source_image} -> {target_image}") + self.docker_load_keywords.tag_docker_image_for_registry( + image_name=source_image, + tag_name=f"{name}:{tag}", + registry=local_registry, + ) + + get_logger().log_info(f"Pushing {target_image}") + self.docker_load_keywords.push_docker_image_to_registry( + tag_name=f"{name}:{tag}", + registry=local_registry, + ) + + def _build_image_references_for_removal(self, image_name: str, image_tag: str, manifest_path: str) -> list: + """ + Build list of image references to attempt removal for. + + Args: + image_name (str): Name of the image. + image_tag (str): Tag of the image. + manifest_path (str): Path to manifest file (for registry resolution). + + Returns: + list: List of image references to try removing. + """ + docker_config = ConfigurationManager.get_docker_config() + local_registry = docker_config.get_registry("local_registry") + + source_registry_name = docker_config.get_effective_source_registry_name({"name": image_name, "tag": image_tag}, manifest_path) + + if not source_registry_name: + get_logger().log_debug(f"Skipping cleanup for image {image_name}:{image_tag} (no source registry resolved)") + return [] + + source_registry = docker_config.get_registry(source_registry_name) + source_url = source_registry.get_registry_url() + + # Always try to remove these two references + refs = [ + f"{local_registry.get_registry_url()}/{image_name}:{image_tag}", + f"{image_name}:{image_tag}", + ] + + # Optionally add full source registry tag if not DockerHub + if "docker.io" not in source_url: + refs.insert(0, f"{source_url}/{image_name}:{image_tag}") + else: + get_logger().log_debug(f"Skipping full docker.io-prefixed tag for {source_url}/{image_name}:{image_tag}") + + return refs + def sync_images_from_manifest(self, manifest_path: str) -> None: """ Syncs Docker images listed in a YAML manifest from a source registry into the local registry. @@ -71,54 +206,16 @@ class DockerSyncImagesKeywords(BaseKeyword): manifest_path (str): Full path to the YAML manifest file. Raises: - KeywordException: If one or more image sync operations fail. - ValueError: If no registry can be resolved for an image. + KeywordException: If one or more image sync operations fail, or if no registry can be resolved for an image. """ - docker_config = ConfigurationManager.get_docker_config() - local_registry = docker_config.get_registry("local_registry") - - with open(manifest_path, "r") as f: - manifest = yaml.safe_load(f) - - if "images" not in manifest: - raise ValueError(f"Manifest at {manifest_path} is missing required 'images' key") - + manifest = self._load_and_validate_manifest(manifest_path) failures = [] for image in manifest["images"]: - name = image["name"] - tag = image["tag"] - - # Resolve the source registry using the config resolution precedence - source_registry_name = docker_config.get_effective_source_registry_name(image, manifest_path) - - if not source_registry_name: - raise ValueError(f"Image '{name}:{tag}' has no registry resolved (manifest: {manifest_path}).") - try: - source_registry = docker_config.get_registry(source_registry_name) - - source_image = f"{source_registry.get_registry_url()}/{name}:{tag}" - target_image = f"{local_registry.get_registry_url()}/{name}:{tag}" - - get_logger().log_info(f"Pulling {source_image}") - self.docker_images_keywords.pull_image(source_image) - - get_logger().log_info(f"Tagging {source_image} -> {target_image}") - self.docker_load_keywords.tag_docker_image_for_registry( - image_name=source_image, - tag_name=f"{name}:{tag}", - registry=local_registry, - ) - - get_logger().log_info(f"Pushing {target_image}") - self.docker_load_keywords.push_docker_image_to_registry( - tag_name=f"{name}:{tag}", - registry=local_registry, - ) - + self._sync_single_image_from_manifest(image, manifest_path) except Exception as e: - error_msg = f"Failed to sync image {name}:{tag} from {source_registry_name}: {e}" + error_msg = f"Failed to sync image {image['name']}:{image['tag']}: {e}" get_logger().log_error(error_msg) failures.append(error_msg) @@ -147,48 +244,20 @@ class DockerSyncImagesKeywords(BaseKeyword): Raises: KeywordException: If the manifest cannot be read, parsed, or is missing required fields. """ - docker_config = ConfigurationManager.get_docker_config() - local_registry = docker_config.get_registry("local_registry") - - try: - with open(manifest_path, "r") as f: - manifest = yaml.safe_load(f) - except Exception as e: - raise KeywordException(f"Failed to load manifest '{manifest_path}': {e}") - - if "images" not in manifest: - raise KeywordException(f"Manifest '{manifest_path}' missing required 'images' key") + manifest = self._load_and_validate_manifest(manifest_path) for image in manifest["images"]: name = image["name"] tag = image["tag"] - source_registry_name = docker_config.get_effective_source_registry_name(image, manifest_path) - if not source_registry_name: - get_logger().log_debug(f"Skipping cleanup for image {name}:{tag} (no source registry resolved)") - continue - - source_registry = docker_config.get_registry(source_registry_name) - source_url = source_registry.get_registry_url() - - # Always try to remove these two references - refs = [ - f"{local_registry.get_registry_url()}/{name}:{tag}", - f"{name}:{tag}", - ] - - # Optionally add full source registry tag if not DockerHub - if "docker.io" not in source_url: - refs.insert(0, f"{source_url}/{name}:{tag}") - else: - get_logger().log_debug(f"Skipping full docker.io-prefixed tag for {source_url}/{name}:{tag}") + refs = self._build_image_references_for_removal(name, tag, manifest_path) for ref in refs: self.docker_images_keywords.remove_image(ref) - def validate_manifest_images_exist(self, manifest_path: str, fail_on_missing: bool = True) -> bool: + def manifest_images_exist_in_local_registry(self, manifest_path: str, fail_on_missing: bool = True) -> bool: """ - Validates that all images listed in a manifest are present in the local Docker registry. + Checks that all images listed in a manifest are present in the local Docker registry. This is typically used after syncing images via `sync_images_from_manifest` to ensure that each expected image was successfully pushed to the local registry (e.g., registry.local:9001). @@ -209,14 +278,7 @@ class DockerSyncImagesKeywords(BaseKeyword): docker_config = ConfigurationManager.get_docker_config() local_registry = docker_config.get_registry("local_registry") - try: - with open(manifest_path, "r") as f: - manifest = yaml.safe_load(f) - except Exception as e: - raise KeywordException(f"Failed to load manifest '{manifest_path}': {e}") - - if "images" not in manifest: - raise KeywordException(f"Manifest '{manifest_path}' missing required 'images' key") + manifest = self._load_and_validate_manifest(manifest_path) images = self.docker_images_keywords.list_images() actual_repos = [img.get_repository() for img in images] @@ -242,3 +304,87 @@ class DockerSyncImagesKeywords(BaseKeyword): return False return True + + def sync_image_from_manifest(self, image_name: str, image_tag: str, manifest_path: str) -> None: + """ + Syncs a single Docker image listed in a manifest from a source registry into the local registry. + + This is similar to sync_images_from_manifest(), but only processes the specified image. + + Args: + image_name (str): Name of the image to sync (without registry). + image_tag (str): Tag of the image to sync. + manifest_path (str): Path to the manifest YAML file. + + Raises: + KeywordException: If the image is not found in the manifest, if multiple entries are found, + or if the sync operation fails. + """ + manifest = self._load_and_validate_manifest(manifest_path) + target_image_entry = self._find_image_in_manifest(manifest, image_name, image_tag, manifest_path) + + self._sync_single_image_from_manifest(target_image_entry, manifest_path) + get_logger().log_info(f"Successfully synced '{image_name}:{image_tag}' from manifest") + + def remove_image_from_manifest(self, image_name: str, image_tag: str, manifest_path: str) -> None: + """ + Removes a single Docker image listed in a manifest from the local system. + + This is similar to remove_images_from_manifest(), but only processes the specified image. + + For each image, removal is attempted for: + 1. source_registry/image:tag (skipped if source is docker.io; see note below) + 2. local_registry/image:tag + 3. image:tag + + Notes: + - docker.io-prefixed references are skipped because Docker stores these as image:tag. + - Removal is attempted even for images that were never successfully synced. + + Args: + image_name (str): Name of the image to remove (without registry). + image_tag (str): Tag of the image to remove. + manifest_path (str): Path to the manifest YAML file. + + Raises: + KeywordException: If the image is not found in the manifest, if multiple entries are found, + or if the removal operation fails. + """ + manifest = self._load_and_validate_manifest(manifest_path) + self._find_image_in_manifest(manifest, image_name, image_tag, manifest_path) + + refs = self._build_image_references_for_removal(image_name, image_tag, manifest_path) + + for ref in refs: + self.docker_images_keywords.remove_image(ref) + + get_logger().log_info(f"Successfully removed '{image_name}:{image_tag}' from local Docker images.") + + def image_exists_in_local_registry(self, image_name: str, image_tag: str) -> bool: + """ + Checks that a Docker image with the specified name and tag exists in the local registry. + + This is typically used after syncing a single image to confirm it was pushed successfully. + + Args: + image_name (str): Name of the image to check (without registry prefix). + image_tag (str): Tag of the image to check. + + Returns: + bool: True if the image exists, False otherwise. + """ + docker_config = ConfigurationManager.get_docker_config() + local_registry = docker_config.get_registry("local_registry") + local_registry_url = local_registry.get_registry_url() + + images = self.docker_images_keywords.list_images() + + expected_repo = f"{local_registry_url}/{image_name}" + matches = [img for img in images if img.get_repository() == expected_repo and img.get_tag() == image_tag] + + if matches: + get_logger().log_info(f"Image '{expected_repo}:{image_tag}' was found in the local registry.") + return True + else: + get_logger().log_warning(f"Image '{expected_repo}:{image_tag}' was NOT found in the local registry.") + return False diff --git a/resources/image_manifests/stx-third-party-test-images.yaml b/resources/image_manifests/stx-third-party-test-images.yaml new file mode 100644 index 00000000..41cac9dc --- /dev/null +++ b/resources/image_manifests/stx-third-party-test-images.yaml @@ -0,0 +1,45 @@ +# ------------------------------------------------------------------------------ +# Image Manifest: `stx-third-party-test-images.yaml` +# +# This manifest defines third-party, publicly available Docker images used +# in StarlingX test suites such as `testcases/cloud_platform/sanity/`. +# +# Images are sourced from registries like DockerHub, registry.k8s.io, and GCR, +# and may be mirrored to the local StarlingX registry (e.g., `registry.local:9001`) +# during system setup or as part of an on-demand test image sync. +# +# Notes: +# - Each image entry must include `name` and `tag`. +# - Image names must include their full namespace (e.g., `google-samples/node-hello`). +# - Registry URLs and credentials are defined in: +# `config/docker/files/default.json5` +# +# Registry resolution priority (from most to least specific): +# 1. `source_registry` field on the individual image entry (recommended) +# 2. `manifest_registry_map` in the Docker config +# 3. `default_source_registry` fallback +# ------------------------------------------------------------------------------ + +images: + # DockerHub images + - name: "busybox" + tag: "1.36.1" + source_registry: "dockerhub" + + - name: "calico/ctl" + tag: "v3.27.0" + source_registry: "dockerhub" + + # k8s images + - name: "pause" + tag: "3.9" + source_registry: "k8s" + + - name: "e2e-test-images/resource-consumer" + tag: "1.10" + source_registry: "k8s" + + # GCR images + - name: "google-samples/node-hello" + tag: "1.0" + source_registry: "gcr" diff --git a/testcases/cloud_platform/images/test_docker_image_sync.py b/testcases/cloud_platform/images/test_docker_image_sync.py index e0af9527..ab8ac25a 100644 --- a/testcases/cloud_platform/images/test_docker_image_sync.py +++ b/testcases/cloud_platform/images/test_docker_image_sync.py @@ -24,6 +24,7 @@ from pytest import FixtureRequest, fail, raises from config.configuration_manager import ConfigurationManager from framework.exceptions.keyword_exception import KeywordException from framework.logging.automation_logger import get_logger +from framework.resources.resource_finder import get_stx_resource_path from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords from keywords.docker.images.docker_images_keywords import DockerImagesKeywords from keywords.docker.images.docker_sync_images_keywords import DockerSyncImagesKeywords @@ -187,3 +188,123 @@ def test_invalid_manifest_logging(request): # Validates that images from a manifest with mixed registries (DockerHub and Harbor) can be pulled and synced into the local registry. # """ # run_manifest_sync_test(request, "stx-test-images-mixed-registries.yaml") + + +def test_sync_single_busybox_image(request: FixtureRequest): + """ + Sync a single image (busybox:1.36.1) from the manifest to the local registry, + validate it exists, and clean up afterwards. + + This test validates the sync_image_from_manifest() method which allows + selective syncing of individual images rather than entire manifests. + """ + manifest_path = get_stx_resource_path("resources/image_manifests/stx-third-party-test-images.yaml") + + ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + docker_sync_keywords = DockerSyncImagesKeywords(ssh_connection) + + def cleanup(): + get_logger().log_teardown_step("Removing busybox image from local registry") + docker_sync_keywords.remove_image_from_manifest(image_name="busybox", image_tag="1.36.1", manifest_path=manifest_path) + + request.addfinalizer(cleanup) + + get_logger().log_test_case_step("Syncing busybox image from manifest") + docker_sync_keywords.sync_image_from_manifest(image_name="busybox", image_tag="1.36.1", manifest_path=manifest_path) + + get_logger().log_test_case_step("Validating busybox image exists in local registry") + assert docker_sync_keywords.image_exists_in_local_registry(image_name="busybox", image_tag="1.36.1") + + +def test_sync_third_party_images_to_local_registry(request: FixtureRequest): + """ + Sync required 3rd party Docker images from source registries to local registry. + + This test ensures all common sanity test images are preloaded into registry.local. + Uses the stx-third-party-test-images.yaml manifest which contains public images + from DockerHub, k8s.gcr.io, and other public registries. + """ + manifest_path = get_stx_resource_path("resources/image_manifests/stx-third-party-test-images.yaml") + + ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + docker_keywords = DockerSyncImagesKeywords(ssh_connection) + + def cleanup(): + get_logger().log_teardown_step("Removing third-party images from manifest") + docker_keywords.remove_images_from_manifest(manifest_path) + + request.addfinalizer(cleanup) + + get_logger().log_test_case_step(f"Syncing Docker images from manifest: {manifest_path}") + docker_keywords.sync_images_from_manifest(manifest_path) + + get_logger().log_test_case_step("Validating all images exist in local registry") + docker_keywords.manifest_images_exist_in_local_registry(manifest_path) + + get_logger().log_info("Successfully synced all 3rd party test images") + + +def test_remove_third_party_images_from_local_registry(): + """ + Remove third-party Docker images from registry.local that were synced from manifest. + + This test ensures cleanup of common sanity/networking images after test execution. + Can be run independently or as part of test cleanup verification. + """ + manifest_path = get_stx_resource_path("resources/image_manifests/stx-third-party-test-images.yaml") + + get_logger().log_test_case_step("Connecting to active controller") + ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + + get_logger().log_test_case_step(f"Removing Docker images listed in: {manifest_path}") + docker_keywords = DockerSyncImagesKeywords(ssh_connection=ssh_connection) + docker_keywords.remove_images_from_manifest(manifest_path) + + get_logger().log_info("Successfully removed all third-party test images") + + +def test_single_image_sync_and_removal_workflow(request: FixtureRequest): + """ + Test the complete workflow of syncing and removing a single image. + + This test validates: + 1. Single image sync from manifest + 2. Image validation in local registry + 3. Single image removal from manifest + 4. Cleanup verification + """ + manifest_path = get_stx_resource_path("resources/image_manifests/stx-third-party-test-images.yaml") + + ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + docker_sync_keywords = DockerSyncImagesKeywords(ssh_connection) + + # Test image: calico/ctl:v3.27.0 (from the third-party manifest) + test_image_name = "calico/ctl" + test_image_tag = "v3.27.0" + + def cleanup(): + get_logger().log_teardown_step(f"Final cleanup: removing {test_image_name}:{test_image_tag}") + try: + docker_sync_keywords.remove_image_from_manifest(image_name=test_image_name, image_tag=test_image_tag, manifest_path=manifest_path) + except Exception as e: + get_logger().log_warning(f"Cleanup failed (expected if test passed): {e}") + + request.addfinalizer(cleanup) + + # Step 1: Sync the single image + get_logger().log_test_case_step(f"Syncing {test_image_name}:{test_image_tag} from manifest") + docker_sync_keywords.sync_image_from_manifest(image_name=test_image_name, image_tag=test_image_tag, manifest_path=manifest_path) + + # Step 2: Validate it exists in local registry + get_logger().log_test_case_step(f"Validating {test_image_name}:{test_image_tag} exists in local registry") + assert docker_sync_keywords.image_exists_in_local_registry(image_name=test_image_name, image_tag=test_image_tag), f"Image {test_image_name}:{test_image_tag} should exist in local registry after sync" + + # Step 3: Remove the single image + get_logger().log_test_case_step(f"Removing {test_image_name}:{test_image_tag} from local registry") + docker_sync_keywords.remove_image_from_manifest(image_name=test_image_name, image_tag=test_image_tag, manifest_path=manifest_path) + + # Step 4: Validate it no longer exists (optional verification) + get_logger().log_test_case_step(f"Verifying {test_image_name}:{test_image_tag} was removed") + assert not docker_sync_keywords.image_exists_in_local_registry(image_name=test_image_name, image_tag=test_image_tag), f"Image {test_image_name}:{test_image_tag} should not exist in local registry after removal" + + get_logger().log_info(f"Successfully completed single image sync/removal workflow for {test_image_name}:{test_image_tag}") diff --git a/unit_tests/keyword/docker/__init__.py b/unit_tests/keyword/docker/__init__.py new file mode 100644 index 00000000..ae448f63 --- /dev/null +++ b/unit_tests/keyword/docker/__init__.py @@ -0,0 +1 @@ +"""Unit tests for Docker keywords.""" diff --git a/unit_tests/keyword/docker/docker_sync_images_keywords_test.py b/unit_tests/keyword/docker/docker_sync_images_keywords_test.py new file mode 100644 index 00000000..30fb375d --- /dev/null +++ b/unit_tests/keyword/docker/docker_sync_images_keywords_test.py @@ -0,0 +1,58 @@ +from unittest.mock import Mock, mock_open, patch + +import pytest +import yaml + +from framework.exceptions.keyword_exception import KeywordException +from keywords.docker.images.docker_sync_images_keywords import DockerSyncImagesKeywords + + +class TestDockerSyncImagesKeywords: + """Unit tests for DockerSyncImagesKeywords private helper methods.""" + + def setup_method(self): + """Set up test fixtures.""" + self.mock_ssh_connection = Mock() + self.docker_sync_keywords = DockerSyncImagesKeywords(self.mock_ssh_connection) + + def test_load_and_validate_manifest_valid(self): + """Test loading a valid manifest file.""" + manifest_content = {"images": [{"name": "busybox", "tag": "1.36.1"}, {"name": "alpine", "tag": "latest"}]} + + with patch("builtins.open", mock_open(read_data=yaml.dump(manifest_content))): + with patch("yaml.safe_load", return_value=manifest_content): + result = DockerSyncImagesKeywords._load_and_validate_manifest(self.docker_sync_keywords, "test_manifest.yaml") + + assert result == manifest_content + assert "images" in result + assert len(result["images"]) == 2 + + def test_load_and_validate_manifest_missing_images_key(self): + """Test loading a manifest without required 'images' key.""" + manifest_content = {"other_key": "value"} + + with patch("builtins.open", mock_open(read_data=yaml.dump(manifest_content))): + with patch("yaml.safe_load", return_value=manifest_content): + with pytest.raises(KeywordException, match="missing required 'images' key"): + DockerSyncImagesKeywords._load_and_validate_manifest(self.docker_sync_keywords, "test_manifest.yaml") + + def test_load_and_validate_manifest_file_not_found(self): + """Test loading a non-existent manifest file.""" + with patch("builtins.open", side_effect=FileNotFoundError("File not found")): + with pytest.raises(KeywordException, match="Failed to load manifest"): + DockerSyncImagesKeywords._load_and_validate_manifest(self.docker_sync_keywords, "nonexistent.yaml") + + def test_find_image_in_manifest_single_match(self): + """Test finding a single image in manifest.""" + manifest = {"images": [{"name": "busybox", "tag": "1.36.1"}, {"name": "alpine", "tag": "latest"}]} + + result = DockerSyncImagesKeywords._find_image_in_manifest(self.docker_sync_keywords, manifest, "busybox", "1.36.1", "test_manifest.yaml") + + assert result == {"name": "busybox", "tag": "1.36.1"} + + def test_find_image_in_manifest_not_found(self): + """Test finding an image that doesn't exist in manifest.""" + manifest = {"images": [{"name": "busybox", "tag": "1.36.1"}]} + + with pytest.raises(KeywordException, match="Image 'nginx:latest' not found"): + DockerSyncImagesKeywords._find_image_in_manifest(self.docker_sync_keywords, manifest, "nginx", "latest", "test_manifest.yaml")