18840025735

Committed 27 Oct 2025 11:53AM UTC coverage: 86.84% (+0.03%) from 86.806%

Build # 18840025735

Build Type

Pull #1083

github

Committed by

web-flow

Commit Message

Merge 7a8145fff into 9cffdf9a6

Pull Request Pull Request #1083: fix: do not pass awaitable in __get_gitlab_image_pull_secret

Run Details

1 of 3 new or added lines in 1 file covered. (33.33%)

4 existing lines in 3 files now uncovered.

22727 of 26171 relevant lines covered (86.84%)

1.52 hits per line

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

77.96

/components/renku_data_services/crc/core.py

"""crc modules converters and validators."""

from typing import Literal, overload
from urllib.parse import urlparse

from ulid import ULID

from renku_data_services.base_models import RESET, ResetType
from renku_data_services.crc import apispec, models
from renku_data_services.errors import errors


def validate_quota(body: apispec.QuotaWithOptionalId) -> models.UnsavedQuota:
    """Validate a quota object."""
    return models.UnsavedQuota(
        cpu=body.cpu,
        memory=body.memory,
        gpu=body.gpu,
    )


def validate_quota_put_patch(body: apispec.QuotaWithId | apispec.QuotaPatch) -> models.QuotaPatch:
    """Validate the put request for a quota."""
    return models.QuotaPatch(
        cpu=body.cpu,
        memory=body.memory,
        gpu=body.gpu,
    )


def validate_resource_class(body: apispec.ResourceClass) -> models.UnsavedResourceClass:
    """Validate a resource class object."""
    if len(body.name) > 40:
        # TODO: Should this be added to the API spec instead?
        raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")
    if body.default_storage > body.max_storage:
        raise errors.ValidationError(message="The default storage cannot be larger than the max allowable storage.")
    # We need to sort node affinities and tolerations to make '__eq__' reliable
    node_affinities = sorted(
        (
            models.NodeAffinity(key=na.key, required_during_scheduling=na.required_during_scheduling)
            for na in body.node_affinities or []
        ),
        key=lambda x: (x.key, x.required_during_scheduling),
    )
    tolerations = sorted(t.root for t in body.tolerations or [])
    return models.UnsavedResourceClass(
        name=body.name,
        cpu=body.cpu,
        memory=body.memory,
        max_storage=body.max_storage,
        gpu=body.gpu,
        default=body.default,
        default_storage=body.default_storage,
        node_affinities=node_affinities,
        tolerations=tolerations,
    )


@overload
def validate_resource_class_patch_or_put(
    body: apispec.ResourceClassPatch, method: Literal["PATCH"]
) -> models.ResourceClassPatch: ...
@overload
def validate_resource_class_patch_or_put(
    body: apispec.ResourceClassPatchWithId, method: Literal["PATCH"]
) -> models.ResourceClassPatchWithId: ...
@overload
def validate_resource_class_patch_or_put(
    body: apispec.ResourceClass, method: Literal["PUT"]
) -> models.ResourceClassPatch: ...
@overload
def validate_resource_class_patch_or_put(
    body: apispec.ResourceClassWithId, method: Literal["PUT"]
) -> models.ResourceClassPatchWithId: ...
def validate_resource_class_patch_or_put(
    body: apispec.ResourceClassPatch
    | apispec.ResourceClassPatchWithId
    | apispec.ResourceClass
    | apispec.ResourceClassWithId,
    method: Literal["PATCH", "PUT"],
) -> models.ResourceClassPatch | models.ResourceClassPatchWithId:
    """Validate the patch to a resource class."""
    rc_id = body.id if isinstance(body, (apispec.ResourceClassPatchWithId, apispec.ResourceClassWithId)) else None
    node_affinities: list[models.NodeAffinity] | None = [] if method == "PUT" else None
    if body.node_affinities:
        node_affinities = sorted(
            (
                models.NodeAffinity(key=na.key, required_during_scheduling=na.required_during_scheduling)
                for na in body.node_affinities or []
            ),
            key=lambda x: (x.key, x.required_during_scheduling),
        )
    tolerations: list[str] | None = [] if method == "PUT" else None
    if body.tolerations:
        tolerations = sorted(t.root for t in body.tolerations or [])
    if rc_id:
        return models.ResourceClassPatchWithId(
            id=rc_id,
            name=body.name,
            cpu=body.cpu,
            memory=body.memory,
            max_storage=body.max_storage,
            gpu=body.gpu,
            default=body.default,
            default_storage=body.default_storage,
            node_affinities=node_affinities,
            tolerations=tolerations,
        )
    return models.ResourceClassPatch(
        name=body.name,
        cpu=body.cpu,
        memory=body.memory,
        max_storage=body.max_storage,
        gpu=body.gpu,
        default=body.default,
        default_storage=body.default_storage,
        node_affinities=node_affinities,
        tolerations=tolerations,
    )


