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

91.16

backend/api/v1/v1_data/management/commands/fake_pending_data_seeder.py

import random
import re
from datetime import timedelta

import pandas as pd
from django.core.management import BaseCommand
from django.utils import timezone
from faker import Faker

from iwsims.settings import COUNTRY_NAME
from api.v1.v1_data.models import (
    PendingFormData,
    PendingAnswers,
    PendingDataApproval,
    PendingDataBatch,
)
from api.v1.v1_forms.constants import QuestionTypes, FormTypes, SubmissionTypes
from api.v1.v1_forms.models import FormApprovalAssignment
from api.v1.v1_forms.models import Forms, UserForms
from api.v1.v1_profile.constants import UserRoleTypes
from api.v1.v1_profile.models import Administration, Access
from api.v1.v1_users.models import SystemUser, Organisation
from api.v1.v1_users.management.commands.demo_approval_flow import (
    create_approver,
)
from api.v1.v1_profile.functions import get_max_administration_level

fake = Faker()


def set_answer_data(data, question):
    name = None
    value = None
    option = None

    if question.type == QuestionTypes.geo:
        option = data.geo
    elif question.type == QuestionTypes.administration:
        name = data.administration.name
        value = data.administration.id
    elif question.type == QuestionTypes.text:
        name = fake.company() if question.meta else fake.sentence(nb_words=3)
    elif question.type == QuestionTypes.number:
        value = fake.random_int(min=10, max=50)
    elif question.type == QuestionTypes.option:
        option = [question.options.order_by("?").first().value]
    elif question.type == QuestionTypes.multiple_option:
        option = list(
            question.options.order_by("?").values_list("value", flat=True)[
                0: fake.random_int(min=1, max=3)
            ]
        )
    elif question.type == QuestionTypes.photo:
        name = fake.image_url()
    elif question.type == QuestionTypes.date:
        name = fake.date_between_dates(
            date_start=timezone.datetime.now().date() - timedelta(days=90),
            date_end=timezone.datetime.now().date(),
        ).strftime("%m/%d/%Y")
    else:
        pass
    return name, value, option


def add_fake_answers(data: PendingFormData):
    form = data.form
    meta_name = []
    for question in form.form_questions.all().order_by("order"):
        name, value, option = set_answer_data(data, question)
        if question.meta:
            if name:
                meta_name.append(name)
            elif option and question.type != QuestionTypes.geo:
                meta_name.append(",".join(option))
            elif value and question.type != QuestionTypes.administration:
                meta_name.append(str(value))
            else:
                pass

        if question.type == QuestionTypes.administration:
            name = None

        PendingAnswers.objects.create(
            pending_data=data,
            question=question,
            name=name,
            value=value,
            options=option,
            created_by=SystemUser.objects.order_by("?").first(),
        )
    data.name = " - ".join(meta_name)
    data.save()


def seed_data(form, fake_geo, repeat, created_by):
    for i in range(repeat):
        administration = created_by.user_access.administration
        mobile_assignment = created_by.mobile_assignments.order_by("?").first()
        geo = fake_geo.iloc[i].to_dict()
        data = PendingFormData.objects.create(
            name=fake.pystr_format(),
            geo=[geo["X"], geo["Y"]],
            form=form,
            administration=administration,
            created_by=created_by,
            submission_type=SubmissionTypes.registration,
            submitter=mobile_assignment.name if mobile_assignment else None,
        )
        add_fake_answers(data)


def create_or_get_submitter(max_adm_level: int = 1):
    organisation = Organisation.objects.first()
    administration = Administration.objects\
        .filter(level__level=max_adm_level).order_by("?").first()
    email = ("{}{}@test.com").format(
        re.sub("[^A-Za-z0-9]+", "", administration.name.lower()),
        random.randint(200, 300),
    )
    submitter, created = SystemUser.objects.get_or_create(
        organisation=organisation,
        email=email,
        first_name=administration.name,
        last_name="User",
    )
    if created:
        submitter.set_password("test")
        submitter.save()
        Access.objects.create(
            user=submitter,
            role=UserRoleTypes.admin,
            administration=administration
        )
    return submitter


