akvo / iwsims-demo / #93

Committed 30 Apr 2025 07:36AM UTC coverage: 86.373% (+0.07%) from 86.3%

Build # #93

Build Type

Pull #29

coveralls-python

Committed by

ifirmawan

Commit Message

[#28] Remove any county, sub-County terms in the backend

Pull Request Pull Request #29: Feature/28 eng 1232 data submission by admin super admin

Run Details

2591 of 3118 branches covered (83.1%)

Branch coverage included in aggregate %.

5915 of 6730 relevant lines covered (87.89%)

0.88 hits per line

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

88.07

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

import random
import re
import pandas as pd
from django.core.management import BaseCommand
from faker import Faker

from iwsims.settings import COUNTRY_NAME
from api.v1.v1_data.models import (
    PendingFormData,
    PendingDataApproval,
    PendingDataBatch,
)
from api.v1.v1_forms.constants import 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
from api.v1.v1_data.functions import add_fake_answers

fake = Faker()


def seed_data(form, fake_geo, repeat, created_by):
    pendings = []
    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()
        pending_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(pending_data)
        pendings.append(pending_data)
    return pendings


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.all()
        for form in forms:
            submitter = create_or_get_submitter(
                max_adm_level=max_adm_level
            )
            max_adm_level -= 1
            seed_data(
                form=form,
                fake_geo=fake_geo,
                repeat=options.get("repeat"),
                created_by=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:
                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"])
                    assign_batch_for_approval(
                        batch=batch,
                        user=submitter,
                        test=test,
                    )

1	import random	1✔
2	import re	1✔
3	import pandas as pd	1✔
4	from django.core.management import BaseCommand	1✔
5	from faker import Faker	1✔
6
7	from iwsims.settings import COUNTRY_NAME	1✔
8	from api.v1.v1_data.models import (	1✔
9	PendingFormData,
10	PendingDataApproval,
11	PendingDataBatch,
12	)
13	from api.v1.v1_forms.constants import SubmissionTypes	1✔
14	from api.v1.v1_forms.models import FormApprovalAssignment	1✔
15	from api.v1.v1_forms.models import Forms, UserForms	1✔
16	from api.v1.v1_profile.constants import UserRoleTypes	1✔
17	from api.v1.v1_profile.models import Administration, Access	1✔
18	from api.v1.v1_users.models import SystemUser, Organisation	1✔
19	from api.v1.v1_users.management.commands.demo_approval_flow import (	1✔
20	create_approver,
21	)
22	from api.v1.v1_profile.functions import get_max_administration_level	1✔
23	from api.v1.v1_data.functions import add_fake_answers	1✔
24
25	fake = Faker()	1✔
26
27
28	def seed_data(form, fake_geo, repeat, created_by):	1✔
29	pendings = []	1✔
30	for i in range(repeat):	1✔
31	administration = created_by.user_access.administration	1✔
32	mobile_assignment = created_by.mobile_assignments.order_by("?").first()	1✔
33	geo = fake_geo.iloc[i].to_dict()	1✔
34	pending_data = PendingFormData.objects.create(	1✔
35	name=fake.pystr_format(),
36	geo=[geo["X"], geo["Y"]],
37	form=form,
38	administration=administration,
39	created_by=created_by,
40	submission_type=SubmissionTypes.registration,
41	submitter=mobile_assignment.name if mobile_assignment else None,
42	)
43	add_fake_answers(pending_data)	1✔
44	pendings.append(pending_data)	1✔
45	return pendings	1✔
46
47
48	def create_or_get_submitter(max_adm_level: int = 1):	1✔
49	organisation = Organisation.objects.first()	1✔
50	administration = Administration.objects\	1✔
51	.filter(level__level=max_adm_level).order_by("?").first()
52	email = ("{}{}@test.com").format(	1✔
53	re.sub("[^A-Za-z0-9]+", "", administration.name.lower()),
54	random.randint(200, 300),
55	)
56	submitter, created = SystemUser.objects.get_or_create(	1✔
57	organisation=organisation,
58	email=email,
59	first_name=administration.name,
60	last_name="User",
61	)
62	if created:	1!
63	submitter.set_password("test")	1✔
64	submitter.save()	1✔
65	Access.objects.create(	1✔
66	user=submitter,
67	role=UserRoleTypes.admin,
68	administration=administration
69	)
70	return submitter	1✔
71
72
73	def assign_batch_for_approval(	1✔
74	batch: PendingDataBatch,
75	user: SystemUser,
76	test: bool = False,
77	):
78	organisation = Organisation.objects.first()	1✔
79	for administration in user.user_access.administration.ancestors.all():	1✔
80	# check if approval assignment for the path is not available
81	assignment = FormApprovalAssignment.objects.filter(	1✔
82	form=batch.form, administration=administration
83	).first()
84	if not assignment:	1✔
85	assignment = create_approver(	1✔
86	form=batch.form,
87	administration=administration,
88	organisation=organisation,
89	)
90	if not test:	1!
91	last_name = (	×
92	"Admin" if administration.level.level == 1 else "Approver"
93	)
94	print(	×
95	"Level: {} ({})".format(
96	administration.level.level, administration.level.name
97	)
98	)
99	print(f"- Administration Name: {administration.name}")	×
100	print(	×
101	"- Approver: {} ({})".format(
102	assignment.user.email, last_name
103	)
104	)
105	PendingDataApproval.objects.create(	1✔
106	batch=batch,
107	user=assignment.user,
108	level=assignment.user.user_access.administration.level,
109	)
110
111
112	def print_info(form, administration, submitter, limit, test):	1✔
113	if not test:	1!
114	print(f"Batch: {limit} datapoints\n")	×
115	print(f"\nForm Name: {form.name}\n")	×
116	print("\nSubmitter:")	×
117	print(f"- Administration: {administration.full_name}")	×
118	print("- Email: {}\n".format(submitter.email))	×
119
120
121	class Command(BaseCommand):	1✔
122	def add_arguments(self, parser):	1✔
123	parser.add_argument(	1✔
124	"-r", "--repeat", nargs="?", const=20, default=5, type=int
125	)
126	parser.add_argument(	1✔
127	"-b", "--batch", nargs="?", const=1, default=5, type=int
128	)
129	parser.add_argument(	1✔
130	"-t", "--test", nargs="?", const=1, default=False, type=bool
131	)
132	parser.add_argument(	1✔
133	"-e", "--email", nargs="?", const=1, default=None, type=str
134	)
135
136	def handle(self, args, *options):	1✔
137	test = options.get("test")	1✔
138	PendingDataApproval.objects.all().delete()	1✔
139	PendingDataBatch.objects.all().delete()	1✔
140	PendingFormData.objects.all().delete()	1✔
141	fake_geo = pd.read_csv(f"./source/{COUNTRY_NAME}_random_points.csv")	1✔
142	max_adm_level = get_max_administration_level()	1✔
143	forms = Forms.objects.all()	1✔
144	for form in forms:	1✔
145	submitter = create_or_get_submitter(	1✔
146	max_adm_level=max_adm_level
147	)
148	max_adm_level -= 1	1✔
149	seed_data(	1✔
150	form=form,
151	fake_geo=fake_geo,
152	repeat=options.get("repeat"),
153	created_by=submitter
154	)
155	administration = submitter.user_access.administration	1✔
156	UserForms.objects.get_or_create(form=form, user=submitter)	1✔
157	limit = options.get("batch")	1✔
158	print_info(form, administration, submitter, limit, test)	1✔
159	if limit:	1!
160	while PendingFormData.objects.filter(	1✔
161	batch__isnull=True, form=form
162	).count():
163	batch = PendingDataBatch.objects.create(	1✔
164	name=fake.catch_phrase(),
165	form=form,
166	administration=submitter.user_access.administration,
167	user=submitter,
168	)
169
170	objs = PendingFormData.objects.filter(	1✔
171	batch__isnull=True, form=form
172	)[:limit]
173	for obj in objs:	1✔
174	obj.batch = batch	1✔
175	PendingFormData.objects.bulk_update(objs, fields=["batch"])	1✔
176	assign_batch_for_approval(	1✔
177	batch=batch,
178	user=submitter,
179	test=test,
180	)

akvo / iwsims-demo / #93

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