19145569596

Committed 06 Nov 2025 06:17PM UTC coverage: 86.819% (-0.03%) from 86.85%

Build # 19145569596

Build Type

Pull #1101

github

Committed by

web-flow

Commit Message

Merge 252d68849 into fabb4724c

Pull Request Pull Request #1101: feat: indicate supported platforms when checking a session image

Run Details

241 of 314 new or added lines in 12 files covered. (76.75%)

8 existing lines in 4 files now uncovered.

23054 of 26554 relevant lines covered (86.82%)

1.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

28.26

/components/renku_data_services/notebooks/oci/utils.py

"""Utilities for OCI images."""

from typing import TYPE_CHECKING

import httpx

from renku_data_services import errors
from renku_data_services.app_config import logging
from renku_data_services.notebooks.oci.image_config import ImageConfig
from renku_data_services.notebooks.oci.image_index import ImageIndex
from renku_data_services.notebooks.oci.image_manifest import ImageManifest
from renku_data_services.notebooks.oci.models import ManifestMediaTypes, Platform

if TYPE_CHECKING:
    from renku_data_services.notebooks.api.classes.image import Image, ImageRepoDockerAPI

logger = logging.getLogger(__name__)


async def get_image_platforms(
    manifest_response: httpx.Response, image: "Image", reg_api: "ImageRepoDockerAPI"
) -> list[Platform] | None:
    """Returns the list of platforms supported by the image manifest."""
    try:
        parsed = parse_manifest_response(manifest_response)
    except Exception as err:
        logger.warning(f"Error parsing image manifest: {err}")
        return None

    if isinstance(parsed, ImageIndex):
        platforms: list[Platform] = []
        for manifest in parsed.manifests:
            # Ignore manifests without a platform
            if (
                manifest.platform is None
                or manifest.platform.os == "unknown"
                or manifest.platform.architecture == "unknown"
            ):
                continue
            platforms.append(
                Platform(
                    architecture=manifest.platform.architecture,
                    os=manifest.platform.os,
                    os_features=list(manifest.platform.os_features) if manifest.platform.os_features else None,
                    os_version=manifest.platform.os_version,
                    variant=manifest.platform.variant,
                )
            )
        platforms = sorted(set(platforms))
        return platforms

    try:
        config_response = await reg_api.get_image_config_from_digest(image=image, config_digest=parsed.config.digest)
    except Exception as err:
        logger.warning(f"Error getting image config: {err}")
        return None

    try:
        parsed_config = parse_config_response(config_response)
    except Exception as err:
        logger.warning(f"Error parsing image config: {err}")
        return None

    platform = Platform(
        architecture=parsed_config.architecture,
        os=parsed_config.os,
        os_features=list(parsed_config.os_features) if parsed_config.os_features else None,
        os_version=parsed_config.os_version,
        variant=parsed_config.variant,
    )
    return [platform]


def parse_manifest_response(response: httpx.Response) -> ImageIndex | ImageManifest:
    """Parse a manifest response."""
    content_type = response.headers.get("Content-Type")
    if content_type not in {
        ManifestMediaTypes.docker_list_v2,
        ManifestMediaTypes.docker_manifest_v2,
        ManifestMediaTypes.oci_index_v1,
        ManifestMediaTypes.oci_manifest_v1,
    }:
        raise errors.ValidationError(message=f"Unexpected content type {content_type}.")

    if content_type in {ManifestMediaTypes.docker_list_v2, ManifestMediaTypes.oci_index_v1}:
        return ImageIndex.model_validate_json(response.content)
    else:
        return ImageManifest.model_validate_json(response.content)


def parse_config_response(response: httpx.Response) -> ImageConfig:
    """Parse a config response."""
    # NOTE: it seems that the "Content-Type" is set to "application/octet-stream" by the Docker registry
    # NOTE: so we do not validate it
    return ImageConfig.model_validate_json(response.content)

