mirror of
https://github.com/suitenumerique/drive.git
synced 2026-08-17 20:15:40 +02:00
✨(backend) filter items list by shared contact
Add a contact filter on the items list returning the items in which the given user is involved through a sharing access, in either direction, so users can find files they share with a given person.
This commit is contained in:
@@ -2,7 +2,7 @@
|
||||
|
||||
from itertools import chain
|
||||
|
||||
from django.db.models import Q, TextChoices
|
||||
from django.db.models import Exists, OuterRef, Q, TextChoices
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
import django_filters
|
||||
@@ -25,10 +25,11 @@ class ItemFilter(django_filters.FilterSet):
|
||||
category = django_filters.ChoiceFilter(
|
||||
method="filter_category", label=_("File type"), choices=enums.FILE_CATEGORY_CHOICES
|
||||
)
|
||||
contact = django_filters.UUIDFilter(method="filter_contact", label=_("Shared with"))
|
||||
|
||||
class Meta:
|
||||
model = models.Item
|
||||
fields = ["title", "type", "category"]
|
||||
fields = ["title", "type", "category", "contact"]
|
||||
|
||||
@staticmethod
|
||||
def _extensions_q(extensions):
|
||||
@@ -63,6 +64,24 @@ class ItemFilter(django_filters.FilterSet):
|
||||
|
||||
return queryset.filter(is_folder | (is_file & matched))
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
def filter_contact(self, queryset, name, value):
|
||||
"""
|
||||
Filter items in which the given contact is involved in sharing, in either
|
||||
direction.
|
||||
|
||||
"Shared with" the contact: they hold an access on the item or one of its
|
||||
ancestors. "Shared by" the contact: they created the item. Both directions
|
||||
are matched.
|
||||
|
||||
Example:
|
||||
- /api/v1.0/items/?contact=<user_id>
|
||||
→ Filters items shared with or by the given user
|
||||
"""
|
||||
contact_access = models.ItemAccess.objects.filter(
|
||||
user_id=value, item__path__ancestors=OuterRef("path")
|
||||
)
|
||||
return queryset.filter(Exists(contact_access) | Q(creator_id=value))
|
||||
|
||||
|
||||
class ItemOrdering(OrderingFilter):
|
||||
|
||||
@@ -1513,3 +1513,25 @@ def test_api_items_children_list_filter_category():
|
||||
assert response.status_code == 200
|
||||
results = response.json()["results"]
|
||||
assert {result["id"] for result in results} == {str(png.id)}
|
||||
|
||||
|
||||
def test_api_items_children_list_filter_contact_inherited():
|
||||
"""Filtering children by contact includes items shared through an ancestor."""
|
||||
user = factories.UserFactory()
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
contact = factories.UserFactory()
|
||||
|
||||
parent = factories.ItemFactory(type=models.ItemTypeChoices.FOLDER, users=[user, contact])
|
||||
child = factories.ItemFactory(
|
||||
parent=parent,
|
||||
type=models.ItemTypeChoices.FILE,
|
||||
filename="doc.txt",
|
||||
update_upload_state=models.ItemUploadStateChoices.READY,
|
||||
)
|
||||
|
||||
response = client.get(f"/api/v1.0/items/{parent.id!s}/children/?contact={contact.id!s}")
|
||||
|
||||
assert response.status_code == 200
|
||||
results = response.json()["results"]
|
||||
assert {result["id"] for result in results} == {str(child.id)}
|
||||
|
||||
@@ -521,3 +521,69 @@ def test_api_items_list_filter_category_invalid():
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
# Filters: contact
|
||||
|
||||
|
||||
def test_api_items_list_filter_contact():
|
||||
"""Filtering by contact should return items shared with that contact."""
|
||||
user, client = _login()
|
||||
contact = factories.UserFactory()
|
||||
|
||||
shared = factories.ItemFactory(users=[user, contact], type=models.ItemTypeChoices.FOLDER)
|
||||
factories.ItemFactory(users=[user], type=models.ItemTypeChoices.FOLDER)
|
||||
|
||||
response = client.get(f"/api/v1.0/items/?contact={contact.id!s}")
|
||||
|
||||
assert response.status_code == 200
|
||||
results = response.json()["results"]
|
||||
assert {result["id"] for result in results} == {str(shared.id)}
|
||||
|
||||
|
||||
def test_api_items_list_filter_contact_no_duplicates():
|
||||
"""An item shared with several users should not be duplicated in the results."""
|
||||
user, client = _login()
|
||||
contact = factories.UserFactory()
|
||||
other = factories.UserFactory()
|
||||
|
||||
factories.ItemFactory(users=[user, contact, other], type=models.ItemTypeChoices.FOLDER)
|
||||
|
||||
response = client.get(f"/api/v1.0/items/?contact={contact.id!s}")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()["results"]) == 1
|
||||
|
||||
|
||||
def test_api_items_list_filter_contact_respects_access_rights():
|
||||
"""Filtering by contact must not leak items the current user cannot access."""
|
||||
_user, client = _login()
|
||||
contact = factories.UserFactory()
|
||||
|
||||
# Item shared with the contact but not with the current user.
|
||||
factories.ItemFactory(
|
||||
users=[contact], link_reach="restricted", type=models.ItemTypeChoices.FOLDER
|
||||
)
|
||||
|
||||
response = client.get(f"/api/v1.0/items/?contact={contact.id!s}")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["results"] == []
|
||||
|
||||
|
||||
def test_api_items_list_filter_contact_as_creator():
|
||||
"""Filtering by contact includes items the contact created and shared (shared by)."""
|
||||
user, client = _login()
|
||||
contact = factories.UserFactory()
|
||||
|
||||
created_by_contact = factories.ItemFactory(
|
||||
users=[user], creator=contact, type=models.ItemTypeChoices.FOLDER
|
||||
)
|
||||
factories.ItemFactory(users=[user], type=models.ItemTypeChoices.FOLDER)
|
||||
|
||||
response = client.get(f"/api/v1.0/items/?contact={contact.id!s}")
|
||||
|
||||
assert response.status_code == 200
|
||||
results = response.json()["results"]
|
||||
assert {result["id"] for result in results} == {str(created_by_contact.id)}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user