3684567753

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/kobotoolbox/kpi/commit/<a class=hub.com/kobotoolbox/kpi/commit/<a class="double-link" href="https://git"><a class=hub.com/kobotoolbox/kpi/commit/<a class="double-link" href="https://git"><a class=hub.com/kobotoolbox/kpi/commit/<a class="double-link" href="https://git"><a class=hub.com/kobotoolbox/kpi/commit/4ba51b2c4824e8d1bfcb963f91ef515d2d80a873">4ba51b2c4">&lt;a href=&quot;https://github.com/kobotoolbox/kpi/commit/</a><a class="double-link" href="https://github.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/4ba51b2c4824e8d1bfcb963f91ef515d2d80a873">4ba51b2c4</a><a href="https://github.com/kobotoolbox/kpi/commit/4ba51b2c4824e8d1bfcb963f91ef515d2d80a873">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/kobotoolbox/kpi/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/4ba51b2c4824e8d1bfcb963f91ef515d2d80a873&quot;&gt;4ba51b2c4&lt;/a&gt;&lt;a href=&quot;https://github.com/kobotoolbox/kpi/commit/4ba51b2c4824e8d1bfcb963f91ef515d2d80a873&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/kobotoolbox/kpi/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/kobotoolbox/kpi/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&am... (continued)

Pull Request Pull Request #4191: Add support for assets list ordering in project views

Run Details

2531 of 4256 branches covered (59.47%)

11 of 11 new or added lines in 2 files covered. (100.0%)

11149 of 14755 relevant lines covered (75.56%)

1.51 hits per line

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

82.59

/kpi/filters.py

# coding: utf-8
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import FieldError
from django.db.models import (
    Case,
    Count,
    F,
    IntegerField,
    Max,
    OuterRef,
    Q,
    Value,
    When,
)
from django.db.models.query import QuerySet
from rest_framework import filters
from rest_framework.request import Request

from kpi.constants import (
    ASSET_SEARCH_DEFAULT_FIELD_LOOKUPS,
    ASSET_STATUS_SHARED,
    ASSET_STATUS_DISCOVERABLE,
    ASSET_STATUS_PRIVATE,
    ASSET_STATUS_PUBLIC,
    PERM_DISCOVER_ASSET,
    PERM_PARTIAL_SUBMISSIONS,
    PERM_VIEW_ASSET,
    PERM_VIEW_SUBMISSIONS,
)
from kpi.exceptions import SearchQueryTooShortException
from kpi.models.asset import UserAssetSubscription
from kpi.models.asset_version import AssetVersion
from kpi.utils.query_parser import get_parsed_parameters, parse, ParseError
from kpi.utils.object_permission import (
    get_objects_for_user,
    get_anonymous_user,
    get_perm_ids_from_code_names,
)
from kpi.utils.permissions import is_user_anonymous
from .models import Asset, ObjectPermission


class AssetOwnerFilterBackend(filters.BaseFilterBackend):
    """
    For use with nested models of Asset.
    Restricts access to items that are owned by the current user
    """

    def filter_queryset(self, request, queryset, view):
        fields = {"asset__owner": request.user}
        return queryset.filter(**fields)


class AssetOrderingFilter(filters.OrderingFilter):

    def filter_queryset(self, request, queryset, view):
        query_params = request.query_params
        collections_first = (
            query_params.get('collections_first', 'false').lower() == 'true'
        )
        ordering = self.get_ordering(request, queryset, view)

        if collections_first:
            # If `collections_first` is `True`, we want the collections to be
            # displayed at the beginning. We add a temporary field to set a
            # priority to 1 for collections and all other asset types to 0.
            # The results are sorted by this priority first and then by what
            # comes next, either:
            # - `ordering` from querystring
            # - model default option
            queryset = queryset.annotate(
                ordering_priority=Case(
                    When(asset_type='collection', then=Value(1)),
                    default=Value(0),
                    output_field=IntegerField(),
                ),
            )
            if ordering is None:
                # Make a copy, we don't want to alter Asset._meta.ordering
                ordering = Asset._meta.ordering.copy()
            ordering.insert(0, '-ordering_priority')

        if ordering:
            if (
                'subscribers_count' in ordering
                or '-subscribers_count' in ordering
            ):
                queryset = queryset.annotate(
                    subscribers_count=Count('userassetsubscription__user')
                )

            if (
                'date_deployed' in ordering
                or '-date_deployed' in ordering
            ):
                subquery_last_deployed = (
                    AssetVersion.objects.values('asset_id')
                    .annotate(last_deployed=Max('date_modified'))
                    .filter(asset_id=OuterRef('pk'))
                    .values('last_deployed')
                )
                queryset = queryset.annotate(date_deployed=subquery_last_deployed)

            return queryset.order_by(*ordering)

        return queryset


