f3cd7aa4-3df2-4a06-826f-ac23522e2baf

Committed 15 Dec 2023 03:18PM UTC coverage: 84.536% (-1.3%) from 85.804%

Build # f3cd7aa4-3df2-4a06-826f-ac23522e2baf

Build Type

push

semaphore

Committed by

web-flow

Commit Message

Merge pull request #789 from akvo/788-add-village-to-level-seeder

[#788] Add Village to Level seeder

Run Details

1902 of 2338 branches covered (0.0%)

Branch coverage included in aggregate %.

4532 of 5273 relevant lines covered (85.95%)

0.86 hits per line

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

87.25

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

import re
import random
from datetime import timedelta

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

from api.v1.v1_data.models import PendingFormData, \
    PendingAnswers, PendingDataApproval, PendingDataBatch
from api.v1.v1_forms.constants import QuestionTypes, FormTypes
from api.v1.v1_forms.models import FormApprovalRule, FormApprovalAssignment
from api.v1.v1_forms.models import Forms, UserForms
from api.v1.v1_profile.constants import UserRoleTypes
from api.v1.v1_profile.management.commands.administration_seeder import (
        MAX_LEVEL_IN_SOURCE_FILE)
from api.v1.v1_profile.models import Administration, Access, Levels
from api.v1.v1_users.models import SystemUser, Organisation

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.question_question_options.order_by('?').first().name
        ]
    elif question.type == QuestionTypes.multiple_option:
        option = list(
            question.question_question_options.order_by('?').values_list(
                'name', 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
        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)
        add_fake_answers(data)


def create_or_get_submitter(role):
    organisation = Organisation.objects.first()
    level = Levels.objects.filter(
            level__lte=MAX_LEVEL_IN_SOURCE_FILE).order_by('-level').first()
    last_name = "User"
    email = "user"
    if role == UserRoleTypes.admin:
        level = Levels.objects.order_by('level')[1]
        last_name = "Admin"
        email = "test"
    administration = Administration.objects.filter(
        level=level).order_by('?').first()
    email = ("{}{}@{}.com").format(
        re.sub('[^A-Za-z0-9]+', '', administration.name.lower()),
        administration.id, email)
    submitter, created = SystemUser.objects.get_or_create(
        organisation=organisation,
        email=email,
        first_name=administration.name,
        last_name=last_name)
    if created:
        submitter.set_password("test")
        submitter.save()
        Access.objects.create(user=submitter,
                              role=role,
                              administration=administration)
    return submitter


def assign_batch_for_approval(batch, user, test):
    organisation = Organisation.objects.first()
    administration = user.user_access.administration
    complete_path = '{0}{1}'.format(administration.path, administration.id)
    complete_path = complete_path.split('.')[1:]
    approval_rule = FormApprovalRule.objects.filter(
        administration_id=complete_path[0], form=batch.form).first()
    levels = None
    if approval_rule:
        levels = approval_rule.levels.all()
    if not approval_rule:
        randoms = Levels.objects.filter(level__gt=1).count()
        randoms = [n + 1 for n in range(randoms)]
        limit = random.choices(randoms)
        levels = Levels.objects.filter(level__gt=1).order_by('?')[:limit[0]]
        levels |= Levels.objects.filter(level=1)
        rule = FormApprovalRule.objects.create(
            form=batch.form,
            administration=Administration.objects.filter(
                id=complete_path[0]).first())
        rule.levels.set(levels)
    administrations = Administration.objects.filter(id__in=complete_path,
                                                    level__in=levels).all()
    for administration in administrations:
        # check if approval assignment for the path is not available
        assignment = FormApprovalAssignment.objects.filter(
            form=batch.form, administration=administration).first()
        if not assignment:
            email = ("{}{}@test.com").format(
                re.sub('[^A-Za-z0-9]+', '', administration.name.lower()),
                administration.id)
            last_name = "Approver"
            role = UserRoleTypes.approver
            if administration.level.level == 1:
                role = UserRoleTypes.admin
                last_name = "Admin"
            # check if someone has access to ancestor adminisration
            approver, created = SystemUser.objects.get_or_create(
                organisation=organisation,
                email=email,
                first_name=administration.name,
                last_name=last_name)
            if created:
                approver.set_password("test")
                approver.save()
                Access.objects.create(user=approver,
                                      role=role,
                                      administration=administration)
            assignment = FormApprovalAssignment.objects.create(
                form=batch.form, administration=administration, user=approver)
            UserForms.objects.get_or_create(form=batch.form, user=approver)
            if not test:
                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("./source/kenya_random_points.csv")
        forms = Forms.objects.all()
        user = None
        if options.get('email'):
            # if user type is 'user' -> seed county form only
            try:
                user = SystemUser.objects.get(email=options.get('email'))
            except SystemUser.DoesNotExist:
                print('User with this {0} is not exists'.format(
                    options.get('email')))
                return
            if user.user_access.role == UserRoleTypes.user:
                forms = Forms.objects.filter(type=FormTypes.county)
            elif user.user_access.role == UserRoleTypes.admin:
                forms = Forms.objects.filter(type=FormTypes.national)
            else:
                print(
                    'User with this {0} is not allowed to submit data '.format(
                        options.get('email')))
                return
        for form in forms:
            submitter = None
            if user:
                submitter = user
            else:
                role = UserRoleTypes.admin
                if form.type == FormTypes.county:
                    role = UserRoleTypes.user
                submitter = create_or_get_submitter(role)
            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, submitter, test)

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

akvo / rtmis / f3cd7aa4-3df2-4a06-826f-ac23522e2baf

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