12047437567

Committed 27 Nov 2024 09:31AM UTC coverage: 86.006% (+0.1%) from 85.882%

Build # 12047437567

Build Type

Pull #545

github

Committed by

web-flow

Commit Message

Merge c434d43ab into 7ae3af62e

Pull Request Pull Request #545: feat!: add support for session secrets

Run Details

216 of 222 new or added lines in 10 files covered. (97.3%)

5 existing lines in 3 files now uncovered.

14682 of 17071 relevant lines covered (86.01%)

1.52 hits per line

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

97.5

/components/renku_data_services/project/core.py

"""Business logic for projects."""

from pathlib import PurePosixPath

from ulid import ULID

from renku_data_services import errors
from renku_data_services.authz.models import Visibility
from renku_data_services.base_models import APIUser, Slug
from renku_data_services.data_connectors.db import DataConnectorProjectLinkRepository
from renku_data_services.project import apispec, models
from renku_data_services.project.db import ProjectRepository
from renku_data_services.session.db import SessionRepository


def validate_project_patch(patch: apispec.ProjectPatch) -> models.ProjectPatch:
    """Validate the update to a project."""
    keywords = [kw.root for kw in patch.keywords] if patch.keywords is not None else None
    return models.ProjectPatch(
        name=patch.name,
        namespace=patch.namespace,
        visibility=Visibility(patch.visibility.value) if patch.visibility is not None else None,
        repositories=patch.repositories,
        description=patch.description,
        keywords=keywords,
        documentation=patch.documentation,
        template_id=None if patch.template_id is None else "",
        is_template=patch.is_template,
    )


async def copy_project(
    project_id: ULID,
    user: APIUser,
    name: str,
    namespace: str,
    slug: str | None,
    description: str | None,
    repositories: list[models.Repository] | None,
    visibility: Visibility | None,
    keywords: list[str],
    project_repo: ProjectRepository,
    session_repo: SessionRepository,
    data_connector_to_project_link_repo: DataConnectorProjectLinkRepository,
) -> models.Project:
    """Create a copy of a given project."""
    template = await project_repo.get_project(user=user, project_id=project_id)

    unsaved_project = models.UnsavedProject(
        name=name,
        namespace=namespace,
        slug=slug or Slug.from_name(name).value,
        description=description or template.description,
        repositories=repositories or template.repositories,
        created_by=user.id,  # type: ignore[arg-type]
        visibility=template.visibility if visibility is None else visibility,
        keywords=keywords or template.keywords,
        template_id=template.id,
    )
    project = await project_repo.insert_project(user, unsaved_project)

    # NOTE: Copy session launchers
    launchers = await session_repo.get_project_launchers(user=user, project_id=project_id)
    for launcher in launchers:
        await session_repo.copy_launcher(user=user, project_id=project.id, launcher=launcher)

    # NOTE: Copy data connector links. If this operation fails due to lack of permission, still proceed to create the
    # copy but return an error code that reflects this
    copy_error = False
    dc_links = await data_connector_to_project_link_repo.get_links_to(user=user, project_id=project_id)
    for dc_link in dc_links:
        try:
            await data_connector_to_project_link_repo.copy_link(user=user, project_id=project.id, link=dc_link)
        except errors.MissingResourceError:
            copy_error = True

    if copy_error:
        raise errors.CopyDataConnectorsError()

    return project


def validate_unsaved_session_secret_slot(
    body: apispec.SessionSecretSlotPost,
) -> models.UnsavedSessionSecretSlot:
    """Validate an unsaved secret slot."""
    _validate_session_launcher_secret_slot_filename(body.filename)
    return models.UnsavedSessionSecretSlot(
        project_id=ULID.from_str(body.project_id),
        name=body.name,
        description=body.description,
        filename=body.filename,
    )


def validate_session_secret_slot_patch(
    body: apispec.SessionSecretSlotPatch,
) -> models.SessionSecretSlotPatch:
    """Validate the update to a secret slot."""
    if body.filename is not None:
        _validate_session_launcher_secret_slot_filename(body.filename)
    return models.SessionSecretSlotPatch(
        name=body.name,
        description=body.description,
        filename=body.filename,
    )


def _validate_session_launcher_secret_slot_filename(filename: str) -> None:
    """Validate the filename field of a secret slot."""
    filename_candidate = PurePosixPath(filename)
    if filename_candidate.name != filename:
        raise errors.ValidationError(message=f"Filename {filename} is not valid.")

