19181077268

Committed 07 Nov 2025 09:00PM UTC coverage: 86.352% (-0.5%) from 86.841%

Build # 19181077268

Build Type

Pull #1059

github

Committed by

web-flow

Commit Message

Merge fb47045e6 into 58a6e4765

Pull Request Pull Request #1059: fix: patching of session custom resources

Run Details

89 of 104 new or added lines in 5 files covered. (85.58%)

577 existing lines in 33 files now uncovered.

22791 of 26393 relevant lines covered (86.35%)

1.52 hits per line

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

94.87

/components/renku_data_services/notebooks/config/static.py

"""Static configuration."""

from dataclasses import dataclass, field
from typing import ClassVar, Self, cast

from marshmallow import INCLUDE, Schema, fields


@dataclass
class _SessionAnnotationName:
    prefix: str
    name: str
    required: bool
    _sep: ClassVar[str] = "/"

    def get_field_name(self, sanitized: bool = False) -> str:
        sep = "" if self.prefix.endswith(self._sep) else self._sep
        if sanitized:
            sanitized_prefix = self.prefix.replace(".", "_")
            sanitized_name = self.name.replace(".", "_")
            return f"{sanitized_prefix}{sep}{sanitized_name}"
        else:
            return f"{self.prefix}{sep}{self.name}"

    def to_marshmallow_field(self) -> fields.Str:
        return fields.Str(required=self.required, data_key=self.get_field_name())

    @classmethod
    def from_str(cls, val: str, required: bool = True) -> Self:
        parts = val.split(cls._sep)
        if len(parts) != 2:
            raise ValueError(f"Expected to find prefix and name in the annotation but found {len(parts)} parts.")
        return cls(
            parts[0],
            parts[1],
            required,
        )


@dataclass
class _ServersGetEndpointAnnotations:
    renku_annotation_prefix: ClassVar[str] = "renku.io/"
    required_annotation_names: list[str] = field(
        default_factory=lambda: [
            "renku.io/namespace",
            "renku.io/projectName",
            "renku.io/branch",
            "renku.io/commit-sha",
            "renku.io/default_image_used",
            "renku.io/repository",
        ]
    )
    optional_annotation_names: list[str] = field(
        default_factory=lambda: [
            "renku.io/hibernation",
            "renku.io/hibernationBranch",
            "renku.io/hibernationCommitSha",
            "renku.io/hibernationDirty",
            "renku.io/hibernationSynchronized",
            "renku.io/hibernationDate",
            "renku.io/hibernatedSecondsThreshold",
            "renku.io/lastActivityDate",
            "renku.io/idleSecondsThreshold",
            "renku.io/servername",
            "renku.io/username",
            "renku.io/git-host",
            "renku.io/gitlabProjectId",
            "renku.io/resourceClassId",
            "jupyter.org/servername",
            "jupyter.org/username",
            # Renku 2.0 annotations
            "renku.io/renkuVersion",
            "renku.io/launcherId",
            "renku.io/projectId",
        ]
    )

    def __post_init__(self) -> None:
        annotations: list[_SessionAnnotationName] = []
        for annotation in self.required_annotation_names:
            annotations.append(_SessionAnnotationName.from_str(annotation, required=True))
        for annotation in self.optional_annotation_names:
            annotations.append(_SessionAnnotationName.from_str(annotation, required=False))
        self.annotations = annotations

        self.schema = Schema.from_dict(
            {
                annotation.get_field_name(sanitized=True): annotation.to_marshmallow_field()
                for annotation in self.annotations
            }
        )(unknown=INCLUDE)

    def sanitize_dict(self, ann_dict: dict[str, str]) -> dict[str, str]:
        return cast(dict[str, str], self.schema.load(ann_dict))

1	"""Static configuration."""
2
3	from dataclasses import dataclass, field	2✔
4	from typing import ClassVar, Self, cast	2✔
5
6	from marshmallow import INCLUDE, Schema, fields	2✔
7
8
9	@dataclass	2✔
10	class _SessionAnnotationName:	2✔
11	prefix: str	2✔
12	name: str	2✔
13	required: bool	2✔
14	_sep: ClassVar[str] = "/"	2✔
15
16	def get_field_name(self, sanitized: bool = False) -> str:	2✔
17	sep = "" if self.prefix.endswith(self._sep) else self._sep	2✔
18	if sanitized:	2✔
19	sanitized_prefix = self.prefix.replace(".", "_")	2✔
20	sanitized_name = self.name.replace(".", "_")	2✔
21	return f"{sanitized_prefix}{sep}{sanitized_name}"	2✔
22	else:
23	return f"{self.prefix}{sep}{self.name}"	2✔
24
25	def to_marshmallow_field(self) -> fields.Str:	2✔
26	return fields.Str(required=self.required, data_key=self.get_field_name())	2✔
27
28	@classmethod	2✔
29	def from_str(cls, val: str, required: bool = True) -> Self:	2✔
30	parts = val.split(cls._sep)	2✔
31	if len(parts) != 2:	2✔
32	raise ValueError(f"Expected to find prefix and name in the annotation but found {len(parts)} parts.")	×
33	return cls(	2✔
34	parts[0],
35	parts[1],
36	required,
37	)
38
39
40	@dataclass	2✔
41	class _ServersGetEndpointAnnotations:	2✔
42	renku_annotation_prefix: ClassVar[str] = "renku.io/"	2✔
43	required_annotation_names: list[str] = field(	2✔
44	default_factory=lambda: [
45	"renku.io/namespace",
46	"renku.io/projectName",
47	"renku.io/branch",
48	"renku.io/commit-sha",
49	"renku.io/default_image_used",
50	"renku.io/repository",
51	]
52	)
53	optional_annotation_names: list[str] = field(	2✔
54	default_factory=lambda: [
55	"renku.io/hibernation",
56	"renku.io/hibernationBranch",
57	"renku.io/hibernationCommitSha",
58	"renku.io/hibernationDirty",
59	"renku.io/hibernationSynchronized",
60	"renku.io/hibernationDate",
61	"renku.io/hibernatedSecondsThreshold",
62	"renku.io/lastActivityDate",
63	"renku.io/idleSecondsThreshold",
64	"renku.io/servername",
65	"renku.io/username",
66	"renku.io/git-host",
67	"renku.io/gitlabProjectId",
68	"renku.io/resourceClassId",
69	"jupyter.org/servername",
70	"jupyter.org/username",
71	# Renku 2.0 annotations
72	"renku.io/renkuVersion",
73	"renku.io/launcherId",
74	"renku.io/projectId",
75	]
76	)
77
78	def __post_init__(self) -> None:	2✔
79	annotations: list[_SessionAnnotationName] = []	2✔
80	for annotation in self.required_annotation_names:	2✔
81	annotations.append(_SessionAnnotationName.from_str(annotation, required=True))	2✔
82	for annotation in self.optional_annotation_names:	2✔
83	annotations.append(_SessionAnnotationName.from_str(annotation, required=False))	2✔
84	self.annotations = annotations	2✔
85
86	self.schema = Schema.from_dict(	2✔
87	{
88	annotation.get_field_name(sanitized=True): annotation.to_marshmallow_field()
89	for annotation in self.annotations
90	}
91	)(unknown=INCLUDE)
92
93	def sanitize_dict(self, ann_dict: dict[str, str]) -> dict[str, str]:	2✔
UNCOV 94	return cast(dict[str, str], self.schema.load(ann_dict))	×

SwissDataScienceCenter / renku-data-services / 19181077268

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