27126983023

Committed 08 Jun 2026 09:01AM UTC coverage: 95.102% (+0.2%) from 94.901%

Build # 27126983023

Build Type

Pull #1606

github

Committed by

web-flow

Commit Message

Merge ac07a122f into e918a7b3b

Pull Request Pull Request #1606: RDMO 2.4.5 🦇

Coverage Stats

2225 of 2318 branches covered (95.99%)

23553 of 24766 relevant lines covered (95.1%)

3.8 hits per line

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

95.35

rdmo/projects/utils.py

import logging
import mimetypes
from collections import defaultdict
from pathlib import Path

from django.conf import settings
from django.contrib.sites.models import Site
from django.core.exceptions import ObjectDoesNotExist
from django.template.loader import render_to_string
from django.urls import reverse
from django.utils.timezone import now

from rdmo.core.mail import send_mail
from rdmo.core.plugins import get_plugins
from rdmo.core.utils import remove_double_newlines

logger = logging.getLogger(__name__)


def get_value_path(project, snapshot=None):
    if snapshot is None:
        return Path('projects') / str(project.id) / 'values'
    else:
        return Path('projects') / str(project.id) / 'snapshots' / str(snapshot.id) / 'values'


def is_last_owner(project, user):
    # check if user is owner
    if user in project.owners:
        # check if the user is the last owner
        return project.owners.count() <= 1
    else:
        return False


def check_conditions(conditions, values, set_prefix=None, set_index=None):
    if conditions:
        for condition in conditions:
            if condition.resolve(values, set_prefix, set_index):
                return True
        return False
    else:
        return True


def check_options(project, value):
    # loop over all values of a question and check if value.option matches the optionsets of the question
    for question in filter(lambda q: q.attribute == value.attribute, project.catalog.questions):
        question_options = [
            option.id
            for optionset in question.optionsets.all()
            for option in optionset.options.all()
        ]

        # fail if question requires an option but value has none
        if question_options and value.option is None:
            return False

        # fail if the value's option is not allowed for this question
        if value.option is not None and value.option.id not in question_options:
            return False

    return True


def copy_project(instance, site, owners):
    from .models import Membership, Project, Value  # to prevent circular inclusion

    timestamp = now()

    tasks = instance.tasks.all()
    views = instance.views.all()

    values = instance.values.filter(snapshot=None)
    snapshots = {
        snapshot: instance.values.filter(snapshot=snapshot)
        for snapshot in instance.snapshots.all()
    }

    # a completely new project instance needs to be created in order for mptt to work
    project = Project.objects.create(
        parent=instance.parent,
        site=site,
        title=instance.title,
        description=instance.description,
        catalog=instance.catalog,
        created=timestamp
    )

    # save project tasks
    for task in tasks:
        project.tasks.add(task)

    # save project views
    for view in views:
        project.views.add(view)

    # save current project values
    project_values = []
    for value in values:
        value.id = None
        value.project = project
        value.updated = timestamp

        if value.file:
            # file values cannot be bulk created since we need their id and only postgres provides that (reliably)
            # https://docs.djangoproject.com/en/4.2/ref/models/querysets/#bulk-create
            value.save()
            value.copy_file(value.file_name, value.file)
        else:
            project_values.append(value)

    # insert the new values using bulk_create
    Value.objects.bulk_create(project_values)

    # save project snapshots
    for snapshot, snapshot_values in snapshots.items():
        snapshot.id = None
        snapshot.project = project
        snapshot.save(copy_values=False)

        project_snapshot_values = []
        for value in snapshot_values:
            value.id = None
            value.project = project
            value.snapshot = snapshot
            value.updated = timestamp

            if value.file:
                value.save()
                value.copy_file(value.file_name, value.file)
            else:
                project_snapshot_values.append(value)

        # insert the new snapshot values using bulk_create
        Value.objects.bulk_create(project_snapshot_values)

    for owner in owners:
        membership = Membership(project=project, user=owner, role='owner')
        membership.save()

    return project


