18099160032

Committed 29 Sep 2025 01:45PM UTC coverage: 83.309% (-3.4%) from 86.756%

Build # 18099160032

Build Type

Pull #1015

github

Committed by

web-flow

Commit Message

Merge a03635015 into 0690bab65

Pull Request Pull Request #1015: fix: loading kube configs

Run Details

23 of 68 new or added lines in 8 files covered. (33.82%)

892 existing lines in 42 files now uncovered.

21422 of 25714 relevant lines covered (83.31%)

1.49 hits per line

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

26.44

/components/renku_data_services/notebooks/api/classes/server.py

"""Jupyter server models."""

from collections.abc import Sequence
from itertools import chain
from pathlib import PurePosixPath
from typing import Any
from urllib.parse import urlparse

from gitlab.v4.objects.projects import Project

from renku_data_services.app_config import logging
from renku_data_services.base_models import AnonymousAPIUser, AuthenticatedAPIUser
from renku_data_services.base_models.core import APIUser
from renku_data_services.k8s.constants import DEFAULT_K8S_CLUSTER
from renku_data_services.notebooks.api.amalthea_patches import cloudstorage as cloudstorage_patches
from renku_data_services.notebooks.api.amalthea_patches import general as general_patches
from renku_data_services.notebooks.api.amalthea_patches import git_proxy as git_proxy_patches
from renku_data_services.notebooks.api.amalthea_patches import git_sidecar as git_sidecar_patches
from renku_data_services.notebooks.api.amalthea_patches import init_containers as init_containers_patches
from renku_data_services.notebooks.api.amalthea_patches import inject_certificates as inject_certificates_patches
from renku_data_services.notebooks.api.amalthea_patches import jupyter_server as jupyter_server_patches
from renku_data_services.notebooks.api.amalthea_patches import ssh as ssh_patches
from renku_data_services.notebooks.api.classes.cloud_storage import ICloudStorageRequest
from renku_data_services.notebooks.api.classes.k8s_client import NotebookK8sClient
from renku_data_services.notebooks.api.classes.repository import GitProvider, Repository
from renku_data_services.notebooks.api.schemas.secrets import K8sUserSecrets
from renku_data_services.notebooks.api.schemas.server_options import ServerOptions
from renku_data_services.notebooks.config import NotebooksConfig
from renku_data_services.notebooks.constants import JUPYTER_SESSION_GVK
from renku_data_services.notebooks.cr_amalthea_session import TlsSecret
from renku_data_services.notebooks.crs import JupyterServerV1Alpha1
from renku_data_services.notebooks.errors.programming import DuplicateEnvironmentVariableError
from renku_data_services.notebooks.errors.user import MissingResourceError

logger = logging.getLogger(__name__)


