mirror of
https://github.com/quay/quay.git
synced 2026-01-26 06:21:37 +03:00
* feat: Add FEATURE_SPARSE_INDEX config for sparse manifest index support When enabled, manifests in an index that cannot be loaded will be skipped if their architecture is not in the SPARSE_INDEX_REQUIRED_ARCHS list. This allows for sparse manifest indexes where not all architectures are required to be present. New config options: - FEATURE_SPARSE_INDEX: Enable sparse manifest index support (default: False) - SPARSE_INDEX_REQUIRED_ARCHS: List of architectures that must be present 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
862 lines
29 KiB
Python
862 lines
29 KiB
Python
"""
|
|
Unit tests for sparse manifest index functionality in LazyManifestLoader.
|
|
|
|
Tests the FEATURE_SPARSE_INDEX and SPARSE_INDEX_REQUIRED_ARCHS configuration options
|
|
which allow manifest lists/indexes to have optional architectures that can be missing.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from image.docker.schema2 import DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE
|
|
from image.docker.schema2.list import DockerSchema2ManifestList
|
|
from image.docker.schema2.manifest import DockerSchema2Manifest
|
|
from image.oci import OCI_IMAGE_INDEX_CONTENT_TYPE, OCI_IMAGE_MANIFEST_CONTENT_TYPE
|
|
from image.oci.index import OCIIndex
|
|
from image.oci.manifest import OCIManifest
|
|
from image.shared import ManifestException
|
|
from image.shared.schemautil import ContentRetrieverForTesting, LazyManifestLoader
|
|
from util.bytes import Bytes
|
|
|
|
# Docker Schema2 manifest bytes for testing (defined locally to avoid import chain)
|
|
DOCKER_SCHEMA2_MANIFEST_BYTES = json.dumps(
|
|
{
|
|
"schemaVersion": 2,
|
|
"mediaType": "application/vnd.docker.distribution.manifest.v2+json",
|
|
"config": {
|
|
"mediaType": "application/vnd.docker.container.image.v1+json",
|
|
"size": 1234,
|
|
"digest": "sha256:b5b2b2c507a0944348e0303114d8d93aaaa081732b86451d9bce1f432a537bc7",
|
|
},
|
|
"layers": [
|
|
{
|
|
"mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip",
|
|
"size": 1234,
|
|
"digest": "sha256:ec4b8955958665577945c89419d1af06b5f7636b4ac3da7f12184802ad867736",
|
|
},
|
|
{
|
|
"mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip",
|
|
"size": 32654,
|
|
"digest": "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f",
|
|
},
|
|
],
|
|
}
|
|
).encode("utf-8")
|
|
|
|
# OCI manifest bytes for testing
|
|
OCI_MANIFEST_BYTES = json.dumps(
|
|
{
|
|
"schemaVersion": 2,
|
|
"config": {
|
|
"mediaType": "application/vnd.oci.image.config.v1+json",
|
|
"size": 7023,
|
|
"digest": "sha256:b5b2b2c507a0944348e0303114d8d93aaaa081732b86451d9bce1f432a537bc7",
|
|
},
|
|
"layers": [
|
|
{
|
|
"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
|
|
"size": 32654,
|
|
"digest": "sha256:9834876dcfb05cb167a5c24953eba58c4ac89b1adf57f28f2f9d09af107ee8f0",
|
|
},
|
|
{
|
|
"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
|
|
"size": 16724,
|
|
"digest": "sha256:3c3a4604a545cdc127456d94e421cd355bca5b528f4a9c1905b15da2eb4a4c6b",
|
|
},
|
|
],
|
|
}
|
|
).encode("utf-8")
|
|
|
|
|
|
# Sample manifest data for testing
|
|
def create_manifest_data(
|
|
digest="sha256:abc123",
|
|
size=946,
|
|
media_type=DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE,
|
|
architecture=None,
|
|
os="linux",
|
|
):
|
|
"""Create manifest data dict with optional platform info."""
|
|
data = {
|
|
"digest": digest,
|
|
"size": size,
|
|
"mediaType": media_type,
|
|
}
|
|
if architecture is not None:
|
|
data["platform"] = {
|
|
"architecture": architecture,
|
|
"os": os,
|
|
}
|
|
return data
|
|
|
|
|
|
# Supported types for testing
|
|
SUPPORTED_TYPES = {
|
|
DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE: DockerSchema2Manifest,
|
|
OCI_IMAGE_MANIFEST_CONTENT_TYPE: OCIManifest,
|
|
}
|
|
|
|
|
|
class TestLazyManifestLoaderArchitecture:
|
|
"""Tests for the architecture property of LazyManifestLoader."""
|
|
|
|
def test_architecture_with_platform_data(self):
|
|
"""Test architecture extraction when platform data is present."""
|
|
manifest_data = create_manifest_data(architecture="amd64")
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
assert loader.architecture == "amd64"
|
|
|
|
def test_architecture_without_platform_data(self):
|
|
"""Test architecture returns None when no platform data exists."""
|
|
manifest_data = create_manifest_data() # No architecture
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
assert loader.architecture is None
|
|
|
|
def test_architecture_with_empty_platform(self):
|
|
"""Test architecture returns None when platform dict is empty."""
|
|
manifest_data = {
|
|
"digest": "sha256:abc123",
|
|
"size": 946,
|
|
"mediaType": DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE,
|
|
"platform": {},
|
|
}
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
assert loader.architecture is None
|
|
|
|
@pytest.mark.parametrize(
|
|
"arch",
|
|
[
|
|
pytest.param("amd64", id="amd64"),
|
|
pytest.param("arm64", id="arm64"),
|
|
pytest.param("ppc64le", id="ppc64le"),
|
|
pytest.param("s390x", id="s390x"),
|
|
pytest.param("386", id="386"),
|
|
],
|
|
)
|
|
def test_architecture_various_values(self, arch):
|
|
"""Test architecture extraction for various architecture values."""
|
|
manifest_data = create_manifest_data(architecture=arch)
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
assert loader.architecture == arch
|
|
|
|
|
|
class TestSparseIndexDisabled:
|
|
"""Tests for behavior when FEATURE_SPARSE_INDEX is disabled (default)."""
|
|
|
|
def test_missing_manifest_raises_exception(self):
|
|
"""When sparse index is disabled, missing manifest raises exception."""
|
|
manifest_data = create_manifest_data(architecture="amd64")
|
|
retriever = ContentRetrieverForTesting() # Empty - manifest not available
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={"FEATURE_SPARSE_INDEX": False},
|
|
)
|
|
|
|
with pytest.raises(ManifestException) as exc_info:
|
|
_ = loader.manifest_obj
|
|
|
|
assert "Could not find child manifest" in str(exc_info.value)
|
|
|
|
def test_missing_manifest_raises_without_config(self):
|
|
"""When config is not set, missing manifest raises exception (default behavior)."""
|
|
manifest_data = create_manifest_data(architecture="arm64")
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={},
|
|
)
|
|
|
|
with pytest.raises(ManifestException) as exc_info:
|
|
_ = loader.manifest_obj
|
|
|
|
assert "Could not find child manifest" in str(exc_info.value)
|
|
|
|
|
|
class TestSparseIndexEnabled:
|
|
"""Tests for behavior when FEATURE_SPARSE_INDEX is enabled."""
|
|
|
|
def test_optional_architecture_returns_none(self):
|
|
"""Optional architecture missing returns None instead of raising."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:missing",
|
|
architecture="arm64",
|
|
)
|
|
retriever = ContentRetrieverForTesting() # Empty
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"], # arm64 is optional
|
|
},
|
|
)
|
|
|
|
result = loader.manifest_obj
|
|
assert result is None
|
|
|
|
def test_required_architecture_raises_exception(self):
|
|
"""Required architecture missing still raises exception."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:missing",
|
|
architecture="amd64",
|
|
)
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64", "arm64"],
|
|
},
|
|
)
|
|
|
|
with pytest.raises(ManifestException) as exc_info:
|
|
_ = loader.manifest_obj
|
|
|
|
assert "Could not find child manifest" in str(exc_info.value)
|
|
|
|
def test_empty_required_archs_treats_all_as_required(self):
|
|
"""When SPARSE_INDEX_REQUIRED_ARCHS is empty, all architectures are required."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:missing",
|
|
architecture="arm64",
|
|
)
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": [], # Empty list = all required
|
|
},
|
|
)
|
|
|
|
with pytest.raises(ManifestException):
|
|
_ = loader.manifest_obj
|
|
|
|
def test_no_architecture_in_manifest_treats_as_required(self):
|
|
"""When manifest has no architecture specified, treat as required."""
|
|
manifest_data = create_manifest_data() # No architecture
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"],
|
|
},
|
|
)
|
|
|
|
with pytest.raises(ManifestException):
|
|
_ = loader.manifest_obj
|
|
|
|
@pytest.mark.parametrize(
|
|
"required_archs,manifest_arch,should_skip",
|
|
[
|
|
pytest.param(["amd64"], "arm64", True, id="arm64_optional_when_amd64_required"),
|
|
pytest.param(["amd64"], "ppc64le", True, id="ppc64le_optional_when_amd64_required"),
|
|
pytest.param(
|
|
["amd64", "arm64"], "s390x", True, id="s390x_optional_when_amd64_arm64_required"
|
|
),
|
|
pytest.param(["amd64"], "amd64", False, id="amd64_required"),
|
|
pytest.param(["arm64"], "arm64", False, id="arm64_required"),
|
|
pytest.param(
|
|
["amd64", "arm64", "ppc64le"], "ppc64le", False, id="ppc64le_in_required_list"
|
|
),
|
|
],
|
|
)
|
|
def test_architecture_required_matrix(self, required_archs, manifest_arch, should_skip):
|
|
"""Test various combinations of required architectures and manifest architectures."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:missing",
|
|
architecture=manifest_arch,
|
|
)
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": required_archs,
|
|
},
|
|
)
|
|
|
|
if should_skip:
|
|
result = loader.manifest_obj
|
|
assert result is None
|
|
else:
|
|
with pytest.raises(ManifestException):
|
|
_ = loader.manifest_obj
|
|
|
|
|
|
class TestManifestLoadingCaching:
|
|
"""Tests for manifest loading caching behavior."""
|
|
|
|
def test_manifest_obj_cached_after_first_load(self):
|
|
"""Test that manifest_obj is cached after first access."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:e6",
|
|
size=len(DOCKER_SCHEMA2_MANIFEST_BYTES),
|
|
)
|
|
retriever = ContentRetrieverForTesting({"sha256:e6": DOCKER_SCHEMA2_MANIFEST_BYTES})
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
# First access
|
|
manifest1 = loader.manifest_obj
|
|
assert manifest1 is not None
|
|
assert isinstance(manifest1, DockerSchema2Manifest)
|
|
|
|
# Second access should return same object
|
|
manifest2 = loader.manifest_obj
|
|
assert manifest1 is manifest2
|
|
|
|
def test_none_manifest_cached(self):
|
|
"""Test that None result is cached for optional architectures."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:missing",
|
|
architecture="arm64",
|
|
)
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"],
|
|
},
|
|
)
|
|
|
|
# First access
|
|
result1 = loader.manifest_obj
|
|
assert result1 is None
|
|
assert loader._load_attempted is True
|
|
|
|
# Second access should not re-attempt loading
|
|
result2 = loader.manifest_obj
|
|
assert result2 is None
|
|
|
|
def test_load_attempted_flag_set(self):
|
|
"""Test that _load_attempted flag is set after first access."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:e6",
|
|
size=len(DOCKER_SCHEMA2_MANIFEST_BYTES),
|
|
)
|
|
retriever = ContentRetrieverForTesting({"sha256:e6": DOCKER_SCHEMA2_MANIFEST_BYTES})
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
assert loader._load_attempted is False
|
|
_ = loader.manifest_obj
|
|
assert loader._load_attempted is True
|
|
|
|
|
|
class TestManifestLoadingSuccess:
|
|
"""Tests for successful manifest loading scenarios."""
|
|
|
|
def test_load_manifest_successfully(self):
|
|
"""Test successful manifest loading."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:e6",
|
|
size=len(DOCKER_SCHEMA2_MANIFEST_BYTES),
|
|
architecture="amd64",
|
|
)
|
|
retriever = ContentRetrieverForTesting({"sha256:e6": DOCKER_SCHEMA2_MANIFEST_BYTES})
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
manifest = loader.manifest_obj
|
|
assert manifest is not None
|
|
assert isinstance(manifest, DockerSchema2Manifest)
|
|
assert manifest.schema_version == 2
|
|
|
|
def test_load_manifest_size_mismatch_raises(self):
|
|
"""Test that size mismatch raises exception."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:e6",
|
|
size=100, # Wrong size
|
|
architecture="amd64",
|
|
)
|
|
retriever = ContentRetrieverForTesting({"sha256:e6": DOCKER_SCHEMA2_MANIFEST_BYTES})
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
with pytest.raises(ManifestException) as exc_info:
|
|
_ = loader.manifest_obj
|
|
|
|
assert "Size of manifest does not match" in str(exc_info.value)
|
|
|
|
def test_load_manifest_unsupported_media_type_raises(self):
|
|
"""Test that unsupported media type raises exception."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:e6",
|
|
size=len(DOCKER_SCHEMA2_MANIFEST_BYTES),
|
|
media_type="application/vnd.unknown.manifest+json",
|
|
)
|
|
retriever = ContentRetrieverForTesting({"sha256:e6": DOCKER_SCHEMA2_MANIFEST_BYTES})
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
)
|
|
|
|
with pytest.raises(ManifestException) as exc_info:
|
|
_ = loader.manifest_obj
|
|
|
|
assert "Unknown or unsupported manifest media type" in str(exc_info.value)
|
|
|
|
|
|
def create_docker_manifest_list_bytes():
|
|
"""Create a Docker Schema2 manifest list with multiple architectures."""
|
|
return json.dumps(
|
|
{
|
|
"schemaVersion": 2,
|
|
"mediaType": "application/vnd.docker.distribution.manifest.list.v2+json",
|
|
"manifests": [
|
|
{
|
|
"mediaType": DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE,
|
|
"size": len(DOCKER_SCHEMA2_MANIFEST_BYTES),
|
|
"digest": "sha256:amd64manifest",
|
|
"platform": {
|
|
"architecture": "amd64",
|
|
"os": "linux",
|
|
},
|
|
},
|
|
{
|
|
"mediaType": DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE,
|
|
"size": len(DOCKER_SCHEMA2_MANIFEST_BYTES),
|
|
"digest": "sha256:arm64manifest",
|
|
"platform": {
|
|
"architecture": "arm64",
|
|
"os": "linux",
|
|
},
|
|
},
|
|
{
|
|
"mediaType": DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE,
|
|
"size": len(DOCKER_SCHEMA2_MANIFEST_BYTES),
|
|
"digest": "sha256:ppc64lemanifest",
|
|
"platform": {
|
|
"architecture": "ppc64le",
|
|
"os": "linux",
|
|
},
|
|
},
|
|
],
|
|
}
|
|
).encode("utf-8")
|
|
|
|
|
|
def create_oci_index_bytes():
|
|
"""Create an OCI index with multiple architectures."""
|
|
return json.dumps(
|
|
{
|
|
"schemaVersion": 2,
|
|
"mediaType": OCI_IMAGE_INDEX_CONTENT_TYPE,
|
|
"manifests": [
|
|
{
|
|
"mediaType": OCI_IMAGE_MANIFEST_CONTENT_TYPE,
|
|
"size": len(OCI_MANIFEST_BYTES),
|
|
"digest": "sha256:amd64manifest",
|
|
"platform": {
|
|
"architecture": "amd64",
|
|
"os": "linux",
|
|
},
|
|
},
|
|
{
|
|
"mediaType": OCI_IMAGE_MANIFEST_CONTENT_TYPE,
|
|
"size": len(OCI_MANIFEST_BYTES),
|
|
"digest": "sha256:arm64manifest",
|
|
"platform": {
|
|
"architecture": "arm64",
|
|
"os": "linux",
|
|
},
|
|
},
|
|
{
|
|
"mediaType": OCI_IMAGE_MANIFEST_CONTENT_TYPE,
|
|
"size": len(OCI_MANIFEST_BYTES),
|
|
"digest": "sha256:ppc64lemanifest",
|
|
"platform": {
|
|
"architecture": "ppc64le",
|
|
"os": "linux",
|
|
},
|
|
},
|
|
],
|
|
}
|
|
).encode("utf-8")
|
|
|
|
|
|
class TestManifestListIndex:
|
|
"""Unit tests for sparse index with DockerSchema2ManifestList and OCIIndex."""
|
|
|
|
@pytest.fixture(
|
|
params=[
|
|
pytest.param(
|
|
(
|
|
DockerSchema2ManifestList,
|
|
create_docker_manifest_list_bytes,
|
|
DockerSchema2Manifest,
|
|
DOCKER_SCHEMA2_MANIFEST_BYTES,
|
|
),
|
|
id="docker_schema2_manifest_list",
|
|
),
|
|
pytest.param(
|
|
(OCIIndex, create_oci_index_bytes, OCIManifest, OCI_MANIFEST_BYTES),
|
|
id="oci_index",
|
|
),
|
|
]
|
|
)
|
|
def manifest_list_config(self, request):
|
|
"""Provide manifest list class, bytes factory, expected manifest type, and child manifest bytes."""
|
|
return request.param
|
|
|
|
def test_manifest_list_all_present(self, manifest_list_config):
|
|
"""Test manifest list/index when all manifests are present."""
|
|
list_class, bytes_factory, manifest_class, child_manifest_bytes = manifest_list_config
|
|
manifest_list_bytes = bytes_factory()
|
|
|
|
retriever = ContentRetrieverForTesting(
|
|
{
|
|
"sha256:amd64manifest": child_manifest_bytes,
|
|
"sha256:arm64manifest": child_manifest_bytes,
|
|
"sha256:ppc64lemanifest": child_manifest_bytes,
|
|
}
|
|
)
|
|
|
|
manifest_list = list_class(Bytes.for_string_or_unicode(manifest_list_bytes))
|
|
|
|
# Explicitly test with sparse index disabled (default behavior)
|
|
with patch("data.model.config.app_config", {"FEATURE_SPARSE_INDEX": False}):
|
|
manifests = manifest_list.manifests(retriever)
|
|
|
|
assert len(manifests) == 3
|
|
for manifest in manifests:
|
|
assert manifest.manifest_obj is not None
|
|
|
|
def test_manifest_list_with_missing_optional_arch(self, manifest_list_config):
|
|
"""Test manifest list/index with sparse index allowing missing optional architectures."""
|
|
list_class, bytes_factory, manifest_class, child_manifest_bytes = manifest_list_config
|
|
manifest_list_bytes = bytes_factory()
|
|
|
|
# Only provide amd64, arm64 and ppc64le are missing
|
|
retriever = ContentRetrieverForTesting(
|
|
{
|
|
"sha256:amd64manifest": child_manifest_bytes,
|
|
}
|
|
)
|
|
|
|
manifest_list = list_class(Bytes.for_string_or_unicode(manifest_list_bytes))
|
|
|
|
config = {
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"], # Only amd64 is required
|
|
}
|
|
with patch("data.model.config.app_config", config):
|
|
manifests = manifest_list.manifests(retriever)
|
|
assert len(manifests) == 3
|
|
|
|
# Count loaded and skipped manifests
|
|
loaded = [m for m in manifests if m.manifest_obj is not None]
|
|
skipped = [m for m in manifests if m.manifest_obj is None]
|
|
|
|
assert len(loaded) == 1 # amd64
|
|
assert len(skipped) == 2 # arm64, ppc64le
|
|
|
|
def test_manifest_list_missing_required_arch_raises(self, manifest_list_config):
|
|
"""Test that missing required architecture raises exception."""
|
|
list_class, bytes_factory, manifest_class, child_manifest_bytes = manifest_list_config
|
|
manifest_list_bytes = bytes_factory()
|
|
|
|
# Only provide arm64, but amd64 is required
|
|
retriever = ContentRetrieverForTesting(
|
|
{
|
|
"sha256:arm64manifest": child_manifest_bytes,
|
|
}
|
|
)
|
|
|
|
manifest_list = list_class(Bytes.for_string_or_unicode(manifest_list_bytes))
|
|
|
|
config = {
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"],
|
|
}
|
|
with patch("data.model.config.app_config", config):
|
|
manifests = manifest_list.manifests(retriever)
|
|
|
|
# Should raise when accessing the amd64 manifest
|
|
with pytest.raises(ManifestException):
|
|
for m in manifests:
|
|
if m.architecture == "amd64":
|
|
_ = m.manifest_obj
|
|
|
|
def test_manifest_list_validate_skips_none_manifests(self, manifest_list_config):
|
|
"""Test that validation skips None manifests from sparse index."""
|
|
list_class, bytes_factory, manifest_class, child_manifest_bytes = manifest_list_config
|
|
manifest_list_bytes = bytes_factory()
|
|
|
|
# Only provide amd64
|
|
retriever = ContentRetrieverForTesting(
|
|
{
|
|
"sha256:amd64manifest": child_manifest_bytes,
|
|
}
|
|
)
|
|
|
|
manifest_list = list_class(Bytes.for_string_or_unicode(manifest_list_bytes))
|
|
|
|
config = {
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"],
|
|
}
|
|
with patch("data.model.config.app_config", config):
|
|
# Validation should not raise even though some manifests are None
|
|
manifest_list.validate(retriever)
|
|
|
|
|
|
class TestIsArchitectureRequiredMethod:
|
|
"""Tests for the _is_architecture_required method."""
|
|
|
|
def test_returns_true_when_sparse_disabled(self):
|
|
"""Returns True when FEATURE_SPARSE_INDEX is disabled."""
|
|
manifest_data = create_manifest_data(architecture="arm64")
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={"FEATURE_SPARSE_INDEX": False},
|
|
)
|
|
|
|
assert loader._is_architecture_required() is True
|
|
|
|
def test_returns_true_when_no_required_archs(self):
|
|
"""Returns True when SPARSE_INDEX_REQUIRED_ARCHS is empty."""
|
|
manifest_data = create_manifest_data(architecture="arm64")
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": [],
|
|
},
|
|
)
|
|
|
|
assert loader._is_architecture_required() is True
|
|
|
|
def test_returns_true_when_no_architecture(self):
|
|
"""Returns True when manifest has no architecture specified."""
|
|
manifest_data = create_manifest_data() # No architecture
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"],
|
|
},
|
|
)
|
|
|
|
assert loader._is_architecture_required() is True
|
|
|
|
def test_returns_true_when_in_required_list(self):
|
|
"""Returns True when architecture is in required list."""
|
|
manifest_data = create_manifest_data(architecture="amd64")
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64", "arm64"],
|
|
},
|
|
)
|
|
|
|
assert loader._is_architecture_required() is True
|
|
|
|
def test_returns_false_when_not_in_required_list(self):
|
|
"""Returns False when architecture is not in required list."""
|
|
manifest_data = create_manifest_data(architecture="ppc64le")
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64", "arm64"],
|
|
},
|
|
)
|
|
|
|
assert loader._is_architecture_required() is False
|
|
|
|
|
|
class TestDebugLogging:
|
|
"""Tests for debug logging behavior."""
|
|
|
|
def test_logs_debug_message_for_skipped_manifest(self, caplog):
|
|
"""Test that debug message is logged when manifest is skipped."""
|
|
manifest_data = create_manifest_data(
|
|
digest="sha256:testdigest",
|
|
architecture="arm64",
|
|
)
|
|
retriever = ContentRetrieverForTesting()
|
|
|
|
loader = LazyManifestLoader(
|
|
manifest_data,
|
|
retriever,
|
|
SUPPORTED_TYPES,
|
|
digest_key="digest",
|
|
size_key="size",
|
|
media_type_key="mediaType",
|
|
app_config={
|
|
"FEATURE_SPARSE_INDEX": True,
|
|
"SPARSE_INDEX_REQUIRED_ARCHS": ["amd64"],
|
|
},
|
|
)
|
|
|
|
with caplog.at_level(logging.DEBUG, logger="image.shared.schemautil"):
|
|
result = loader.manifest_obj
|
|
assert result is None
|
|
|
|
assert "Skipping manifest with digest" in caplog.text
|
|
assert "sha256:testdigest" in caplog.text
|
|
assert "arm64" in caplog.text
|