16070961563

Committed 04 Jul 2025 09:48AM UTC coverage: 94.796% (+4.0%) from 90.817%

Build # 16070961563

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1361 from rdmorganiser/2.3.2

RDMO 2.3.2 🐛🐛

Run Details

2086 of 2191 branches covered (95.21%)

22353 of 23580 relevant lines covered (94.8%)

3.79 hits per line

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

90.48

rdmo/projects/managers.py

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)
        )


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)

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

rdmorganiser / rdmo / 16070961563

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