5948296099

Committed 23 Aug 2023 07:23AM UTC coverage: 85.801% (+0.04%) from 85.766%

Build # 5948296099

Build Type

Pull #3601

github-actions

Committed by

olevski

Commit Message

chore: run poetry lock

Pull Request Pull Request #3601: hotfix: v2.6.1

Run Details

40 of 48 new or added lines in 10 files covered. (83.33%)

285 existing lines in 25 files now uncovered.

25875 of 30157 relevant lines covered (85.8%)

4.9 hits per line

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

94.05

/renku/ui/service/serializers/cache.py

#
# Copyright 2020 - Swiss Data Science Center (SDSC)
# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Renku service cache serializers."""
import time
import uuid
from urllib.parse import urlparse

from marshmallow import Schema, ValidationError, fields, post_load, pre_load, validates_schema
from marshmallow_oneofschema import OneOfSchema
from werkzeug.utils import secure_filename

from renku.core import errors
from renku.core.util.os import normalize_to_ascii
from renku.domain_model.git import GitURL
from renku.ui.service.config import PROJECT_CLONE_DEPTH_DEFAULT
from renku.ui.service.serializers.common import (
    ArchiveSchema,
    AsyncSchema,
    ErrorResponse,
    FileDetailsSchema,
    LocalRepositorySchema,
    RemoteRepositorySchema,
    RenkuSyncSchema,
)
from renku.ui.service.serializers.rpc import JsonRPCResponse


def extract_file(request):
    """Extract file from Flask request.

    Raises:
        `ValidationError`: If file data or filename is missing in request.
    """
    files = request.files
    if "file" not in files:
        raise ValidationError("missing key: file")

    file = files["file"]
    if file and not file.filename:
        raise ValidationError(f"wrong filename: {file.filename}")

    if file:
        file.filename = secure_filename(file.filename)
        return file


class FileUploadRequest(ArchiveSchema):
    """Request schema for file upload."""

    override_existing = fields.Boolean(
        load_default=False,
        metadata={"description": "Overried files. Useful when extracting from archives."},
    )

    chunked_id = fields.String(data_key="dzuuid", load_default=None, metadata={"description": "Dropzone upload id."})
    chunk_index = fields.Integer(
        data_key="dzchunkindex", load_default=None, metadata={"description": "Dropzone chunk index."}
    )
    chunk_count = fields.Integer(
        data_key="dztotalchunkcount", load_default=None, metadata={"description": "Dropzone total chunk count."}
    )
    chunk_size = fields.Integer(
        data_key="dzchunksize", load_default=None, metadata={"description": "Dropzone chunk size."}
    )
    chunk_byte_offset = fields.Integer(
        data_key="dzchunkbyteoffset", load_default=None, metadata={"description": "Dropzone chunk offset."}
    )
    chunked_content_type = fields.String(
        load_default=None, metadata={"description": "Content type of file file for chunked uploads."}
    )
    total_size = fields.Integer(
        data_key="dztotalfilesize", load_default=None, metadata={"description": "Dropzone total file size."}
    )

    @validates_schema
    def validate_requires(self, data, **kwargs):
        """Validate chunk upload data."""
        if data.get("dzuuid") is not None and ("dzchunkindex" not in data or "dztotalchunkcount" not in data):
            raise ValidationError("'dzchunkindex' and 'dztotalchunkcount' are required when 'dzuuid' is set.")


class FileUploadResponse(Schema):
    """Response schema for file upload."""

    files = fields.List(fields.Nested(FileDetailsSchema))


class FileUploadResponseRPC(JsonRPCResponse):
    """RPC response schema for file upload response."""

    result = fields.Nested(FileUploadResponse)


class FileChunksDeleteRequest(Schema):
    """Request schema for deleting uploaded chunks."""

    chunked_id = fields.String(data_key="dzuuid", load_default=None, metadata={"description": "Dropzone upload id."})


class FileChunksDeleteResponseRPC(JsonRPCResponse):
    """RPC response schema for file upload response."""

    result = fields.String()


class FileListResponse(Schema):
    """Response schema for files listing."""

    files = fields.List(fields.Nested(FileDetailsSchema), required=True)


