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

96.15

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

import pandas as pd
from django.core.management import BaseCommand
from faker import Faker
from api.v1.v1_data.models import (
    PendingFormData,
    PendingDataBatch,
    PendingDataApproval,
)
from api.v1.v1_forms.models import Forms, FormApprovalAssignment
from api.v1.v1_forms.constants import SubmissionTypes
from api.v1.v1_data.management.commands.fake_data_seeder import (
    add_fake_answers,
)
from api.v1.v1_profile.models import Administration
from api.v1.v1_data.tasks import seed_approved_data
from api.v1.v1_data.constants import DataApprovalStatus

fake = Faker()


def seed_data(form, datapoint, user, repeat, approved):
    pendings = []
    for i in range(repeat):
        pending_data = PendingFormData.objects.create(
            name=datapoint.name,
            geo=datapoint.geo,
            uuid=datapoint.uuid,
            form=datapoint.form,
            administration=datapoint.administration,
            created_by=user,
            submission_type=SubmissionTypes.monitoring,
        )
        add_fake_answers(pending_data)
        pendings.append(pending_data)
    pending_items = [
        {"administration_id": pending.administration.id, "instance": pending}
        for pending in pendings
    ]
    df = pd.DataFrame(pending_items)
    grouped_data = df.groupby(["administration_id"]).apply(
        lambda x: x.to_dict("records")
    )
    for administration_id, items in grouped_data.items():
        [dp] = items[:1]
        batch_name = fake.sentence(nb_words=3)
        batch = PendingDataBatch.objects.create(
            form=dp["instance"].form,
            administration=dp["instance"].administration,
            user=dp["instance"].created_by,
            name=f"{batch_name}-{administration_id}",
            approved=approved,
        )
        batch_items = [i["instance"] for i in items]
        batch.batch_pending_data_batch.add(*batch_items)
        path = "{0}{1}".format(
            user.user_access.administration.path,
            user.user_access.administration_id,
        )
        parent_adms = Administration.objects.filter(id__in=path.split("."))
        for adm in parent_adms:
            assignment = FormApprovalAssignment.objects.filter(
                form_id=form.id, administration=adm
            ).first()
            if assignment:
                level = assignment.user.user_access.administration.level_id
                approval_status = (
                    DataApprovalStatus.approved
                    if batch.approved
                    else DataApprovalStatus.pending
                )
                PendingDataApproval.objects.create(
                    batch=batch,
                    user=assignment.user,
                    level_id=level,
                    status=approval_status,
                )
        if batch.approved:
            for pending in batch.batch_pending_data_batch.all():
                seed_approved_data(pending)


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument(
            "-r", "--repeat", nargs="?", const=20, default=20, type=int
        )
        parser.add_argument(
            "-t", "--test", nargs="?", const=False, default=False, type=bool
        )
        parser.add_argument(
            "-a",
            "--approved",
            nargs="?",
            const=False,
            default=True,
            type=bool,
        )

    def handle(self, *args, **options):
        test = options.get("test")
        repeat = options.get("repeat")
        approved = options.get("approved")

        forms = Forms.objects.all()
        for form in forms:
            if not test:
                print(f"Seeding - {form.name}")
            for dp in form.form_form_data.all():
                seed_data(
                    form=form,
                    datapoint=dp,
                    user=dp.created_by,
                    repeat=repeat,
                    approved=approved,
                )

1	import pandas as pd	1✔
2	from django.core.management import BaseCommand	1✔
3	from faker import Faker	1✔
4	from api.v1.v1_data.models import (	1✔
5	PendingFormData,
6	PendingDataBatch,
7	PendingDataApproval,
8	)
9	from api.v1.v1_forms.models import Forms, FormApprovalAssignment	1✔
10	from api.v1.v1_forms.constants import SubmissionTypes	1✔
11	from api.v1.v1_data.management.commands.fake_data_seeder import (	1✔
12	add_fake_answers,
13	)
14	from api.v1.v1_profile.models import Administration	1✔
15	from api.v1.v1_data.tasks import seed_approved_data	1✔
16	from api.v1.v1_data.constants import DataApprovalStatus	1✔
17
18	fake = Faker()	1✔
19
20
21	def seed_data(form, datapoint, user, repeat, approved):	1✔
22	pendings = []	1✔
23	for i in range(repeat):	1✔
24	pending_data = PendingFormData.objects.create(	1✔
25	name=datapoint.name,
26	geo=datapoint.geo,
27	uuid=datapoint.uuid,
28	form=datapoint.form,
29	administration=datapoint.administration,
30	created_by=user,
31	submission_type=SubmissionTypes.monitoring,
32	)
33	add_fake_answers(pending_data)	1✔
34	pendings.append(pending_data)	1✔
35	pending_items = [	1✔
36	{"administration_id": pending.administration.id, "instance": pending}
37	for pending in pendings
38	]
39	df = pd.DataFrame(pending_items)	1✔
40	grouped_data = df.groupby(["administration_id"]).apply(	1✔
41	lambda x: x.to_dict("records")
42	)
43	for administration_id, items in grouped_data.items():	1✔
44	[dp] = items[:1]	1✔
45	batch_name = fake.sentence(nb_words=3)	1✔
46	batch = PendingDataBatch.objects.create(	1✔
47	form=dp["instance"].form,
48	administration=dp["instance"].administration,
49	user=dp["instance"].created_by,
50	name=f"{batch_name}-{administration_id}",
51	approved=approved,
52	)
53	batch_items = [i["instance"] for i in items]	1✔
54	batch.batch_pending_data_batch.add(*batch_items)	1✔
55	path = "{0}{1}".format(	1✔
56	user.user_access.administration.path,
57	user.user_access.administration_id,
58	)
59	parent_adms = Administration.objects.filter(id__in=path.split("."))	1✔
60	for adm in parent_adms:	1✔
61	assignment = FormApprovalAssignment.objects.filter(	1✔
62	form_id=form.id, administration=adm
63	).first()
64	if assignment:	1✔
65	level = assignment.user.user_access.administration.level_id	1✔
66	approval_status = (	1✔
67	DataApprovalStatus.approved
68	if batch.approved
69	else DataApprovalStatus.pending
70	)
71	PendingDataApproval.objects.create(	1✔
72	batch=batch,
73	user=assignment.user,
74	level_id=level,
75	status=approval_status,
76	)
77	if batch.approved:	1!
78	for pending in batch.batch_pending_data_batch.all():	1✔
79	seed_approved_data(pending)	1✔
80
81
82	class Command(BaseCommand):	1✔
83	def add_arguments(self, parser):	1✔
84	parser.add_argument(	1✔
85	"-r", "--repeat", nargs="?", const=20, default=20, type=int
86	)
87	parser.add_argument(	1✔
88	"-t", "--test", nargs="?", const=False, default=False, type=bool
89	)
90	parser.add_argument(	1✔
91	"-a",
92	"--approved",
93	nargs="?",
94	const=False,
95	default=True,
96	type=bool,
97	)
98
99	def handle(self, args, *options):	1✔
100	test = options.get("test")	1✔
101	repeat = options.get("repeat")	1✔
102	approved = options.get("approved")	1✔
103
104	forms = Forms.objects.all()	1✔
105	for form in forms:	1✔
106	if not test:	1!
107	print(f"Seeding - {form.name}")	×
108	for dp in form.form_form_data.all():	1✔
109	seed_data(	1✔
110	form=form,
111	datapoint=dp,
112	user=dp.created_by,
113	repeat=repeat,
114	approved=approved,
115	)

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