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

84.7

backend/api/v1/v1_visualization/formula.py

"""Pure-function formula evaluator for the dashboard map filter.

Each formula has a list of ``buckets``; for each bucket we test that
**all** conditions in ``all_of`` pass against the supplied
``answers_by_qname`` mapping. The first bucket whose conditions all pass
wins; otherwise the formula's ``default`` bucket is returned.

``answers_by_qname`` maps ``question_name`` to whatever Answers row should
be considered the "current" value — for repeatable groups the caller
selects the latest repeat (highest ``index``) per question and passes
*that* row in.

No Django imports here so the unit-test surface stays trivial.
"""

NUMERIC_OPS = {
    "<": lambda a, b: a < b,
    "<=": lambda a, b: a <= b,
    ">": lambda a, b: a > b,
    ">=": lambda a, b: a >= b,
    "==": lambda a, b: a == b,
    "!=": lambda a, b: a != b,
}


def _answer_numeric(answer):
    """Return the numeric value of an Answer-shaped object or None."""
    value = getattr(answer, "value", None)
    if value is None and isinstance(answer, dict):
        value = answer.get("value")
    return value


def _answer_options(answer):
    """Return the options list of an Answer-shaped object or []."""
    options = getattr(answer, "options", None)
    if options is None and isinstance(answer, dict):
        options = answer.get("options")
    return options or []


def _match(condition, answers_by_qname):
    """Return True iff the condition passes against the answers map."""
    qname = condition.get("question_name")
    op = condition.get("op")
    answer = answers_by_qname.get(qname)
    if answer is None:
        return False

    if op in NUMERIC_OPS:
        value = _answer_numeric(answer)
        if value is None:
            return False
        return NUMERIC_OPS[op](value, condition["value"])

    if op == "between":
        value = _answer_numeric(answer)
        if value is None:
            return False
        return condition["min"] <= value <= condition["max"]

    if op == "option_equals":
        return condition["value"] in _answer_options(answer)

    if op == "option_in":
        opts = _answer_options(answer)
        return any(v in opts for v in condition.get("values", []))

    return False


def evaluate(formula, answers_by_qname):
    """Return the bucket value matching the answers, or the default.

    ``formula`` shape::

        {
          "buckets": [
            {"value": "...", "label": "...", "all_of": [<cond>, ...]},
            ...
          ],
          "default": {"value": "...", "label": "..."}
        }
    """
    for bucket in formula.get("buckets", []):
        conditions = bucket.get("all_of") or []
        if all(_match(c, answers_by_qname) for c in conditions):
            return bucket["value"]
    return formula.get("default", {}).get("value")


def pick_latest_repeat(answers):
    """Pick the highest-index Answers row per ``question_name``.

    Accepts either a list of Answers-shaped objects (with a
    ``question_name`` attribute) or a list of dicts with at least
    ``question_name`` and ``index`` keys. The rows must therefore carry
    ``question_name`` — source them from ``mv_answer_denormalized``, not
    the base ``Answers`` table.
    Returns ``dict[question_name -> answer_row]``.
    """
    out = {}
    best_idx = {}
    for ans in answers:
        if hasattr(ans, "question_name"):
            qname = ans.question_name
        elif isinstance(ans, dict):
            qname = ans.get("question_name")
        else:
            qname = None
        if qname is None:
            continue
        idx = getattr(ans, "index", None)
        if idx is None and isinstance(ans, dict):
            idx = ans.get("index", 0)
        idx = idx or 0
        if qname not in best_idx or idx > best_idx[qname]:
            best_idx[qname] = idx
            out[qname] = ans
    return out