class UserServer:
    """Represents a Renku server session."""

    def __init__(
        self,
        user: AnonymousAPIUser | AuthenticatedAPIUser,
        server_name: str,
        image: str | None,
        server_options: ServerOptions,
        environment_variables: dict[str, str],
        user_secrets: K8sUserSecrets | None,
        cloudstorage: Sequence[ICloudStorageRequest],
        k8s_client: NotebookK8sClient[JupyterServerV1Alpha1],
        workspace_mount_path: PurePosixPath,
        work_dir: PurePosixPath,
        config: NotebooksConfig,
        internal_gitlab_user: APIUser,
        host: str,
        namespace: str,
        using_default_image: bool = False,
        is_image_private: bool = False,
        repositories: list[Repository] | None = None,
    ):
        self._user = user
        self.server_name = server_name
        self._k8s_client = k8s_client
        self.safe_username = self._user.id
        self.image = image
        self.server_options = server_options
        self.environment_variables = environment_variables
        self.user_secrets = user_secrets
        self.using_default_image = using_default_image
        self.workspace_mount_path = workspace_mount_path
        self.work_dir = work_dir
        self.cloudstorage = cloudstorage
        self.is_image_private = is_image_private
        self.host = host
        self.__namespace = namespace
        self.config = config
        self.internal_gitlab_user = internal_gitlab_user

        if self.server_options.idle_threshold_seconds is not None:
            self.idle_seconds_threshold = self.server_options.idle_threshold_seconds
        else:
            self.idle_seconds_threshold = (
                config.sessions.culling.registered.idle_seconds
                if isinstance(self._user, AuthenticatedAPIUser)
                else config.sessions.culling.anonymous.idle_seconds
            )

        if self.server_options.hibernation_threshold_seconds is not None:
            self.hibernated_seconds_threshold: int = self.server_options.hibernation_threshold_seconds
        else:
            self.hibernated_seconds_threshold = (
                config.sessions.culling.registered.hibernated_seconds
                if isinstance(user, AuthenticatedAPIUser)
                else config.sessions.culling.anonymous.hibernated_seconds
            )
        self._repositories: list[Repository] = repositories or []
        self._git_providers: list[GitProvider] | None = None
        self._has_configured_git_providers = False

        self.server_url = f"https://{self.host}/sessions/{self.server_name}"
        if not self._user.is_authenticated:
            self.server_url = f"{self.server_url}?token={self._user.id}"

    def k8s_namespace(self) -> str:
        """Get the preferred namespace for a server."""
        return self.__namespace

    @property
    def user(self) -> AnonymousAPIUser | AuthenticatedAPIUser:
        """Getter for server's user."""
        return self._user

    async def repositories(self) -> list[Repository]:
        """Get the list of repositories in the project."""
        # Configure git repository providers based on matching URLs.
        if not self._has_configured_git_providers:
            git_providers = await self.git_providers()
            for repo in self._repositories:
                found_provider = None
                for provider in git_providers:
                    if urlparse(provider.url).netloc == urlparse(repo.url).netloc:
                        found_provider = provider
                        break
                if found_provider is not None:
                    repo.provider = found_provider.id
            self._has_configured_git_providers = True

        return self._repositories

    async def git_providers(self) -> list[GitProvider]:
        """The list of git providers."""
        if self._git_providers is None:
            self._git_providers = await self.config.git_provider_helper.get_providers(user=self.user)
        return self._git_providers

    async def required_git_providers(self) -> list[GitProvider]:
        """The list of required git providers."""
        repositories = await self.repositories()
        required_provider_ids: set[str] = set(r.provider for r in repositories if r.provider)
        providers = await self.git_providers()
        return [p for p in providers if p.id in required_provider_ids]

    def __str__(self) -> str:
        return f"<UserServer user: {self._user.id} server_name: {self.server_name}>"

    async def start(self) -> JupyterServerV1Alpha1 | None:
        """Create the jupyterserver resource in k8s."""
        errors = self._get_start_errors()
        if errors:
            raise MissingResourceError(
                message=(
                    "Cannot start the session because the following Git "
                    f"or Docker resources are missing: {', '.join(errors)}"
                )
            )
        session_manifest = await self._get_session_manifest()
        manifest = JupyterServerV1Alpha1.model_validate(session_manifest)
        return await self._k8s_client.create_session(manifest, self.user)

    @staticmethod
    def _check_environment_variables_overrides(patches_list: list[dict[str, Any]]) -> None:
        """Check if any patch overrides server's environment variables.

        Checks if it overrides with a different value or if two patches create environment variables with different
        values.
        """
        env_vars: dict[tuple[str, str], str] = {}

        for patch_list in patches_list:
            patches = patch_list["patch"]

            for patch in patches:
                path = str(patch["path"]).lower()
                if path.endswith("/env/-"):
                    name = str(patch["value"]["name"])
                    value = str(patch["value"]["value"])
                    key = (path, name)

                    if key in env_vars and env_vars[key] != value:
                        raise DuplicateEnvironmentVariableError(
                            message=f"Environment variable {path}::{name} is being overridden by multiple patches"
                        )
                    else:
                        env_vars[key] = value

    def _get_start_errors(self) -> list[str]:
        """Check if there are any errors before starting the server."""
        errors: list[str] = []
        if self.image is None:
            errors.append(f"image {self.image} does not exist or cannot be accessed")
        return errors

    async def _get_session_manifest(self) -> dict[str, Any]:
        """Compose the body of the user session for the k8s operator."""
        patches = await self._get_patches()
        self._check_environment_variables_overrides(patches)

        # Storage
        if self.config.sessions.storage.pvs_enabled:
            storage: dict[str, Any] = {
                "size": self.server_options.storage,
                "pvc": {
                    "enabled": True,
                    # We should check against the cluster, but as this is only used by V1 sessions, we ignore this
                    # use-case.
                    "storageClassName": self.config.sessions.storage.pvs_storage_class,
                    "mountPath": self.workspace_mount_path.as_posix(),
                },
            }
        else:
            storage_size = self.server_options.storage if self.config.sessions.storage.use_empty_dir_size_limit else ""
            storage = {
                "size": storage_size,
                "pvc": {
                    "enabled": False,
                    "mountPath": self.workspace_mount_path.as_posix(),
                },
            }
        # Authentication
        if isinstance(self._user, AuthenticatedAPIUser):
            session_auth = {
                "token": "",
                "oidc": {
                    "enabled": True,
                    "clientId": self.config.sessions.oidc.client_id,
                    "clientSecret": {"value": self.config.sessions.oidc.client_secret},
                    "issuerUrl": self.config.sessions.oidc.issuer_url,
                    "authorizedEmails": [self._user.email],
                },
            }
        else:
            session_auth = {
                "token": self._user.id,
                "oidc": {"enabled": False},
            }

        cluster = await self.config.k8s_client.cluster_by_class_id(self.server_options.resource_class_id, self._user)

        if cluster.id != DEFAULT_K8S_CLUSTER:
            cluster_settings = await self.config.cluster_rp.select(cluster.id)
            (
                base_server_path,
                _,
                _,
                host,
                tls_secret,
                ingress_annotations,
            ) = cluster_settings.get_ingress_parameters(self.server_name)
        else:
            # Fallback to global, main cluster parameters
            host = self.config.sessions.ingress.host
            base_server_path = self.config.sessions.ingress.base_path(self.server_name)
            ingress_annotations = self.config.sessions.ingress.annotations

            tls_name = self.config.sessions.ingress.tls_secret
            tls_secret = None if tls_name is None else TlsSecret(adopt=False, name=tls_name)

        # Combine everything into the manifest
        manifest = {
            "apiVersion": JUPYTER_SESSION_GVK.group_version,
            "kind": JUPYTER_SESSION_GVK.kind,
            "metadata": {
                "name": self.server_name,
                "labels": self.get_labels(),
                "annotations": self.get_annotations(),
            },
            "spec": {
                "auth": session_auth,
                "culling": {
                    "idleSecondsThreshold": self.idle_seconds_threshold,
                    "maxAgeSecondsThreshold": (
                        self.config.sessions.culling.registered.max_age_seconds
                        if isinstance(self._user, AuthenticatedAPIUser)
                        else self.config.sessions.culling.anonymous.max_age_seconds
                    ),
                    "hibernatedSecondsThreshold": self.hibernated_seconds_threshold,
                },
                "jupyterServer": {
                    "defaultUrl": self.server_options.default_url,
                    "image": self.image,
                    "rootDir": self.work_dir.as_posix(),
                    "resources": self.server_options.to_k8s_resources(
                        enforce_cpu_limits=self.config.sessions.enforce_cpu_limits
                    ),
                },
                "routing": {
                    "host": host,
                    "path": base_server_path,
                    "ingressAnnotations": ingress_annotations,
                    "tls": {
                        "enabled": tls_secret is not None,
                        "secretName": tls_secret.name if tls_secret is not None else "",
                    },
                },
                "storage": storage,
                "patches": patches,
            },
        }
        return manifest

    def _get_renku_annotation_prefix(self) -> str:
        return self.config.session_get_endpoint_annotations.renku_annotation_prefix

    async def _get_patches(self) -> list[dict[str, Any]]:
        return list(
            chain(
                general_patches.test(self),
                general_patches.session_tolerations(self),
                general_patches.session_affinity(self),
                general_patches.session_node_selector(self),
                general_patches.priority_class(self),
                general_patches.dev_shm(self),
                jupyter_server_patches.args(),
                jupyter_server_patches.env(self),
                jupyter_server_patches.image_pull_secret(self, self.internal_gitlab_user.access_token),
                jupyter_server_patches.disable_service_links(),
                jupyter_server_patches.rstudio_env_variables(self),
                await git_proxy_patches.main(self),
                await git_sidecar_patches.main(self),
                general_patches.oidc_unverified_email(self),
                ssh_patches.main(self.config),
                # init container for certs must come before all other init containers
                # so that it runs first before all other init containers
                init_containers_patches.certificates(self.config),
                init_containers_patches.download_image(self),
                await init_containers_patches.git_clone(self),
                inject_certificates_patches.proxy(self),
                # Cloud Storage needs to patch the git clone sidecar spec and so should come after
                # the sidecars
                # WARN: this patch depends on the index of the sidecar and so needs to be updated
                # if sidecars are added or removed
                await cloudstorage_patches.main(self),
                # NOTE: User secrets adds an init container, volume and mounts, so it may affect
                # indices in other patches.
                jupyter_server_patches.user_secrets(self),
            )
        )

    def get_labels(self) -> dict[str, str | None]:
        """Get the labels for the session."""
        prefix = self._get_renku_annotation_prefix()
        labels = {
            "app": "jupyter",
            "component": "singleuser-server",
            f"{prefix}commit-sha": None,
            f"{prefix}gitlabProjectId": None,
            f"{prefix}safe-username": self.safe_username,
            f"{prefix}quota": self.server_options.priority_class
            if self.server_options.priority_class is not None
            else "",
            f"{prefix}userId": self._user.id,
        }
        return labels

    def get_annotations(self) -> dict[str, str | None]:
        """Get the annotations for the session."""
        prefix = self._get_renku_annotation_prefix()
        username = self._user.id
        if isinstance(self.user, AuthenticatedAPIUser) and self._user.email:
            username = self._user.email
        annotations = {
            f"{prefix}commit-sha": None,
            f"{prefix}gitlabProjectId": None,
            f"{prefix}safe-username": self._user.id,
            f"{prefix}username": username,
            f"{prefix}userId": self._user.id,
            f"{prefix}servername": self.server_name,
            f"{prefix}branch": None,
            f"{prefix}git-host": None,
            f"{prefix}namespace": None,
            f"{prefix}projectName": None,
            f"{prefix}requested-image": self.image,
            f"{prefix}repository": None,
            f"{prefix}hibernation": "",
            f"{prefix}hibernationBranch": "",
            f"{prefix}hibernationCommitSha": "",
            f"{prefix}hibernationDirty": "",
            f"{prefix}hibernationSynchronized": "",
            f"{prefix}hibernationDate": "",
            f"{prefix}hibernatedSecondsThreshold": str(self.hibernated_seconds_threshold),
            f"{prefix}lastActivityDate": "",
            f"{prefix}idleSecondsThreshold": str(self.idle_seconds_threshold),
        }
        if self.server_options.resource_class_id:
            annotations[f"{prefix}resourceClassId"] = str(self.server_options.resource_class_id)
        return annotations