1	"""Business logic for projects."""
2
3	from pathlib import PurePosixPath	2✔
4
5	from ulid import ULID	2✔
6
7	from renku_data_services import errors	2✔
8	from renku_data_services.authz.models import Visibility	2✔
9	from renku_data_services.base_models import APIUser, Slug	2✔
10	from renku_data_services.data_connectors.db import DataConnectorProjectLinkRepository	2✔
11	from renku_data_services.project import apispec, models	2✔
12	from renku_data_services.project.db import ProjectRepository	2✔
13	from renku_data_services.session.db import SessionRepository	2✔
14
15
16	def validate_project_patch(patch: apispec.ProjectPatch) -> models.ProjectPatch:	2✔
17	"""Validate the update to a project."""
18	keywords = [kw.root for kw in patch.keywords] if patch.keywords is not None else None	2✔
19	return models.ProjectPatch(	2✔
20	name=patch.name,
21	namespace=patch.namespace,
22	visibility=Visibility(patch.visibility.value) if patch.visibility is not None else None,
23	repositories=patch.repositories,
24	description=patch.description,
25	keywords=keywords,
26	documentation=patch.documentation,
27	template_id=None if patch.template_id is None else "",
28	is_template=patch.is_template,
29	)
30
31
32	async def copy_project(	2✔
33	project_id: ULID,
34	user: APIUser,
35	name: str,
36	namespace: str,
37	slug: str \| None,
38	description: str \| None,
39	repositories: list[models.Repository] \| None,
40	visibility: Visibility \| None,
41	keywords: list[str],
42	project_repo: ProjectRepository,
43	session_repo: SessionRepository,
44	data_connector_to_project_link_repo: DataConnectorProjectLinkRepository,
45	) -> models.Project:
46	"""Create a copy of a given project."""
47	template = await project_repo.get_project(user=user, project_id=project_id)	2✔
48
49	unsaved_project = models.UnsavedProject(	1✔
50	name=name,
51	namespace=namespace,
52	slug=slug or Slug.from_name(name).value,
53	description=description or template.description,
54	repositories=repositories or template.repositories,
55	created_by=user.id, # type: ignore[arg-type]
56	visibility=template.visibility if visibility is None else visibility,
57	keywords=keywords or template.keywords,
58	template_id=template.id,
59	)
60	project = await project_repo.insert_project(user, unsaved_project)	1✔
61
62	# NOTE: Copy session launchers
63	launchers = await session_repo.get_project_launchers(user=user, project_id=project_id)	1✔
64	for launcher in launchers:	1✔
65	await session_repo.copy_launcher(user=user, project_id=project.id, launcher=launcher)	1✔
66
67	# NOTE: Copy data connector links. If this operation fails due to lack of permission, still proceed to create the
68	# copy but return an error code that reflects this
69	copy_error = False	1✔
70	dc_links = await data_connector_to_project_link_repo.get_links_to(user=user, project_id=project_id)	1✔
71	for dc_link in dc_links:	1✔
72	try:	1✔
73	await data_connector_to_project_link_repo.copy_link(user=user, project_id=project.id, link=dc_link)	1✔
74	except errors.MissingResourceError:	1✔
75	copy_error = True	1✔
76
77	if copy_error:	1✔
78	raise errors.CopyDataConnectorsError()	1✔
79
80	return project	1✔
81
82
83	def validate_unsaved_session_secret_slot(	2✔
84	body: apispec.SessionSecretSlotPost,
85	) -> models.UnsavedSessionSecretSlot:
86	"""Validate an unsaved secret slot."""
87	_validate_session_launcher_secret_slot_filename(body.filename)	2✔
88	return models.UnsavedSessionSecretSlot(	2✔
89	project_id=ULID.from_str(body.project_id),
90	name=body.name,
91	description=body.description,
92	filename=body.filename,
93	)
94
95
96	def validate_session_secret_slot_patch(	2✔
97	body: apispec.SessionSecretSlotPatch,
98	) -> models.SessionSecretSlotPatch:
99	"""Validate the update to a secret slot."""
100	if body.filename is not None:	2✔
101	_validate_session_launcher_secret_slot_filename(body.filename)	2✔
102	return models.SessionSecretSlotPatch(	2✔
103	name=body.name,
104	description=body.description,
105	filename=body.filename,
106	)
107
108
109	def _validate_session_launcher_secret_slot_filename(filename: str) -> None:	2✔
110	"""Validate the filename field of a secret slot."""
111	filename_candidate = PurePosixPath(filename)	2✔
112	if filename_candidate.name != filename:	2✔
NEW 113	raise errors.ValidationError(message=f"Filename {filename} is not valid.")	×

SwissDataScienceCenter / renku-data-services / 12047437567

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