class KpiObjectPermissionsFilter:

    STATUS_PARAMETER = 'status'
    PARENT_UID_PARAMETER = 'parent__uid'
    DATA_SHARING_PARAMETER = 'data_sharing__enabled'

    def filter_queryset(self, request, queryset, view):
        user = request.user
        if user.is_superuser and view.action != 'list':
            # For a list, we won't deluge the superuser with everyone else's
            # stuff. This isn't a list, though, so return it all
            return queryset

        queryset = self._get_queryset_for_collection_statuses(request, queryset)
        if self._return_queryset:
            return queryset.distinct()

        queryset = self._get_queryset_for_data_sharing_enabled(
            request, queryset
        )
        if self._return_queryset:
            return queryset.distinct()

        # Getting the children of discoverable public collections
        queryset = self._get_queryset_for_discoverable_child_assets(
            request, queryset
        )
        if self._return_queryset:
            return queryset.distinct()

        owned_and_explicit_shared = self._get_owned_and_explicitly_shared(user)

        if view.action != 'list':
            # Not a list, so discoverability doesn't matter
            assets = owned_and_explicit_shared.union(self._get_publics())
            return queryset.filter(pk__in=assets)

        subscribed = self._get_subscribed(user)

        # As other places in the code, coerce `asset_ids` as a list to force
        # the query to be processed right now. Otherwise, because queryset is
        # a lazy query, Django creates (left) joins on tables when queryset is
        # interpreted and it is way slower than running this extra query.
        asset_ids = list(
            (
                owned_and_explicit_shared
                    .union(subscribed)
                    # Since user would be subscribed to a collection and not
                    # the assets themselves, we append children of subscribed
                    # collections to the queryset in order for `?q=parent__uid`
                    # queries to return the collection's children
                    .union(queryset.filter(parent__in=subscribed).values('pk'))
            ).values_list('id', flat=True)
        )
        return queryset.filter(pk__in=asset_ids)

    def _get_queryset_for_data_sharing_enabled(
        self, request: Request, queryset: QuerySet
    ) -> QuerySet:
        """
        Returns a queryset containing the assets that the user is allowed
        to view data.
        I.e., user needs 'view_submissions' or 'partial_submissions'.
        """

        self._return_queryset = False
        parameters = self.__get_parsed_parameters(request)
        try:
            data_sharing = parameters[self.DATA_SHARING_PARAMETER][0]
        except KeyError:
            return queryset

        if not data_sharing:  # No reason to be False, but just in case
            return queryset

        self._return_queryset = True
        required_perm_ids = get_perm_ids_from_code_names(
            [PERM_VIEW_SUBMISSIONS, PERM_PARTIAL_SUBMISSIONS]
        )
        user = request.user
        if is_user_anonymous(user):
            # Avoid giving anonymous users special treatment when viewing
            # public objects
            perms = ObjectPermission.objects.none()
        else:
            perms = ObjectPermission.objects.filter(
                deny=False,
                user=user,
                permission_id__in=required_perm_ids)

        asset_ids = perms.values('asset')

        # `SearchFilter` handles further filtering to include only assets with
        # `data_sharing__enabled` (`self.DATA_SHARING_PARAMETER`) set to true
        return queryset.filter(pk__in=asset_ids)

    def _get_discoverable(self, queryset):
        # We were asked not to consider subscriptions; return all
        # discoverable objects
        return get_objects_for_user(
            get_anonymous_user(), PERM_DISCOVER_ASSET,
            queryset
        )

    def _get_queryset_for_collection_statuses(self, request, queryset):
        """
        Narrow down the queryset based on `status` parameter.
        It is useful when fetching Assets of type `collection`

        If `status` is not detected in `q`, it returns the queryset as is.
        Otherwise, there are 4 scenarios:
        - `status` == 'private': collections owned by user
        - `status` == 'shared': collections user can view
        - `status` == 'public': collections user has subscribed to (ONLY)
        - `status` == 'public-discoverable': all public collections

        Args:
            request
            queryset
        Returns:
            QuerySet
        """

        # Governs whether returned queryset should be processed immediately
        # and should stop other filtering on `queryset` in parent method of
        # `_get_queryset_for_collection_statuses()`
        self._return_queryset = False
        user = request.user

        try:
            status = request.query_params[self.STATUS_PARAMETER].strip()
        except KeyError:
            return queryset

        if status == ASSET_STATUS_PRIVATE:
            self._return_queryset = True
            return queryset.filter(owner=request.user)

        elif status == ASSET_STATUS_SHARED:
            self._return_queryset = True
            return get_objects_for_user(user, PERM_VIEW_ASSET, queryset)

        elif status == ASSET_STATUS_PUBLIC:
            self._return_queryset = True
            return queryset.filter(pk__in=self._get_subscribed(user))

        elif status == ASSET_STATUS_DISCOVERABLE:
            self._return_queryset = True
            discoverable = self._get_discoverable(queryset)
            # We were asked not to consider subscriptions; return all
            # discoverable objects
            return discoverable

        return queryset

    def _get_queryset_for_discoverable_child_assets(
        self, request: Request, queryset: QuerySet
    ) -> QuerySet:
        """
        Returns a queryset containing the children of publically discoverable
        assets based on the discoverability of the child's parent. The parent
        uid is passed in the request query string.
        """

        self._return_queryset = False

        parameters = self.__get_parsed_parameters(request)
        try:
            parent_uids = parameters[self.PARENT_UID_PARAMETER]
        except KeyError:
            return queryset

        # `self.__get_parsed_parameters()` returns a list for each parameters
        # but we should only search only with one parent uid
        parent_obj = queryset.get(uid=parent_uids[0])

        if not isinstance(parent_obj, Asset):
            return queryset

        if parent_obj.has_perm(get_anonymous_user(), PERM_DISCOVER_ASSET):
            self._return_queryset = True
            return queryset.filter(
                pk__in=self._get_publics(), parent=parent_obj
            )

        return queryset

    @staticmethod
    def _get_owned_and_explicitly_shared(user):
        view_asset_perm_id = get_perm_ids_from_code_names(PERM_VIEW_ASSET)
        if is_user_anonymous(user):
            # Avoid giving anonymous users special treatment when viewing
            # public objects
            perms = ObjectPermission.objects.none()
        else:
            perms = ObjectPermission.objects.filter(
                deny=False,
                user=user,
                permission_id=view_asset_perm_id)

        return perms.values('asset')

    @staticmethod
    def _get_publics():
        view_asset_perm_id = get_perm_ids_from_code_names(PERM_VIEW_ASSET)
        return ObjectPermission.objects.filter(
            deny=False,
            user=get_anonymous_user(),
            permission_id=view_asset_perm_id).values('asset')

    @classmethod
    def _get_subscribed(cls, user):
        # Of the public objects, determine to which the user has subscribed
        if is_user_anonymous(user):
            user = get_anonymous_user()

        return UserAssetSubscription.objects.filter(
            asset__in=cls._get_publics(), user=user
        ).values('asset')

    def __get_parsed_parameters(self, request: Request) -> dict:
        """
        Returns the results of `get_parsed_parameters` utility.
        Useful in this class to detect specific parameters and to return
        according queryset
        """
        try:
            q = request.query_params['q']
        except KeyError:
            return {}

        try:
            q_obj = parse(
                q, default_field_lookups=ASSET_SEARCH_DEFAULT_FIELD_LOOKUPS
            )
        except (ParseError, SearchQueryTooShortException):
            # Let's `SearchFilter` handle errors
            return {}
        else:
            return get_parsed_parameters(q_obj)


