add rpc method find_collection_from_path (d1c1cc89) · Commits · certic / Jama

src/jama/init.py

+1 −1

Original line number	Diff line number	Diff line
		__version__ = "0.6.6"
		__version__ = "0.6.7"

src/rpc/methods.py

+48 −8

Original line number	Diff line number	Diff line
		@@ -4,7 +4,7 @@ import re
		from datetime import timedelta
		from functools import wraps as _wraps
		from glob import glob as _glob
		from typing import List, Dict, Iterator, Union, Tuple
		from typing import List, Dict, Iterator, Union, Tuple, Optional
		import cv2
		import numpy
		import pyvips
		@@ -212,6 +212,19 @@ def _collections_with_serialization_data(query_set: QuerySet) -> QuerySet:
		)


		def _collection_path_segments(path: str) -> List[str]:
		segments = []
		for segment in path.split("/"):
		segment = segment.strip()
		if segment:
		segments.append(segment)
		return segments


		def _ascii_collection_path_segments(path: str) -> List[str]:
		return [unidecode.unidecode(segment) for segment in _collection_path_segments(path)]


		def _resources_with_serialization_data(
		query_set: QuerySet, include_metas: bool
		) -> QuerySet:
		@@ -545,6 +558,38 @@ def collection(user: User, collection_id: int) -> Dict:
		return serializers.collection(collection_instance, cache=SerializerCache())


		@_rpc_groups(["Collections"])
		def find_collection_from_path(user: User, path: str, project_id: int) -> Optional[Dict]:
		"""
		Return the collection corresponding to 'path' in the project, or null
		if no matching collection exists.

		The path is resolved from the implicit project root collection. Each
		path item is interpreted as the title of a nested collection.
		"""
		if not path:
		raise ServiceException(NO_PATH)
		if not project_id:
		raise ServiceException(NO_PROJECT_ID)
		project = Project.objects.filter(pk=project_id).first()
		if not project:
		raise ServiceException(NO_SUCH_PROJECT)

		collection_instance = project.root_collection
		for segment in _collection_path_segments(path):
		collection_instance = Collection.objects.filter(
		title=segment,
		parent=collection_instance,
		project=project,
		deleted_at__isnull=True,
		).first()
		if not collection_instance:
		return None

		UserAccess(user, project).check_read(collection_instance)
		return serializers.collection(collection_instance, cache=SerializerCache())


		@_rpc_groups(["Collections"])
		def add_collection(user: User, title: str, parent_id: int) -> Dict:
		"""
		@@ -666,10 +711,7 @@ def add_collection_from_path(user: User, path: str, project_id: int) -> List[Dic
		hierarchy = []
		previous_dir = project.root_collection
		serializer_cache = SerializerCache()
		for dir_name in path.split("/"):
		# force ascii representation of unicode strings
		dir_name = unidecode.unidecode(dir_name.strip())
		if dir_name:
		for dir_name in _ascii_collection_path_segments(path):
		previous_dir, created = Collection.objects.get_or_create(
		title=dir_name, parent=previous_dir, project=project
		)
		@@ -677,9 +719,7 @@ def add_collection_from_path(user: User, path: str, project_id: int) -> List[Dic
		if previous_dir.deleted_at:
		previous_dir.deleted_at = None
		previous_dir.save()
		hierarchy.append(
		serializers.collection(previous_dir, cache=serializer_cache)
		)
		hierarchy.append(serializers.collection(previous_dir, cache=serializer_cache))
		return hierarchy

src/rpc/tests.py

+57 −0

Original line number	Diff line number	Diff line
		@@ -641,6 +641,63 @@ class ServiceTestCase(TestCase):
		idx = idx + 1
		self.assertEqual(models.Collection.objects.filter(title="root").count(), 1)

		def test_find_collection_from_path(self):
		created_collections = methods.add_collection_from_path(
		self.test_user, "/some/example/collection/", self.test_project.pk
		)

		collection_with_slash = methods.find_collection_from_path(
		self.test_user, "/some/example/collection/", self.test_project.pk
		)
		collection_without_slash = methods.find_collection_from_path(
		self.test_user, "some/example/collection/", self.test_project.pk
		)

		self.assertEqual(collection_with_slash["id"], created_collections[-1]["id"])
		self.assertEqual(collection_without_slash["id"], created_collections[-1]["id"])
		self.assertEqual(collection_with_slash["title"], "collection")

		def test_find_collection_from_path_preserves_diacritics(self):
		parent = models.Collection.objects.create(
		title="été",
		parent=self.test_project.root_collection,
		project=self.test_project,
		)
		child = models.Collection.objects.create(
		title="Noël",
		parent=parent,
		project=self.test_project,
		)

		collection = methods.find_collection_from_path(
		self.test_user, "/été/Noël/", self.test_project.pk
		)

		self.assertEqual(collection["id"], child.pk)

		def test_find_collection_from_path_returns_none_when_missing(self):
		methods.add_collection_from_path(
		self.test_user, "/some/example/", self.test_project.pk
		)

		collection = methods.find_collection_from_path(
		self.test_user, "/some/example/missing/", self.test_project.pk
		)

		self.assertIsNone(collection)

		def test_find_collection_from_path_ignores_deleted_collection(self):
		created_collections = methods.add_collection_from_path(
		self.test_user, "/some/deleted/", self.test_project.pk
		)
		models.Collection.objects.get(pk=created_collections[-1]["id"]).soft_delete()

		collection = methods.find_collection_from_path(
		self.test_user, "/some/deleted/", self.test_project.pk
		)

		self.assertIsNone(collection)

		def test_require_superuser(self):
		with self.assertRaises(ServiceException):
		methods.delete_role(self.test_user, 1)