def save_import_values(project, values, checked):
    for value in values:
        if value.attribute:
            value_key = f'{value.attribute.uri}[{value.set_prefix}][{value.set_index}][{value.collection_index}]'

            if value_key in checked:
                current_value = value.current
                if current_value is None:
                    # assert that this is a new value
                    assert value.pk is None

                    value.project = project
                    value.save()

                    if value.file:
                        value.copy_file(value.file_name, value.file)
                    else:
                        try:
                            name = value.file_import.get('name')
                            file = value.file_import.get('file')
                            value.file.save(name, file)
                        except AttributeError:
                            pass

                else:
                    # make sure we have the correct value
                    assert current_value.snapshot is None
                    assert current_value.attribute == value.attribute
                    assert current_value.set_prefix == value.set_prefix
                    assert current_value.set_index == value.set_index
                    assert current_value.collection_index == value.collection_index

                    # assert that this is an new value
                    assert current_value.pk is not None

                    current_value.text = value.text
                    current_value.option = value.option
                    current_value.value_type = value.value_type
                    current_value.unit = value.unit
                    current_value.save()

                    if value.file:
                        current_value.file.delete()
                        current_value.copy_file(value.file_name, value.file)
                    else:
                        try:
                            name = value.file_import.get('name')
                            file = value.file_import.get('file')
                            current_value.file.delete()
                            current_value.file.save(name, file)
                        except AttributeError:
                            pass


def save_import_snapshot_values(project, snapshots, checked):
    for snapshot in snapshots:
        # assert that this is a new snapshot
        assert snapshot.pk is None

        snapshot.project = project
        snapshot.save(copy_values=False)

        for value in snapshot.snapshot_values:
            if value.attribute:
                value_key = f"{value.attribute.uri}[{snapshot.snapshot_index}][{value.set_prefix}][{value.set_index}][{value.collection_index}]" # noqa: E501

                if value_key in checked:
                    # assert that this is a new value
                    assert value.pk is None

                    value.project = project
                    value.snapshot = snapshot
                    value.save()

                    if value.file:
                        value.copy_file(value.file_name, value.file)
                    else:
                        try:
                            name = value.file_import.get('name')
                            file = value.file_import.get('file')
                            value.file.save(name, file)
                        except AttributeError:
                            pass


def save_import_tasks(project, tasks):
    for task in tasks:
        project.tasks.add(task)


def save_import_views(project, views):
    for view in views:
        project.views.add(view)


def get_invite_email_project_path(request, invite):
    project_invite_path = reverse('project_join', args=[invite.token])

    if settings.PROJECT_INVITE_USE_PROJECT_SITE:
        return f'{request.scheme}://{invite.project.site.domain}{project_invite_path}'

    elif settings.MULTISITE and invite.user is not None:
        current_site = Site.objects.get_current()

        if not invite.user.role.member.filter(id=current_site.id).exists():
            # take users first site
            user_first_site = invite.user.role.member.first()
            if user_first_site:
                return f'{request.scheme}://{user_first_site.domain}{project_invite_path}'

    return request.build_absolute_uri(project_invite_path)


def send_invite_email(request, invite):
    context = {
        'invite_url': get_invite_email_project_path(request, invite),
        'invite_user': invite.user,
        'invite_email': invite.email,
        'project': invite.project,
        'user': request.user,
        'site': Site.objects.get_current()
    }

    subject = render_to_string('projects/email/project_invite_subject.txt', context)
    message = render_to_string('projects/email/project_invite_message.txt', context)

    # send the email
    send_mail(subject, message, to=[invite.email])


def set_context_querystring_with_filter_and_page(context: dict) -> dict:
    '''prepares the filter part of the querystring for the next and previous hyperlinks in the pagination'''
    if context["filter"].data:
        querystring = context["filter"].data.copy()
        if context["filter"].data.get('page'):
            del querystring['page']
        context['querystring'] = querystring.urlencode()
    return context