def validate_shape(formula):
    """Lightweight structural validation; raises ValueError on bad input.

    Used by the request serializer to reject malformed formulas before
    the view starts iterating.
    """
    if not isinstance(formula, dict):
        raise ValueError("formula must be an object")
    buckets = formula.get("buckets")
    if not isinstance(buckets, list) or not buckets:
        raise ValueError("formula.buckets must be a non-empty array")
    for i, bucket in enumerate(buckets):
        if not isinstance(bucket, dict):
            raise ValueError(f"buckets[{i}] must be an object")
        if "value" not in bucket or "label" not in bucket:
            raise ValueError(
                f"buckets[{i}] must include 'value' and 'label'"
            )
        conditions = bucket.get("all_of")
        if not isinstance(conditions, list) or not conditions:
            raise ValueError(
                f"buckets[{i}].all_of must be a non-empty array"
            )
        for j, cond in enumerate(conditions):
            if not isinstance(cond, dict):
                raise ValueError(
                    f"buckets[{i}].all_of[{j}] must be an object"
                )
            if "question_name" not in cond or "op" not in cond:
                raise ValueError(
                    f"buckets[{i}].all_of[{j}] requires "
                    "'question_name' and 'op'"
                )
            op = cond["op"]
            if op == "between":
                if "min" not in cond or "max" not in cond:
                    raise ValueError(
                        f"buckets[{i}].all_of[{j}] op 'between' "
                        "requires 'min' and 'max'"
                    )
                if not isinstance(cond["min"], (int, float)) or \
                        not isinstance(cond["max"], (int, float)):
                    raise ValueError(
                        f"buckets[{i}].all_of[{j}] op 'between' "
                        "'min' and 'max' must be numbers"
                    )
            elif op == "option_in":
                if not isinstance(cond.get("values"), list):
                    raise ValueError(
                        f"buckets[{i}].all_of[{j}] op 'option_in' "
                        "requires 'values' array"
                    )
            elif op in NUMERIC_OPS:
                if "value" not in cond:
                    raise ValueError(
                        f"buckets[{i}].all_of[{j}] op '{op}' "
                        "requires 'value'"
                    )
                if not isinstance(cond["value"], (int, float)):
                    raise ValueError(
                        f"buckets[{i}].all_of[{j}] op '{op}' "
                        "'value' must be a number"
                    )
            elif op == "option_equals":
                if "value" not in cond:
                    raise ValueError(
                        f"buckets[{i}].all_of[{j}] op '{op}' "
                        "requires 'value'"
                    )
            else:
                raise ValueError(
                    f"buckets[{i}].all_of[{j}] unsupported op '{op}'"
                )
    default = formula.get("default")
    if not isinstance(default, dict):
        raise ValueError("formula.default must be an object")
    if "value" not in default or "label" not in default:
        raise ValueError(
            "formula.default must include 'value' and 'label'"
        )
    return formula

