20227548215

Committed 15 Dec 2025 09:43AM UTC coverage: 94.814% (+0.02%) from 94.796%

Build # 20227548215

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1427 from rdmorganiser/2.4.0

RDMO 2.4.0 🎆

Run Details

2124 of 2229 branches covered (95.29%)

22688 of 23929 relevant lines covered (94.81%)

3.79 hits per line

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

89.45

rdmo/projects/managers.py

from collections import defaultdict

from django.conf import settings
from django.db import models
from django.db.models import Q

from mptt.models import TreeManager
from mptt.querysets import TreeQuerySet

from rdmo.accounts.utils import is_site_manager
from rdmo.core.managers import CurrentSiteManagerMixin


class ProjectQuerySet(TreeQuerySet):

    def filter_user(self, user, filter_for_user=False):
        if user.is_authenticated:
            if user.has_perm('projects.view_project') and not filter_for_user:
                # return all projects for admins or users with model permissions (unless filter_for_user is set)
                return self.all()
            else:
                # create a filter for all project where this user is a member
                user_filter = Q(user=user)

                # create a filter for all projects which are visible for this user
                groups = user.groups.all()
                sites_filter = Q(visibility__sites=None) | Q(visibility__sites=settings.SITE_ID)
                groups_filter = Q(visibility__groups=None) | Q(visibility__groups__in=groups)
                visibility_filter = Q(visibility__isnull=False) & sites_filter & groups_filter

                # create a filter for all projects of this site (unless filter_for_user is set)
                if (user.has_perm('projects.view_project') or is_site_manager(user)) and not filter_for_user:
                    current_site_filter = Q(site=settings.SITE_ID)
                else:
                    current_site_filter = Q()

                # create queryset by combining all three filters
                queryset = self.filter(user_filter | visibility_filter | current_site_filter)

                # add descendant projects
                for instance in queryset:
                    queryset |= instance.get_descendants()
                return queryset.distinct()
        else:
            return self.none()

    def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):
        catalogs_filter = Q()
        if exclude_null:
          catalogs_filter &= Q(catalog__isnull=False)
        if catalogs:
            catalogs_filter &= Q(catalog__in=catalogs)
        if exclude_catalogs:
            catalogs_filter &= ~Q(catalog__in=exclude_catalogs)
        return self.filter(catalogs_filter)

    def filter_groups(self, groups):
        if not groups:
            return self

        # need to import here to prevent circular import
        # queryset methods need to be refactored in modules otherwise
        from django.contrib.auth import get_user_model

        from rdmo.projects.models import Membership

        # users in the given groups
        users = get_user_model().objects.filter(groups__in=groups)
        # memberships for those users with role 'owner'
        memberships = Membership.objects.filter(role='owner', user__in=users)
        # projects that have those memberships
        return self.filter(memberships__in=memberships).distinct()

    def filter_projects_for_task_or_view(self, instance):
        # projects that have an unavailable catalog should be disregarded
        qs = self.filter(catalog__available=True)

        # when View/Task is not available it should not show for any project
        if not instance.available:
            return self.none()

        # when View/Task has any catalogs it can be filtered for those
        if instance.catalogs.exists():
            qs = qs.filter(catalog__in=instance.catalogs.all())

        # when View/Task has any sites it can be filtered for those
        if instance.sites.exists():
            qs = qs.filter(site__in=instance.sites.all())
        elif settings.MULTISITE:
            # when View/Task has no sites in a multi-site, it should not appear at all
            return self.none()

        # when  has any groups it can be filtered for those
        if instance.groups.exists():
            qs = qs.filter_groups(instance.groups.all())

        return qs


class MembershipQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_membership'):
                return self.all()
            elif is_site_manager(user):
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.objects.none()


class IssueQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_integration'):
                return self.all()
            elif is_site_manager(user):
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()


class IntegrationQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_issue'):
                return self.all()
            elif is_site_manager(user):
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()


class InviteQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_invite'):
                return self.all()
            elif is_site_manager(user):
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter(memberships__user=user, memberships__role='owner')
                return self.filter(project__in=projects)
        else:
            return self.none()


class SnapshotQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_snapshot'):
                return self.all()
            elif is_site_manager(user):
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()


class ValueQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_value'):
                return self.all()
            elif is_site_manager(user):
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()

    def filter_empty(self):
        return self.filter((Q(text='') | Q(text=None)) & Q(option=None) & (Q(file='') | Q(file=None)))

    def exclude_empty(self):
        return self.exclude((Q(text='') | Q(text=None)) & Q(option=None) & (Q(file='') | Q(file=None)))

    def exclude_empty_optional(self, catalog):
        optional_values = self.filter(attribute__in=[q.attribute for q in catalog.optional_questions])
        return self.exclude(id__in=optional_values.filter_empty().values_list('id', flat=True))

    def distinct_list(self):
        return self.order_by('attribute').values_list('attribute', 'set_prefix', 'set_index').distinct()

    def filter_set(self, set_value):
        # get the catalog and prefetch most elements of the catalog
        catalog = set_value.project.catalog
        catalog.prefetch_elements()

        # Get all attributes from matching elements and their descendants
        attributes = {
            descendant.attribute
            for element in (catalog.pages + catalog.questions)
            if element.attribute == set_value.attribute
            for descendant in element.descendants
        }

        # construct the set_prefix for descendants for this set
        descendants_set_prefix = \
            f'{set_value.set_prefix}|{set_value.set_index}' if set_value.set_prefix else str(set_value.set_index)

        # collect all values for this set and all descendants
        return self.filter(project=set_value.project, snapshot=set_value.snapshot, attribute__in=attributes).filter(
            Q(set_prefix=set_value.set_prefix, set_index=set_value.set_index) |
            Q(set_prefix__startswith=descendants_set_prefix)
        )

    def compute_sets(self):
        sets = defaultdict(set)
        for attribute, set_prefix, set_index in self.distinct_list():
            sets[attribute].add((set_prefix, set_index))
        return sets

class ProjectManager(CurrentSiteManagerMixin, TreeManager):

    def get_queryset(self):
        return ProjectQuerySet(self.model, using=self._db)

    def filter_user(self, user, filter_for_user=False):
        return self.get_queryset().filter_user(user, filter_for_user)

    def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):
        return self.get_queryset().filter_catalogs(catalogs=catalogs, exclude_catalogs=exclude_catalogs,
                                                   exclude_null=exclude_null)
    def filter_groups(self, groups):
        return self.get_queryset().filter_groups(groups)

    def filter_projects_for_task_or_view(self, instance):
        return self.get_queryset().filter_projects_for_task_or_view(instance)


class MembershipManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return MembershipQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class IssueManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return IssueQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)

    def active(self):
        return self.get_queryset().active()


class IntegrationManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return IntegrationQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class InviteManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return InviteQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class SnapshotManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return SnapshotQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class ValueManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return ValueQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)

    def compute_sets(self):
        return self.get_queryset().compute_sets()

1	from collections import defaultdict	4✔
2
3	from django.conf import settings	4✔
4	from django.db import models	4✔
5	from django.db.models import Q	4✔
6
7	from mptt.models import TreeManager	4✔
8	from mptt.querysets import TreeQuerySet	4✔
9
10	from rdmo.accounts.utils import is_site_manager	4✔
11	from rdmo.core.managers import CurrentSiteManagerMixin	4✔
12
13
14	class ProjectQuerySet(TreeQuerySet):	4✔
15
16	def filter_user(self, user, filter_for_user=False):	4✔
17	if user.is_authenticated:	4✔
18	if user.has_perm('projects.view_project') and not filter_for_user:	4✔
19	# return all projects for admins or users with model permissions (unless filter_for_user is set)
20	return self.all()	4✔
21	else:
22	# create a filter for all project where this user is a member
23	user_filter = Q(user=user)	4✔
24
25	# create a filter for all projects which are visible for this user
26	groups = user.groups.all()	4✔
27	sites_filter = Q(visibility__sites=None) \| Q(visibility__sites=settings.SITE_ID)	4✔
28	groups_filter = Q(visibility__groups=None) \| Q(visibility__groups__in=groups)	4✔
29	visibility_filter = Q(visibility__isnull=False) & sites_filter & groups_filter	4✔
30
31	# create a filter for all projects of this site (unless filter_for_user is set)
32	if (user.has_perm('projects.view_project') or is_site_manager(user)) and not filter_for_user:	4✔
33	current_site_filter = Q(site=settings.SITE_ID)	4✔
34	else:
35	current_site_filter = Q()	4✔
36
37	# create queryset by combining all three filters
38	queryset = self.filter(user_filter \| visibility_filter \| current_site_filter)	4✔
39
40	# add descendant projects
41	for instance in queryset:	4✔
42	queryset \|= instance.get_descendants()	4✔
43	return queryset.distinct()	4✔
44	else:
45	return self.none()	4✔
46
47	def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):	4✔
48	catalogs_filter = Q()	×
49	if exclude_null:	×
50	catalogs_filter &= Q(catalog__isnull=False)	×
51	if catalogs:	×
52	catalogs_filter &= Q(catalog__in=catalogs)	×
53	if exclude_catalogs:	×
54	catalogs_filter &= ~Q(catalog__in=exclude_catalogs)	×
55	return self.filter(catalogs_filter)	×
56
57	def filter_groups(self, groups):	4✔
58	if not groups:	4✔
59	return self	×
60
61	# need to import here to prevent circular import
62	# queryset methods need to be refactored in modules otherwise
63	from django.contrib.auth import get_user_model	4✔
64
65	from rdmo.projects.models import Membership	4✔
66
67	# users in the given groups
68	users = get_user_model().objects.filter(groups__in=groups)	4✔
69	# memberships for those users with role 'owner'
70	memberships = Membership.objects.filter(role='owner', user__in=users)	4✔
71	# projects that have those memberships
72	return self.filter(memberships__in=memberships).distinct()	4✔
73
74	def filter_projects_for_task_or_view(self, instance):	4✔
75	# projects that have an unavailable catalog should be disregarded
76	qs = self.filter(catalog__available=True)	4✔
77
78	# when View/Task is not available it should not show for any project
79	if not instance.available:	4✔
80	return self.none()	4✔
81
82	# when View/Task has any catalogs it can be filtered for those
83	if instance.catalogs.exists():	4✔
84	qs = qs.filter(catalog__in=instance.catalogs.all())	4✔
85
86	# when View/Task has any sites it can be filtered for those
87	if instance.sites.exists():	4✔
88	qs = qs.filter(site__in=instance.sites.all())	4✔
89	elif settings.MULTISITE:	4✔
90	# when View/Task has no sites in a multi-site, it should not appear at all
91	return self.none()	4✔
92
93	# when has any groups it can be filtered for those
94	if instance.groups.exists():	4✔
95	qs = qs.filter_groups(instance.groups.all())	4✔
96
97	return qs	4✔
98
99
100	class MembershipQuerySet(models.QuerySet):	4✔
101
102	def filter_current_site(self):	4✔
103	return self.filter(project__site=settings.SITE_ID)	4✔
104
105	def filter_user(self, user):	4✔
106	if user.is_authenticated:	4✔
107	if user.has_perm('projects.view_membership'):	4✔
108	return self.all()	4✔
109	elif is_site_manager(user):	4✔
110	return self.filter_current_site()	4✔
111	else:
112	from .models import Project	4✔
113	projects = Project.objects.filter_user(user)	4✔
114	return self.filter(project__in=projects)	4✔
115	else:
116	return self.objects.none()	×
117
118
119	class IssueQuerySet(models.QuerySet):	4✔
120
121	def filter_current_site(self):	4✔
122	return self.filter(project__site=settings.SITE_ID)	4✔
123
124	def filter_user(self, user):	4✔
125	if user.is_authenticated:	4✔
126	if user.has_perm('projects.view_integration'):	4✔
127	return self.all()	4✔
128	elif is_site_manager(user):	4✔
129	return self.filter_current_site()	4✔
130	else:
131	from .models import Project	4✔
132	projects = Project.objects.filter_user(user)	4✔
133	return self.filter(project__in=projects)	4✔
134	else:
135	return self.none()	×
136
137
138	class IntegrationQuerySet(models.QuerySet):	4✔
139
140	def filter_current_site(self):	4✔
141	return self.filter(project__site=settings.SITE_ID)	4✔
142
143	def filter_user(self, user):	4✔
144	if user.is_authenticated:	4✔
145	if user.has_perm('projects.view_issue'):	4✔
146	return self.all()	4✔
147	elif is_site_manager(user):	4✔
148	return self.filter_current_site()	4✔
149	else:
150	from .models import Project	4✔
151	projects = Project.objects.filter_user(user)	4✔
152	return self.filter(project__in=projects)	4✔
153	else:
154	return self.none()	×
155
156
157	class InviteQuerySet(models.QuerySet):	4✔
158
159	def filter_current_site(self):	4✔
160	return self.filter(project__site=settings.SITE_ID)	4✔
161
162	def filter_user(self, user):	4✔
163	if user.is_authenticated:	4✔
164	if user.has_perm('projects.view_invite'):	4✔
165	return self.all()	4✔
166	elif is_site_manager(user):	4✔
167	return self.filter_current_site()	4✔
168	else:
169	from .models import Project	4✔
170	projects = Project.objects.filter(memberships__user=user, memberships__role='owner')	4✔
171	return self.filter(project__in=projects)	4✔
172	else:
173	return self.none()	×
174
175
176	class SnapshotQuerySet(models.QuerySet):	4✔
177
178	def filter_current_site(self):	4✔
179	return self.filter(project__site=settings.SITE_ID)	4✔
180
181	def filter_user(self, user):	4✔
182	if user.is_authenticated:	4✔
183	if user.has_perm('projects.view_snapshot'):	4✔
184	return self.all()	4✔
185	elif is_site_manager(user):	4✔
186	return self.filter_current_site()	4✔
187	else:
188	from .models import Project	4✔
189	projects = Project.objects.filter_user(user)	4✔
190	return self.filter(project__in=projects)	4✔
191	else:
192	return self.none()	×
193
194
195	class ValueQuerySet(models.QuerySet):	4✔
196
197	def filter_current_site(self):	4✔
198	return self.filter(project__site=settings.SITE_ID)	4✔
199
200	def filter_user(self, user):	4✔
201	if user.is_authenticated:	4✔
202	if user.has_perm('projects.view_value'):	4✔
203	return self.all()	4✔
204	elif is_site_manager(user):	4✔
205	return self.filter_current_site()	4✔
206	else:
207	from .models import Project	4✔
208	projects = Project.objects.filter_user(user)	4✔
209	return self.filter(project__in=projects)	4✔
210	else:
211	return self.none()	×
212
213	def filter_empty(self):	4✔
214	return self.filter((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
215
216	def exclude_empty(self):	4✔
217	return self.exclude((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
218
219	def exclude_empty_optional(self, catalog):	4✔
220	optional_values = self.filter(attribute__in=[q.attribute for q in catalog.optional_questions])	×
221	return self.exclude(id__in=optional_values.filter_empty().values_list('id', flat=True))	×
222
223	def distinct_list(self):	4✔
224	return self.order_by('attribute').values_list('attribute', 'set_prefix', 'set_index').distinct()	4✔
225
226	def filter_set(self, set_value):	4✔
227	# get the catalog and prefetch most elements of the catalog
228	catalog = set_value.project.catalog	4✔
229	catalog.prefetch_elements()	4✔
230
231	# Get all attributes from matching elements and their descendants
232	attributes = {	4✔
233	descendant.attribute
234	for element in (catalog.pages + catalog.questions)
235	if element.attribute == set_value.attribute
236	for descendant in element.descendants
237	}
238
239	# construct the set_prefix for descendants for this set
240	descendants_set_prefix = \	4✔
241	f'{set_value.set_prefix}\|{set_value.set_index}' if set_value.set_prefix else str(set_value.set_index)
242
243	# collect all values for this set and all descendants
244	return self.filter(project=set_value.project, snapshot=set_value.snapshot, attribute__in=attributes).filter(	4✔
245	Q(set_prefix=set_value.set_prefix, set_index=set_value.set_index) \|
246	Q(set_prefix__startswith=descendants_set_prefix)
247	)
248
249	def compute_sets(self):	4✔
250	sets = defaultdict(set)	4✔
251	for attribute, set_prefix, set_index in self.distinct_list():	4✔
252	sets[attribute].add((set_prefix, set_index))	4✔
253	return sets	4✔
254
255	class ProjectManager(CurrentSiteManagerMixin, TreeManager):	4✔
256
257	def get_queryset(self):	4✔
258	return ProjectQuerySet(self.model, using=self._db)	4✔
259
260	def filter_user(self, user, filter_for_user=False):	4✔
261	return self.get_queryset().filter_user(user, filter_for_user)	4✔
262
263	def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):	4✔
264	return self.get_queryset().filter_catalogs(catalogs=catalogs, exclude_catalogs=exclude_catalogs,	×
265	exclude_null=exclude_null)
266	def filter_groups(self, groups):	4✔
267	return self.get_queryset().filter_groups(groups)	×
268
269	def filter_projects_for_task_or_view(self, instance):	4✔
270	return self.get_queryset().filter_projects_for_task_or_view(instance)	4✔
271
272
273	class MembershipManager(CurrentSiteManagerMixin, models.Manager):	4✔
274
275	def get_queryset(self):	4✔
276	return MembershipQuerySet(self.model, using=self._db)	4✔
277
278	def filter_user(self, user):	4✔
279	return self.get_queryset().filter_user(user)	4✔
280
281
282	class IssueManager(CurrentSiteManagerMixin, models.Manager):	4✔
283
284	def get_queryset(self):	4✔
285	return IssueQuerySet(self.model, using=self._db)	4✔
286
287	def filter_user(self, user):	4✔
288	return self.get_queryset().filter_user(user)	4✔
289
290	def active(self):	4✔
291	return self.get_queryset().active()	×
292
293
294	class IntegrationManager(CurrentSiteManagerMixin, models.Manager):	4✔
295
296	def get_queryset(self):	4✔
297	return IntegrationQuerySet(self.model, using=self._db)	4✔
298
299	def filter_user(self, user):	4✔
300	return self.get_queryset().filter_user(user)	4✔
301
302
303	class InviteManager(CurrentSiteManagerMixin, models.Manager):	4✔
304
305	def get_queryset(self):	4✔
306	return InviteQuerySet(self.model, using=self._db)	4✔
307
308	def filter_user(self, user):	4✔
309	return self.get_queryset().filter_user(user)	4✔
310
311
312	class SnapshotManager(CurrentSiteManagerMixin, models.Manager):	4✔
313
314	def get_queryset(self):	4✔
315	return SnapshotQuerySet(self.model, using=self._db)	4✔
316
317	def filter_user(self, user):	4✔
318	return self.get_queryset().filter_user(user)	4✔
319
320
321	class ValueManager(CurrentSiteManagerMixin, models.Manager):	4✔
322
323	def get_queryset(self):	4✔
324	return ValueQuerySet(self.model, using=self._db)	4✔
325
326	def filter_user(self, user):	4✔
327	return self.get_queryset().filter_user(user)	4✔
328
329	def compute_sets(self):	4✔
330	return self.get_queryset().compute_sets()	×

rdmorganiser / rdmo / 20227548215

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