def get_upload_accept():
    accept = defaultdict(set)
    for import_plugin in get_plugins('PROJECT_IMPORTS').values():
        if isinstance(import_plugin.accept, dict):
            for mime_type, suffixes in import_plugin.accept.items():
                accept[mime_type].update(suffixes)

        elif isinstance(import_plugin.accept, str):
            # legacy fallback for pre 2.3.0 RDMO, e.g. `accept = '.xml'`
            suffix = import_plugin.accept
            mime_type, _encoding = mimetypes.guess_type(f'example{suffix}')
            if mime_type:
                accept[mime_type].update([suffix])

        elif import_plugin.upload is True:
            # if one of the plugins does not have the accept field, but is marked as upload plugin
            # all file types are allowed
            return {}

    return accept


def compute_set_prefix_from_set_value(set_value, value):
    set_prefix_length = len(set_value.set_prefix.split('|')) if set_value.set_prefix else 0
    return '|'.join([
        str(set_value.set_index) if (index == set_prefix_length) else value
        for index, value in enumerate(value.set_prefix.split('|'))
    ])


def get_contact_message(request, project):
    project_url = project.get_absolute_url()
    project_interview_url = reverse('project_interview', args=[project.id])

    context = {
        'user': request.user,
        'site': Site.objects.get_current(),
        'project': project,
        'project_url': request.build_absolute_uri(project_url)
    }

    page_id = request.GET.get('page')
    if page_id:
        page = project.catalog.get_page(page_id)
        if page:
            context.update({
                'page': page,
                'page_url': request.build_absolute_uri(f'{project_interview_url}{page_id}/')
            })

    questionset_id = request.GET.get('questionset')
    if questionset_id:
        context['questionset'] = project.catalog.get_questionset(questionset_id)

    question_id = request.GET.get('question')
    if question_id:
        context['question'] = project.catalog.get_question(question_id)

    value_ids = request.GET.getlist('values')
    if value_ids:
        values = project.values.filter(snapshot=None).filter(id__in=value_ids)
        context['values'] = values

        if page_id and page and page.is_collection and page.attribute is not None:
            value = values.filter(set_prefix='').first()
            if value:
                try:
                    context['set_value'] = project.values.filter(snapshot=None).get(
                        attribute=page.attribute,
                        set_prefix='',
                        set_index=value.set_index,
                        collection_index=0
                    )
                except ObjectDoesNotExist:
                    pass

    subject = render_to_string('projects/email/project_contact_subject.txt', context)
    message = render_to_string('projects/email/project_contact_message.txt', context)

    return {
        'subject': subject,
        'message': remove_double_newlines(message)
    }


def send_contact_message(request, subject, message):
    send_mail(subject, message,
              to=settings.PROJECT_CONTACT_RECIPIENTS,
              cc=[request.user.email], reply_to=[request.user.email])

