14404330126

Committed 11 Apr 2025 01:31PM UTC coverage: 90.789% (+0.3%) from 90.478%

Build # 14404330126

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1195 from rdmorganiser/2.3.0

RDMO 2.3.0 ⭐

Run Details

989 of 1076 branches covered (91.91%)

9176 of 10107 relevant lines covered (90.79%)

3.63 hits per line

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

93.98

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_current_site(self):
        return self.filter(site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_project'):
                return self.all()
            else:
                queryset = self.filter_visibility(user)
                for instance in queryset:
                    queryset |= instance.get_descendants()
                return queryset.distinct()
        else:
            return self.none()

    def filter_visibility(self, 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
        current_site_filter = Q(site=settings.SITE_ID) if is_site_manager(user) else Q()

        return self.filter(Q(user=user) | visibility_filter | current_site_filter)

    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):
        return self.get_queryset().filter_user(user)

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

rdmorganiser / rdmo / 14404330126

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