15062782620

Committed 16 May 2025 07:10AM UTC coverage: 90.817% (+0.03%) from 90.789%

Build # 15062782620

Build Type

Pull #1325

github

Committed by

web-flow

Commit Message

Merge 2e12b53df into 628fa2788

Pull Request Pull Request #1325: RDMO 2.3.1 🐛

Run Details

990 of 1077 branches covered (91.92%)

9188 of 10117 relevant lines covered (90.82%)

3.63 hits per line

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

95.09

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)


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


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()	4✔
47	if exclude_null:	4✔
48	catalogs_filter &= Q(catalog__isnull=False)	4✔
49	if catalogs:	4✔
50	catalogs_filter &= Q(catalog__in=catalogs)	4✔
51	if exclude_catalogs:	4✔
52	catalogs_filter &= ~Q(catalog__in=exclude_catalogs)	×
53	return self.filter(catalogs_filter)	4✔
54
55
56	class MembershipQuerySet(models.QuerySet):	4✔
57
58	def filter_current_site(self):	4✔
59	return self.filter(project__site=settings.SITE_ID)	4✔
60
61	def filter_user(self, user):	4✔
62	if user.is_authenticated:	4✔
63	if user.has_perm('projects.view_membership'):	4✔
64	return self.all()	4✔
65	elif is_site_manager(user):	4✔
66	return self.filter_current_site()	4✔
67	else:
68	from .models import Project	4✔
69	projects = Project.objects.filter_user(user)	4✔
70	return self.filter(project__in=projects)	4✔
71	else:
72	return self.objects.none()	×
73
74
75	class IssueQuerySet(models.QuerySet):	4✔
76
77	def filter_current_site(self):	4✔
78	return self.filter(project__site=settings.SITE_ID)	4✔
79
80	def filter_user(self, user):	4✔
81	if user.is_authenticated:	4✔
82	if user.has_perm('projects.view_integration'):	4✔
83	return self.all()	4✔
84	elif is_site_manager(user):	4✔
85	return self.filter_current_site()	4✔
86	else:
87	from .models import Project	4✔
88	projects = Project.objects.filter_user(user)	4✔
89	return self.filter(project__in=projects)	4✔
90	else:
91	return self.none()	×
92
93
94	class IntegrationQuerySet(models.QuerySet):	4✔
95
96	def filter_current_site(self):	4✔
97	return self.filter(project__site=settings.SITE_ID)	4✔
98
99	def filter_user(self, user):	4✔
100	if user.is_authenticated:	4✔
101	if user.has_perm('projects.view_issue'):	4✔
102	return self.all()	4✔
103	elif is_site_manager(user):	4✔
104	return self.filter_current_site()	4✔
105	else:
106	from .models import Project	4✔
107	projects = Project.objects.filter_user(user)	4✔
108	return self.filter(project__in=projects)	4✔
109	else:
110	return self.none()	×
111
112
113	class InviteQuerySet(models.QuerySet):	4✔
114
115	def filter_current_site(self):	4✔
116	return self.filter(project__site=settings.SITE_ID)	4✔
117
118	def filter_user(self, user):	4✔
119	if user.is_authenticated:	4✔
120	if user.has_perm('projects.view_invite'):	4✔
121	return self.all()	4✔
122	elif is_site_manager(user):	4✔
123	return self.filter_current_site()	4✔
124	else:
125	from .models import Project	4✔
126	projects = Project.objects.filter(memberships__user=user, memberships__role='owner')	4✔
127	return self.filter(project__in=projects)	4✔
128	else:
129	return self.none()	×
130
131
132	class SnapshotQuerySet(models.QuerySet):	4✔
133
134	def filter_current_site(self):	4✔
135	return self.filter(project__site=settings.SITE_ID)	4✔
136
137	def filter_user(self, user):	4✔
138	if user.is_authenticated:	4✔
139	if user.has_perm('projects.view_snapshot'):	4✔
140	return self.all()	4✔
141	elif is_site_manager(user):	4✔
142	return self.filter_current_site()	4✔
143	else:
144	from .models import Project	4✔
145	projects = Project.objects.filter_user(user)	4✔
146	return self.filter(project__in=projects)	4✔
147	else:
148	return self.none()	×
149
150
151	class ValueQuerySet(models.QuerySet):	4✔
152
153	def filter_current_site(self):	4✔
154	return self.filter(project__site=settings.SITE_ID)	4✔
155
156	def filter_user(self, user):	4✔
157	if user.is_authenticated:	4✔
158	if user.has_perm('projects.view_value'):	4✔
159	return self.all()	4✔
160	elif is_site_manager(user):	4✔
161	return self.filter_current_site()	4✔
162	else:
163	from .models import Project	4✔
164	projects = Project.objects.filter_user(user)	4✔
165	return self.filter(project__in=projects)	4✔
166	else:
167	return self.none()	×
168
169	def filter_empty(self):	4✔
170	return self.filter((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
171
172	def exclude_empty(self):	4✔
173	return self.exclude((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
174
175	def distinct_list(self):	4✔
176	return self.order_by('attribute').values_list('attribute', 'set_prefix', 'set_index').distinct()	4✔
177
178	def filter_set(self, set_value):	4✔
179	# get the catalog and prefetch most elements of the catalog
180	catalog = set_value.project.catalog	4✔
181	catalog.prefetch_elements()	4✔
182
183	# Get all attributes from matching elements and their descendants
184	attributes = {	4✔
185	descendant.attribute
186	for element in (catalog.pages + catalog.questions)
187	if element.attribute == set_value.attribute
188	for descendant in element.descendants
189	}
190
191	# construct the set_prefix for descendants for this set
192	descendants_set_prefix = \	4✔
193	f'{set_value.set_prefix}\|{set_value.set_index}' if set_value.set_prefix else str(set_value.set_index)
194
195	# collect all values for this set and all descendants
196	return self.filter(project=set_value.project, snapshot=set_value.snapshot, attribute__in=attributes).filter(	4✔
197	Q(set_prefix=set_value.set_prefix, set_index=set_value.set_index) \|
198	Q(set_prefix__startswith=descendants_set_prefix)
199	)
200
201
202	class ProjectManager(CurrentSiteManagerMixin, TreeManager):	4✔
203
204	def get_queryset(self):	4✔
205	return ProjectQuerySet(self.model, using=self._db)	4✔
206
207	def filter_user(self, user, filter_for_user=False):	4✔
208	return self.get_queryset().filter_user(user, filter_for_user)	4✔
209
210	def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):	4✔
211	return self.get_queryset().filter_catalogs(catalogs=catalogs, exclude_catalogs=exclude_catalogs,	4✔
212	exclude_null=exclude_null)
213
214
215	class MembershipManager(CurrentSiteManagerMixin, models.Manager):	4✔
216
217	def get_queryset(self):	4✔
218	return MembershipQuerySet(self.model, using=self._db)	4✔
219
220	def filter_user(self, user):	4✔
221	return self.get_queryset().filter_user(user)	4✔
222
223
224	class IssueManager(CurrentSiteManagerMixin, models.Manager):	4✔
225
226	def get_queryset(self):	4✔
227	return IssueQuerySet(self.model, using=self._db)	4✔
228
229	def filter_user(self, user):	4✔
230	return self.get_queryset().filter_user(user)	4✔
231
232	def active(self):	4✔
233	return self.get_queryset().active()	×
234
235
236	class IntegrationManager(CurrentSiteManagerMixin, models.Manager):	4✔
237
238	def get_queryset(self):	4✔
239	return IntegrationQuerySet(self.model, using=self._db)	4✔
240
241	def filter_user(self, user):	4✔
242	return self.get_queryset().filter_user(user)	4✔
243
244
245	class InviteManager(CurrentSiteManagerMixin, models.Manager):	4✔
246
247	def get_queryset(self):	4✔
248	return InviteQuerySet(self.model, using=self._db)	4✔
249
250	def filter_user(self, user):	4✔
251	return self.get_queryset().filter_user(user)	4✔
252
253
254	class SnapshotManager(CurrentSiteManagerMixin, models.Manager):	4✔
255
256	def get_queryset(self):	4✔
257	return SnapshotQuerySet(self.model, using=self._db)	4✔
258
259	def filter_user(self, user):	4✔
260	return self.get_queryset().filter_user(user)	4✔
261
262
263	class ValueManager(CurrentSiteManagerMixin, models.Manager):	4✔
264
265	def get_queryset(self):	4✔
266	return ValueQuerySet(self.model, using=self._db)	4✔
267
268	def filter_user(self, user):	4✔
269	return self.get_queryset().filter_user(user)	4✔

rdmorganiser / rdmo / 15062782620

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