Commit 33353ffa authored by Mickaël Desfrênes's avatar Mickaël Desfrênes
Browse files

check project membership and boundaries

parent 608ece3f
Loading
Loading
Loading
Loading
+32 −1
Original line number Diff line number Diff line
@@ -1320,6 +1320,34 @@ def advanced_search(
    if not project:
        raise ServiceException(NO_SUCH_PROJECT)
    acl = UserAccess(user, project)
    acl.check_project_access()

    if collection_id:
        try:
            collection_id = int(collection_id)
        except (TypeError, ValueError):
            raise ServiceException(WRONG_ARGUMENT)
        if not Collection.objects.filter(
            pk=collection_id,
            project=project,
            deleted_at__isnull=True,
        ).exists():
            raise ServiceException(NO_SUCH_COLLECTION)

    try:
        metadata_ids = {
            int(search_term[key])
            for search_term in search_terms
            for key in ("meta", "exclude_meta")
            if key in search_term
        }
    except (TypeError, ValueError):
        raise ServiceException(WRONG_ARGUMENT)
    if metadata_ids and (
        Metadata.objects.filter(project=project, pk__in=metadata_ids).count()
        != len(metadata_ids)
    ):
        raise ServiceException(NO_SUCH_METADATA)

    results = {
        "collections": [],
@@ -2919,7 +2947,10 @@ def meta_count(user: User, metadata_id: int, collection_id: int) -> dict:
    if not collection_instance:
        raise ServiceException(NO_SUCH_COLLECTION)
    UserAccess(user, collection_instance.project).check_read(collection_instance)
    meta = Metadata.objects.filter(pk=metadata_id).first()
    meta = Metadata.objects.filter(
        pk=metadata_id,
        project=collection_instance.project,
    ).first()
    if not meta:
        raise ServiceException(NO_SUCH_METADATA)
    collections_ids = collection_instance.descendants_and_self_ids()
+92 −1
Original line number Diff line number Diff line
@@ -1276,12 +1276,89 @@ class ServiceTestCase(TestCase):
            terms,
            self.test_project.pk,
            include_metas=True,
            collection_id=321,
            collection_id=self.test_project_root_collection.pk,
            limit_from=0,
            limit_to=2000,
        )
        methods.simple_search(self.public_user, "pouet", self.test_project.pk)

    def test_advanced_search_requires_membership_and_project_scoped_ids(self):
        resource = models.Resource.objects.create(
            title="search membership oracle",
            ptr_project=self.test_project,
        )
        no_access_user = User.objects.create(username="no_search_membership")
        search_terms = [
            {
                "property": "title",
                "term": "is",
                "value": resource.title,
            }
        ]

        with self.assertRaises(ServiceException):
            methods.advanced_search(
                no_access_user,
                search_terms,
                self.test_project.pk,
            )

        project_member = User.objects.create(username="search_project_member")
        membership_role = models.Role.objects.create(
            label="search membership",
            project=self.test_project,
        )
        models.ProjectAccess.objects.create(
            project=self.test_project,
            user=project_member,
            role=membership_role,
        )
        results = methods.advanced_search(
            project_member,
            search_terms,
            self.test_project.pk,
            fetch_collections=False,
        )
        self.assertEqual(results["resources_count"], 1)
        self.assertEqual(results["resources"], [])

        other_project = models.Project.objects.create(label="foreign search project")
        other_collection = models.Collection.objects.create(
            title="foreign search collection",
            project=other_project,
        )
        other_metadata_set = models.MetadataSet.objects.create(
            title="foreign search metadata set",
            project=other_project,
        )
        other_metadata = models.Metadata.objects.create(
            title="foreign search metadata",
            set=other_metadata_set,
            project=other_project,
        )

        with self.assertRaises(ServiceException) as context:
            methods.advanced_search(
                self.test_user,
                [],
                self.test_project.pk,
                collection_id=other_collection.pk,
            )
        self.assertEqual(context.exception.message, methods.NO_SUCH_COLLECTION)

        for foreign_metadata_term in (
            {"meta": other_metadata.pk, "term": "is", "value": "value"},
            {"exclude_meta": other_metadata.pk},
        ):
            with self.subTest(search_term=foreign_metadata_term):
                with self.assertRaises(ServiceException) as context:
                    methods.advanced_search(
                        self.test_user,
                        [foreign_metadata_term],
                        self.test_project.pk,
                    )
                self.assertEqual(context.exception.message, methods.NO_SUCH_METADATA)

    def test_descendants_resources_count(self):
        col = models.Collection.objects.create(
            title="desc col",
@@ -1423,6 +1500,20 @@ class ServiceTestCase(TestCase):
        result = methods.meta_count(self.admin_user, metadata_id, col.pk)
        self.assertEqual(result["test meta value C"], 1)

        other_project = models.Project.objects.create(label="meta count other project")
        other_metadata_set = models.MetadataSet.objects.create(
            project=other_project,
            title="meta count other set",
        )
        other_metadata = models.Metadata.objects.create(
            project=other_project,
            set=other_metadata_set,
            title="meta count other metadata",
        )
        with self.assertRaises(ServiceException) as context:
            methods.meta_count(self.admin_user, other_metadata.pk, col.pk)
        self.assertEqual(context.exception.message, methods.NO_SUCH_METADATA)

    def test_project_properties(self):
        test_property = methods.set_project_property(
            self.admin_user,