akvo / iwsims-demo / #72

Committed 28 Apr 2025 03:28AM UTC coverage: 86.134% (+1.1%) from 85.024%

Build # #72

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #20 from akvo/feature/19-eng-1231-dynamic-level-approval

Feature/19 eng 1231 dynamic level approval

Run Details

2646 of 3188 branches covered (83.0%)

Branch coverage included in aggregate %.

5995 of 6844 relevant lines covered (87.59%)

0.88 hits per line

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

63.86

backend/api/v1/v1_users/functions.py

from django.utils import timezone
from typing import List
from api.v1.v1_profile.constants import UserRoleTypes
from api.v1.v1_forms.constants import FormAccessTypes
from api.v1.v1_forms.models import (
    FormApprovalAssignment,
    FormAccess,
    UserForms,
    Forms,
)
from api.v1.v1_data.models import PendingDataBatch, PendingDataApproval
from api.v1.v1_data.constants import DataApprovalStatus
from api.v1.v1_profile.models import Administration
from api.v1.v1_users.models import SystemUser


def is_has_approver(role: int, access_forms: list = None):
    # Check if user has approver access through access_forms
    has_approver_access = False
    if access_forms:
        for form_access in access_forms:
            if form_access.get('access_type') == FormAccessTypes.approve:
                has_approver_access = True
                break
    is_approver = role == UserRoleTypes.admin and has_approver_access
    is_super_admin = role == UserRoleTypes.super_admin
    return is_approver or is_super_admin


def check_form_approval_assigned(
    role: int,
    administration: Administration,
    user: SystemUser = None,
    access_forms: list = None
):
    forms = [
        item["form_id"]
        for item in access_forms
    ]
    # Check if user is super admin
    # Check if user is admin and has approver access
    unique_user = is_has_approver(
        role=role,
        access_forms=access_forms
    )
    # Check if user is not super admin and has no approver access
    # and is editing user
    if not unique_user and not user:
        return False
    # Check if form id x in y administration has approver assignment
    # send a message to FE 403
    form_approval_assignment_obj = FormApprovalAssignment.objects
    form_approval_assignment = form_approval_assignment_obj.filter(
        administration=administration
    )
    if not user:
        form_approval_assignment = form_approval_assignment.filter(
            form__in=forms
        )
    if user:  # for edited user
        # if administration updated
        if user.user_access.administration_id != administration.id:
            # check approver tree with prev administration and prev forms
            # if any, delete previous approver tree
            prev_approval_assignment = form_approval_assignment_obj.filter(
                administration_id=user.user_access.administration_id,
                form_id__in=[uf.form_id for uf in user.user_form.all()],
                user=user,
            )
            if prev_approval_assignment.count():
                # find administration path
                user_adm = user.user_access.administration
                filter_batch = {
                    "form_id__in": [uf.form_id for uf in user.user_form.all()],
                    "approved": False,
                }
                if user_adm.level.level in [0, 3]:
                    filter_batch.update({"administration_id": user_adm.id})
                else:
                    adm_path = f"{user_adm.path}{user_adm.id}"
                    filter_batch.update(
                        {"administration__path__startswith": adm_path}
                    )
                prev_pending_batch = PendingDataBatch.objects.filter(
                    **filter_batch
                ).values_list("id", flat=True)
                # check pending batch approval for prev user
                # find pending data approval by prev_pending_batch
                prev_pending_approval = PendingDataApproval.objects.filter(
                    batch_id__in=prev_pending_batch,
                    status=DataApprovalStatus.pending,
                    user=user,
                )
                if prev_pending_approval.count():
                    # raise an error to prevent administration update
                    # when edited user has pending data approval
                    prev_pending_approval.delete()
                prev_approval_assignment.delete()

        # check if updated user already have form assigned
        form_assigned = (
            form_approval_assignment.filter(user=user)
            .distinct("form")
            .values_list("form_id", flat=True)
        )
        form_assigned_to_delete = []
        form_to_assign = [
            item["form_id"].id
            for item in access_forms
            if item["access_type"] == FormAccessTypes.approve
        ]
        for fa in form_assigned:
            if fa not in form_to_assign:
                form_assigned_to_delete.append(fa)
            else:
                # remove assigned form from form filter
                form_to_assign.remove(fa)
        form_approval_assignment = form_approval_assignment.filter(
            form_id__in=form_to_assign
        )
        # delete approval assigned
        if form_assigned_to_delete:
            FormApprovalAssignment.objects.filter(
                administration=administration,
                form_id__in=form_assigned_to_delete,
                user=user,
            ).delete()

    form_approval_assignment = form_approval_assignment.distinct(
        "form", "administration"
    ).all()
    if form_approval_assignment:
        message_detail = [
            {"form": fa.form.name, "administration": fa.administration.name}
            for fa in form_approval_assignment
        ]
        return message_detail
    return False


