starlingx/test
Code Issues Proposed changes

Merge "Add granular Docker image sync operations"

Browse Source
This commit is contained in:
Zuul
2025-07-15 17:36:36 +00:00
committed by Gerrit Code Review
parent 28795f9f9b 58f64e27f7
commit 7517557829
7 changed files with 482 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
config/docker/files/default.json5
View File

@@ -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",

46
keywords/docker/images/docker_images_keywords.py
View File

@@ -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")

310
keywords/docker/images/docker_sync_images_keywords.py
View File

@@ -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

45
resources/image_manifests/stx-third-party-test-images.yaml Normal file
View File

@@ -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"

121
testcases/cloud_platform/images/test_docker_image_sync.py
View File

@@ -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}")

1
unit_tests/keyword/docker/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Unit tests for Docker keywords."""

58
unit_tests/keyword/docker/docker_sync_images_keywords_test.py Normal file
View File

@@ -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")