class FileListResponseRPC(JsonRPCResponse):
    """RPC response schema for files listing."""

    result = fields.Nested(FileListResponse)


class RepositoryCloneRequest(RemoteRepositorySchema):
    """Request schema for repository clone."""

    depth = fields.Integer(metadata={"description": "Git fetch depth"}, load_default=PROJECT_CLONE_DEPTH_DEFAULT)


class ProjectCloneContext(RepositoryCloneRequest):
    """Context schema for project clone."""

    # measured in ms
    timestamp = fields.Integer(load_default=time.time() * 1e3)

    # user data
    name = fields.String()
    slug = fields.String()
    fullname = fields.String()
    email = fields.String()
    owner = fields.String()
    token = fields.String()

    @pre_load()
    def set_missing_id(self, data, **kwargs):
        """Set project_id when missing."""
        if not data.get("project_id"):
            data["project_id"] = uuid.uuid4().hex

        return data

    @pre_load()
    def set_owner_name(self, data, **kwargs):
        """Set owner and name fields."""
        try:
            git_url = GitURL.parse(data["git_url"])
        except UnicodeError as e:
            raise ValidationError("`git_url` contains unsupported characters") from e
        except errors.InvalidGitURL as e:
            raise ValidationError("Invalid `git_url`") from e

        if git_url.owner is None:
            raise ValidationError("Invalid `git_url`")
        data["owner"] = git_url.owner

        if git_url.name is None:
            raise ValidationError("Invalid `git_url`")
        data["name"] = git_url.name
        data["slug"] = normalize_to_ascii(data["name"])

        return data

    def format_url(self, data):
        """Format url with auth."""
        git_url = urlparse(data["git_url"])

        url = "oauth2:{}@{}".format(data["token"], git_url.netloc)
        return git_url._replace(netloc=url).geturl()

    @post_load
    def finalize_data(self, data, **kwargs):
        """Finalize data."""
        data["url_with_auth"] = self.format_url(data)

        if not data["depth"]:
            # NOTE: In case of `depth=None` or `depth=0` we set to default depth.
            data["depth"] = PROJECT_CLONE_DEPTH_DEFAULT

        try:
            depth = int(data["depth"])

            if depth < 0:
                # NOTE: In case of `depth<0` we remove the depth limit.
                data["depth"] = None

        except ValueError:
            data["depth"] = PROJECT_CLONE_DEPTH_DEFAULT

        return data


class ProjectCloneResponse(Schema):
    """Response schema for project clone."""

    project_id = fields.String(required=True)
    git_url = fields.String(required=True)
    initialized = fields.Boolean(dump_default=False)


class ProjectCloneResponseRPC(JsonRPCResponse):
    """RPC response schema for project clone response."""

    result = fields.Nested(ProjectCloneResponse)


class ProjectListResponse(Schema):
    """Response schema for project listing."""

    projects = fields.List(fields.Nested(ProjectCloneResponse), required=True)


class ProjectListResponseRPC(JsonRPCResponse):
    """RPC response schema for project listing."""

    result = fields.Nested(ProjectListResponse)


class ProjectMigrateRequest(AsyncSchema, LocalRepositorySchema, RemoteRepositorySchema):
    """Request schema for project migrate."""

    force_template_update = fields.Boolean(dump_default=False)
    skip_template_update = fields.Boolean(dump_default=False)
    skip_docker_update = fields.Boolean(dump_default=False)
    skip_migrations = fields.Boolean(dump_default=False)


class ProjectMigrateResponse(RenkuSyncSchema):
    """Response schema for project migrate."""

    was_migrated = fields.Boolean()
    template_migrated = fields.Boolean()
    docker_migrated = fields.Boolean()
    messages = fields.List(fields.String)
    warnings = fields.List(fields.String)
    errors = fields.List(fields.String)


class ProjectMigrateResponseRPC(JsonRPCResponse):
    """RPC response schema for project migrate."""

    result = fields.Nested(ProjectMigrateResponse)


class ProjectMigrationCheckRequest(LocalRepositorySchema, RemoteRepositorySchema):
    """Request schema for project migration check."""