def assign_form_approval(
    role: int,
    forms: List[Forms],
    administration: Administration,
    user: SystemUser,
    access_forms: List = None
):
    unique_user = is_has_approver(
        role=role,
        access_forms=access_forms
    )
    if not unique_user:
        return False

    form_to_assign = []
    for fr in forms:
        # Check if the user has approver access for this form
        user_form = UserForms.objects.filter(user=user, form=fr).first()
        has_approver_access = False

        # Check if the user has approver access through FormAccess
        if user_form:
            user_form_access = FormAccess.objects.filter(
                user_form=user_form,
                access_type=FormAccessTypes.approve
            ).exists()
            if user_form_access:
                has_approver_access = True

        # Only assign forms where the user has approver access
        if has_approver_access:
            form_to_assign.append(fr)

    # check if forms already assigned to user
    check = FormApprovalAssignment.objects.filter(
        administration=administration, form__in=form_to_assign, user=user
    )

    # Update existing assignments
    if check:
        # Remove already assigned forms from the list to assign
        form_to_assign = [
            fr for fr in form_to_assign
            if fr.id not in check.values_list("form_id", flat=True)
        ]

        # Update timestamps for existing assignments
        for fa in check.all():
            fa.updated = timezone.now()
            fa.save()
    # Add user value to approval assignment table (approval tree)
    form_approval_obj = [
        FormApprovalAssignment(
            form=fr, administration=administration, user=user
        )
        for fr in form_to_assign
    ]
    approval = FormApprovalAssignment.objects.bulk_create(form_approval_obj)

    # Assign to previous batch
    has_pending_data_batch = PendingDataBatch.objects.filter(
        approved=False
    ).count()
    if has_pending_data_batch:
        batch_filter = {"approved": False}
        if administration.path:
            batch_filter.update(
                {"administration__path__startswith": administration.path}
            )
        else:
            batch_filter.update({"administration": 1})
        current_batch = PendingDataBatch.objects.filter(**batch_filter).all()
        if current_batch.count():
            for batch in current_batch:
                # Only process forms that are in the approved list
                if batch.form in forms and batch.form in form_to_assign:
                    approver = PendingDataApproval.objects.filter(
                        level=administration.level, batch=batch
                    ).first()
                    if not approver:
                        approver = PendingDataApproval(
                            level=administration.level, user=user, batch=batch
                        )
                    else:
                        approver.user = user
                    approver.save()
                else:
                    approver = PendingDataApproval.objects.filter(
                        batch=batch, user=user
                    ).all()
                    approver.delete()
    return approval

