akvo / iwsims / #59

Committed 18 Jun 2026 07:20AM UTC coverage: 88.033% (-0.1%) from 88.13%

Build # #59

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 5dfcb298b into a6f6761c9

Coverage Stats

5183 of 6053 branches covered (85.63%)

Branch coverage included in aggregate %.

9979 of 11170 relevant lines covered (89.34%)

0.89 hits per line

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

81.67

backend/utils/functions.py

import json
import os

from django.utils import timezone
from api.v1.v1_data.models import Answers, AnswerHistory
from api.v1.v1_forms.constants import QuestionTypes
from api.v1.v1_profile.models import Administration


def atomic_write(path: str, content: str) -> None:
    """Write content to path atomically via tmp+rename.

    Prevents concurrent readers from seeing a zero-byte file during the
    truncate/write window of `open(path, "w")`.
    """
    tmp = f"{path}.tmp.{os.getpid()}"
    with open(tmp, "w") as f:
        f.write(content)
    os.replace(tmp, path)


def atomic_write_json(path: str, obj) -> None:
    atomic_write(path, json.dumps(obj, indent=2))


def update_date_time_format(date):
    if date:
        # date = timezone.datetime.strptime(date, "%Y-%m-%d").date()
        if not timezone.is_naive(date):
            return timezone.localtime(date).strftime("%Y-%m-%d %I:%M %p")
        return date.strftime("%Y-%m-%d %I:%M %p")
    return None


def _typed_value(question_type, options, value, name):
    """Shared typed extraction for get_answer_value / get_answer_history.

    Handles the branches both functions agree on: geo / option /
    multiple_option return the options list, number returns the numeric
    value, everything else (text, date, cascade, …) returns the name.
    The administration type is handled by the callers, which differ on
    webform expansion and None handling.
    """
    if question_type in (
        QuestionTypes.geo,
        QuestionTypes.option,
        QuestionTypes.multiple_option,
    ):
        return options
    if question_type == QuestionTypes.number:
        return value
    return name


def get_answer_value(answer: Answers, webform: bool = False):
    if answer.question.type == QuestionTypes.administration:
        if webform:
            adm = Administration.objects.filter(id=answer.value).first()
            if adm:
                return [
                    a.id
                    for a in adm.ancestors.exclude(parent__isnull=True).all()
                ] + [adm.id]
            return answer.value
        return int(float(answer.value)) if answer.value else None
    return _typed_value(
        answer.question.type, answer.options, answer.value, answer.name
    )


def get_answer_history(answer_history: AnswerHistory):
    created = update_date_time_format(answer_history.created)
    created_by = answer_history.created_by.get_full_name()
    if answer_history.question.type == QuestionTypes.administration:
        value = int(float(answer_history.value))
    else:
        value = _typed_value(
            answer_history.question.type,
            answer_history.options,
            answer_history.value,
            answer_history.name,
        )
    return {"value": value, "created": created, "created_by": created_by}

1	import json	1✔
2	import os	1✔
3
4	from django.utils import timezone	1✔
5	from api.v1.v1_data.models import Answers, AnswerHistory	1✔
6	from api.v1.v1_forms.constants import QuestionTypes	1✔
7	from api.v1.v1_profile.models import Administration	1✔
8
9
10	def atomic_write(path: str, content: str) -> None:	1✔
11	"""Write content to path atomically via tmp+rename.
12
13	Prevents concurrent readers from seeing a zero-byte file during the
14	truncate/write window of `open(path, "w")`.
15	"""
16	tmp = f"{path}.tmp.{os.getpid()}"	1✔
17	with open(tmp, "w") as f:	1✔
18	f.write(content)	1✔
19	os.replace(tmp, path)	1✔
20
21
22	def atomic_write_json(path: str, obj) -> None:	1✔
23	atomic_write(path, json.dumps(obj, indent=2))	1✔
24
25
26	def update_date_time_format(date):	1✔
27	if date:	1✔
28	# date = timezone.datetime.strptime(date, "%Y-%m-%d").date()
29	if not timezone.is_naive(date):	1✔
30	return timezone.localtime(date).strftime("%Y-%m-%d %I:%M %p")	1✔
31	return date.strftime("%Y-%m-%d %I:%M %p")	1✔
32	return None	1✔
33
34
35	def _typed_value(question_type, options, value, name):	1✔
36	"""Shared typed extraction for get_answer_value / get_answer_history.
37
38	Handles the branches both functions agree on: geo / option /
39	multiple_option return the options list, number returns the numeric
40	value, everything else (text, date, cascade, …) returns the name.
41	The administration type is handled by the callers, which differ on
42	webform expansion and None handling.
43	"""
44	if question_type in (	1✔
45	QuestionTypes.geo,
46	QuestionTypes.option,
47	QuestionTypes.multiple_option,
48	):
49	return options	1✔
50	if question_type == QuestionTypes.number:	1✔
51	return value	1✔
52	return name	1✔
53
54
55	def get_answer_value(answer: Answers, webform: bool = False):	1✔
56	if answer.question.type == QuestionTypes.administration:	1✔
57	if webform:	1!
58	adm = Administration.objects.filter(id=answer.value).first()	×
59	if adm:	×
60	return [	×
61	a.id
62	for a in adm.ancestors.exclude(parent__isnull=True).all()
63	] + [adm.id]
64	return answer.value	×
65	return int(float(answer.value)) if answer.value else None	1✔
66	return _typed_value(	1✔
67	answer.question.type, answer.options, answer.value, answer.name
68	)
69
70
71	def get_answer_history(answer_history: AnswerHistory):	1✔
72	created = update_date_time_format(answer_history.created)	1✔
73	created_by = answer_history.created_by.get_full_name()	1✔
74	if answer_history.question.type == QuestionTypes.administration:	1!
75	value = int(float(answer_history.value))	×
76	else:
77	value = _typed_value(	1✔
78	answer_history.question.type,
79	answer_history.options,
80	answer_history.value,
81	answer_history.name,
82	)
83	return {"value": value, "created": created, "created_by": created_by}	1✔

akvo / iwsims / #59

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