class ProjectCompatibilityResponseDetail(Schema):
    """Response schema outlining service compatibility for migrations check."""

    project_metadata_version = fields.String(
        metadata={"description": "Current version of the Renku metadata in the project."}
    )
    current_metadata_version = fields.String(
        metadata={"description": "Highest metadata version supported by this service."}
    )
    migration_required = fields.Boolean(
        metadata={"description": "Whether or not a metadata migration is required to be compatible with this service."}
    )


class ProjectCompatibilityResponse(OneOfSchema):
    """Combined schema of DockerfileStatusResponseDetail or Exception."""

    type_schemas = {"detail": ProjectCompatibilityResponseDetail, "error": ErrorResponse}

    def get_obj_type(self, obj):
        """Get type from object."""
        from renku.command.migrate import CoreStatusResult

        if isinstance(obj, CoreStatusResult) or (isinstance(obj, dict) and "userMessage" not in obj):
            return "detail"

        return "error"


class DockerfileStatusResponseDetail(Schema):
    """Response schema outlining dockerfile status for migrations check."""

    newer_renku_available = fields.Boolean(
        metadata={
            "description": "Whether the version of Renku in this service is newer than the one in the Dockerfile."
        }
    )
    automated_dockerfile_update = fields.Boolean(
        metadata={"description": "Whether or not the Dockerfile supports automated Renku version updates."}
    )
    latest_renku_version = fields.String(
        metadata={"description": "The current version of Renku available in this service."}
    )
    dockerfile_renku_version = fields.String(metadata={"description": "Version of Renku specified in the Dockerfile."})


class DockerfileStatusResponse(OneOfSchema):
    """Combined schema of DockerfileStatusResponseDetail or Exception."""

    type_schemas = {"detail": DockerfileStatusResponseDetail, "error": ErrorResponse}

    def get_obj_type(self, obj):
        """Get type from object."""
        from renku.command.migrate import DockerfileStatusResult

        if isinstance(obj, DockerfileStatusResult) or (isinstance(obj, dict) and "userMessage" not in obj):
            return "detail"

        return "error"


class TemplateStatusResponseDetail(Schema):
    """Response schema outlining template status for migrations check."""

    automated_template_update = fields.Boolean(
        metadata={"description": "Whether or not the project template explicitly supports automated updates."}
    )
    newer_template_available = fields.Boolean(
        metadata={
            "description": "Whether or not the current version of the project template differs from the "
            "one used in the project."
        }
    )
    template_source = fields.String(
        metadata={
            "description": "Source of the template repository, "
            "either a Git URL or 'renku' if an embedded template was used."
        }
    )
    template_ref = fields.String(
        metadata={
            "description": "The branch/tag/commit from the template_source repository "
            "that was used to create this project."
        }
    )
    template_id = fields.String(metadata={"description": "The id of the template in the template repository."})

    project_template_version = fields.String(
        allow_none=True, metadata={"description": "The version of the template last used in the user's project."}
    )
    latest_template_version = fields.String(
        allow_none=True, metadata={"description": "The current version of the template in the template repository."}
    )

    ssh_supported = fields.Boolean(
        metadata={"description": "Whether this project supports ssh connections to sessions or not."}
    )


class TemplateStatusResponse(OneOfSchema):
    """Combined schema of TemplateStatusResponseDetail or Exception."""

    type_schemas = {"detail": TemplateStatusResponseDetail, "error": ErrorResponse}

    def get_obj_type(self, obj):
        """Get type from object."""
        from renku.command.migrate import TemplateStatusResult

        if isinstance(obj, TemplateStatusResult) or (isinstance(obj, dict) and "userMessage" not in obj):
            return "detail"

        return "error"


class ProjectMigrationCheckResponse(Schema):
    """Response schema for project migration check."""

    project_supported = fields.Boolean(
        metadata={
            "description": "Determines whether this project is a Renku project that is supported by the version "
            "running on this service (not made with a newer version)."
        }
    )
    core_renku_version = fields.String(metadata={"description": "Version of Renku running in this service."})
    project_renku_version = fields.String(metadata={"description": "Version of Renku last used to change the project."})

    core_compatibility_status = fields.Nested(
        ProjectCompatibilityResponse,
        metadata={"description": "Fields detailing the compatibility of the project with this core service version."},
    )
    dockerfile_renku_status = fields.Nested(
        DockerfileStatusResponse,
        metadata={"description": "Fields detailing the status of the Dockerfile in the project."},
    )
    template_status = fields.Nested(
        TemplateStatusResponse,
        metadata={"description": "Fields detailing the status of the project template used by this project."},
    )