def validate_resource_class_update(
    existing: models.ResourceClass,
    update: models.ResourceClassPatch,
) -> None:
    """Validate the update to a resource class."""
    name = update.name if update.name is not None else existing.name
    max_storage = update.max_storage if update.max_storage is not None else existing.max_storage
    default_storage = update.default_storage if update.default_storage is not None else existing.default_storage

    if update.default is not None and existing.default != update.default:
        raise errors.ValidationError(message="Changing the default class in a resource pool is not supported.")

    if len(name) > 40:
        # TODO: Should this be added to the API spec instead?
        raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")
    if default_storage > max_storage:
        raise errors.ValidationError(message="The default storage cannot be larger than the max allowable storage.")


def validate_resource_pool(body: apispec.ResourcePool) -> models.UnsavedResourcePool:
    """Validate a resource pool object."""
    if len(body.name) > 40:
        # TODO: Should this be added to the API spec instead?
        raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")
    if body.default and not body.public:
        raise errors.ValidationError(message="The default resource pool has to be public.")
    if body.default and body.quota is not None:
        raise errors.ValidationError(message="A default resource pool cannot have a quota.")
    if body.remote and body.default:
        raise errors.ValidationError(message="The default resource pool cannot start remote sessions.")
    if body.remote and body.public:
        raise errors.ValidationError(message="A resource pool which starts remote sessions cannot be public.")
    if (body.idle_threshold and body.idle_threshold < 0) or (
        body.hibernation_threshold and body.hibernation_threshold < 0
    ):
        raise errors.ValidationError(message="Idle threshold and hibernation threshold need to be larger than 0.")

    idle_threshold = body.idle_threshold
    if idle_threshold == 0:
        idle_threshold = None
    hibernation_threshold = body.hibernation_threshold
    if hibernation_threshold == 0:
        hibernation_threshold = None

    quota = validate_quota(body=body.quota) if body.quota else None
    classes = [validate_resource_class(body=new_cls) for new_cls in body.classes]

    default_classes: list[models.UnsavedResourceClass] = []
    for cls in classes:
        if quota is not None and not quota.is_resource_class_compatible(cls):
            raise errors.ValidationError(
                message=f"The resource class with name {cls.name} is not compatible with the quota."
            )
        if cls.default:
            default_classes.append(cls)
    if len(default_classes) != 1:
        raise errors.ValidationError(message="One default class is required in each resource pool.")

    remote = validate_remote(body=body.remote) if body.remote else None

    return models.UnsavedResourcePool(
        name=body.name,
        classes=classes,
        quota=quota,
        idle_threshold=idle_threshold,
        hibernation_threshold=hibernation_threshold,
        default=body.default,
        public=body.public,
        remote=remote,
        cluster_id=ULID.from_str(body.cluster_id) if body.cluster_id else None,
    )


def validate_resource_pool_patch(body: apispec.ResourcePoolPatch) -> models.ResourcePoolPatch:
    """Validate the patch to a resource pool."""
    classes = (
        [validate_resource_class_patch_or_put(body=rc, method="PATCH") for rc in body.classes] if body.classes else None
    )
    quota = validate_quota_put_patch(body=body.quota) if body.quota else None
    remote = validate_remote_patch(body=body.remote) if body.remote else None
    return models.ResourcePoolPatch(
        name=body.name,
        classes=classes,
        quota=quota,
        idle_threshold=body.idle_threshold,
        hibernation_threshold=body.hibernation_threshold,
        default=body.default,
        public=body.public,
        remote=remote,
        cluster_id=ULID.from_str(body.cluster_id) if body.cluster_id else None,
    )


def validate_resource_pool_put(body: apispec.ResourcePoolPut) -> models.ResourcePoolPatch:
    """Validate the put request for a resource pool."""
    # NOTE: the current behavior is to do a PATCH-style update on nested classes for "PUT resource pool"
    classes = (
        [validate_resource_class_patch_or_put(body=rc, method="PUT") for rc in body.classes] if body.classes else None
    )
    quota = validate_quota_put_patch(body=body.quota) if body.quota else RESET
    remote = validate_remote_put(body=body.remote)
    return models.ResourcePoolPatch(
        name=body.name,
        classes=classes,
        quota=quota,
        idle_threshold=body.idle_threshold,
        hibernation_threshold=body.hibernation_threshold,
        default=body.default,
        public=body.public,
        remote=remote,
        cluster_id=ULID.from_str(body.cluster_id) if body.cluster_id else RESET,
    )