1	"""Utilities for OCI images."""
2
3	from typing import TYPE_CHECKING	2✔
4
5	import httpx	2✔
6
7	from renku_data_services import errors	2✔
8	from renku_data_services.app_config import logging	2✔
9	from renku_data_services.notebooks.oci.image_config import ImageConfig	2✔
10	from renku_data_services.notebooks.oci.image_index import ImageIndex	2✔
11	from renku_data_services.notebooks.oci.image_manifest import ImageManifest	2✔
12	from renku_data_services.notebooks.oci.models import ManifestMediaTypes, Platform	2✔
13
14	if TYPE_CHECKING:	2✔
NEW 15	from renku_data_services.notebooks.api.classes.image import Image, ImageRepoDockerAPI	×
16
17	logger = logging.getLogger(__name__)	2✔
18
19
20	async def get_image_platforms(	2✔
21	manifest_response: httpx.Response, image: "Image", reg_api: "ImageRepoDockerAPI"
22	) -> list[Platform] \| None:
23	"""Returns the list of platforms supported by the image manifest."""
NEW 24	try:	×
NEW 25	parsed = parse_manifest_response(manifest_response)	×
NEW 26	except Exception as err:	×
NEW 27	logger.warning(f"Error parsing image manifest: {err}")	×
NEW 28	return None	×
29
NEW 30	if isinstance(parsed, ImageIndex):	×
NEW 31	platforms: list[Platform] = []	×
NEW 32	for manifest in parsed.manifests:	×
33	# Ignore manifests without a platform
NEW 34	if (	×
35	manifest.platform is None
36	or manifest.platform.os == "unknown"
37	or manifest.platform.architecture == "unknown"
38	):
NEW 39	continue	×
NEW 40	platforms.append(	×
41	Platform(
42	architecture=manifest.platform.architecture,
43	os=manifest.platform.os,
44	os_features=list(manifest.platform.os_features) if manifest.platform.os_features else None,
45	os_version=manifest.platform.os_version,
46	variant=manifest.platform.variant,
47	)
48	)
NEW 49	platforms = sorted(set(platforms))	×
NEW 50	return platforms	×
51
NEW 52	try:	×
NEW 53	config_response = await reg_api.get_image_config_from_digest(image=image, config_digest=parsed.config.digest)	×
NEW 54	except Exception as err:	×
NEW 55	logger.warning(f"Error getting image config: {err}")	×
NEW 56	return None	×
57
NEW 58	try:	×
NEW 59	parsed_config = parse_config_response(config_response)	×
NEW 60	except Exception as err:	×
NEW 61	logger.warning(f"Error parsing image config: {err}")	×
NEW 62	return None	×
63
NEW 64	platform = Platform(	×
65	architecture=parsed_config.architecture,
66	os=parsed_config.os,
67	os_features=list(parsed_config.os_features) if parsed_config.os_features else None,
68	os_version=parsed_config.os_version,
69	variant=parsed_config.variant,
70	)
NEW 71	return [platform]	×
72
73
74	def parse_manifest_response(response: httpx.Response) -> ImageIndex \| ImageManifest:	2✔
75	"""Parse a manifest response."""
NEW 76	content_type = response.headers.get("Content-Type")	×
NEW 77	if content_type not in {	×
78	ManifestMediaTypes.docker_list_v2,
79	ManifestMediaTypes.docker_manifest_v2,
80	ManifestMediaTypes.oci_index_v1,
81	ManifestMediaTypes.oci_manifest_v1,
82	}:
NEW 83	raise errors.ValidationError(message=f"Unexpected content type {content_type}.")	×
84
NEW 85	if content_type in {ManifestMediaTypes.docker_list_v2, ManifestMediaTypes.oci_index_v1}:	×
NEW 86	return ImageIndex.model_validate_json(response.content)	×
87	else:
NEW 88	return ImageManifest.model_validate_json(response.content)	×
89
90
91	def parse_config_response(response: httpx.Response) -> ImageConfig:	2✔
92	"""Parse a config response."""
93	# NOTE: it seems that the "Content-Type" is set to "application/octet-stream" by the Docker registry
94	# NOTE: so we do not validate it
NEW 95	return ImageConfig.model_validate_json(response.content)	×

SwissDataScienceCenter / renku-data-services / 19145569596

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous