19672034171

Committed 25 Nov 2025 01:58PM UTC coverage: 94.784% (-0.01%) from 94.796%

Build # 19672034171

Build Type

Pull #1427

github

Committed by

web-flow

Commit Message

Merge 499177e07 into 79917de8d

Pull Request Pull Request #1427: RDMO 2.4.0 🎆

Run Details

2109 of 2214 branches covered (95.26%)

22533 of 23773 relevant lines covered (94.78%)

3.79 hits per line

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

89.34

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):
        # if View/Task is not available it should not show for any project
        if not instance.available:
            return self.none()

        # projects that have an unavailable catalog should be disregarded
        qs = self.filter(catalog__available=True)

        # when instance.catalogs is empty it applies to all
        if instance.catalogs.exists():
            qs = qs.filter(catalog__in=instance.catalogs.all())

        # when instance.sites is empty it applies to all
        if instance.sites.exists():
            qs = qs.filter(site__in=instance.sites.all())

        # when instance.groups is empty it applies to all
        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	# if View/Task is not available it should not show for any project
76	if not instance.available:	4✔
77	return self.none()	4✔
78
79	# projects that have an unavailable catalog should be disregarded
80	qs = self.filter(catalog__available=True)	4✔
81
82	# when instance.catalogs is empty it applies to all
83	if instance.catalogs.exists():	4✔
84	qs = qs.filter(catalog__in=instance.catalogs.all())	4✔
85
86	# when instance.sites is empty it applies to all
87	if instance.sites.exists():	4✔
88	qs = qs.filter(site__in=instance.sites.all())	4✔
89
90	# when instance.groups is empty it applies to all
91	if instance.groups.exists():	4✔
92	qs = qs.filter_groups(instance.groups.all())	4✔
93
94	return qs	4✔
95
96
97	class MembershipQuerySet(models.QuerySet):	4✔
98
99	def filter_current_site(self):	4✔
100	return self.filter(project__site=settings.SITE_ID)	4✔
101
102	def filter_user(self, user):	4✔
103	if user.is_authenticated:	4✔
104	if user.has_perm('projects.view_membership'):	4✔
105	return self.all()	4✔
106	elif is_site_manager(user):	4✔
107	return self.filter_current_site()	4✔
108	else:
109	from .models import Project	4✔
110	projects = Project.objects.filter_user(user)	4✔
111	return self.filter(project__in=projects)	4✔
112	else:
113	return self.objects.none()	×
114
115
116	class IssueQuerySet(models.QuerySet):	4✔
117
118	def filter_current_site(self):	4✔
119	return self.filter(project__site=settings.SITE_ID)	4✔
120
121	def filter_user(self, user):	4✔
122	if user.is_authenticated:	4✔
123	if user.has_perm('projects.view_integration'):	4✔
124	return self.all()	4✔
125	elif is_site_manager(user):	4✔
126	return self.filter_current_site()	4✔
127	else:
128	from .models import Project	4✔
129	projects = Project.objects.filter_user(user)	4✔
130	return self.filter(project__in=projects)	4✔
131	else:
132	return self.none()	×
133
134
135	class IntegrationQuerySet(models.QuerySet):	4✔
136
137	def filter_current_site(self):	4✔
138	return self.filter(project__site=settings.SITE_ID)	4✔
139
140	def filter_user(self, user):	4✔
141	if user.is_authenticated:	4✔
142	if user.has_perm('projects.view_issue'):	4✔
143	return self.all()	4✔
144	elif is_site_manager(user):	4✔
145	return self.filter_current_site()	4✔
146	else:
147	from .models import Project	4✔
148	projects = Project.objects.filter_user(user)	4✔
149	return self.filter(project__in=projects)	4✔
150	else:
151	return self.none()	×
152
153
154	class InviteQuerySet(models.QuerySet):	4✔
155
156	def filter_current_site(self):	4✔
157	return self.filter(project__site=settings.SITE_ID)	4✔
158
159	def filter_user(self, user):	4✔
160	if user.is_authenticated:	4✔
161	if user.has_perm('projects.view_invite'):	4✔
162	return self.all()	4✔
163	elif is_site_manager(user):	4✔
164	return self.filter_current_site()	4✔
165	else:
166	from .models import Project	4✔
167	projects = Project.objects.filter(memberships__user=user, memberships__role='owner')	4✔
168	return self.filter(project__in=projects)	4✔
169	else:
170	return self.none()	×
171
172
173	class SnapshotQuerySet(models.QuerySet):	4✔
174
175	def filter_current_site(self):	4✔
176	return self.filter(project__site=settings.SITE_ID)	4✔
177
178	def filter_user(self, user):	4✔
179	if user.is_authenticated:	4✔
180	if user.has_perm('projects.view_snapshot'):	4✔
181	return self.all()	4✔
182	elif is_site_manager(user):	4✔
183	return self.filter_current_site()	4✔
184	else:
185	from .models import Project	4✔
186	projects = Project.objects.filter_user(user)	4✔
187	return self.filter(project__in=projects)	4✔
188	else:
189	return self.none()	×
190
191
192	class ValueQuerySet(models.QuerySet):	4✔
193
194	def filter_current_site(self):	4✔
195	return self.filter(project__site=settings.SITE_ID)	4✔
196
197	def filter_user(self, user):	4✔
198	if user.is_authenticated:	4✔
199	if user.has_perm('projects.view_value'):	4✔
200	return self.all()	4✔
201	elif is_site_manager(user):	4✔
202	return self.filter_current_site()	4✔
203	else:
204	from .models import Project	4✔
205	projects = Project.objects.filter_user(user)	4✔
206	return self.filter(project__in=projects)	4✔
207	else:
208	return self.none()	×
209
210	def filter_empty(self):	4✔
211	return self.filter((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
212
213	def exclude_empty(self):	4✔
214	return self.exclude((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
215
216	def exclude_empty_optional(self, catalog):	4✔
217	optional_values = self.filter(attribute__in=[q.attribute for q in catalog.optional_questions])	×
218	return self.exclude(id__in=optional_values.filter_empty().values_list('id', flat=True))	×
219
220	def distinct_list(self):	4✔
221	return self.order_by('attribute').values_list('attribute', 'set_prefix', 'set_index').distinct()	4✔
222
223	def filter_set(self, set_value):	4✔
224	# get the catalog and prefetch most elements of the catalog
225	catalog = set_value.project.catalog	4✔
226	catalog.prefetch_elements()	4✔
227
228	# Get all attributes from matching elements and their descendants
229	attributes = {	4✔
230	descendant.attribute
231	for element in (catalog.pages + catalog.questions)
232	if element.attribute == set_value.attribute
233	for descendant in element.descendants
234	}
235
236	# construct the set_prefix for descendants for this set
237	descendants_set_prefix = \	4✔
238	f'{set_value.set_prefix}\|{set_value.set_index}' if set_value.set_prefix else str(set_value.set_index)
239
240	# collect all values for this set and all descendants
241	return self.filter(project=set_value.project, snapshot=set_value.snapshot, attribute__in=attributes).filter(	4✔
242	Q(set_prefix=set_value.set_prefix, set_index=set_value.set_index) \|
243	Q(set_prefix__startswith=descendants_set_prefix)
244	)
245
246	def compute_sets(self):	4✔
247	sets = defaultdict(set)	4✔
248	for attribute, set_prefix, set_index in self.distinct_list():	4✔
249	sets[attribute].add((set_prefix, set_index))	4✔
250	return sets	4✔
251
252	class ProjectManager(CurrentSiteManagerMixin, TreeManager):	4✔
253
254	def get_queryset(self):	4✔
255	return ProjectQuerySet(self.model, using=self._db)	4✔
256
257	def filter_user(self, user, filter_for_user=False):	4✔
258	return self.get_queryset().filter_user(user, filter_for_user)	4✔
259
260	def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):	4✔
261	return self.get_queryset().filter_catalogs(catalogs=catalogs, exclude_catalogs=exclude_catalogs,	×
262	exclude_null=exclude_null)
263	def filter_groups(self, groups):	4✔
264	return self.get_queryset().filter_groups(groups)	×
265
266	def filter_projects_for_task_or_view(self, instance):	4✔
267	return self.get_queryset().filter_projects_for_task_or_view(instance)	4✔
268
269
270	class MembershipManager(CurrentSiteManagerMixin, models.Manager):	4✔
271
272	def get_queryset(self):	4✔
273	return MembershipQuerySet(self.model, using=self._db)	4✔
274
275	def filter_user(self, user):	4✔
276	return self.get_queryset().filter_user(user)	4✔
277
278
279	class IssueManager(CurrentSiteManagerMixin, models.Manager):	4✔
280
281	def get_queryset(self):	4✔
282	return IssueQuerySet(self.model, using=self._db)	4✔
283
284	def filter_user(self, user):	4✔
285	return self.get_queryset().filter_user(user)	4✔
286
287	def active(self):	4✔
288	return self.get_queryset().active()	×
289
290
291	class IntegrationManager(CurrentSiteManagerMixin, models.Manager):	4✔
292
293	def get_queryset(self):	4✔
294	return IntegrationQuerySet(self.model, using=self._db)	4✔
295
296	def filter_user(self, user):	4✔
297	return self.get_queryset().filter_user(user)	4✔
298
299
300	class InviteManager(CurrentSiteManagerMixin, models.Manager):	4✔
301
302	def get_queryset(self):	4✔
303	return InviteQuerySet(self.model, using=self._db)	4✔
304
305	def filter_user(self, user):	4✔
306	return self.get_queryset().filter_user(user)	4✔
307
308
309	class SnapshotManager(CurrentSiteManagerMixin, models.Manager):	4✔
310
311	def get_queryset(self):	4✔
312	return SnapshotQuerySet(self.model, using=self._db)	4✔
313
314	def filter_user(self, user):	4✔
315	return self.get_queryset().filter_user(user)	4✔
316
317
318	class ValueManager(CurrentSiteManagerMixin, models.Manager):	4✔
319
320	def get_queryset(self):	4✔
321	return ValueQuerySet(self.model, using=self._db)	4✔
322
323	def filter_user(self, user):	4✔
324	return self.get_queryset().filter_user(user)	4✔
325
326	def compute_sets(self):	4✔
327	return self.get_queryset().compute_sets()	×

rdmorganiser / rdmo / 19672034171

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