class Renku1UserServer(UserServer):
    """Represents a Renku 1.0 server session."""

    def __init__(
        self,
        user: AnonymousAPIUser | AuthenticatedAPIUser,
        server_name: str,
        gl_namespace: str,
        project: str,
        branch: str,
        commit_sha: str,
        image: str | None,
        server_options: ServerOptions,
        environment_variables: dict[str, str],
        user_secrets: K8sUserSecrets | None,
        cloudstorage: Sequence[ICloudStorageRequest],
        k8s_client: NotebookK8sClient,
        workspace_mount_path: PurePosixPath,
        work_dir: PurePosixPath,
        config: NotebooksConfig,
        host: str,
        namespace: str,
        gitlab_project: Project | None,
        internal_gitlab_user: APIUser,
        using_default_image: bool = False,
        is_image_private: bool = False,
        **_: dict,  # Required to ignore unused arguments, among which repositories
    ):
        repositories = [
            Repository(
                url=p.http_url_to_repo,
                dirname=p.path,
                branch=branch,
                commit_sha=commit_sha,
            )
            for p in [gitlab_project]
            if p is not None
        ]

        super().__init__(
            user=user,
            server_name=server_name,
            image=image,
            server_options=server_options,
            environment_variables=environment_variables,
            user_secrets=user_secrets,
            cloudstorage=cloudstorage,
            k8s_client=k8s_client,
            workspace_mount_path=workspace_mount_path,
            work_dir=work_dir,
            using_default_image=using_default_image,
            is_image_private=is_image_private,
            repositories=repositories,
            host=host,
            namespace=namespace,
            config=config,
            internal_gitlab_user=internal_gitlab_user,
        )

        self.gl_namespace = gl_namespace
        self.project = project
        self.branch = branch
        self.commit_sha = commit_sha
        self.git_host = urlparse(config.git.url).netloc
        self.gitlab_project = gitlab_project

    def _get_start_errors(self) -> list[str]:
        """Check if there are any errors before starting the server."""
        errors = super()._get_start_errors()
        if self.gitlab_project is None:
            errors.append(f"project {self.project} does not exist")
        if not self._branch_exists():
            errors.append(f"branch {self.branch} does not exist")
        if not self._commit_sha_exists():
            errors.append(f"commit {self.commit_sha} does not exist")
        return errors

    def _branch_exists(self) -> bool:
        """Check if a specific branch exists in the user's gitlab project.

        The branch name is not required by the API and therefore
        passing None to this function will return True.
        """
        if self.branch is not None and self.gitlab_project is not None:
            try:
                self.gitlab_project.branches.get(self.branch)
            except Exception as err:
                logger.warning(f"Branch {self.branch} cannot be verified or does not exist. {err}")
            else:
                return True
        return False

    def _commit_sha_exists(self) -> bool:
        """Check if a specific commit sha exists in the user's gitlab project."""
        if self.commit_sha is not None and self.gitlab_project is not None:
            try:
                self.gitlab_project.commits.get(self.commit_sha)
            except Exception as err:
                logger.warning(f"Commit {self.commit_sha} cannot be verified or does not exist. {err}")
            else:
                return True
        return False

    def get_labels(self) -> dict[str, str | None]:
        """Get the labels of the jupyter server."""
        prefix = self._get_renku_annotation_prefix()
        labels = super().get_labels()
        labels[f"{prefix}commit-sha"] = self.commit_sha
        if self.gitlab_project is not None:
            labels[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)
        return labels

    def get_annotations(self) -> dict[str, str | None]:
        """Get the annotations of the jupyter server."""
        prefix = self._get_renku_annotation_prefix()
        annotations = super().get_annotations()
        annotations[f"{prefix}commit-sha"] = self.commit_sha
        annotations[f"{prefix}branch"] = self.branch
        annotations[f"{prefix}git-host"] = self.git_host
        annotations[f"{prefix}namespace"] = self.gl_namespace
        annotations[f"{prefix}projectName"] = self.project
        if self.gitlab_project is not None:
            annotations[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)
            annotations[f"{prefix}repository"] = self.gitlab_project.web_url
        return annotations