1	"""Pure-function formula evaluator for the dashboard map filter.
2
3	Each formula has a list of ``buckets``; for each bucket we test that
4	all conditions in ``all_of`` pass against the supplied
5	``answers_by_qname`` mapping. The first bucket whose conditions all pass
6	wins; otherwise the formula's ``default`` bucket is returned.
7
8	``answers_by_qname`` maps ``question_name`` to whatever Answers row should
9	be considered the "current" value — for repeatable groups the caller
10	selects the latest repeat (highest ``index``) per question and passes
11	that row in.
12
13	No Django imports here so the unit-test surface stays trivial.
14	"""
15
16	NUMERIC_OPS = {	1✔
17	"<": lambda a, b: a < b,
18	"<=": lambda a, b: a <= b,
19	">": lambda a, b: a > b,
20	">=": lambda a, b: a >= b,
21	"==": lambda a, b: a == b,
22	"!=": lambda a, b: a != b,
23	}
24
25
26	def _answer_numeric(answer):	1✔
27	"""Return the numeric value of an Answer-shaped object or None."""
28	value = getattr(answer, "value", None)	1✔
29	if value is None and isinstance(answer, dict):	1✔
30	value = answer.get("value")	1✔
31	return value	1✔
32
33
34	def _answer_options(answer):	1✔
35	"""Return the options list of an Answer-shaped object or []."""
36	options = getattr(answer, "options", None)	1✔
37	if options is None and isinstance(answer, dict):	1✔
38	options = answer.get("options")	1✔
39	return options or []	1✔
40
41
42	def _match(condition, answers_by_qname):	1✔
43	"""Return True iff the condition passes against the answers map."""
44	qname = condition.get("question_name")	1✔
45	op = condition.get("op")	1✔
46	answer = answers_by_qname.get(qname)	1✔
47	if answer is None:	1✔
48	return False	1✔
49
50	if op in NUMERIC_OPS:	1✔
51	value = _answer_numeric(answer)	1✔
52	if value is None:	1!
53	return False	×
54	return NUMERIC_OPS[op](value, condition["value"])	1✔
55
56	if op == "between":	1✔
57	value = _answer_numeric(answer)	1✔
58	if value is None:	1!
59	return False	×
60	return condition["min"] <= value <= condition["max"]	1✔
61
62	if op == "option_equals":	1✔
63	return condition["value"] in _answer_options(answer)	1✔
64
65	if op == "option_in":	1✔
66	opts = _answer_options(answer)	1✔
67	return any(v in opts for v in condition.get("values", []))	1!
68
69	return False	1✔
70
71
72	def evaluate(formula, answers_by_qname):	1✔
73	"""Return the bucket value matching the answers, or the default.
74
75	``formula`` shape::
76
77	{
78	"buckets": [
79	{"value": "...", "label": "...", "all_of": [<cond>, ...]},
80	...
81	],
82	"default": {"value": "...", "label": "..."}
83	}
84	"""
85	for bucket in formula.get("buckets", []):	1✔
86	conditions = bucket.get("all_of") or []	1✔
87	if all(_match(c, answers_by_qname) for c in conditions):	1✔
88	return bucket["value"]	1✔
89	return formula.get("default", {}).get("value")	1✔
90
91
92	def pick_latest_repeat(answers):	1✔
93	"""Pick the highest-index Answers row per ``question_name``.
94
95	Accepts either a list of Answers-shaped objects (with a
96	``question_name`` attribute) or a list of dicts with at least
97	``question_name`` and ``index`` keys. The rows must therefore carry
98	``question_name`` — source them from ``mv_answer_denormalized``, not
99	the base ``Answers`` table.
100	Returns ``dict[question_name -> answer_row]``.
101	"""
102	out = {}	1✔
103	best_idx = {}	1✔
104	for ans in answers:	1✔
105	if hasattr(ans, "question_name"):	1✔
106	qname = ans.question_name	1✔
107	elif isinstance(ans, dict):	1!
108	qname = ans.get("question_name")	1✔
109	else:
110	qname = None	×
111	if qname is None:	1!
112	continue	×
113	idx = getattr(ans, "index", None)	1✔
114	if idx is None and isinstance(ans, dict):	1✔
115	idx = ans.get("index", 0)	1✔
116	idx = idx or 0	1✔
117	if qname not in best_idx or idx > best_idx[qname]:	1✔
118	best_idx[qname] = idx	1✔
119	out[qname] = ans	1✔
120	return out	1✔
121
122
123	def validate_shape(formula):	1✔
124	"""Lightweight structural validation; raises ValueError on bad input.
125
126	Used by the request serializer to reject malformed formulas before
127	the view starts iterating.
128	"""
129	if not isinstance(formula, dict):	1!
130	raise ValueError("formula must be an object")	×
131	buckets = formula.get("buckets")	1✔
132	if not isinstance(buckets, list) or not buckets:	1✔
133	raise ValueError("formula.buckets must be a non-empty array")	1✔
134	for i, bucket in enumerate(buckets):	1✔
135	if not isinstance(bucket, dict):	1!
136	raise ValueError(f"buckets[{i}] must be an object")	×
137	if "value" not in bucket or "label" not in bucket:	1✔
138	raise ValueError(	1✔
139	f"buckets[{i}] must include 'value' and 'label'"
140	)
141	conditions = bucket.get("all_of")	1✔
142	if not isinstance(conditions, list) or not conditions:	1!
143	raise ValueError(	×
144	f"buckets[{i}].all_of must be a non-empty array"
145	)
146	for j, cond in enumerate(conditions):	1✔
147	if not isinstance(cond, dict):	1!
148	raise ValueError(	×
149	f"buckets[{i}].all_of[{j}] must be an object"
150	)
151	if "question_name" not in cond or "op" not in cond:	1✔
152	raise ValueError(	1✔
153	f"buckets[{i}].all_of[{j}] requires "
154	"'question_name' and 'op'"
155	)
156	op = cond["op"]	1✔
157	if op == "between":	1✔
158	if "min" not in cond or "max" not in cond:	1✔
159	raise ValueError(	1✔
160	f"buckets[{i}].all_of[{j}] op 'between' "
161	"requires 'min' and 'max'"
162	)
163	if not isinstance(cond["min"], (int, float)) or \	1✔
164	not isinstance(cond["max"], (int, float)):
165	raise ValueError(	1✔
166	f"buckets[{i}].all_of[{j}] op 'between' "
167	"'min' and 'max' must be numbers"
168	)
169	elif op == "option_in":	1!
170	if not isinstance(cond.get("values"), list):	×
171	raise ValueError(	×
172	f"buckets[{i}].all_of[{j}] op 'option_in' "
173	"requires 'values' array"
174	)
175	elif op in NUMERIC_OPS:	1✔
176	if "value" not in cond:	1!
177	raise ValueError(	×
178	f"buckets[{i}].all_of[{j}] op '{op}' "
179	"requires 'value'"
180	)
181	if not isinstance(cond["value"], (int, float)):	1✔
182	raise ValueError(	1✔
183	f"buckets[{i}].all_of[{j}] op '{op}' "
184	"'value' must be a number"
185	)
186	elif op == "option_equals":	1✔
187	if "value" not in cond:	1!
188	raise ValueError(	×
189	f"buckets[{i}].all_of[{j}] op '{op}' "
190	"requires 'value'"
191	)
192	else:
193	raise ValueError(	1✔
194	f"buckets[{i}].all_of[{j}] unsupported op '{op}'"
195	)
196	default = formula.get("default")	1✔
197	if not isinstance(default, dict):	1✔
198	raise ValueError("formula.default must be an object")	1✔
199	if "value" not in default or "label" not in default:	1!
200	raise ValueError(	×
201	"formula.default must include 'value' and 'label'"
202	)
203	return formula	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