class RelatedAssetPermissionsFilter(KpiObjectPermissionsFilter):
    """
    Uses KpiObjectPermissionsFilter to determine which assets the user
    may access, and then filters the provided queryset to include only objects
    related to those assets. The queryset's model must be related to `Asset`
    via a field named `asset`.
    """

    def filter_queryset(self, request, queryset, view):
        available_assets = super().filter_queryset(
            request=request,
            queryset=Asset.objects.all(),
            view=view
        )
        return queryset.filter(asset__in=available_assets)


class SearchFilter(filters.BaseFilterBackend):
    """
    If the request includes a `q` parameter specifying a Boolean search string
    with a Whoosh-like syntax, filter the queryset accordingly using the ORM.
    If no `q` is present, return the queryset untouched. If `q` is not
    parseable, references a field that does not exist, or specifies an invalid
    value for a field (e.g. text for an integer field), return an empty
    queryset to make the problem obvious.
    """

    def filter_queryset(self, request, queryset, view):
        try:
            q = request.query_params['q']
        except KeyError:
            return queryset

        try:
            q_obj = parse(
                q,
                default_field_lookups=view.search_default_field_lookups,
                min_search_characters=getattr(
                    view, 'min_search_characters', None
                ),
            )
        except ParseError:
            return queryset.model.objects.none()
        except SearchQueryTooShortException as e:
            # raising an exception if the default search query without a
            # specified field is less than a set length of characters -
            # currently 3 (see `settings.MINIMUM_DEFAULT_SEARCH_CHARACTERS`)
            raise e

        try:
            # If no search field is specified, the search term is compared
            # to several default fields and therefore may return a copies
            # of the same match, therefore the `distinct()` method is required
            return queryset.filter(q_obj).distinct()
        except (FieldError, ValueError):
            return queryset.model.objects.none()