1	"""Jupyter server models."""
2
3	from collections.abc import Sequence	2✔
4	from itertools import chain	2✔
5	from pathlib import PurePosixPath	2✔
6	from typing import Any	2✔
7	from urllib.parse import urlparse	2✔
8
9	from gitlab.v4.objects.projects import Project	2✔
10
11	from renku_data_services.app_config import logging	2✔
12	from renku_data_services.base_models import AnonymousAPIUser, AuthenticatedAPIUser	2✔
13	from renku_data_services.base_models.core import APIUser	2✔
14	from renku_data_services.k8s.constants import DEFAULT_K8S_CLUSTER	2✔
15	from renku_data_services.notebooks.api.amalthea_patches import cloudstorage as cloudstorage_patches	2✔
16	from renku_data_services.notebooks.api.amalthea_patches import general as general_patches	2✔
17	from renku_data_services.notebooks.api.amalthea_patches import git_proxy as git_proxy_patches	2✔
18	from renku_data_services.notebooks.api.amalthea_patches import git_sidecar as git_sidecar_patches	2✔
19	from renku_data_services.notebooks.api.amalthea_patches import init_containers as init_containers_patches	2✔
20	from renku_data_services.notebooks.api.amalthea_patches import inject_certificates as inject_certificates_patches	2✔
21	from renku_data_services.notebooks.api.amalthea_patches import jupyter_server as jupyter_server_patches	2✔
22	from renku_data_services.notebooks.api.amalthea_patches import ssh as ssh_patches	2✔
23	from renku_data_services.notebooks.api.classes.cloud_storage import ICloudStorageRequest	2✔
24	from renku_data_services.notebooks.api.classes.k8s_client import NotebookK8sClient	2✔
25	from renku_data_services.notebooks.api.classes.repository import GitProvider, Repository	2✔
26	from renku_data_services.notebooks.api.schemas.secrets import K8sUserSecrets	2✔
27	from renku_data_services.notebooks.api.schemas.server_options import ServerOptions	2✔
28	from renku_data_services.notebooks.config import NotebooksConfig	2✔
29	from renku_data_services.notebooks.constants import JUPYTER_SESSION_GVK	2✔
30	from renku_data_services.notebooks.cr_amalthea_session import TlsSecret	2✔
31	from renku_data_services.notebooks.crs import JupyterServerV1Alpha1	2✔
32	from renku_data_services.notebooks.errors.programming import DuplicateEnvironmentVariableError	2✔
33	from renku_data_services.notebooks.errors.user import MissingResourceError	2✔
34
35	logger = logging.getLogger(__name__)	2✔
36
37
38	class UserServer:	2✔
39	"""Represents a Renku server session."""
40
41	def __init__(	2✔
42	self,
43	user: AnonymousAPIUser \| AuthenticatedAPIUser,
44	server_name: str,
45	image: str \| None,
46	server_options: ServerOptions,
47	environment_variables: dict[str, str],
48	user_secrets: K8sUserSecrets \| None,
49	cloudstorage: Sequence[ICloudStorageRequest],
50	k8s_client: NotebookK8sClient[JupyterServerV1Alpha1],
51	workspace_mount_path: PurePosixPath,
52	work_dir: PurePosixPath,
53	config: NotebooksConfig,
54	internal_gitlab_user: APIUser,
55	host: str,
56	namespace: str,
57	using_default_image: bool = False,
58	is_image_private: bool = False,
59	repositories: list[Repository] \| None = None,
60	):
UNCOV 61	self._user = user	×
UNCOV 62	self.server_name = server_name	×
UNCOV 63	self._k8s_client = k8s_client	×
UNCOV 64	self.safe_username = self._user.id	×
UNCOV 65	self.image = image	×
UNCOV 66	self.server_options = server_options	×
UNCOV 67	self.environment_variables = environment_variables	×
UNCOV 68	self.user_secrets = user_secrets	×
UNCOV 69	self.using_default_image = using_default_image	×
UNCOV 70	self.workspace_mount_path = workspace_mount_path	×
UNCOV 71	self.work_dir = work_dir	×
UNCOV 72	self.cloudstorage = cloudstorage	×
UNCOV 73	self.is_image_private = is_image_private	×
UNCOV 74	self.host = host	×
UNCOV 75	self.__namespace = namespace	×
UNCOV 76	self.config = config	×
UNCOV 77	self.internal_gitlab_user = internal_gitlab_user	×
78
UNCOV 79	if self.server_options.idle_threshold_seconds is not None:	×
80	self.idle_seconds_threshold = self.server_options.idle_threshold_seconds	×
81	else:
UNCOV 82	self.idle_seconds_threshold = (	×
83	config.sessions.culling.registered.idle_seconds
84	if isinstance(self._user, AuthenticatedAPIUser)
85	else config.sessions.culling.anonymous.idle_seconds
86	)
87
UNCOV 88	if self.server_options.hibernation_threshold_seconds is not None:	×
89	self.hibernated_seconds_threshold: int = self.server_options.hibernation_threshold_seconds	×
90	else:
UNCOV 91	self.hibernated_seconds_threshold = (	×
92	config.sessions.culling.registered.hibernated_seconds
93	if isinstance(user, AuthenticatedAPIUser)
94	else config.sessions.culling.anonymous.hibernated_seconds
95	)
UNCOV 96	self._repositories: list[Repository] = repositories or []	×
UNCOV 97	self._git_providers: list[GitProvider] \| None = None	×
UNCOV 98	self._has_configured_git_providers = False	×
99
UNCOV 100	self.server_url = f"https://{self.host}/sessions/{self.server_name}"	×
UNCOV 101	if not self._user.is_authenticated:	×
102	self.server_url = f"{self.server_url}?token={self._user.id}"	×
103
104	def k8s_namespace(self) -> str:	2✔
105	"""Get the preferred namespace for a server."""
UNCOV 106	return self.__namespace	×
107
108	@property	2✔
109	def user(self) -> AnonymousAPIUser \| AuthenticatedAPIUser:	2✔
110	"""Getter for server's user."""
UNCOV 111	return self._user	×
112
113	async def repositories(self) -> list[Repository]:	2✔
114	"""Get the list of repositories in the project."""
115	# Configure git repository providers based on matching URLs.
UNCOV 116	if not self._has_configured_git_providers:	×
UNCOV 117	git_providers = await self.git_providers()	×
UNCOV 118	for repo in self._repositories:	×
UNCOV 119	found_provider = None	×
UNCOV 120	for provider in git_providers:	×
121	if urlparse(provider.url).netloc == urlparse(repo.url).netloc:	×
122	found_provider = provider	×
123	break	×
UNCOV 124	if found_provider is not None:	×
125	repo.provider = found_provider.id	×
UNCOV 126	self._has_configured_git_providers = True	×
127
UNCOV 128	return self._repositories	×
129
130	async def git_providers(self) -> list[GitProvider]:	2✔
131	"""The list of git providers."""
UNCOV 132	if self._git_providers is None:	×
UNCOV 133	self._git_providers = await self.config.git_provider_helper.get_providers(user=self.user)	×
UNCOV 134	return self._git_providers	×
135
136	async def required_git_providers(self) -> list[GitProvider]:	2✔
137	"""The list of required git providers."""
UNCOV 138	repositories = await self.repositories()	×
UNCOV 139	required_provider_ids: set[str] = set(r.provider for r in repositories if r.provider)	×
UNCOV 140	providers = await self.git_providers()	×
UNCOV 141	return [p for p in providers if p.id in required_provider_ids]	×
142
143	def __str__(self) -> str:	2✔
144	return f"<UserServer user: {self._user.id} server_name: {self.server_name}>"	×
145
146	async def start(self) -> JupyterServerV1Alpha1 \| None:	2✔
147	"""Create the jupyterserver resource in k8s."""
UNCOV 148	errors = self._get_start_errors()	×
UNCOV 149	if errors:	×
150	raise MissingResourceError(	×
151	message=(
152	"Cannot start the session because the following Git "
153	f"or Docker resources are missing: {', '.join(errors)}"
154	)
155	)
UNCOV 156	session_manifest = await self._get_session_manifest()	×
UNCOV 157	manifest = JupyterServerV1Alpha1.model_validate(session_manifest)	×
UNCOV 158	return await self._k8s_client.create_session(manifest, self.user)	×
159
160	@staticmethod	2✔
161	def _check_environment_variables_overrides(patches_list: list[dict[str, Any]]) -> None:	2✔
162	"""Check if any patch overrides server's environment variables.
163
164	Checks if it overrides with a different value or if two patches create environment variables with different
165	values.
166	"""
UNCOV 167	env_vars: dict[tuple[str, str], str] = {}	×
168
UNCOV 169	for patch_list in patches_list:	×
UNCOV 170	patches = patch_list["patch"]	×
171
UNCOV 172	for patch in patches:	×
UNCOV 173	path = str(patch["path"]).lower()	×
UNCOV 174	if path.endswith("/env/-"):	×
UNCOV 175	name = str(patch["value"]["name"])	×
UNCOV 176	value = str(patch["value"]["value"])	×
UNCOV 177	key = (path, name)	×
178
UNCOV 179	if key in env_vars and env_vars[key] != value:	×
180	raise DuplicateEnvironmentVariableError(	×
181	message=f"Environment variable {path}::{name} is being overridden by multiple patches"
182	)
183	else:
UNCOV 184	env_vars[key] = value	×
185
186	def _get_start_errors(self) -> list[str]:	2✔
187	"""Check if there are any errors before starting the server."""
UNCOV 188	errors: list[str] = []	×
UNCOV 189	if self.image is None:	×
190	errors.append(f"image {self.image} does not exist or cannot be accessed")	×
UNCOV 191	return errors	×
192
193	async def _get_session_manifest(self) -> dict[str, Any]:	2✔
194	"""Compose the body of the user session for the k8s operator."""
UNCOV 195	patches = await self._get_patches()	×
UNCOV 196	self._check_environment_variables_overrides(patches)	×
197
198	# Storage
UNCOV 199	if self.config.sessions.storage.pvs_enabled:	×
UNCOV 200	storage: dict[str, Any] = {	×
201	"size": self.server_options.storage,
202	"pvc": {
203	"enabled": True,
204	# We should check against the cluster, but as this is only used by V1 sessions, we ignore this
205	# use-case.
206	"storageClassName": self.config.sessions.storage.pvs_storage_class,
207	"mountPath": self.workspace_mount_path.as_posix(),
208	},
209	}
210	else:
211	storage_size = self.server_options.storage if self.config.sessions.storage.use_empty_dir_size_limit else ""	×
212	storage = {	×
213	"size": storage_size,
214	"pvc": {
215	"enabled": False,
216	"mountPath": self.workspace_mount_path.as_posix(),
217	},
218	}
219	# Authentication
UNCOV 220	if isinstance(self._user, AuthenticatedAPIUser):	×
221	session_auth = {	×
222	"token": "",
223	"oidc": {
224	"enabled": True,
225	"clientId": self.config.sessions.oidc.client_id,
226	"clientSecret": {"value": self.config.sessions.oidc.client_secret},
227	"issuerUrl": self.config.sessions.oidc.issuer_url,
228	"authorizedEmails": [self._user.email],
229	},
230	}
231	else:
UNCOV 232	session_auth = {	×
233	"token": self._user.id,
234	"oidc": {"enabled": False},
235	}
236
UNCOV 237	cluster = await self.config.k8s_client.cluster_by_class_id(self.server_options.resource_class_id, self._user)	×
238
UNCOV 239	if cluster.id != DEFAULT_K8S_CLUSTER:	×
240	cluster_settings = await self.config.cluster_rp.select(cluster.id)	×
241	(	×
242	base_server_path,
243	_,
244	_,
245	host,
246	tls_secret,
247	ingress_annotations,
248	) = cluster_settings.get_ingress_parameters(self.server_name)
249	else:
250	# Fallback to global, main cluster parameters
UNCOV 251	host = self.config.sessions.ingress.host	×
UNCOV 252	base_server_path = self.config.sessions.ingress.base_path(self.server_name)	×
UNCOV 253	ingress_annotations = self.config.sessions.ingress.annotations	×
254
NEW 255	tls_name = self.config.sessions.ingress.tls_secret	×
256	tls_secret = None if tls_name is None else TlsSecret(adopt=False, name=tls_name)	×
257
258	# Combine everything into the manifest
UNCOV 259	manifest = {	×
260	"apiVersion": JUPYTER_SESSION_GVK.group_version,
261	"kind": JUPYTER_SESSION_GVK.kind,
262	"metadata": {
263	"name": self.server_name,
264	"labels": self.get_labels(),
265	"annotations": self.get_annotations(),
266	},
267	"spec": {
268	"auth": session_auth,
269	"culling": {
270	"idleSecondsThreshold": self.idle_seconds_threshold,
271	"maxAgeSecondsThreshold": (
272	self.config.sessions.culling.registered.max_age_seconds
273	if isinstance(self._user, AuthenticatedAPIUser)
274	else self.config.sessions.culling.anonymous.max_age_seconds
275	),
276	"hibernatedSecondsThreshold": self.hibernated_seconds_threshold,
277	},
278	"jupyterServer": {
279	"defaultUrl": self.server_options.default_url,
280	"image": self.image,
281	"rootDir": self.work_dir.as_posix(),
282	"resources": self.server_options.to_k8s_resources(
283	enforce_cpu_limits=self.config.sessions.enforce_cpu_limits
284	),
285	},
286	"routing": {
287	"host": host,
288	"path": base_server_path,
289	"ingressAnnotations": ingress_annotations,
290	"tls": {
291	"enabled": tls_secret is not None,
292	"secretName": tls_secret.name if tls_secret is not None else "",
293	},
294	},
295	"storage": storage,
296	"patches": patches,
297	},
298	}
UNCOV 299	return manifest	×
300
301	def _get_renku_annotation_prefix(self) -> str:	2✔
UNCOV 302	return self.config.session_get_endpoint_annotations.renku_annotation_prefix	×
303
304	async def _get_patches(self) -> list[dict[str, Any]]:	2✔
UNCOV 305	return list(	×
306	chain(
307	general_patches.test(self),
308	general_patches.session_tolerations(self),
309	general_patches.session_affinity(self),
310	general_patches.session_node_selector(self),
311	general_patches.priority_class(self),
312	general_patches.dev_shm(self),
313	jupyter_server_patches.args(),
314	jupyter_server_patches.env(self),
315	jupyter_server_patches.image_pull_secret(self, self.internal_gitlab_user.access_token),
316	jupyter_server_patches.disable_service_links(),
317	jupyter_server_patches.rstudio_env_variables(self),
318	await git_proxy_patches.main(self),
319	await git_sidecar_patches.main(self),
320	general_patches.oidc_unverified_email(self),
321	ssh_patches.main(self.config),
322	# init container for certs must come before all other init containers
323	# so that it runs first before all other init containers
324	init_containers_patches.certificates(self.config),
325	init_containers_patches.download_image(self),
326	await init_containers_patches.git_clone(self),
327	inject_certificates_patches.proxy(self),
328	# Cloud Storage needs to patch the git clone sidecar spec and so should come after
329	# the sidecars
330	# WARN: this patch depends on the index of the sidecar and so needs to be updated
331	# if sidecars are added or removed
332	await cloudstorage_patches.main(self),
333	# NOTE: User secrets adds an init container, volume and mounts, so it may affect
334	# indices in other patches.
335	jupyter_server_patches.user_secrets(self),
336	)
337	)
338
339	def get_labels(self) -> dict[str, str \| None]:	2✔
340	"""Get the labels for the session."""
UNCOV 341	prefix = self._get_renku_annotation_prefix()	×
UNCOV 342	labels = {	×
343	"app": "jupyter",
344	"component": "singleuser-server",
345	f"{prefix}commit-sha": None,
346	f"{prefix}gitlabProjectId": None,
347	f"{prefix}safe-username": self.safe_username,
348	f"{prefix}quota": self.server_options.priority_class
349	if self.server_options.priority_class is not None
350	else "",
351	f"{prefix}userId": self._user.id,
352	}
UNCOV 353	return labels	×
354
355	def get_annotations(self) -> dict[str, str \| None]:	2✔
356	"""Get the annotations for the session."""
UNCOV 357	prefix = self._get_renku_annotation_prefix()	×
UNCOV 358	username = self._user.id	×
UNCOV 359	if isinstance(self.user, AuthenticatedAPIUser) and self._user.email:	×
360	username = self._user.email	×
UNCOV 361	annotations = {	×
362	f"{prefix}commit-sha": None,
363	f"{prefix}gitlabProjectId": None,
364	f"{prefix}safe-username": self._user.id,
365	f"{prefix}username": username,
366	f"{prefix}userId": self._user.id,
367	f"{prefix}servername": self.server_name,
368	f"{prefix}branch": None,
369	f"{prefix}git-host": None,
370	f"{prefix}namespace": None,
371	f"{prefix}projectName": None,
372	f"{prefix}requested-image": self.image,
373	f"{prefix}repository": None,
374	f"{prefix}hibernation": "",
375	f"{prefix}hibernationBranch": "",
376	f"{prefix}hibernationCommitSha": "",
377	f"{prefix}hibernationDirty": "",
378	f"{prefix}hibernationSynchronized": "",
379	f"{prefix}hibernationDate": "",
380	f"{prefix}hibernatedSecondsThreshold": str(self.hibernated_seconds_threshold),
381	f"{prefix}lastActivityDate": "",
382	f"{prefix}idleSecondsThreshold": str(self.idle_seconds_threshold),
383	}
UNCOV 384	if self.server_options.resource_class_id:	×
UNCOV 385	annotations[f"{prefix}resourceClassId"] = str(self.server_options.resource_class_id)	×
UNCOV 386	return annotations	×
387
388
389	class Renku1UserServer(UserServer):	2✔
390	"""Represents a Renku 1.0 server session."""
391
392	def __init__(	2✔
393	self,
394	user: AnonymousAPIUser \| AuthenticatedAPIUser,
395	server_name: str,
396	gl_namespace: str,
397	project: str,
398	branch: str,
399	commit_sha: str,
400	image: str \| None,
401	server_options: ServerOptions,
402	environment_variables: dict[str, str],
403	user_secrets: K8sUserSecrets \| None,
404	cloudstorage: Sequence[ICloudStorageRequest],
405	k8s_client: NotebookK8sClient,
406	workspace_mount_path: PurePosixPath,
407	work_dir: PurePosixPath,
408	config: NotebooksConfig,
409	host: str,
410	namespace: str,
411	gitlab_project: Project \| None,
412	internal_gitlab_user: APIUser,
413	using_default_image: bool = False,
414	is_image_private: bool = False,
415	**_: dict, # Required to ignore unused arguments, among which repositories
416	):
UNCOV 417	repositories = [	×
418	Repository(
419	url=p.http_url_to_repo,
420	dirname=p.path,
421	branch=branch,
422	commit_sha=commit_sha,
423	)
424	for p in [gitlab_project]
425	if p is not None
426	]
427
UNCOV 428	super().__init__(	×
429	user=user,
430	server_name=server_name,
431	image=image,
432	server_options=server_options,
433	environment_variables=environment_variables,
434	user_secrets=user_secrets,
435	cloudstorage=cloudstorage,
436	k8s_client=k8s_client,
437	workspace_mount_path=workspace_mount_path,
438	work_dir=work_dir,
439	using_default_image=using_default_image,
440	is_image_private=is_image_private,
441	repositories=repositories,
442	host=host,
443	namespace=namespace,
444	config=config,
445	internal_gitlab_user=internal_gitlab_user,
446	)
447
UNCOV 448	self.gl_namespace = gl_namespace	×
UNCOV 449	self.project = project	×
UNCOV 450	self.branch = branch	×
UNCOV 451	self.commit_sha = commit_sha	×
UNCOV 452	self.git_host = urlparse(config.git.url).netloc	×
UNCOV 453	self.gitlab_project = gitlab_project	×
454
455	def _get_start_errors(self) -> list[str]:	2✔
456	"""Check if there are any errors before starting the server."""
UNCOV 457	errors = super()._get_start_errors()	×
UNCOV 458	if self.gitlab_project is None:	×
459	errors.append(f"project {self.project} does not exist")	×
UNCOV 460	if not self._branch_exists():	×
461	errors.append(f"branch {self.branch} does not exist")	×
UNCOV 462	if not self._commit_sha_exists():	×
463	errors.append(f"commit {self.commit_sha} does not exist")	×
UNCOV 464	return errors	×
465
466	def _branch_exists(self) -> bool:	2✔
467	"""Check if a specific branch exists in the user's gitlab project.
468
469	The branch name is not required by the API and therefore
470	passing None to this function will return True.
471	"""
UNCOV 472	if self.branch is not None and self.gitlab_project is not None:	×
UNCOV 473	try:	×
UNCOV 474	self.gitlab_project.branches.get(self.branch)	×
475	except Exception as err:	×
476	logger.warning(f"Branch {self.branch} cannot be verified or does not exist. {err}")	×
477	else:
UNCOV 478	return True	×
479	return False	×
480
481	def _commit_sha_exists(self) -> bool:	2✔
482	"""Check if a specific commit sha exists in the user's gitlab project."""
UNCOV 483	if self.commit_sha is not None and self.gitlab_project is not None:	×
UNCOV 484	try:	×
UNCOV 485	self.gitlab_project.commits.get(self.commit_sha)	×
486	except Exception as err:	×
487	logger.warning(f"Commit {self.commit_sha} cannot be verified or does not exist. {err}")	×
488	else:
UNCOV 489	return True	×
490	return False	×
491
492	def get_labels(self) -> dict[str, str \| None]:	2✔
493	"""Get the labels of the jupyter server."""
UNCOV 494	prefix = self._get_renku_annotation_prefix()	×
UNCOV 495	labels = super().get_labels()	×
UNCOV 496	labels[f"{prefix}commit-sha"] = self.commit_sha	×
UNCOV 497	if self.gitlab_project is not None:	×
UNCOV 498	labels[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	×
UNCOV 499	return labels	×
500
501	def get_annotations(self) -> dict[str, str \| None]:	2✔
502	"""Get the annotations of the jupyter server."""
UNCOV 503	prefix = self._get_renku_annotation_prefix()	×
UNCOV 504	annotations = super().get_annotations()	×
UNCOV 505	annotations[f"{prefix}commit-sha"] = self.commit_sha	×
UNCOV 506	annotations[f"{prefix}branch"] = self.branch	×
UNCOV 507	annotations[f"{prefix}git-host"] = self.git_host	×
UNCOV 508	annotations[f"{prefix}namespace"] = self.gl_namespace	×
UNCOV 509	annotations[f"{prefix}projectName"] = self.project	×
UNCOV 510	if self.gitlab_project is not None:	×
UNCOV 511	annotations[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	×
UNCOV 512	annotations[f"{prefix}repository"] = self.gitlab_project.web_url	×
UNCOV 513	return annotations	×

SwissDataScienceCenter / renku-data-services / 18099160032

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