1	import logging	4✔
2	import mimetypes	4✔
3	from collections import defaultdict	4✔
4	from pathlib import Path	4✔
5
6	from django.conf import settings	4✔
7	from django.contrib.sites.models import Site	4✔
8	from django.core.exceptions import ObjectDoesNotExist	4✔
9	from django.template.loader import render_to_string	4✔
10	from django.urls import reverse	4✔
11	from django.utils.timezone import now	4✔
12
13	from rdmo.core.mail import send_mail	4✔
14	from rdmo.core.plugins import get_plugins	4✔
15	from rdmo.core.utils import remove_double_newlines	4✔
16
17	logger = logging.getLogger(__name__)	4✔
18
19
20	def get_value_path(project, snapshot=None):	4✔
21	if snapshot is None:	4✔
22	return Path('projects') / str(project.id) / 'values'	4✔
23	else:
24	return Path('projects') / str(project.id) / 'snapshots' / str(snapshot.id) / 'values'	4✔
25
26
27	def is_last_owner(project, user):	4✔
28	# check if user is owner
29	if user in project.owners:	4✔
30	# check if the user is the last owner
31	return project.owners.count() <= 1	4✔
32	else:
33	return False	4✔
34
35
36	def check_conditions(conditions, values, set_prefix=None, set_index=None):	4✔
37	if conditions:	4✔
38	for condition in conditions:	4✔
39	if condition.resolve(values, set_prefix, set_index):	4✔
40	return True	4✔
41	return False	4✔
42	else:
43	return True	×
44
45
46	def check_options(project, value):	4✔
47	# loop over all values of a question and check if value.option matches the optionsets of the question
48	for question in filter(lambda q: q.attribute == value.attribute, project.catalog.questions):	4✔
49	question_options = [	4✔
50	option.id
51	for optionset in question.optionsets.all()
52	for option in optionset.options.all()
53	]
54
55	# fail if question requires an option but value has none
56	if question_options and value.option is None:	4✔
57	return False	4✔
58
59	# fail if the value's option is not allowed for this question
60	if value.option is not None and value.option.id not in question_options:	4✔
61	return False	4✔
62
63	return True	4✔
64
65
66	def copy_project(instance, site, owners):	4✔
67	from .models import Membership, Project, Value # to prevent circular inclusion	4✔
68
69	timestamp = now()	4✔
70
71	tasks = instance.tasks.all()	4✔
72	views = instance.views.all()	4✔
73
74	values = instance.values.filter(snapshot=None)	4✔
75	snapshots = {	4✔
76	snapshot: instance.values.filter(snapshot=snapshot)
77	for snapshot in instance.snapshots.all()
78	}
79
80	# a completely new project instance needs to be created in order for mptt to work
81	project = Project.objects.create(	4✔
82	parent=instance.parent,
83	site=site,
84	title=instance.title,
85	description=instance.description,
86	catalog=instance.catalog,
87	created=timestamp
88	)
89
90	# save project tasks
91	for task in tasks:	4✔
92	project.tasks.add(task)	4✔
93
94	# save project views
95	for view in views:	4✔
96	project.views.add(view)	4✔
97
98	# save current project values
99	project_values = []	4✔
100	for value in values:	4✔
101	value.id = None	4✔
102	value.project = project	4✔
103	value.updated = timestamp	4✔
104
105	if value.file:	4✔
106	# file values cannot be bulk created since we need their id and only postgres provides that (reliably)
107	# https://docs.djangoproject.com/en/4.2/ref/models/querysets/#bulk-create
108	value.save()	4✔
109	value.copy_file(value.file_name, value.file)	4✔
110	else:
111	project_values.append(value)	4✔
112
113	# insert the new values using bulk_create
114	Value.objects.bulk_create(project_values)	4✔
115
116	# save project snapshots
117	for snapshot, snapshot_values in snapshots.items():	4✔
118	snapshot.id = None	4✔
119	snapshot.project = project	4✔
120	snapshot.save(copy_values=False)	4✔
121
122	project_snapshot_values = []	4✔
123	for value in snapshot_values:	4✔
124	value.id = None	4✔
125	value.project = project	4✔
126	value.snapshot = snapshot	4✔
127	value.updated = timestamp	4✔
128
129	if value.file:	4✔
130	value.save()	4✔
131	value.copy_file(value.file_name, value.file)	4✔
132	else:
133	project_snapshot_values.append(value)	4✔
134
135	# insert the new snapshot values using bulk_create
136	Value.objects.bulk_create(project_snapshot_values)	4✔
137
138	for owner in owners:	4✔
139	membership = Membership(project=project, user=owner, role='owner')	4✔
140	membership.save()	4✔
141
142	return project	4✔
143
144
145	def save_import_values(project, values, checked):	4✔
146	for value in values:	4✔
147	if value.attribute:	4✔
148	value_key = f'{value.attribute.uri}[{value.set_prefix}][{value.set_index}][{value.collection_index}]'	4✔
149
150	if value_key in checked:	4✔
151	current_value = value.current	4✔
152	if current_value is None:	4✔
153	# assert that this is a new value
154	assert value.pk is None	4✔
155
156	value.project = project	4✔
157	value.save()	4✔
158
159	if value.file:	4✔
160	value.copy_file(value.file_name, value.file)	4✔
161	else:
162	try:	4✔
163	name = value.file_import.get('name')	4✔
164	file = value.file_import.get('file')	4✔
165	value.file.save(name, file)	4✔
166	except AttributeError:	4✔
167	pass	4✔
168
169	else:
170	# make sure we have the correct value
171	assert current_value.snapshot is None	4✔
172	assert current_value.attribute == value.attribute	4✔
173	assert current_value.set_prefix == value.set_prefix	4✔
174	assert current_value.set_index == value.set_index	4✔
175	assert current_value.collection_index == value.collection_index	4✔
176
177	# assert that this is an new value
178	assert current_value.pk is not None	4✔
179
180	current_value.text = value.text	4✔
181	current_value.option = value.option	4✔
182	current_value.value_type = value.value_type	4✔
183	current_value.unit = value.unit	4✔
184	current_value.save()	4✔
185
186	if value.file:	4✔
187	current_value.file.delete()	4✔
188	current_value.copy_file(value.file_name, value.file)	4✔
189	else:
190	try:	4✔
191	name = value.file_import.get('name')	4✔
192	file = value.file_import.get('file')	4✔
193	current_value.file.delete()	4✔
194	current_value.file.save(name, file)	4✔
195	except AttributeError:	4✔
196	pass	4✔
197
198
199	def save_import_snapshot_values(project, snapshots, checked):	4✔
200	for snapshot in snapshots:	4✔
201	# assert that this is a new snapshot
202	assert snapshot.pk is None	4✔
203
204	snapshot.project = project	4✔
205	snapshot.save(copy_values=False)	4✔
206
207	for value in snapshot.snapshot_values:	4✔
208	if value.attribute:	4✔
209	value_key = f"{value.attribute.uri}[{snapshot.snapshot_index}][{value.set_prefix}][{value.set_index}][{value.collection_index}]" # noqa: E501	4✔
210
211	if value_key in checked:	4✔
212	# assert that this is a new value
213	assert value.pk is None	4✔
214
215	value.project = project	4✔
216	value.snapshot = snapshot	4✔
217	value.save()	4✔
218
219	if value.file:	4✔
220	value.copy_file(value.file_name, value.file)	×
221	else:
222	try:	4✔
223	name = value.file_import.get('name')	4✔
224	file = value.file_import.get('file')	4✔
225	value.file.save(name, file)	4✔
226	except AttributeError:	4✔
227	pass	4✔
228
229
230	def save_import_tasks(project, tasks):	4✔
231	for task in tasks:	4✔
232	project.tasks.add(task)	4✔
233
234
235	def save_import_views(project, views):	4✔
236	for view in views:	4✔
237	project.views.add(view)	4✔
238
239
240	def get_invite_email_project_path(request, invite):	4✔
241	project_invite_path = reverse('project_join', args=[invite.token])	4✔
242
243	if settings.PROJECT_INVITE_USE_PROJECT_SITE:	4✔
244	return f'{request.scheme}://{invite.project.site.domain}{project_invite_path}'	4✔
245
246	elif settings.MULTISITE and invite.user is not None:	4✔
247	current_site = Site.objects.get_current()	4✔
248
249	if not invite.user.role.member.filter(id=current_site.id).exists():	4✔
250	# take users first site
251	user_first_site = invite.user.role.member.first()	4✔
252	if user_first_site:	4✔
253	return f'{request.scheme}://{user_first_site.domain}{project_invite_path}'	4✔
254
255	return request.build_absolute_uri(project_invite_path)	4✔
256
257
258	def send_invite_email(request, invite):	4✔
259	context = {	4✔
260	'invite_url': get_invite_email_project_path(request, invite),
261	'invite_user': invite.user,
262	'invite_email': invite.email,
263	'project': invite.project,
264	'user': request.user,
265	'site': Site.objects.get_current()
266	}
267
268	subject = render_to_string('projects/email/project_invite_subject.txt', context)	4✔
269	message = render_to_string('projects/email/project_invite_message.txt', context)	4✔
270
271	# send the email
272	send_mail(subject, message, to=[invite.email])	4✔
273
274
275	def set_context_querystring_with_filter_and_page(context: dict) -> dict:	4✔
276	'''prepares the filter part of the querystring for the next and previous hyperlinks in the pagination'''
277	if context["filter"].data:	4✔
278	querystring = context["filter"].data.copy()	4✔
279	if context["filter"].data.get('page'):	4✔
280	del querystring['page']	4✔
281	context['querystring'] = querystring.urlencode()	4✔
282	return context	4✔
283
284
285	def get_upload_accept():	4✔
286	accept = defaultdict(set)	4✔
287	for import_plugin in get_plugins('PROJECT_IMPORTS').values():	4✔
288	if isinstance(import_plugin.accept, dict):	4✔
289	for mime_type, suffixes in import_plugin.accept.items():	4✔
290	accept[mime_type].update(suffixes)	4✔
291
292	elif isinstance(import_plugin.accept, str):	4✔
293	# legacy fallback for pre 2.3.0 RDMO, e.g. `accept = '.xml'`
294	suffix = import_plugin.accept	×
295	mime_type, _encoding = mimetypes.guess_type(f'example{suffix}')	×
296	if mime_type:	×
297	accept[mime_type].update([suffix])	×
298
299	elif import_plugin.upload is True:	4✔
300	# if one of the plugins does not have the accept field, but is marked as upload plugin
301	# all file types are allowed
302	return {}	×
303
304	return accept	4✔
305
306
307	def compute_set_prefix_from_set_value(set_value, value):	4✔
308	set_prefix_length = len(set_value.set_prefix.split('\|')) if set_value.set_prefix else 0	4✔
309	return '\|'.join([	4✔
310	str(set_value.set_index) if (index == set_prefix_length) else value
311	for index, value in enumerate(value.set_prefix.split('\|'))
312	])
313
314
315	def get_contact_message(request, project):	4✔
316	project_url = project.get_absolute_url()	4✔
317	project_interview_url = reverse('project_interview', args=[project.id])	4✔
318
319	context = {	4✔
320	'user': request.user,
321	'site': Site.objects.get_current(),
322	'project': project,
323	'project_url': request.build_absolute_uri(project_url)
324	}
325
326	page_id = request.GET.get('page')	4✔
327	if page_id:	4✔
328	page = project.catalog.get_page(page_id)	4✔
329	if page:	4✔
330	context.update({	4✔
331	'page': page,
332	'page_url': request.build_absolute_uri(f'{project_interview_url}{page_id}/')
333	})
334
335	questionset_id = request.GET.get('questionset')	4✔
336	if questionset_id:	4✔
337	context['questionset'] = project.catalog.get_questionset(questionset_id)	×
338
339	question_id = request.GET.get('question')	4✔
340	if question_id:	4✔
341	context['question'] = project.catalog.get_question(question_id)	4✔
342
343	value_ids = request.GET.getlist('values')	4✔
344	if value_ids:	4✔
345	values = project.values.filter(snapshot=None).filter(id__in=value_ids)	4✔
346	context['values'] = values	4✔
347
348	if page_id and page and page.is_collection and page.attribute is not None:	4✔
349	value = values.filter(set_prefix='').first()	4✔
350	if value:	4✔
351	try:	4✔
352	context['set_value'] = project.values.filter(snapshot=None).get(	4✔
353	attribute=page.attribute,
354	set_prefix='',
355	set_index=value.set_index,
356	collection_index=0
357	)
358	except ObjectDoesNotExist:	×
359	pass	×
360
361	subject = render_to_string('projects/email/project_contact_subject.txt', context)	4✔
362	message = render_to_string('projects/email/project_contact_message.txt', context)	4✔
363
364	return {	4✔
365	'subject': subject,
366	'message': remove_double_newlines(message)
367	}
368
369
370	def send_contact_message(request, subject, message):	4✔
371	send_mail(subject, message,	4✔
372	to=settings.PROJECT_CONTACT_RECIPIENTS,
373	cc=[request.user.email], reply_to=[request.user.email])

rdmorganiser / rdmo / 27126983023

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