def validate_resource_pool_update(existing: models.ResourcePool, update: models.ResourcePoolPatch) -> None:
    """Validate the update to a resource pool."""
    name = update.name if update.name is not None else existing.name
    classes = existing.classes
    for rc in update.classes or []:
        found = next(filter(lambda tup: tup[1].id == rc.id, enumerate(classes)), None)
        if found is None:
            raise errors.ValidationError(
                message=f"Resource class '{rc.id}' does not exist in resource pool '{existing.id}'."
            )
        idx, existing_rc = found
        classes[idx] = models.ResourceClass(
            name=rc.name if rc.name is not None else existing_rc.name,
            cpu=rc.cpu if rc.cpu is not None else existing_rc.cpu,
            memory=rc.memory if rc.memory is not None else existing_rc.memory,
            max_storage=rc.max_storage if rc.max_storage is not None else existing_rc.max_storage,
            gpu=rc.gpu if rc.gpu is not None else existing_rc.gpu,
            id=existing_rc.id,
            default=rc.default if rc.default is not None else existing_rc.default,
            default_storage=rc.default_storage if rc.default_storage is not None else existing_rc.default_storage,
            node_affinities=rc.node_affinities if rc.node_affinities is not None else existing_rc.node_affinities,
            tolerations=rc.tolerations if rc.tolerations is not None else existing_rc.tolerations,
        )
    quota: models.Quota | models.UnsavedQuota | ResetType = existing.quota if existing.quota else RESET
    if update.quota is RESET:
        quota = RESET
    elif isinstance(update.quota, models.QuotaPatch) and existing.quota is None:
        # The quota patch needs to contain all required fields
        cpu = update.quota.cpu
        if cpu is None:
            raise errors.ValidationError(message="The 'quota.cpu' field is required when creating a new quota.")
        memory = update.quota.memory
        if memory is None:
            raise errors.ValidationError(message="The 'quota.memory' field is required when creating a new quota.")
        gpu = update.quota.gpu
        if gpu is None:
            raise errors.ValidationError(message="The 'quota.gpu' field is required when creating a new quota.")
        quota = models.UnsavedQuota(
            cpu=cpu,
            memory=memory,
            gpu=gpu,
        )
    elif isinstance(update.quota, models.QuotaPatch):
        assert existing.quota is not None
        quota = models.Quota(
            cpu=update.quota.cpu if update.quota.cpu is not None else existing.quota.cpu,
            memory=update.quota.memory if update.quota.memory is not None else existing.quota.memory,
            gpu=update.quota.gpu if update.quota.gpu is not None else existing.quota.gpu,
            gpu_kind=update.quota.gpu_kind if update.quota.gpu_kind is not None else existing.quota.gpu_kind,
            id=existing.quota.id,
        )
    idle_threshold = update.idle_threshold if update.idle_threshold is not None else existing.idle_threshold
    hibernation_threshold = (
        update.hibernation_threshold if update.hibernation_threshold is not None else existing.hibernation_threshold
    )
    default = update.default if update.default is not None else existing.default
    public = update.public if update.public is not None else existing.public
    remote: models.RemoteConfigurationFirecrest | ResetType = existing.remote if existing.remote else RESET
    if update.remote is RESET:
        remote = RESET
    elif isinstance(update.remote, models.RemoteConfigurationFirecrestPatch) and existing.remote is None:
        # The remote patch needs to contain all required fields
        kind = update.remote.kind
        if kind is None:
            raise errors.ValidationError(message="The 'remote.kind' field is required when creating a new remote.")
        api_url = update.remote.api_url
        if api_url is None:
            raise errors.ValidationError(message="The 'remote.api_url' field is required when creating a new remote.")
        system_name = update.remote.system_name
        if system_name is None:
            raise errors.ValidationError(
                message="The 'remote.system_name' field is required when creating a new remote."
            )
        remote = models.RemoteConfigurationFirecrest(
            kind=kind,
            provider_id=update.remote.provider_id,
            api_url=api_url,
            system_name=system_name,
            partition=update.remote.partition,
        )
    elif isinstance(update.remote, models.RemoteConfigurationFirecrestPatch):
        assert existing.remote is not None
        remote = models.RemoteConfigurationFirecrest(
            kind=update.remote.kind if update.remote.kind is not None else existing.remote.kind,
            provider_id=update.remote.provider_id
            if update.remote.provider_id is not None
            else existing.remote.provider_id,
            api_url=update.remote.api_url if update.remote.api_url is not None else existing.remote.api_url,
            system_name=update.remote.system_name
            if update.remote.system_name is not None
            else existing.remote.system_name,
            partition=update.remote.partition if update.remote.partition is not None else existing.remote.partition,
        )

    if len(name) > 40:
        # TODO: Should this be added to the API spec instead?
        raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")
    if default and not public:
        raise errors.ValidationError(message="The default resource pool has to be public.")
    if default and quota is not RESET:
        raise errors.ValidationError(message="A default resource pool cannot have a quota.")
    if isinstance(remote, models.RemoteConfigurationFirecrest) and default:
        raise errors.ValidationError(message="The default resource pool cannot start remote sessions.")
    if isinstance(remote, models.RemoteConfigurationFirecrest) and public:
        raise errors.ValidationError(message="A resource pool which starts remote sessions cannot be public.")
    if (idle_threshold and idle_threshold < 0) or (hibernation_threshold and hibernation_threshold < 0):
        raise errors.ValidationError(message="Idle threshold and hibernation threshold need to be larger than 0.")

    default_classes: list[models.ResourceClass] = []
    for cls in classes:
        if (isinstance(quota, (models.Quota, models.UnsavedQuota))) and not quota.is_resource_class_compatible(cls):
            raise errors.ValidationError(
                message=f"The resource class with name {cls.name} is not compatible with the quota."
            )
        if cls.default:
            default_classes.append(cls)
    if len(default_classes) != 1:
        raise errors.ValidationError(message="One default class is required in each resource pool.")