1	from django.utils import timezone	1✔
2	from typing import List	1✔
3	from api.v1.v1_profile.constants import UserRoleTypes	1✔
4	from api.v1.v1_forms.constants import FormAccessTypes	1✔
5	from api.v1.v1_forms.models import (	1✔
6	FormApprovalAssignment,
7	FormAccess,
8	UserForms,
9	Forms,
10	)
11	from api.v1.v1_data.models import PendingDataBatch, PendingDataApproval	1✔
12	from api.v1.v1_data.constants import DataApprovalStatus	1✔
13	from api.v1.v1_profile.models import Administration	1✔
14	from api.v1.v1_users.models import SystemUser	1✔
15
16
17	def is_has_approver(role: int, access_forms: list = None):	1✔
18	# Check if user has approver access through access_forms
19	has_approver_access = False	1✔
20	if access_forms:	1✔
21	for form_access in access_forms:	1✔
22	if form_access.get('access_type') == FormAccessTypes.approve:	1✔
23	has_approver_access = True	1✔
24	break	1✔
25	is_approver = role == UserRoleTypes.admin and has_approver_access	1✔
26	is_super_admin = role == UserRoleTypes.super_admin	1✔
27	return is_approver or is_super_admin	1✔
28
29
30	def check_form_approval_assigned(	1✔
31	role: int,
32	administration: Administration,
33	user: SystemUser = None,
34	access_forms: list = None
35	):
36	forms = [	1✔
37	item["form_id"]
38	for item in access_forms
39	]
40	# Check if user is super admin
41	# Check if user is admin and has approver access
42	unique_user = is_has_approver(	1✔
43	role=role,
44	access_forms=access_forms
45	)
46	# Check if user is not super admin and has no approver access
47	# and is editing user
48	if not unique_user and not user:	1✔
49	return False	1✔
50	# Check if form id x in y administration has approver assignment
51	# send a message to FE 403
52	form_approval_assignment_obj = FormApprovalAssignment.objects	1✔
53	form_approval_assignment = form_approval_assignment_obj.filter(	1✔
54	administration=administration
55	)
56	if not user:	1✔
57	form_approval_assignment = form_approval_assignment.filter(	1✔
58	form__in=forms
59	)
60	if user: # for edited user	1✔
61	# if administration updated
62	if user.user_access.administration_id != administration.id:	1✔
63	# check approver tree with prev administration and prev forms
64	# if any, delete previous approver tree
65	prev_approval_assignment = form_approval_assignment_obj.filter(	1✔
66	administration_id=user.user_access.administration_id,
67	form_id__in=[uf.form_id for uf in user.user_form.all()],
68	user=user,
69	)
70	if prev_approval_assignment.count():	1!
71	# find administration path
72	user_adm = user.user_access.administration	×
73	filter_batch = {	×
74	"form_id__in": [uf.form_id for uf in user.user_form.all()],
75	"approved": False,
76	}
77	if user_adm.level.level in [0, 3]:	×
78	filter_batch.update({"administration_id": user_adm.id})	×
79	else:
80	adm_path = f"{user_adm.path}{user_adm.id}"	×
81	filter_batch.update(	×
82	{"administration__path__startswith": adm_path}
83	)
84	prev_pending_batch = PendingDataBatch.objects.filter(	×
85	**filter_batch
86	).values_list("id", flat=True)
87	# check pending batch approval for prev user
88	# find pending data approval by prev_pending_batch
89	prev_pending_approval = PendingDataApproval.objects.filter(	×
90	batch_id__in=prev_pending_batch,
91	status=DataApprovalStatus.pending,
92	user=user,
93	)
94	if prev_pending_approval.count():	×
95	# raise an error to prevent administration update
96	# when edited user has pending data approval
97	prev_pending_approval.delete()	×
98	prev_approval_assignment.delete()	×
99
100	# check if updated user already have form assigned
101	form_assigned = (	1✔
102	form_approval_assignment.filter(user=user)
103	.distinct("form")
104	.values_list("form_id", flat=True)
105	)
106	form_assigned_to_delete = []	1✔
107	form_to_assign = [	1✔
108	item["form_id"].id
109	for item in access_forms
110	if item["access_type"] == FormAccessTypes.approve
111	]
112	for fa in form_assigned:	1✔
113	if fa not in form_to_assign:	1!
114	form_assigned_to_delete.append(fa)	1✔
115	else:
116	# remove assigned form from form filter
117	form_to_assign.remove(fa)	×
118	form_approval_assignment = form_approval_assignment.filter(	1✔
119	form_id__in=form_to_assign
120	)
121	# delete approval assigned
122	if form_assigned_to_delete:	1✔
123	FormApprovalAssignment.objects.filter(	1✔
124	administration=administration,
125	form_id__in=form_assigned_to_delete,
126	user=user,
127	).delete()
128
129	form_approval_assignment = form_approval_assignment.distinct(	1✔
130	"form", "administration"
131	).all()
132	if form_approval_assignment:	1✔
133	message_detail = [	1✔
134	{"form": fa.form.name, "administration": fa.administration.name}
135	for fa in form_approval_assignment
136	]
137	return message_detail	1✔
138	return False	1✔
139
140
141	def assign_form_approval(	1✔
142	role: int,
143	forms: List[Forms],
144	administration: Administration,
145	user: SystemUser,
146	access_forms: List = None
147	):
148	unique_user = is_has_approver(	1✔
149	role=role,
150	access_forms=access_forms
151	)
152	if not unique_user:	1!
153	return False	×
154
155	form_to_assign = []	1✔
156	for fr in forms:	1✔
157	# Check if the user has approver access for this form
158	user_form = UserForms.objects.filter(user=user, form=fr).first()	1✔
159	has_approver_access = False	1✔
160
161	# Check if the user has approver access through FormAccess
162	if user_form:	1!
163	user_form_access = FormAccess.objects.filter(	1✔
164	user_form=user_form,
165	access_type=FormAccessTypes.approve
166	).exists()
167	if user_form_access:	1!
168	has_approver_access = True	1✔
169
170	# Only assign forms where the user has approver access
171	if has_approver_access:	1!
172	form_to_assign.append(fr)	1✔
173
174	# check if forms already assigned to user
175	check = FormApprovalAssignment.objects.filter(	1✔
176	administration=administration, form__in=form_to_assign, user=user
177	)
178
179	# Update existing assignments
180	if check:	1!
181	# Remove already assigned forms from the list to assign
182	form_to_assign = [	×
183	fr for fr in form_to_assign
184	if fr.id not in check.values_list("form_id", flat=True)
185	]
186
187	# Update timestamps for existing assignments
188	for fa in check.all():	×
189	fa.updated = timezone.now()	×
190	fa.save()	×
191	# Add user value to approval assignment table (approval tree)
192	form_approval_obj = [	1✔
193	FormApprovalAssignment(
194	form=fr, administration=administration, user=user
195	)
196	for fr in form_to_assign
197	]
198	approval = FormApprovalAssignment.objects.bulk_create(form_approval_obj)	1✔
199
200	# Assign to previous batch
201	has_pending_data_batch = PendingDataBatch.objects.filter(	1✔
202	approved=False
203	).count()
204	if has_pending_data_batch:	1!
205	batch_filter = {"approved": False}	×
206	if administration.path:	×
207	batch_filter.update(	×
208	{"administration__path__startswith": administration.path}
209	)
210	else:
211	batch_filter.update({"administration": 1})	×
212	current_batch = PendingDataBatch.objects.filter(**batch_filter).all()	×
213	if current_batch.count():	×
214	for batch in current_batch:	×
215	# Only process forms that are in the approved list
216	if batch.form in forms and batch.form in form_to_assign:	×
217	approver = PendingDataApproval.objects.filter(	×
218	level=administration.level, batch=batch
219	).first()
220	if not approver:	×
221	approver = PendingDataApproval(	×
222	level=administration.level, user=user, batch=batch
223	)
224	else:
225	approver.user = user	×
226	approver.save()	×
227	else:
228	approver = PendingDataApproval.objects.filter(	×
229	batch=batch, user=user
230	).all()
231	approver.delete()	×
232	return approval	1✔

akvo / iwsims-demo / #72

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