class KpiAssignedObjectPermissionsFilter(filters.BaseFilterBackend):
    """
    Used by kpi.views.v1.object_permission.ObjectPermissionViewSet only
    """

    def filter_queryset(self, request, queryset, view):
        # TODO: omit objects for which the user has only a deny permission
        user = request.user
        if isinstance(request.user, AnonymousUser):
            user = get_anonymous_user()

        if user.is_superuser:
            # Superuser sees all
            return queryset
        if user.pk == settings.ANONYMOUS_USER_ID:
            # Hide permissions from anonymous users
            return queryset.none()
        """
        A regular user sees their own permissions and the owner's permissions
        for objects to which they have access. For example, if Alana and John
        have view access to an object owned by Richard, John should see all of
        his own permissions and Richard's permissions, but not any of Alana's
        permissions.
        """
        result = ObjectPermission.objects.filter(
            Q(asset__owner=user)  # owner sees everything
            | Q(user=user)  # everyone with access sees their own
            | Q(
                # everyone with access sees the owner's
                asset__permissions__user=user, user=F('asset__owner')
            )
        ).distinct()
        return result

1	# coding: utf-8
2	from django.conf import settings	2✔
3	from django.contrib.auth.models import AnonymousUser	2✔
4	from django.core.exceptions import FieldError	2✔
5	from django.db.models import (	2✔
6	Case,
7	Count,
8	F,
9	IntegerField,
10	Max,
11	OuterRef,
12	Q,
13	Value,
14	When,
15	)
16	from django.db.models.query import QuerySet	2✔
17	from rest_framework import filters	2✔
18	from rest_framework.request import Request	2✔
19
20	from kpi.constants import (	2✔
21	ASSET_SEARCH_DEFAULT_FIELD_LOOKUPS,
22	ASSET_STATUS_SHARED,
23	ASSET_STATUS_DISCOVERABLE,
24	ASSET_STATUS_PRIVATE,
25	ASSET_STATUS_PUBLIC,
26	PERM_DISCOVER_ASSET,
27	PERM_PARTIAL_SUBMISSIONS,
28	PERM_VIEW_ASSET,
29	PERM_VIEW_SUBMISSIONS,
30	)
31	from kpi.exceptions import SearchQueryTooShortException	2✔
32	from kpi.models.asset import UserAssetSubscription	2✔
33	from kpi.models.asset_version import AssetVersion	2✔
34	from kpi.utils.query_parser import get_parsed_parameters, parse, ParseError	2✔
35	from kpi.utils.object_permission import (	2✔
36	get_objects_for_user,
37	get_anonymous_user,
38	get_perm_ids_from_code_names,
39	)
40	from kpi.utils.permissions import is_user_anonymous	2✔
41	from .models import Asset, ObjectPermission	2✔
42
43
44	class AssetOwnerFilterBackend(filters.BaseFilterBackend):	2✔
45	"""
46	For use with nested models of Asset.
47	Restricts access to items that are owned by the current user
48	"""
49
50	def filter_queryset(self, request, queryset, view):	2✔
51	fields = {"asset__owner": request.user}	2✔
52	return queryset.filter(**fields)	2✔
53
54
55	class AssetOrderingFilter(filters.OrderingFilter):	2✔
56
57	def filter_queryset(self, request, queryset, view):	2✔
58	query_params = request.query_params	2✔
59	collections_first = (	2✔
60	query_params.get('collections_first', 'false').lower() == 'true'
61	)
62	ordering = self.get_ordering(request, queryset, view)	2✔
63
64	if collections_first:	2✔
65	# If `collections_first` is `True`, we want the collections to be
66	# displayed at the beginning. We add a temporary field to set a
67	# priority to 1 for collections and all other asset types to 0.
68	# The results are sorted by this priority first and then by what
69	# comes next, either:
70	# - `ordering` from querystring
71	# - model default option
72	queryset = queryset.annotate(	2✔
73	ordering_priority=Case(
74	When(asset_type='collection', then=Value(1)),
75	default=Value(0),
76	output_field=IntegerField(),
77	),
78	)
79	if ordering is None:	2!
80	# Make a copy, we don't want to alter Asset._meta.ordering
81	ordering = Asset._meta.ordering.copy()	×
82	ordering.insert(0, '-ordering_priority')	2✔
83
84	if ordering:	2✔
85	if (	2!
86	'subscribers_count' in ordering
87	or '-subscribers_count' in ordering
88	):
89	queryset = queryset.annotate(	×
90	subscribers_count=Count('userassetsubscription__user')
91	)
92
93	if (	2✔
94	'date_deployed' in ordering
95	or '-date_deployed' in ordering
96	):
97	subquery_last_deployed = (	2✔
98	AssetVersion.objects.values('asset_id')
99	.annotate(last_deployed=Max('date_modified'))
100	.filter(asset_id=OuterRef('pk'))
101	.values('last_deployed')
102	)
103	queryset = queryset.annotate(date_deployed=subquery_last_deployed)	2✔
104
105	return queryset.order_by(*ordering)	2✔
106
107	return queryset	2✔
108
109
110	class KpiObjectPermissionsFilter:	2✔
111
112	STATUS_PARAMETER = 'status'	2✔
113	PARENT_UID_PARAMETER = 'parent__uid'	2✔
114	DATA_SHARING_PARAMETER = 'data_sharing__enabled'	2✔
115
116	def filter_queryset(self, request, queryset, view):	2✔
117	user = request.user	2✔
118	if user.is_superuser and view.action != 'list':	2✔
119	# For a list, we won't deluge the superuser with everyone else's
120	# stuff. This isn't a list, though, so return it all
121	return queryset	2✔
122
123	queryset = self._get_queryset_for_collection_statuses(request, queryset)	2✔
124	if self._return_queryset:	2✔
125	return queryset.distinct()	2✔
126
127	queryset = self._get_queryset_for_data_sharing_enabled(	2✔
128	request, queryset
129	)
130	if self._return_queryset:	2!
131	return queryset.distinct()	×
132
133	# Getting the children of discoverable public collections
134	queryset = self._get_queryset_for_discoverable_child_assets(	2✔
135	request, queryset
136	)
137	if self._return_queryset:	2✔
138	return queryset.distinct()	2✔
139
140	owned_and_explicit_shared = self._get_owned_and_explicitly_shared(user)	2✔
141
142	if view.action != 'list':	2✔
143	# Not a list, so discoverability doesn't matter
144	assets = owned_and_explicit_shared.union(self._get_publics())	2✔
145	return queryset.filter(pk__in=assets)	2✔
146
147	subscribed = self._get_subscribed(user)	2✔
148
149	# As other places in the code, coerce `asset_ids` as a list to force
150	# the query to be processed right now. Otherwise, because queryset is
151	# a lazy query, Django creates (left) joins on tables when queryset is
152	# interpreted and it is way slower than running this extra query.
153	asset_ids = list(	2✔
154	(
155	owned_and_explicit_shared
156	.union(subscribed)
157	# Since user would be subscribed to a collection and not
158	# the assets themselves, we append children of subscribed
159	# collections to the queryset in order for `?q=parent__uid`
160	# queries to return the collection's children
161	.union(queryset.filter(parent__in=subscribed).values('pk'))
162	).values_list('id', flat=True)
163	)
164	return queryset.filter(pk__in=asset_ids)	2✔
165
166	def _get_queryset_for_data_sharing_enabled(	2✔
167	self, request: Request, queryset: QuerySet
168	) -> QuerySet:
169	"""
170	Returns a queryset containing the assets that the user is allowed
171	to view data.
172	I.e., user needs 'view_submissions' or 'partial_submissions'.
173	"""
174
175	self._return_queryset = False	2✔
176	parameters = self.__get_parsed_parameters(request)	2✔
177	try:	2✔
178	data_sharing = parameters[self.DATA_SHARING_PARAMETER][0]	2✔
179	except KeyError:	2✔
180	return queryset	2✔
181
182	if not data_sharing: # No reason to be False, but just in case	×
183	return queryset	×
184
185	self._return_queryset = True	×
186	required_perm_ids = get_perm_ids_from_code_names(	×
187	[PERM_VIEW_SUBMISSIONS, PERM_PARTIAL_SUBMISSIONS]
188	)
189	user = request.user	×
190	if is_user_anonymous(user):	×
191	# Avoid giving anonymous users special treatment when viewing
192	# public objects
193	perms = ObjectPermission.objects.none()	×
194	else:
195	perms = ObjectPermission.objects.filter(	×
196	deny=False,
197	user=user,
198	permission_id__in=required_perm_ids)
199
200	asset_ids = perms.values('asset')	×
201
202	# `SearchFilter` handles further filtering to include only assets with
203	# `data_sharing__enabled` (`self.DATA_SHARING_PARAMETER`) set to true
204	return queryset.filter(pk__in=asset_ids)	×
205
206	def _get_discoverable(self, queryset):	2✔
207	# We were asked not to consider subscriptions; return all
208	# discoverable objects
209	return get_objects_for_user(	2✔
210	get_anonymous_user(), PERM_DISCOVER_ASSET,
211	queryset
212	)
213
214	def _get_queryset_for_collection_statuses(self, request, queryset):	2✔
215	"""
216	Narrow down the queryset based on `status` parameter.
217	It is useful when fetching Assets of type `collection`
218
219	If `status` is not detected in `q`, it returns the queryset as is.
220	Otherwise, there are 4 scenarios:
221	- `status` == 'private': collections owned by user
222	- `status` == 'shared': collections user can view
223	- `status` == 'public': collections user has subscribed to (ONLY)
224	- `status` == 'public-discoverable': all public collections
225
226	Args:
227	request
228	queryset
229	Returns:
230	QuerySet
231	"""
232
233	# Governs whether returned queryset should be processed immediately
234	# and should stop other filtering on `queryset` in parent method of
235	# `_get_queryset_for_collection_statuses()`
236	self._return_queryset = False	2✔
237	user = request.user	2✔
238
239	try:	2✔
240	status = request.query_params[self.STATUS_PARAMETER].strip()	2✔
241	except KeyError:	2✔
242	return queryset	2✔
243
244	if status == ASSET_STATUS_PRIVATE:	2!
245	self._return_queryset = True	×
246	return queryset.filter(owner=request.user)	×
247
248	elif status == ASSET_STATUS_SHARED:	2!
249	self._return_queryset = True	×
250	return get_objects_for_user(user, PERM_VIEW_ASSET, queryset)	×
251
252	elif status == ASSET_STATUS_PUBLIC:	2!
253	self._return_queryset = True	×
254	return queryset.filter(pk__in=self._get_subscribed(user))	×
255
256	elif status == ASSET_STATUS_DISCOVERABLE:	2!
257	self._return_queryset = True	2✔
258	discoverable = self._get_discoverable(queryset)	2✔
259	# We were asked not to consider subscriptions; return all
260	# discoverable objects
261	return discoverable	2✔
262
263	return queryset	×
264
265	def _get_queryset_for_discoverable_child_assets(	2✔
266	self, request: Request, queryset: QuerySet
267	) -> QuerySet:
268	"""
269	Returns a queryset containing the children of publically discoverable
270	assets based on the discoverability of the child's parent. The parent
271	uid is passed in the request query string.
272	"""
273
274	self._return_queryset = False	2✔
275
276	parameters = self.__get_parsed_parameters(request)	2✔
277	try:	2✔
278	parent_uids = parameters[self.PARENT_UID_PARAMETER]	2✔
279	except KeyError:	2✔
280	return queryset	2✔
281
282	# `self.__get_parsed_parameters()` returns a list for each parameters
283	# but we should only search only with one parent uid
284	parent_obj = queryset.get(uid=parent_uids[0])	2✔
285
286	if not isinstance(parent_obj, Asset):	2!
287	return queryset	×
288
289	if parent_obj.has_perm(get_anonymous_user(), PERM_DISCOVER_ASSET):	2!
290	self._return_queryset = True	2✔
291	return queryset.filter(	2✔
292	pk__in=self._get_publics(), parent=parent_obj
293	)
294
295	return queryset	×
296
297	@staticmethod	2✔
298	def _get_owned_and_explicitly_shared(user):	2✔
299	view_asset_perm_id = get_perm_ids_from_code_names(PERM_VIEW_ASSET)	2✔
300	if is_user_anonymous(user):	2✔
301	# Avoid giving anonymous users special treatment when viewing
302	# public objects
303	perms = ObjectPermission.objects.none()	2✔
304	else:
305	perms = ObjectPermission.objects.filter(	2✔
306	deny=False,
307	user=user,
308	permission_id=view_asset_perm_id)
309
310	return perms.values('asset')	2✔
311
312	@staticmethod	2✔
313	def _get_publics():	2✔
314	view_asset_perm_id = get_perm_ids_from_code_names(PERM_VIEW_ASSET)	2✔
315	return ObjectPermission.objects.filter(	2✔
316	deny=False,
317	user=get_anonymous_user(),
318	permission_id=view_asset_perm_id).values('asset')
319
320	@classmethod	2✔
321	def _get_subscribed(cls, user):	2✔
322	# Of the public objects, determine to which the user has subscribed
323	if is_user_anonymous(user):	2✔
324	user = get_anonymous_user()	2✔
325
326	return UserAssetSubscription.objects.filter(	2✔
327	asset__in=cls._get_publics(), user=user
328	).values('asset')
329
330	def __get_parsed_parameters(self, request: Request) -> dict:	2✔
331	"""
332	Returns the results of `get_parsed_parameters` utility.
333	Useful in this class to detect specific parameters and to return
334	according queryset
335	"""
336	try:	2✔
337	q = request.query_params['q']	2✔
338	except KeyError:	2✔
339	return {}	2✔
340
341	try:	2✔
342	q_obj = parse(	2✔
343	q, default_field_lookups=ASSET_SEARCH_DEFAULT_FIELD_LOOKUPS
344	)
345	except (ParseError, SearchQueryTooShortException):	×
346	# Let's `SearchFilter` handle errors
347	return {}	×
348	else:
349	return get_parsed_parameters(q_obj)	2✔
350
351
352	class RelatedAssetPermissionsFilter(KpiObjectPermissionsFilter):	2✔
353	"""
354	Uses KpiObjectPermissionsFilter to determine which assets the user
355	may access, and then filters the provided queryset to include only objects
356	related to those assets. The queryset's model must be related to `Asset`
357	via a field named `asset`.
358	"""
359
360	def filter_queryset(self, request, queryset, view):	2✔
361	available_assets = super().filter_queryset(	2✔
362	request=request,
363	queryset=Asset.objects.all(),
364	view=view
365	)
366	return queryset.filter(asset__in=available_assets)	2✔
367
368
369	class SearchFilter(filters.BaseFilterBackend):	2✔
370	"""
371	If the request includes a `q` parameter specifying a Boolean search string
372	with a Whoosh-like syntax, filter the queryset accordingly using the ORM.
373	If no `q` is present, return the queryset untouched. If `q` is not
374	parseable, references a field that does not exist, or specifies an invalid
375	value for a field (e.g. text for an integer field), return an empty
376	queryset to make the problem obvious.
377	"""
378
379	def filter_queryset(self, request, queryset, view):	2✔
380	try:	2✔
381	q = request.query_params['q']	2✔
382	except KeyError:	2✔
383	return queryset	2✔
384
385	try:	2✔
386	q_obj = parse(	2✔
387	q,
388	default_field_lookups=view.search_default_field_lookups,
389	min_search_characters=getattr(
390	view, 'min_search_characters', None
391	),
392	)
393	except ParseError:	2!
394	return queryset.model.objects.none()	×
395	except SearchQueryTooShortException as e:	2✔
396	# raising an exception if the default search query without a
397	# specified field is less than a set length of characters -
398	# currently 3 (see `settings.MINIMUM_DEFAULT_SEARCH_CHARACTERS`)
399	raise e	2✔
400
401	try:	2✔
402	# If no search field is specified, the search term is compared
403	# to several default fields and therefore may return a copies
404	# of the same match, therefore the `distinct()` method is required
405	return queryset.filter(q_obj).distinct()	2✔
406	except (FieldError, ValueError):	2✔
407	return queryset.model.objects.none()	2✔
408
409
410	class KpiAssignedObjectPermissionsFilter(filters.BaseFilterBackend):	2✔
411	"""
412	Used by kpi.views.v1.object_permission.ObjectPermissionViewSet only
413	"""
414
415	def filter_queryset(self, request, queryset, view):	2✔
416	# TODO: omit objects for which the user has only a deny permission
417	user = request.user	2✔
418	if isinstance(request.user, AnonymousUser):	2✔
419	user = get_anonymous_user()	2✔
420
421	if user.is_superuser:	2✔
422	# Superuser sees all
423	return queryset	2✔
424	if user.pk == settings.ANONYMOUS_USER_ID:	2✔
425	# Hide permissions from anonymous users
426	return queryset.none()	2✔
427	"""	1✔
428	A regular user sees their own permissions and the owner's permissions
429	for objects to which they have access. For example, if Alana and John
430	have view access to an object owned by Richard, John should see all of
431	his own permissions and Richard's permissions, but not any of Alana's
432	permissions.
433	"""
434	result = ObjectPermission.objects.filter(	2✔
435	Q(asset__owner=user) # owner sees everything
436	\| Q(user=user) # everyone with access sees their own
437	\| Q(
438	# everyone with access sees the owner's
439	asset__permissions__user=user, user=F('asset__owner')
440	)
441	).distinct()
442	return result	2✔

kobotoolbox / kpi / 3684567753

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