def validate_cluster(body: apispec.Cluster) -> models.ClusterSettings:
    """Convert a REST API Cluster object to a model Cluster object."""
    return models.ClusterSettings(
        name=body.name,
        config_name=body.config_name,
        session_protocol=models.SessionProtocol(body.session_protocol.value),
        session_host=body.session_host,
        session_port=body.session_port,
        session_path=body.session_path,
        session_ingress_class_name=body.session_ingress_class_name,
        session_ingress_annotations=body.session_ingress_annotations.model_dump(),
        session_tls_secret_name=body.session_tls_secret_name,
        session_storage_class=body.session_storage_class,
        service_account_name=body.service_account_name,
    )


def validate_cluster_patch(patch: apispec.ClusterPatch) -> models.ClusterPatch:
    """Convert a REST API Cluster object patch to a model Cluster object."""

    return models.ClusterPatch(
        name=patch.name,
        config_name=patch.config_name,
        session_protocol=models.SessionProtocol(patch.session_protocol.value)
        if patch.session_protocol is not None
        else None,
        session_host=patch.session_host,
        session_port=patch.session_port,
        session_path=patch.session_path,
        session_ingress_class_name=patch.session_ingress_class_name,
        session_ingress_annotations=patch.session_ingress_annotations.model_dump()
        if patch.session_ingress_annotations is not None
        else None,
        session_tls_secret_name=patch.session_tls_secret_name,
        session_storage_class=patch.session_storage_class,
        service_account_name=patch.service_account_name,
    )


def validate_remote(body: apispec.RemoteConfigurationFirecrest) -> models.RemoteConfigurationFirecrest:
    """Validate a remote configuration object."""
    kind = models.RemoteConfigurationKind(body.kind.value)
    if kind != models.RemoteConfigurationKind.firecrest:
        raise errors.ValidationError(message=f"The kind '{kind}' of remote configuration is not supported.", quiet=True)
    validate_firecrest_api_url(body.api_url)
    return models.RemoteConfigurationFirecrest(
        kind=kind,
        provider_id=body.provider_id,
        api_url=body.api_url,
        system_name=body.system_name,
        partition=body.partition,
    )


def validate_remote_put(
    body: apispec.RemoteConfigurationFirecrest | None,
) -> models.RemoteConfigurationPatch:
    """Validate the PUT update to a remote configuration object."""
    if body is None:
        return RESET
    remote = validate_remote(body=body)
    return models.RemoteConfigurationFirecrestPatch(
        kind=remote.kind,
        provider_id=remote.provider_id,
        api_url=remote.api_url,
        system_name=remote.system_name,
        partition=remote.partition,
    )


def validate_remote_patch(
    body: apispec.RemoteConfigurationPatchReset | apispec.RemoteConfigurationFirecrestPatch,
) -> models.RemoteConfigurationPatch:
    """Validate the patch to a remote configuration object."""
    if isinstance(body, apispec.RemoteConfigurationPatchReset):
        return RESET
    kind = models.RemoteConfigurationKind(body.kind.value) if body.kind else None
    if kind and kind != models.RemoteConfigurationKind.firecrest:
        raise errors.ValidationError(message=f"The kind '{kind}' of remote configuration is not supported.", quiet=True)
    if body.api_url:
        validate_firecrest_api_url(body.api_url)
    return models.RemoteConfigurationFirecrestPatch(
        kind=kind,
        provider_id=body.provider_id,
        api_url=body.api_url,
        system_name=body.system_name,
        partition=body.partition,
    )


def validate_firecrest_api_url(url: str) -> None:
    """Validate the URL to the FirecREST API."""
    parsed = urlparse(url)
    if not parsed.netloc:
        raise errors.ValidationError(
            message=f"The host for the firecrest api url {url} is not valid, expected a non-empty value.",
            quiet=True,
        )
    accepted_schemes = ["https"]
    if parsed.scheme not in accepted_schemes:
        raise errors.ValidationError(
            message=f"The scheme for the firecrest api url {url} is not valid, expected one of {accepted_schemes}",
            quiet=True,
        )