class ProjectMigrationCheckResponseRPC(JsonRPCResponse):
    """RPC response schema for project migration check."""

    result = fields.Nested(ProjectMigrationCheckResponse)

1	#
2	# Copyright 2020 - Swiss Data Science Center (SDSC)
3	# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
4	# Eidgenössische Technische Hochschule Zürich (ETHZ).
5	#
6	# Licensed under the Apache License, Version 2.0 (the "License");
7	# you may not use this file except in compliance with the License.
8	# You may obtain a copy of the License at
9	#
10	# http://www.apache.org/licenses/LICENSE-2.0
11	#
12	# Unless required by applicable law or agreed to in writing, software
13	# distributed under the License is distributed on an "AS IS" BASIS,
14	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15	# See the License for the specific language governing permissions and
16	# limitations under the License.
17	"""Renku service cache serializers."""	5✔
18	import time	5✔
19	import uuid	5✔
20	from urllib.parse import urlparse	5✔
21
22	from marshmallow import Schema, ValidationError, fields, post_load, pre_load, validates_schema	5✔
23	from marshmallow_oneofschema import OneOfSchema	5✔
24	from werkzeug.utils import secure_filename	5✔
25
26	from renku.core import errors	5✔
27	from renku.core.util.os import normalize_to_ascii	5✔
28	from renku.domain_model.git import GitURL	5✔
29	from renku.ui.service.config import PROJECT_CLONE_DEPTH_DEFAULT	5✔
30	from renku.ui.service.serializers.common import (	5✔
31	ArchiveSchema,
32	AsyncSchema,
33	ErrorResponse,
34	FileDetailsSchema,
35	LocalRepositorySchema,
36	RemoteRepositorySchema,
37	RenkuSyncSchema,
38	)
39	from renku.ui.service.serializers.rpc import JsonRPCResponse	5✔
40
41
42	def extract_file(request):	5✔
43	"""Extract file from Flask request.
44
45	Raises:
46	`ValidationError`: If file data or filename is missing in request.
47	"""
48	files = request.files	2✔
49	if "file" not in files:	2✔
UNCOV 50	raise ValidationError("missing key: file")	×
51
52	file = files["file"]	2✔
53	if file and not file.filename:	2✔
UNCOV 54	raise ValidationError(f"wrong filename: {file.filename}")	×
55
56	if file:	2✔
57	file.filename = secure_filename(file.filename)	2✔
58	return file	2✔
59
60
61	class FileUploadRequest(ArchiveSchema):	5✔
62	"""Request schema for file upload."""
63
64	override_existing = fields.Boolean(	5✔
65	load_default=False,
66	metadata={"description": "Overried files. Useful when extracting from archives."},
67	)
68
69	chunked_id = fields.String(data_key="dzuuid", load_default=None, metadata={"description": "Dropzone upload id."})	5✔
70	chunk_index = fields.Integer(	5✔
71	data_key="dzchunkindex", load_default=None, metadata={"description": "Dropzone chunk index."}
72	)
73	chunk_count = fields.Integer(	5✔
74	data_key="dztotalchunkcount", load_default=None, metadata={"description": "Dropzone total chunk count."}
75	)
76	chunk_size = fields.Integer(	5✔
77	data_key="dzchunksize", load_default=None, metadata={"description": "Dropzone chunk size."}
78	)
79	chunk_byte_offset = fields.Integer(	5✔
80	data_key="dzchunkbyteoffset", load_default=None, metadata={"description": "Dropzone chunk offset."}
81	)
82	chunked_content_type = fields.String(	5✔
83	load_default=None, metadata={"description": "Content type of file file for chunked uploads."}
84	)
85	total_size = fields.Integer(	5✔
86	data_key="dztotalfilesize", load_default=None, metadata={"description": "Dropzone total file size."}
87	)
88
89	@validates_schema	5✔
90	def validate_requires(self, data, **kwargs):	5✔
91	"""Validate chunk upload data."""
92	if data.get("dzuuid") is not None and ("dzchunkindex" not in data or "dztotalchunkcount" not in data):	2✔
UNCOV 93	raise ValidationError("'dzchunkindex' and 'dztotalchunkcount' are required when 'dzuuid' is set.")	×
94
95
96	class FileUploadResponse(Schema):	5✔
97	"""Response schema for file upload."""
98
99	files = fields.List(fields.Nested(FileDetailsSchema))	5✔
100
101
102	class FileUploadResponseRPC(JsonRPCResponse):	5✔
103	"""RPC response schema for file upload response."""
104
105	result = fields.Nested(FileUploadResponse)	5✔
106
107
108	class FileChunksDeleteRequest(Schema):	5✔
109	"""Request schema for deleting uploaded chunks."""
110
111	chunked_id = fields.String(data_key="dzuuid", load_default=None, metadata={"description": "Dropzone upload id."})	5✔
112
113
114	class FileChunksDeleteResponseRPC(JsonRPCResponse):	5✔
115	"""RPC response schema for file upload response."""
116
117	result = fields.String()	5✔
118
119
120	class FileListResponse(Schema):	5✔
121	"""Response schema for files listing."""
122
123	files = fields.List(fields.Nested(FileDetailsSchema), required=True)	5✔
124
125
126	class FileListResponseRPC(JsonRPCResponse):	5✔
127	"""RPC response schema for files listing."""
128
129	result = fields.Nested(FileListResponse)	5✔
130
131
132	class RepositoryCloneRequest(RemoteRepositorySchema):	5✔
133	"""Request schema for repository clone."""
134
135	depth = fields.Integer(metadata={"description": "Git fetch depth"}, load_default=PROJECT_CLONE_DEPTH_DEFAULT)	5✔
136
137
138	class ProjectCloneContext(RepositoryCloneRequest):	5✔
139	"""Context schema for project clone."""
140
141	# measured in ms
142	timestamp = fields.Integer(load_default=time.time() * 1e3)	5✔
143
144	# user data
145	name = fields.String()	5✔
146	slug = fields.String()	5✔
147	fullname = fields.String()	5✔
148	email = fields.String()	5✔
149	owner = fields.String()	5✔
150	token = fields.String()	5✔
151
152	@pre_load()	5✔
153	def set_missing_id(self, data, **kwargs):	5✔
154	"""Set project_id when missing."""
155	if not data.get("project_id"):	3✔
156	data["project_id"] = uuid.uuid4().hex	3✔
157
158	return data	3✔
159
160	@pre_load()	5✔
161	def set_owner_name(self, data, **kwargs):	5✔
162	"""Set owner and name fields."""
163	try:	3✔
164	git_url = GitURL.parse(data["git_url"])	3✔
165	except UnicodeError as e:	1✔
UNCOV 166	raise ValidationError("`git_url` contains unsupported characters") from e	×
167	except errors.InvalidGitURL as e:	1✔
168	raise ValidationError("Invalid `git_url`") from e	1✔
169
170	if git_url.owner is None:	3✔
171	raise ValidationError("Invalid `git_url`")	2✔
172	data["owner"] = git_url.owner	3✔
173
174	if git_url.name is None:	3✔
UNCOV 175	raise ValidationError("Invalid `git_url`")	×
176	data["name"] = git_url.name	3✔
177	data["slug"] = normalize_to_ascii(data["name"])	3✔
178
179	return data	3✔
180
181	def format_url(self, data):	5✔
182	"""Format url with auth."""
183	git_url = urlparse(data["git_url"])	3✔
184
185	url = "oauth2:{}@{}".format(data["token"], git_url.netloc)	3✔
186	return git_url._replace(netloc=url).geturl()	3✔
187
188	@post_load	5✔
189	def finalize_data(self, data, **kwargs):	5✔
190	"""Finalize data."""
191	data["url_with_auth"] = self.format_url(data)	3✔
192
193	if not data["depth"]:	3✔
194	# NOTE: In case of `depth=None` or `depth=0` we set to default depth.
UNCOV 195	data["depth"] = PROJECT_CLONE_DEPTH_DEFAULT	×
196
197	try:	3✔
198	depth = int(data["depth"])	3✔
199
200	if depth < 0:	3✔
201	# NOTE: In case of `depth<0` we remove the depth limit.
202	data["depth"] = None	1✔
203
204	except ValueError:	×
UNCOV 205	data["depth"] = PROJECT_CLONE_DEPTH_DEFAULT	×
206
207	return data	3✔
208
209
210	class ProjectCloneResponse(Schema):	5✔
211	"""Response schema for project clone."""
212
213	project_id = fields.String(required=True)	5✔
214	git_url = fields.String(required=True)	5✔
215	initialized = fields.Boolean(dump_default=False)	5✔
216
217
218	class ProjectCloneResponseRPC(JsonRPCResponse):	5✔
219	"""RPC response schema for project clone response."""
220
221	result = fields.Nested(ProjectCloneResponse)	5✔
222
223
224	class ProjectListResponse(Schema):	5✔
225	"""Response schema for project listing."""
226
227	projects = fields.List(fields.Nested(ProjectCloneResponse), required=True)	5✔
228
229
230	class ProjectListResponseRPC(JsonRPCResponse):	5✔
231	"""RPC response schema for project listing."""
232
233	result = fields.Nested(ProjectListResponse)	5✔
234
235
236	class ProjectMigrateRequest(AsyncSchema, LocalRepositorySchema, RemoteRepositorySchema):	5✔
237	"""Request schema for project migrate."""
238
239	force_template_update = fields.Boolean(dump_default=False)	5✔
240	skip_template_update = fields.Boolean(dump_default=False)	5✔
241	skip_docker_update = fields.Boolean(dump_default=False)	5✔
242	skip_migrations = fields.Boolean(dump_default=False)	5✔
243
244
245	class ProjectMigrateResponse(RenkuSyncSchema):	5✔
246	"""Response schema for project migrate."""
247
248	was_migrated = fields.Boolean()	5✔
249	template_migrated = fields.Boolean()	5✔
250	docker_migrated = fields.Boolean()	5✔
251	messages = fields.List(fields.String)	5✔
252	warnings = fields.List(fields.String)	5✔
253	errors = fields.List(fields.String)	5✔
254
255
256	class ProjectMigrateResponseRPC(JsonRPCResponse):	5✔
257	"""RPC response schema for project migrate."""
258
259	result = fields.Nested(ProjectMigrateResponse)	5✔
260
261
262	class ProjectMigrationCheckRequest(LocalRepositorySchema, RemoteRepositorySchema):	5✔
263	"""Request schema for project migration check."""
264
265
266	class ProjectCompatibilityResponseDetail(Schema):	5✔
267	"""Response schema outlining service compatibility for migrations check."""
268
269	project_metadata_version = fields.String(	5✔
270	metadata={"description": "Current version of the Renku metadata in the project."}
271	)
272	current_metadata_version = fields.String(	5✔
273	metadata={"description": "Highest metadata version supported by this service."}
274	)
275	migration_required = fields.Boolean(	5✔
276	metadata={"description": "Whether or not a metadata migration is required to be compatible with this service."}
277	)
278
279
280	class ProjectCompatibilityResponse(OneOfSchema):	5✔
281	"""Combined schema of DockerfileStatusResponseDetail or Exception."""
282
283	type_schemas = {"detail": ProjectCompatibilityResponseDetail, "error": ErrorResponse}	5✔
284
285	def get_obj_type(self, obj):	5✔
286	"""Get type from object."""
287	from renku.command.migrate import CoreStatusResult	1✔
288
289	if isinstance(obj, CoreStatusResult) or (isinstance(obj, dict) and "userMessage" not in obj):	1✔
290	return "detail"	1✔
291
UNCOV 292	return "error"	×
293
294
295	class DockerfileStatusResponseDetail(Schema):	5✔
296	"""Response schema outlining dockerfile status for migrations check."""
297
298	newer_renku_available = fields.Boolean(	5✔
299	metadata={
300	"description": "Whether the version of Renku in this service is newer than the one in the Dockerfile."
301	}
302	)
303	automated_dockerfile_update = fields.Boolean(	5✔
304	metadata={"description": "Whether or not the Dockerfile supports automated Renku version updates."}
305	)
306	latest_renku_version = fields.String(	5✔
307	metadata={"description": "The current version of Renku available in this service."}
308	)
309	dockerfile_renku_version = fields.String(metadata={"description": "Version of Renku specified in the Dockerfile."})	5✔
310
311
312	class DockerfileStatusResponse(OneOfSchema):	5✔
313	"""Combined schema of DockerfileStatusResponseDetail or Exception."""
314
315	type_schemas = {"detail": DockerfileStatusResponseDetail, "error": ErrorResponse}	5✔
316
317	def get_obj_type(self, obj):	5✔
318	"""Get type from object."""
319	from renku.command.migrate import DockerfileStatusResult	1✔
320
321	if isinstance(obj, DockerfileStatusResult) or (isinstance(obj, dict) and "userMessage" not in obj):	1✔
322	return "detail"	1✔
323
UNCOV 324	return "error"	×
325
326
327	class TemplateStatusResponseDetail(Schema):	5✔
328	"""Response schema outlining template status for migrations check."""
329
330	automated_template_update = fields.Boolean(	5✔
331	metadata={"description": "Whether or not the project template explicitly supports automated updates."}
332	)
333	newer_template_available = fields.Boolean(	5✔
334	metadata={
335	"description": "Whether or not the current version of the project template differs from the "
336	"one used in the project."
337	}
338	)
339	template_source = fields.String(	5✔
340	metadata={
341	"description": "Source of the template repository, "
342	"either a Git URL or 'renku' if an embedded template was used."
343	}
344	)
345	template_ref = fields.String(	5✔
346	metadata={
347	"description": "The branch/tag/commit from the template_source repository "
348	"that was used to create this project."
349	}
350	)
351	template_id = fields.String(metadata={"description": "The id of the template in the template repository."})	5✔
352
353	project_template_version = fields.String(	5✔
354	allow_none=True, metadata={"description": "The version of the template last used in the user's project."}
355	)
356	latest_template_version = fields.String(	5✔
357	allow_none=True, metadata={"description": "The current version of the template in the template repository."}
358	)
359
360	ssh_supported = fields.Boolean(	5✔
361	metadata={"description": "Whether this project supports ssh connections to sessions or not."}
362	)
363
364
365	class TemplateStatusResponse(OneOfSchema):	5✔
366	"""Combined schema of TemplateStatusResponseDetail or Exception."""
367
368	type_schemas = {"detail": TemplateStatusResponseDetail, "error": ErrorResponse}	5✔
369
370	def get_obj_type(self, obj):	5✔
371	"""Get type from object."""
372	from renku.command.migrate import TemplateStatusResult	1✔
373
374	if isinstance(obj, TemplateStatusResult) or (isinstance(obj, dict) and "userMessage" not in obj):	1✔
375	return "detail"	1✔
376
377	return "error"	1✔
378
379
380	class ProjectMigrationCheckResponse(Schema):	5✔
381	"""Response schema for project migration check."""
382
383	project_supported = fields.Boolean(	5✔
384	metadata={
385	"description": "Determines whether this project is a Renku project that is supported by the version "
386	"running on this service (not made with a newer version)."
387	}
388	)
389	core_renku_version = fields.String(metadata={"description": "Version of Renku running in this service."})	5✔
390	project_renku_version = fields.String(metadata={"description": "Version of Renku last used to change the project."})	5✔
391
392	core_compatibility_status = fields.Nested(	5✔
393	ProjectCompatibilityResponse,
394	metadata={"description": "Fields detailing the compatibility of the project with this core service version."},
395	)
396	dockerfile_renku_status = fields.Nested(	5✔
397	DockerfileStatusResponse,
398	metadata={"description": "Fields detailing the status of the Dockerfile in the project."},
399	)
400	template_status = fields.Nested(	5✔
401	TemplateStatusResponse,
402	metadata={"description": "Fields detailing the status of the project template used by this project."},
403	)
404
405
406	class ProjectMigrationCheckResponseRPC(JsonRPCResponse):	5✔
407	"""RPC response schema for project migration check."""
408
409	result = fields.Nested(ProjectMigrationCheckResponse)	5✔

SwissDataScienceCenter / renku-python / 5948296099

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