def assign_batch_for_approval(
    batch: PendingDataBatch,
    user: SystemUser,
    test: bool = False,
):
    organisation = Organisation.objects.first()
    for administration in user.user_access.administration.ancestors.all():
        # check if approval assignment for the path is not available
        assignment = FormApprovalAssignment.objects.filter(
            form=batch.form, administration=administration
        ).first()
        if not assignment:
            assignment = create_approver(
                form=batch.form,
                administration=administration,
                organisation=organisation,
            )
            if not test:
                last_name = (
                    "Admin" if administration.level.level == 1 else "Approver"
                )
                print(
                    "Level: {} ({})".format(
                        administration.level.level, administration.level.name
                    )
                )
                print(f"- Administration Name: {administration.name}")
                print(
                    "- Approver: {} ({})".format(
                        assignment.user.email, last_name
                    )
                )
        PendingDataApproval.objects.create(
            batch=batch,
            user=assignment.user,
            level=assignment.user.user_access.administration.level,
        )


def print_info(form, administration, submitter, limit, test):
    if not test:
        print(f"Batch: {limit} datapoints\n")
        print(f"\nForm Name: {form.name}\n")
        print("\nSubmitter:")
        print(f"- Administration: {administration.full_name}")
        print("- Email: {}\n".format(submitter.email))


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument(
            "-r", "--repeat", nargs="?", const=20, default=5, type=int
        )
        parser.add_argument(
            "-b", "--batch", nargs="?", const=1, default=5, type=int
        )
        parser.add_argument(
            "-t", "--test", nargs="?", const=1, default=False, type=bool
        )
        parser.add_argument(
            "-e", "--email", nargs="?", const=1, default=None, type=str
        )

    def handle(self, *args, **options):
        test = options.get("test")
        PendingDataApproval.objects.all().delete()
        PendingDataBatch.objects.all().delete()
        PendingFormData.objects.all().delete()
        fake_geo = pd.read_csv(f"./source/{COUNTRY_NAME}_random_points.csv")
        max_adm_level = get_max_administration_level()
        forms = Forms.objects.filter(type=FormTypes.county).all()
        for form in forms:
            submitter = create_or_get_submitter(
                max_adm_level=max_adm_level
            )
            max_adm_level -= 1
            seed_data(form, fake_geo, options.get("repeat"), submitter)
            administration = submitter.user_access.administration
            UserForms.objects.get_or_create(form=form, user=submitter)
            limit = options.get("batch")
            print_info(form, administration, submitter, limit, test)
            if limit:
                if form.type == FormTypes.county and not test:
                    print("Approvers:\n")
                while PendingFormData.objects.filter(
                    batch__isnull=True, form=form
                ).count():
                    batch = PendingDataBatch.objects.create(
                        name=fake.catch_phrase(),
                        form=form,
                        administration=submitter.user_access.administration,
                        user=submitter,
                    )

                    objs = PendingFormData.objects.filter(
                        batch__isnull=True, form=form
                    )[:limit]
                    for obj in objs:
                        obj.batch = batch
                    PendingFormData.objects.bulk_update(objs, fields=["batch"])
                    if form.type == FormTypes.county:
                        assign_batch_for_approval(
                            batch=batch,
                            user=submitter,
                            test=test,
                        )