1	"""crc modules converters and validators."""
2
3	from typing import Literal, overload	2✔
4	from urllib.parse import urlparse	2✔
5
6	from ulid import ULID	2✔
7
8	from renku_data_services.base_models import RESET, ResetType	2✔
9	from renku_data_services.crc import apispec, models	2✔
10	from renku_data_services.errors import errors	2✔
11
12
13	def validate_quota(body: apispec.QuotaWithOptionalId) -> models.UnsavedQuota:	2✔
14	"""Validate a quota object."""
15	return models.UnsavedQuota(	2✔
16	cpu=body.cpu,
17	memory=body.memory,
18	gpu=body.gpu,
19	)
20
21
22	def validate_quota_put_patch(body: apispec.QuotaWithId \| apispec.QuotaPatch) -> models.QuotaPatch:	2✔
23	"""Validate the put request for a quota."""
24	return models.QuotaPatch(	2✔
25	cpu=body.cpu,
26	memory=body.memory,
27	gpu=body.gpu,
28	)
29
30
31	def validate_resource_class(body: apispec.ResourceClass) -> models.UnsavedResourceClass:	2✔
32	"""Validate a resource class object."""
33	if len(body.name) > 40:	2✔
34	# TODO: Should this be added to the API spec instead?
35	raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")	×
36	if body.default_storage > body.max_storage:	2✔
37	raise errors.ValidationError(message="The default storage cannot be larger than the max allowable storage.")	1✔
38	# We need to sort node affinities and tolerations to make '__eq__' reliable
39	node_affinities = sorted(	2✔
40	(
41	models.NodeAffinity(key=na.key, required_during_scheduling=na.required_during_scheduling)
42	for na in body.node_affinities or []
43	),
44	key=lambda x: (x.key, x.required_during_scheduling),
45	)
46	tolerations = sorted(t.root for t in body.tolerations or [])	2✔
47	return models.UnsavedResourceClass(	2✔
48	name=body.name,
49	cpu=body.cpu,
50	memory=body.memory,
51	max_storage=body.max_storage,
52	gpu=body.gpu,
53	default=body.default,
54	default_storage=body.default_storage,
55	node_affinities=node_affinities,
56	tolerations=tolerations,
57	)
58
59
60	@overload	2✔
61	def validate_resource_class_patch_or_put(	2✔
62	body: apispec.ResourceClassPatch, method: Literal["PATCH"]
63	) -> models.ResourceClassPatch: ...
64	@overload	2✔
65	def validate_resource_class_patch_or_put(	2✔
66	body: apispec.ResourceClassPatchWithId, method: Literal["PATCH"]
67	) -> models.ResourceClassPatchWithId: ...
68	@overload	2✔
69	def validate_resource_class_patch_or_put(	2✔
70	body: apispec.ResourceClass, method: Literal["PUT"]
71	) -> models.ResourceClassPatch: ...
72	@overload	2✔
73	def validate_resource_class_patch_or_put(	2✔
74	body: apispec.ResourceClassWithId, method: Literal["PUT"]
75	) -> models.ResourceClassPatchWithId: ...
76	def validate_resource_class_patch_or_put(	2✔
77	body: apispec.ResourceClassPatch
78	\| apispec.ResourceClassPatchWithId
79	\| apispec.ResourceClass
80	\| apispec.ResourceClassWithId,
81	method: Literal["PATCH", "PUT"],
82	) -> models.ResourceClassPatch \| models.ResourceClassPatchWithId:
83	"""Validate the patch to a resource class."""
84	rc_id = body.id if isinstance(body, (apispec.ResourceClassPatchWithId, apispec.ResourceClassWithId)) else None	2✔
85	node_affinities: list[models.NodeAffinity] \| None = [] if method == "PUT" else None	2✔
86	if body.node_affinities:	2✔
87	node_affinities = sorted(	2✔
88	(
89	models.NodeAffinity(key=na.key, required_during_scheduling=na.required_during_scheduling)
90	for na in body.node_affinities or []
91	),
92	key=lambda x: (x.key, x.required_during_scheduling),
93	)
94	tolerations: list[str] \| None = [] if method == "PUT" else None	2✔
95	if body.tolerations:	2✔
96	tolerations = sorted(t.root for t in body.tolerations or [])	2✔
97	if rc_id:	2✔
98	return models.ResourceClassPatchWithId(	2✔
99	id=rc_id,
100	name=body.name,
101	cpu=body.cpu,
102	memory=body.memory,
103	max_storage=body.max_storage,
104	gpu=body.gpu,
105	default=body.default,
106	default_storage=body.default_storage,
107	node_affinities=node_affinities,
108	tolerations=tolerations,
109	)
110	return models.ResourceClassPatch(	2✔
111	name=body.name,
112	cpu=body.cpu,
113	memory=body.memory,
114	max_storage=body.max_storage,
115	gpu=body.gpu,
116	default=body.default,
117	default_storage=body.default_storage,
118	node_affinities=node_affinities,
119	tolerations=tolerations,
120	)
121
122
123	def validate_resource_class_update(	2✔
124	existing: models.ResourceClass,
125	update: models.ResourceClassPatch,
126	) -> None:
127	"""Validate the update to a resource class."""
128	name = update.name if update.name is not None else existing.name	1✔
129	max_storage = update.max_storage if update.max_storage is not None else existing.max_storage	1✔
130	default_storage = update.default_storage if update.default_storage is not None else existing.default_storage	1✔
131
132	if update.default is not None and existing.default != update.default:	1✔
133	raise errors.ValidationError(message="Changing the default class in a resource pool is not supported.")	×
134
135	if len(name) > 40:	1✔
136	# TODO: Should this be added to the API spec instead?
137	raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")	×
138	if default_storage > max_storage:	1✔
139	raise errors.ValidationError(message="The default storage cannot be larger than the max allowable storage.")	×
140
141
142	def validate_resource_pool(body: apispec.ResourcePool) -> models.UnsavedResourcePool:	2✔
143	"""Validate a resource pool object."""
144	if len(body.name) > 40:	2✔
145	# TODO: Should this be added to the API spec instead?
146	raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")	×
147	if body.default and not body.public:	2✔
UNCOV 148	raise errors.ValidationError(message="The default resource pool has to be public.")	×
149	if body.default and body.quota is not None:	2✔
150	raise errors.ValidationError(message="A default resource pool cannot have a quota.")	×
151	if body.remote and body.default:	2✔
152	raise errors.ValidationError(message="The default resource pool cannot start remote sessions.")	×
153	if body.remote and body.public:	2✔
154	raise errors.ValidationError(message="A resource pool which starts remote sessions cannot be public.")	×
155	if (body.idle_threshold and body.idle_threshold < 0) or (	2✔
156	body.hibernation_threshold and body.hibernation_threshold < 0
157	):
158	raise errors.ValidationError(message="Idle threshold and hibernation threshold need to be larger than 0.")	×
159
160	idle_threshold = body.idle_threshold	2✔
161	if idle_threshold == 0:	2✔
162	idle_threshold = None	×
163	hibernation_threshold = body.hibernation_threshold	2✔
164	if hibernation_threshold == 0:	2✔
165	hibernation_threshold = None	×
166
167	quota = validate_quota(body=body.quota) if body.quota else None	2✔
168	classes = [validate_resource_class(body=new_cls) for new_cls in body.classes]	2✔
169
170	default_classes: list[models.UnsavedResourceClass] = []	2✔
171	for cls in classes:	2✔
172	if quota is not None and not quota.is_resource_class_compatible(cls):	2✔
173	raise errors.ValidationError(	1✔
174	message=f"The resource class with name {cls.name} is not compatible with the quota."
175	)
176	if cls.default:	2✔
177	default_classes.append(cls)	1✔
178	if len(default_classes) != 1:	2✔
179	raise errors.ValidationError(message="One default class is required in each resource pool.")	1✔
180
181	remote = validate_remote(body=body.remote) if body.remote else None	1✔
182
183	return models.UnsavedResourcePool(	1✔
184	name=body.name,
185	classes=classes,
186	quota=quota,
187	idle_threshold=idle_threshold,
188	hibernation_threshold=hibernation_threshold,
189	default=body.default,
190	public=body.public,
191	remote=remote,
192	cluster_id=ULID.from_str(body.cluster_id) if body.cluster_id else None,
193	)
194
195
196	def validate_resource_pool_patch(body: apispec.ResourcePoolPatch) -> models.ResourcePoolPatch:	2✔
197	"""Validate the patch to a resource pool."""
198	classes = (	2✔
199	[validate_resource_class_patch_or_put(body=rc, method="PATCH") for rc in body.classes] if body.classes else None
200	)
201	quota = validate_quota_put_patch(body=body.quota) if body.quota else None	2✔
202	remote = validate_remote_patch(body=body.remote) if body.remote else None	2✔
203	return models.ResourcePoolPatch(	2✔
204	name=body.name,
205	classes=classes,
206	quota=quota,
207	idle_threshold=body.idle_threshold,
208	hibernation_threshold=body.hibernation_threshold,
209	default=body.default,
210	public=body.public,
211	remote=remote,
212	cluster_id=ULID.from_str(body.cluster_id) if body.cluster_id else None,
213	)
214
215
216	def validate_resource_pool_put(body: apispec.ResourcePoolPut) -> models.ResourcePoolPatch:	2✔
217	"""Validate the put request for a resource pool."""
218	# NOTE: the current behavior is to do a PATCH-style update on nested classes for "PUT resource pool"
219	classes = (	2✔
220	[validate_resource_class_patch_or_put(body=rc, method="PUT") for rc in body.classes] if body.classes else None
221	)
222	quota = validate_quota_put_patch(body=body.quota) if body.quota else RESET	2✔
223	remote = validate_remote_put(body=body.remote)	2✔
224	return models.ResourcePoolPatch(	2✔
225	name=body.name,
226	classes=classes,
227	quota=quota,
228	idle_threshold=body.idle_threshold,
229	hibernation_threshold=body.hibernation_threshold,
230	default=body.default,
231	public=body.public,
232	remote=remote,
233	cluster_id=ULID.from_str(body.cluster_id) if body.cluster_id else RESET,
234	)
235
236
237	def validate_resource_pool_update(existing: models.ResourcePool, update: models.ResourcePoolPatch) -> None:	2✔
238	"""Validate the update to a resource pool."""
239	name = update.name if update.name is not None else existing.name	1✔
240	classes = existing.classes	1✔
241	for rc in update.classes or []:	1✔
242	found = next(filter(lambda tup: tup[1].id == rc.id, enumerate(classes)), None)	1✔
243	if found is None:	1✔
244	raise errors.ValidationError(	×
245	message=f"Resource class '{rc.id}' does not exist in resource pool '{existing.id}'."
246	)
247	idx, existing_rc = found	1✔
248	classes[idx] = models.ResourceClass(	1✔
249	name=rc.name if rc.name is not None else existing_rc.name,
250	cpu=rc.cpu if rc.cpu is not None else existing_rc.cpu,
251	memory=rc.memory if rc.memory is not None else existing_rc.memory,
252	max_storage=rc.max_storage if rc.max_storage is not None else existing_rc.max_storage,
253	gpu=rc.gpu if rc.gpu is not None else existing_rc.gpu,
254	id=existing_rc.id,
255	default=rc.default if rc.default is not None else existing_rc.default,
256	default_storage=rc.default_storage if rc.default_storage is not None else existing_rc.default_storage,
257	node_affinities=rc.node_affinities if rc.node_affinities is not None else existing_rc.node_affinities,
258	tolerations=rc.tolerations if rc.tolerations is not None else existing_rc.tolerations,
259	)
260	quota: models.Quota \| models.UnsavedQuota \| ResetType = existing.quota if existing.quota else RESET	1✔
261	if update.quota is RESET:	1✔
262	quota = RESET	×
263	elif isinstance(update.quota, models.QuotaPatch) and existing.quota is None:	1✔
264	# The quota patch needs to contain all required fields
265	cpu = update.quota.cpu	×
266	if cpu is None:	×
267	raise errors.ValidationError(message="The 'quota.cpu' field is required when creating a new quota.")	×
268	memory = update.quota.memory	×
269	if memory is None:	×
270	raise errors.ValidationError(message="The 'quota.memory' field is required when creating a new quota.")	×
271	gpu = update.quota.gpu	×
272	if gpu is None:	×
273	raise errors.ValidationError(message="The 'quota.gpu' field is required when creating a new quota.")	×
274	quota = models.UnsavedQuota(	×
275	cpu=cpu,
276	memory=memory,
277	gpu=gpu,
278	)
279	elif isinstance(update.quota, models.QuotaPatch):	1✔
280	assert existing.quota is not None	1✔
281	quota = models.Quota(	1✔
282	cpu=update.quota.cpu if update.quota.cpu is not None else existing.quota.cpu,
283	memory=update.quota.memory if update.quota.memory is not None else existing.quota.memory,
284	gpu=update.quota.gpu if update.quota.gpu is not None else existing.quota.gpu,
285	gpu_kind=update.quota.gpu_kind if update.quota.gpu_kind is not None else existing.quota.gpu_kind,
286	id=existing.quota.id,
287	)
288	idle_threshold = update.idle_threshold if update.idle_threshold is not None else existing.idle_threshold	1✔
289	hibernation_threshold = (	1✔
290	update.hibernation_threshold if update.hibernation_threshold is not None else existing.hibernation_threshold
291	)
292	default = update.default if update.default is not None else existing.default	1✔
293	public = update.public if update.public is not None else existing.public	1✔
294	remote: models.RemoteConfigurationFirecrest \| ResetType = existing.remote if existing.remote else RESET	1✔
295	if update.remote is RESET:	1✔
296	remote = RESET	1✔
297	elif isinstance(update.remote, models.RemoteConfigurationFirecrestPatch) and existing.remote is None:	1✔
298	# The remote patch needs to contain all required fields
299	kind = update.remote.kind	1✔
300	if kind is None:	1✔
301	raise errors.ValidationError(message="The 'remote.kind' field is required when creating a new remote.")	×
302	api_url = update.remote.api_url	1✔
303	if api_url is None:	1✔
304	raise errors.ValidationError(message="The 'remote.api_url' field is required when creating a new remote.")	×
305	system_name = update.remote.system_name	1✔
306	if system_name is None:	1✔
307	raise errors.ValidationError(	×
308	message="The 'remote.system_name' field is required when creating a new remote."
309	)
310	remote = models.RemoteConfigurationFirecrest(	1✔
311	kind=kind,
312	provider_id=update.remote.provider_id,
313	api_url=api_url,
314	system_name=system_name,
315	partition=update.remote.partition,
316	)
317	elif isinstance(update.remote, models.RemoteConfigurationFirecrestPatch):	1✔
318	assert existing.remote is not None	×
319	remote = models.RemoteConfigurationFirecrest(	×
320	kind=update.remote.kind if update.remote.kind is not None else existing.remote.kind,
321	provider_id=update.remote.provider_id
322	if update.remote.provider_id is not None
323	else existing.remote.provider_id,
324	api_url=update.remote.api_url if update.remote.api_url is not None else existing.remote.api_url,
325	system_name=update.remote.system_name
326	if update.remote.system_name is not None
327	else existing.remote.system_name,
328	partition=update.remote.partition if update.remote.partition is not None else existing.remote.partition,
329	)
330
331	if len(name) > 40:	1✔
332	# TODO: Should this be added to the API spec instead?
333	raise errors.ValidationError(message="'name' cannot be longer than 40 characters.")	×
334	if default and not public:	1✔
335	raise errors.ValidationError(message="The default resource pool has to be public.")	×
336	if default and quota is not RESET:	1✔
337	raise errors.ValidationError(message="A default resource pool cannot have a quota.")	×
338	if isinstance(remote, models.RemoteConfigurationFirecrest) and default:	1✔
339	raise errors.ValidationError(message="The default resource pool cannot start remote sessions.")	×
340	if isinstance(remote, models.RemoteConfigurationFirecrest) and public:	1✔
341	raise errors.ValidationError(message="A resource pool which starts remote sessions cannot be public.")	×
342	if (idle_threshold and idle_threshold < 0) or (hibernation_threshold and hibernation_threshold < 0):	1✔
343	raise errors.ValidationError(message="Idle threshold and hibernation threshold need to be larger than 0.")	×
344
345	default_classes: list[models.ResourceClass] = []	1✔
346	for cls in classes:	1✔
347	if (isinstance(quota, (models.Quota, models.UnsavedQuota))) and not quota.is_resource_class_compatible(cls):	1✔
348	raise errors.ValidationError(	×
349	message=f"The resource class with name {cls.name} is not compatible with the quota."
350	)
351	if cls.default:	1✔
352	default_classes.append(cls)	1✔
353	if len(default_classes) != 1:	1✔
354	raise errors.ValidationError(message="One default class is required in each resource pool.")	×
355
356
357	def validate_cluster(body: apispec.Cluster) -> models.ClusterSettings:	2✔
358	"""Convert a REST API Cluster object to a model Cluster object."""
359	return models.ClusterSettings(	2✔
360	name=body.name,
361	config_name=body.config_name,
362	session_protocol=models.SessionProtocol(body.session_protocol.value),
363	session_host=body.session_host,
364	session_port=body.session_port,
365	session_path=body.session_path,
366	session_ingress_class_name=body.session_ingress_class_name,
367	session_ingress_annotations=body.session_ingress_annotations.model_dump(),
368	session_tls_secret_name=body.session_tls_secret_name,
369	session_storage_class=body.session_storage_class,
370	service_account_name=body.service_account_name,
371	)
372
373
374	def validate_cluster_patch(patch: apispec.ClusterPatch) -> models.ClusterPatch:	2✔
375	"""Convert a REST API Cluster object patch to a model Cluster object."""
376
377	return models.ClusterPatch(	2✔
378	name=patch.name,
379	config_name=patch.config_name,
380	session_protocol=models.SessionProtocol(patch.session_protocol.value)
381	if patch.session_protocol is not None
382	else None,
383	session_host=patch.session_host,
384	session_port=patch.session_port,
385	session_path=patch.session_path,
386	session_ingress_class_name=patch.session_ingress_class_name,
387	session_ingress_annotations=patch.session_ingress_annotations.model_dump()
388	if patch.session_ingress_annotations is not None
389	else None,
390	session_tls_secret_name=patch.session_tls_secret_name,
391	session_storage_class=patch.session_storage_class,
392	service_account_name=patch.service_account_name,
393	)
394
395
396	def validate_remote(body: apispec.RemoteConfigurationFirecrest) -> models.RemoteConfigurationFirecrest:	2✔
397	"""Validate a remote configuration object."""
398	kind = models.RemoteConfigurationKind(body.kind.value)	2✔
399	if kind != models.RemoteConfigurationKind.firecrest:	2✔
400	raise errors.ValidationError(message=f"The kind '{kind}' of remote configuration is not supported.", quiet=True)	×
401	validate_firecrest_api_url(body.api_url)	2✔
402	return models.RemoteConfigurationFirecrest(	2✔
403	kind=kind,
404	provider_id=body.provider_id,
405	api_url=body.api_url,
406	system_name=body.system_name,
407	partition=body.partition,
408	)
409
410
411	def validate_remote_put(	2✔
412	body: apispec.RemoteConfigurationFirecrest \| None,
413	) -> models.RemoteConfigurationPatch:
414	"""Validate the PUT update to a remote configuration object."""
415	if body is None:	2✔
416	return RESET	2✔
417	remote = validate_remote(body=body)	1✔
418	return models.RemoteConfigurationFirecrestPatch(	1✔
419	kind=remote.kind,
420	provider_id=remote.provider_id,
421	api_url=remote.api_url,
422	system_name=remote.system_name,
423	partition=remote.partition,
424	)
425
426
427	def validate_remote_patch(	2✔
428	body: apispec.RemoteConfigurationPatchReset \| apispec.RemoteConfigurationFirecrestPatch,
429	) -> models.RemoteConfigurationPatch:
430	"""Validate the patch to a remote configuration object."""
431	if isinstance(body, apispec.RemoteConfigurationPatchReset):	2✔
432	return RESET	1✔
433	kind = models.RemoteConfigurationKind(body.kind.value) if body.kind else None	2✔
434	if kind and kind != models.RemoteConfigurationKind.firecrest:	2✔
435	raise errors.ValidationError(message=f"The kind '{kind}' of remote configuration is not supported.", quiet=True)	×
436	if body.api_url:	2✔
437	validate_firecrest_api_url(body.api_url)	2✔
438	return models.RemoteConfigurationFirecrestPatch(	2✔
439	kind=kind,
440	provider_id=body.provider_id,
441	api_url=body.api_url,
442	system_name=body.system_name,
443	partition=body.partition,
444	)
445
446
447	def validate_firecrest_api_url(url: str) -> None:	2✔
448	"""Validate the URL to the FirecREST API."""
449	parsed = urlparse(url)	2✔
450	if not parsed.netloc:	2✔
451	raise errors.ValidationError(	×
452	message=f"The host for the firecrest api url {url} is not valid, expected a non-empty value.",
453	quiet=True,
454	)
455	accepted_schemes = ["https"]	2✔
456	if parsed.scheme not in accepted_schemes:	2✔
457	raise errors.ValidationError(	×
458	message=f"The scheme for the firecrest api url {url} is not valid, expected one of {accepted_schemes}",
459	quiet=True,
460	)

SwissDataScienceCenter / renku-data-services / 18840025735

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