1	import random	1✔
2	import re	1✔
3	from datetime import timedelta	1✔
4
5	import pandas as pd	1✔
6	from django.core.management import BaseCommand	1✔
7	from django.utils import timezone	1✔
8	from faker import Faker	1✔
9
10	from iwsims.settings import COUNTRY_NAME	1✔
11	from api.v1.v1_data.models import (	1✔
12	PendingFormData,
13	PendingAnswers,
14	PendingDataApproval,
15	PendingDataBatch,
16	)
17	from api.v1.v1_forms.constants import QuestionTypes, FormTypes, SubmissionTypes	1✔
18	from api.v1.v1_forms.models import FormApprovalAssignment	1✔
19	from api.v1.v1_forms.models import Forms, UserForms	1✔
20	from api.v1.v1_profile.constants import UserRoleTypes	1✔
21	from api.v1.v1_profile.models import Administration, Access	1✔
22	from api.v1.v1_users.models import SystemUser, Organisation	1✔
23	from api.v1.v1_users.management.commands.demo_approval_flow import (	1✔
24	create_approver,
25	)
26	from api.v1.v1_profile.functions import get_max_administration_level	1✔
27
28	fake = Faker()	1✔
29
30
31	def set_answer_data(data, question):	1✔
32	name = None	1✔
33	value = None	1✔
34	option = None	1✔
35
36	if question.type == QuestionTypes.geo:	1✔
37	option = data.geo	1✔
38	elif question.type == QuestionTypes.administration:	1✔
39	name = data.administration.name	1✔
40	value = data.administration.id	1✔
41	elif question.type == QuestionTypes.text:	1✔
42	name = fake.company() if question.meta else fake.sentence(nb_words=3)	1✔
43	elif question.type == QuestionTypes.number:	1✔
44	value = fake.random_int(min=10, max=50)	1✔
45	elif question.type == QuestionTypes.option:	1✔
46	option = [question.options.order_by("?").first().value]	1✔
47	elif question.type == QuestionTypes.multiple_option:	1✔
48	option = list(	1✔
49	question.options.order_by("?").values_list("value", flat=True)[
50	0: fake.random_int(min=1, max=3)
51	]
52	)
53	elif question.type == QuestionTypes.photo:	1✔
54	name = fake.image_url()	1✔
55	elif question.type == QuestionTypes.date:	1✔
56	name = fake.date_between_dates(	1✔
57	date_start=timezone.datetime.now().date() - timedelta(days=90),
58	date_end=timezone.datetime.now().date(),
59	).strftime("%m/%d/%Y")
60	else:
61	pass
62	return name, value, option	1✔
63
64
65	def add_fake_answers(data: PendingFormData):	1✔
66	form = data.form	1✔
67	meta_name = []	1✔
68	for question in form.form_questions.all().order_by("order"):	1✔
69	name, value, option = set_answer_data(data, question)	1✔
70	if question.meta:	1✔
71	if name:	1✔
72	meta_name.append(name)	1✔
73	elif option and question.type != QuestionTypes.geo:	1✔
74	meta_name.append(",".join(option))	1✔
75	elif value and question.type != QuestionTypes.administration:	1✔
76	meta_name.append(str(value))	1✔
77	else:
78	pass
79
80	if question.type == QuestionTypes.administration:	1✔
81	name = None	1✔
82
83	PendingAnswers.objects.create(	1✔
84	pending_data=data,
85	question=question,
86	name=name,
87	value=value,
88	options=option,
89	created_by=SystemUser.objects.order_by("?").first(),
90	)
91	data.name = " - ".join(meta_name)	1✔
92	data.save()	1✔
93
94
95	def seed_data(form, fake_geo, repeat, created_by):	1✔
96	for i in range(repeat):	1✔
97	administration = created_by.user_access.administration	1✔
98	mobile_assignment = created_by.mobile_assignments.order_by("?").first()	1✔
99	geo = fake_geo.iloc[i].to_dict()	1✔
100	data = PendingFormData.objects.create(	1✔
101	name=fake.pystr_format(),
102	geo=[geo["X"], geo["Y"]],
103	form=form,
104	administration=administration,
105	created_by=created_by,
106	submission_type=SubmissionTypes.registration,
107	submitter=mobile_assignment.name if mobile_assignment else None,
108	)
109	add_fake_answers(data)	1✔
110
111
112	def create_or_get_submitter(max_adm_level: int = 1):	1✔
113	organisation = Organisation.objects.first()	1✔
114	administration = Administration.objects\	1✔
115	.filter(level__level=max_adm_level).order_by("?").first()
116	email = ("{}{}@test.com").format(	1✔
117	re.sub("[^A-Za-z0-9]+", "", administration.name.lower()),
118	random.randint(200, 300),
119	)
120	submitter, created = SystemUser.objects.get_or_create(	1✔
121	organisation=organisation,
122	email=email,
123	first_name=administration.name,
124	last_name="User",
125	)
126	if created:	1!
127	submitter.set_password("test")	1✔
128	submitter.save()	1✔
129	Access.objects.create(	1✔
130	user=submitter,
131	role=UserRoleTypes.admin,
132	administration=administration
133	)
134	return submitter	1✔
135
136
137	def assign_batch_for_approval(	1✔
138	batch: PendingDataBatch,
139	user: SystemUser,
140	test: bool = False,
141	):
142	organisation = Organisation.objects.first()	1✔
143	for administration in user.user_access.administration.ancestors.all():	1✔
144	# check if approval assignment for the path is not available
145	assignment = FormApprovalAssignment.objects.filter(	1✔
146	form=batch.form, administration=administration
147	).first()
148	if not assignment:	1✔
149	assignment = create_approver(	1✔
150	form=batch.form,
151	administration=administration,
152	organisation=organisation,
153	)
154	if not test:	1!
155	last_name = (	×
156	"Admin" if administration.level.level == 1 else "Approver"
157	)
158	print(	×
159	"Level: {} ({})".format(
160	administration.level.level, administration.level.name
161	)
162	)
163	print(f"- Administration Name: {administration.name}")	×
164	print(	×
165	"- Approver: {} ({})".format(
166	assignment.user.email, last_name
167	)
168	)
169	PendingDataApproval.objects.create(	1✔
170	batch=batch,
171	user=assignment.user,
172	level=assignment.user.user_access.administration.level,
173	)
174
175
176	def print_info(form, administration, submitter, limit, test):	1✔
177	if not test:	1!
178	print(f"Batch: {limit} datapoints\n")	×
179	print(f"\nForm Name: {form.name}\n")	×
180	print("\nSubmitter:")	×
181	print(f"- Administration: {administration.full_name}")	×
182	print("- Email: {}\n".format(submitter.email))	×
183
184
185	class Command(BaseCommand):	1✔
186	def add_arguments(self, parser):	1✔
187	parser.add_argument(	1✔
188	"-r", "--repeat", nargs="?", const=20, default=5, type=int
189	)
190	parser.add_argument(	1✔
191	"-b", "--batch", nargs="?", const=1, default=5, type=int
192	)
193	parser.add_argument(	1✔
194	"-t", "--test", nargs="?", const=1, default=False, type=bool
195	)
196	parser.add_argument(	1✔
197	"-e", "--email", nargs="?", const=1, default=None, type=str
198	)
199
200	def handle(self, args, *options):	1✔
201	test = options.get("test")	1✔
202	PendingDataApproval.objects.all().delete()	1✔
203	PendingDataBatch.objects.all().delete()	1✔
204	PendingFormData.objects.all().delete()	1✔
205	fake_geo = pd.read_csv(f"./source/{COUNTRY_NAME}_random_points.csv")	1✔
206	max_adm_level = get_max_administration_level()	1✔
207	forms = Forms.objects.filter(type=FormTypes.county).all()	1✔
208	for form in forms:	1✔
209	submitter = create_or_get_submitter(	1✔
210	max_adm_level=max_adm_level
211	)
212	max_adm_level -= 1	1✔
213	seed_data(form, fake_geo, options.get("repeat"), submitter)	1✔
214	administration = submitter.user_access.administration	1✔
215	UserForms.objects.get_or_create(form=form, user=submitter)	1✔
216	limit = options.get("batch")	1✔
217	print_info(form, administration, submitter, limit, test)	1✔
218	if limit:	1!
219	if form.type == FormTypes.county and not test:	1!
220	print("Approvers:\n")	×
221	while PendingFormData.objects.filter(	1✔
222	batch__isnull=True, form=form
223	).count():
224	batch = PendingDataBatch.objects.create(	1✔
225	name=fake.catch_phrase(),
226	form=form,
227	administration=submitter.user_access.administration,
228	user=submitter,
229	)
230
231	objs = PendingFormData.objects.filter(	1✔
232	batch__isnull=True, form=form
233	)[:limit]
234	for obj in objs:	1✔
235	obj.batch = batch	1✔
236	PendingFormData.objects.bulk_update(objs, fields=["batch"])	1✔
237	if form.type == FormTypes.county:	1!
238	assign_batch_for_approval(	1✔
239	batch=batch,
240	user=submitter,
241	test=test,
242	)

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