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

88.51

backend/api/v1/v1_visualization/dashboard_serializers.py

from rest_framework import serializers
from api.v1.v1_visualization.constants import (
    VALID_GROUP_BY,
    VALID_MONITORING,
    VALID_VALUE_TYPE,
    VALID_REPEAT_AGG,
    VALID_STACK_BY,
    VALID_CRITERIA_TYPES,
    VALID_VALUES_CRITERIA_TYPES,
    VALID_COLUMN_SOURCES,
    VALID_PROGRESS_FORMULAS,
    SUPPORTED_QUESTION_TYPES,
)
from api.v1.v1_visualization.functions import (
    parse_criteria_string,
    validate_qname,
)
from api.v1.v1_forms.models import Forms, Questions

__all__ = ["validate_qname"]


class ValuesFilterSerializer(serializers.Serializer):
    """Validates query parameters for /visualization/values endpoint."""

    form_id = serializers.IntegerField(required=False)
    question_id = serializers.IntegerField(required=False)
    question_name = serializers.CharField(required=False, max_length=255)
    parent_form_id = serializers.IntegerField(required=False)
    monitoring = serializers.ChoiceField(
        choices=list(VALID_MONITORING),
        default="latest",
    )
    group_by = serializers.ChoiceField(
        choices=list(VALID_GROUP_BY),
        required=False,
        allow_null=True,
    )
    stack_by = serializers.ChoiceField(
        choices=list(VALID_STACK_BY),
        required=False,
        allow_null=True,
    )
    sum_by = serializers.ChoiceField(
        choices=["id", "parent_id"],
        required=False,
        allow_null=True,
    )
    value_type = serializers.ChoiceField(
        choices=list(VALID_VALUE_TYPE),
        default="number",
    )
    repeat_agg = serializers.ChoiceField(
        choices=list(VALID_REPEAT_AGG),
        default="average",
    )
    rolling_months = serializers.IntegerField(
        required=False, min_value=1,
    )
    from_date = serializers.DateField(required=False)
    to_date = serializers.DateField(required=False)
    date_question_name = serializers.CharField(required=False)
    administration_id = serializers.IntegerField(required=False)
    option_value = serializers.CharField(required=False)
    criteria = serializers.CharField(required=False)
    include_unanswered = serializers.BooleanField(
        required=False,
        default=False,
    )
    include_empty = serializers.BooleanField(
        required=False,
        default=False,
    )

    def validate_criteria(self, value):
        try:
            return parse_criteria_string(
                value, VALID_VALUES_CRITERIA_TYPES,
            )
        except ValueError as e:
            raise serializers.ValidationError(str(e))

    def validate_form_id(self, value):
        if not Forms.objects.filter(pk=value).exists():
            raise serializers.ValidationError(
                f"Form {value} not found."
            )
        return value

    def validate(self, data):
        question_name = data.get("question_name")
        form_id = data.get("form_id")

        if not question_name and not form_id:
            raise serializers.ValidationError(
                "Either question_name or form_id is required."
            )

        # Global cross-form path: question_name without form_id.
        if question_name and not form_id:
            validate_qname(question_name)  # numeric id -> 400
            return data

        question_id = data.get("question_id")
        stack_by = data.get("stack_by")
        group_by = data.get("group_by")

        # Resolve the target question — by name (form-scoped name path)
        # or by id (legacy rich path). Both must belong to the form and
        # be a supported type.
        question = None
        if question_name:
            validate_qname(question_name)  # numeric id -> 400
            question = Questions.objects.filter(
                name=question_name,
                form_id=form_id,
            ).first()
            if not question:
                raise serializers.ValidationError({
                    "question_name": (
                        f"Question '{question_name}' not found"
                        f" on form {form_id}."
                    ),
                })
        elif question_id:
            question = Questions.objects.filter(
                pk=question_id,
                form_id=form_id,
            ).first()
            if not question:
                raise serializers.ValidationError({
                    "question_id": (
                        f"Question {question_id} not found"
                        f" on form {form_id}."
                    ),
                })
        if question:
            if question.type not in SUPPORTED_QUESTION_TYPES:
                raise serializers.ValidationError({
                    "question": (
                        f"Question type {question.type}"
                        " is not supported."
                    ),
                })
            data["question"] = question

        # Split criteria into same-form and parent-form buckets.
        # names on form_id → criteria; names on parent form → parent_criteria.
        criteria = data.get("criteria") or []
        if criteria:
            qnames = {c["parts"][0] for c in criteria}
            on_form = set(
                Questions.objects.filter(
                    name__in=qnames, form_id=form_id,
                ).values_list("name", flat=True)
            )
            remaining = qnames - on_form
            parent_form = Forms.objects.filter(
                pk=form_id,
            ).values_list("parent_id", flat=True).first()
            on_parent = set()
            if remaining and parent_form:
                on_parent = set(
                    Questions.objects.filter(
                        name__in=remaining,
                        form_id=parent_form,
                    ).values_list("name", flat=True)
                )
            unknown = remaining - on_parent
            if unknown:
                raise serializers.ValidationError({
                    "criteria": (
                        "question_name(s) not on form "
                        f"{form_id} or its parent: "
                        f"{sorted(unknown)}"
                    ),
                })
            data["criteria"] = [
                c for c in criteria
                if c["parts"][0] in on_form
            ] or None
            data["parent_criteria"] = [
                c for c in criteria
                if c["parts"][0] in on_parent
            ] or None

        # stack_by requires group_by and a resolved question (id or name)
        if stack_by:
            if not group_by:
                raise serializers.ValidationError({
                    "stack_by": "stack_by requires group_by.",
                })
            if not data.get("question"):
                raise serializers.ValidationError({
                    "stack_by": "stack_by requires a question.",
                })

        return data


class EscalationFilterSerializer(serializers.Serializer):
    """Validates query parameters for /visualization/escalation.

    Criteria format: comma-separated, colon-delimited.
      option_equals:{qid}:{value}
      threshold_gt:{qid}:{value}
      threshold_lt:{qid}:{value}
      overdue:{completion_qid}:{deadline_qid}

    Columns format: comma-separated, colon-delimited.
      {key}:parent_name
      {key}:administration
      {key}:answer:{qid}
      {key}:latest_date:{date_qid}
    """

    monitoring_form_id = serializers.IntegerField(required=True)
    criteria = serializers.CharField(required=True)
    columns = serializers.CharField(required=True)
    page = serializers.IntegerField(default=1, min_value=1)
    page_size = serializers.IntegerField(
        default=20, min_value=1, max_value=100,
    )
    from_date = serializers.DateField(required=False)
    to_date = serializers.DateField(required=False)
    date_question_name = serializers.CharField(required=False)
    administration_id = serializers.IntegerField(required=False)
    filter_criteria = serializers.CharField(required=False)

    def validate_filter_criteria(self, value):
        try:
            return parse_criteria_string(
                value, VALID_VALUES_CRITERIA_TYPES,
            )
        except ValueError as e:
            raise serializers.ValidationError(str(e))

    def validate_criteria(self, value):
        """Parse and validate criteria string."""
        parsed = []
        for item in value.split(","):
            parts = item.strip().split(":")
            if len(parts) < 3:
                raise serializers.ValidationError(
                    f"Invalid criteria format: '{item}'."
                    " Expected type:qid:value"
                )
            ctype = parts[0]
            if ctype not in VALID_CRITERIA_TYPES:
                raise serializers.ValidationError(
                    f"Invalid criteria type: '{ctype}'."
                    f" Options: {VALID_CRITERIA_TYPES}"
                )

            try:
                if ctype == "option_equals":
                    qname = validate_qname(parts[1])
                    normalized = [qname, parts[2]]
                elif ctype in ("threshold_gt", "threshold_lt"):
                    qname = validate_qname(parts[1])
                    threshold = float(parts[2])
                    normalized = [qname, threshold]
                elif ctype == "overdue":
                    completion_qname = validate_qname(parts[1])
                    deadline_qname = validate_qname(parts[2])
                    normalized = [completion_qname, deadline_qname]
            except ValueError:
                raise serializers.ValidationError(
                    f"Invalid numeric value in criteria: '{item}'."
                )

            parsed.append({
                "type": ctype,
                "parts": normalized,
            })
        return parsed

    def validate_columns(self, value):
        """Parse and validate columns string."""
        qid_required_sources = {
            "answer", "parent_answer", "latest_date",
        }
        parsed = []
        for item in value.split(","):
            parts = item.strip().split(":")
            if len(parts) < 2:
                raise serializers.ValidationError(
                    f"Invalid column format: '{item}'."
                    " Expected key:source[:qid]"
                )
            key = parts[0]
            source = parts[1]
            if source not in VALID_COLUMN_SOURCES:
                raise serializers.ValidationError(
                    f"Invalid column source: '{source}'."
                    f" Options: {VALID_COLUMN_SOURCES}"
                )
            col = {"key": key, "source": source}
            if source in qid_required_sources and len(parts) < 3:
                raise serializers.ValidationError(
                    f"Column source '{source}' requires a"
                    f" question_name: '{item}'"
                )
            if len(parts) > 2:
                col["question_name"] = validate_qname(parts[2])
            parsed.append(col)
        return parsed


class ProgressFilterSerializer(serializers.Serializer):
    """Validates query parameters for /visualization/progress.

    Components format: comma-separated, colon-delimited.
      {key}:{formula}:{qid1}:{qid2}:...:{total_items}

    Example:
      base:any_yes:111:222:333,tank:completed_binary:444,
      pipes:ratio:555,security:multi_select_proportion:666:3
    """

    monitoring_form_id = serializers.IntegerField(
        required=True
    )
    components = serializers.CharField(required=True)
    filter_question_name = serializers.CharField(
        required=False
    )
    filter_option_value = serializers.CharField(
        required=False
    )
    scope_question_name = serializers.CharField(
        required=False
    )
    from_date = serializers.DateField(required=False)
    to_date = serializers.DateField(required=False)
    date_question_name = serializers.CharField(
        required=False
    )
    administration_id = serializers.IntegerField(
        required=False
    )
    criteria = serializers.CharField(required=False)

    def validate_criteria(self, value):
        try:
            return parse_criteria_string(
                value, VALID_VALUES_CRITERIA_TYPES,
            )
        except ValueError as e:
            raise serializers.ValidationError(str(e))

    def validate_components(self, value):
        """Parse and validate components string.

        Format: key:formula:qid[:qid...][@type1|type2|...]
        The optional @-suffix lists applicable project types.
        """
        parsed = []
        for item in value.split(","):
            raw = item.strip()

            # Split off optional @applicable_types suffix
            applicable_types = None
            if "@" in raw:
                raw, types_str = raw.split("@", 1)
                applicable_types = [
                    t.strip() for t in types_str.split("|")
                    if t.strip()
                ]

            parts = raw.split(":")
            if len(parts) < 3:
                raise serializers.ValidationError(
                    f"Invalid component format: '{item}'."
                    " Expected key:formula:qid[:qid...]"
                )
            key = parts[0]
            formula = parts[1]
            if formula not in VALID_PROGRESS_FORMULAS:
                raise serializers.ValidationError(
                    f"Invalid formula: '{formula}'."
                    f" Options: {VALID_PROGRESS_FORMULAS}"
                )

            comp = {"key": key, "formula": formula}

            if formula == "multi_select_proportion":
                # Last segment is total_items
                if len(parts) < 4:
                    raise serializers.ValidationError(
                        f"{formula} requires total_items:"
                        f" '{item}'"
                    )
                comp["question_names"] = [
                    validate_qname(q) for q in parts[2:-1]
                ]
                try:
                    total_items = int(parts[-1])
                except ValueError:
                    raise serializers.ValidationError(
                        f"Invalid total_items in component: '{item}'."
                    )
                if total_items < 1:
                    raise serializers.ValidationError(
                        f"total_items must be >= 1: '{item}'"
                    )
                comp["total_items"] = total_items
            elif formula == "ratio":
                # ratio requires implemented_qid:planned_qid
                if len(parts) != 4:
                    raise serializers.ValidationError(
                        f"{formula} requires exactly two"
                        " question ids"
                        " (implemented:planned):"
                        f" '{item}'"
                    )
                comp["question_names"] = [
                    validate_qname(parts[2]), validate_qname(parts[3]),
                ]
            else:
                comp["question_names"] = [
                    validate_qname(q) for q in parts[2:]
                ]

            if applicable_types:
                comp["applicable_types"] = applicable_types

            parsed.append(comp)
        return parsed


# -- Response serializers (documentation only) --------------------
#
# These serializers describe the shape of JSON bodies returned by
# the dashboard endpoints. They are referenced from @extend_schema
# `responses=...` to replace Swagger's "No response body" with a
# concrete schema, and live alongside the request serializers for
# locality. None of them are used for actual DRF serialization —
# the views build plain dicts — so they only need field + type
# metadata that drf-spectacular can introspect.


class ValuesDataItemSerializer(serializers.Serializer):
    """One row in the /values `data` array.

    Shape varies with `group_by`:
    - none / stack_by: extra numeric columns are keyed dynamically,
      so downstream readers should treat unknown keys as stack cells.
    - option: `group` + `color` are populated.
    - month / date: `group` is the machine-readable bucket key.
    """

    value = serializers.FloatField(
        required=False, allow_null=True,
        help_text="Numeric aggregate (or percentage when requested).",
    )
    label = serializers.CharField(
        required=False,
        help_text="Human-readable label for this row.",
    )
    group = serializers.CharField(
        required=False,
        help_text=(
            "Machine-readable key (option value, YYYY-MM, parent id,"
            " …). Stable across translations."
        ),
    )
    color = serializers.CharField(
        required=False,
        help_text=(
            "Hex color from QuestionOptions.color"
            " (only when group_by=option)."
        ),
    )


class ValuesResponseSerializer(serializers.Serializer):
    """/visualization/values response envelope.

    For stacked responses (`stack_by=option|parent_id`), each row in
    `data` additionally carries one numeric column per stack — those
    keys are dynamic and therefore not enumerable here. `stack_labels`
    and `colors` are only present in that mode.
    """

    data = ValuesDataItemSerializer(many=True)
    labels = serializers.ListField(
        child=serializers.CharField(),
        help_text=(
            "Ordered axis / legend labels — parallel to `data[].label`."
        ),
    )
    stack_labels = serializers.ListField(
        child=serializers.CharField(), required=False,
        help_text="Legend entries. Present only when stack_by is set.",
    )
    colors = serializers.ListField(
        child=serializers.CharField(), required=False,
        help_text="Per-stack colors when stack_by=option.",
    )


class EscalationResultItemSerializer(serializers.Serializer):
    """One row from /escalation `results`.

    Column keys are driven by the request's `columns=` param, so only
    `id` is guaranteed. Other keys are documented per column in the
    API spec; values are strings, numbers, or null.
    """

    id = serializers.IntegerField()


class EscalationResponseSerializer(serializers.Serializer):
    """/visualization/escalation paginated envelope."""

    count = serializers.IntegerField()
    next = serializers.CharField(
        allow_null=True, required=False,
    )
    previous = serializers.CharField(
        allow_null=True, required=False,
    )
    results = EscalationResultItemSerializer(many=True)


class ProgressHistogramBucketSerializer(serializers.Serializer):
    progress = serializers.CharField(
        help_text="Bucket label, e.g. '0-10%', '11-20%'.",
    )
    count = serializers.IntegerField()


class ProgressDetailItemSerializer(serializers.Serializer):
    """One EPS / parent in the /progress `details` array."""

    label = serializers.CharField()
    group = serializers.CharField(
        help_text="Parent FormData.id as string.",
    )
    components = serializers.DictField(
        child=serializers.FloatField(),
        help_text=(
            "Per-component percent score, keyed by the component"
            " `key` from the request."
        ),
    )
    overall = serializers.FloatField(
        help_text="Average across components, 0-100.",
    )


class ProgressResponseSerializer(serializers.Serializer):
    """/visualization/progress response envelope."""

    histogram = ProgressHistogramBucketSerializer(many=True)
    details = ProgressDetailItemSerializer(many=True)

1	from rest_framework import serializers	1✔
2	from api.v1.v1_visualization.constants import (	1✔
3	VALID_GROUP_BY,
4	VALID_MONITORING,
5	VALID_VALUE_TYPE,
6	VALID_REPEAT_AGG,
7	VALID_STACK_BY,
8	VALID_CRITERIA_TYPES,
9	VALID_VALUES_CRITERIA_TYPES,
10	VALID_COLUMN_SOURCES,
11	VALID_PROGRESS_FORMULAS,
12	SUPPORTED_QUESTION_TYPES,
13	)
14	from api.v1.v1_visualization.functions import (	1✔
15	parse_criteria_string,
16	validate_qname,
17	)
18	from api.v1.v1_forms.models import Forms, Questions	1✔
19
20	__all__ = ["validate_qname"]	1✔
21
22
23	class ValuesFilterSerializer(serializers.Serializer):	1✔
24	"""Validates query parameters for /visualization/values endpoint."""
25
26	form_id = serializers.IntegerField(required=False)	1✔
27	question_id = serializers.IntegerField(required=False)	1✔
28	question_name = serializers.CharField(required=False, max_length=255)	1✔
29	parent_form_id = serializers.IntegerField(required=False)	1✔
30	monitoring = serializers.ChoiceField(	1✔
31	choices=list(VALID_MONITORING),
32	default="latest",
33	)
34	group_by = serializers.ChoiceField(	1✔
35	choices=list(VALID_GROUP_BY),
36	required=False,
37	allow_null=True,
38	)
39	stack_by = serializers.ChoiceField(	1✔
40	choices=list(VALID_STACK_BY),
41	required=False,
42	allow_null=True,
43	)
44	sum_by = serializers.ChoiceField(	1✔
45	choices=["id", "parent_id"],
46	required=False,
47	allow_null=True,
48	)
49	value_type = serializers.ChoiceField(	1✔
50	choices=list(VALID_VALUE_TYPE),
51	default="number",
52	)
53	repeat_agg = serializers.ChoiceField(	1✔
54	choices=list(VALID_REPEAT_AGG),
55	default="average",
56	)
57	rolling_months = serializers.IntegerField(	1✔
58	required=False, min_value=1,
59	)
60	from_date = serializers.DateField(required=False)	1✔
61	to_date = serializers.DateField(required=False)	1✔
62	date_question_name = serializers.CharField(required=False)	1✔
63	administration_id = serializers.IntegerField(required=False)	1✔
64	option_value = serializers.CharField(required=False)	1✔
65	criteria = serializers.CharField(required=False)	1✔
66	include_unanswered = serializers.BooleanField(	1✔
67	required=False,
68	default=False,
69	)
70	include_empty = serializers.BooleanField(	1✔
71	required=False,
72	default=False,
73	)
74
75	def validate_criteria(self, value):	1✔
76	try:	1✔
77	return parse_criteria_string(	1✔
78	value, VALID_VALUES_CRITERIA_TYPES,
79	)
80	except ValueError as e:	1✔
81	raise serializers.ValidationError(str(e))	1✔
82
83	def validate_form_id(self, value):	1✔
84	if not Forms.objects.filter(pk=value).exists():	1✔
85	raise serializers.ValidationError(	1✔
86	f"Form {value} not found."
87	)
88	return value	1✔
89
90	def validate(self, data):	1✔
91	question_name = data.get("question_name")	1✔
92	form_id = data.get("form_id")	1✔
93
94	if not question_name and not form_id:	1✔
95	raise serializers.ValidationError(	1✔
96	"Either question_name or form_id is required."
97	)
98
99	# Global cross-form path: question_name without form_id.
100	if question_name and not form_id:	1✔
101	validate_qname(question_name) # numeric id -> 400	1✔
102	return data	1✔
103
104	question_id = data.get("question_id")	1✔
105	stack_by = data.get("stack_by")	1✔
106	group_by = data.get("group_by")	1✔
107
108	# Resolve the target question — by name (form-scoped name path)
109	# or by id (legacy rich path). Both must belong to the form and
110	# be a supported type.
111	question = None	1✔
112	if question_name:	1✔
113	validate_qname(question_name) # numeric id -> 400	1✔
114	question = Questions.objects.filter(	1✔
115	name=question_name,
116	form_id=form_id,
117	).first()
118	if not question:	1✔
119	raise serializers.ValidationError({	1✔
120	"question_name": (
121	f"Question '{question_name}' not found"
122	f" on form {form_id}."
123	),
124	})
125	elif question_id:	1✔
126	question = Questions.objects.filter(	1✔
127	pk=question_id,
128	form_id=form_id,
129	).first()
130	if not question:	1✔
131	raise serializers.ValidationError({	1✔
132	"question_id": (
133	f"Question {question_id} not found"
134	f" on form {form_id}."
135	),
136	})
137	if question:	1✔
138	if question.type not in SUPPORTED_QUESTION_TYPES:	1✔
139	raise serializers.ValidationError({	1✔
140	"question": (
141	f"Question type {question.type}"
142	" is not supported."
143	),
144	})
145	data["question"] = question	1✔
146
147	# Split criteria into same-form and parent-form buckets.
148	# names on form_id → criteria; names on parent form → parent_criteria.
149	criteria = data.get("criteria") or []	1✔
150	if criteria:	1✔
151	qnames = {c["parts"][0] for c in criteria}	1✔
152	on_form = set(	1✔
153	Questions.objects.filter(
154	name__in=qnames, form_id=form_id,
155	).values_list("name", flat=True)
156	)
157	remaining = qnames - on_form	1✔
158	parent_form = Forms.objects.filter(	1✔
159	pk=form_id,
160	).values_list("parent_id", flat=True).first()
161	on_parent = set()	1✔
162	if remaining and parent_form:	1✔
163	on_parent = set(	1✔
164	Questions.objects.filter(
165	name__in=remaining,
166	form_id=parent_form,
167	).values_list("name", flat=True)
168	)
169	unknown = remaining - on_parent	1✔
170	if unknown:	1!
171	raise serializers.ValidationError({	×
172	"criteria": (
173	"question_name(s) not on form "
174	f"{form_id} or its parent: "
175	f"{sorted(unknown)}"
176	),
177	})
178	data["criteria"] = [	1✔
179	c for c in criteria
180	if c["parts"][0] in on_form
181	] or None
182	data["parent_criteria"] = [	1✔
183	c for c in criteria
184	if c["parts"][0] in on_parent
185	] or None
186
187	# stack_by requires group_by and a resolved question (id or name)
188	if stack_by:	1✔
189	if not group_by:	1✔
190	raise serializers.ValidationError({	1✔
191	"stack_by": "stack_by requires group_by.",
192	})
193	if not data.get("question"):	1✔
194	raise serializers.ValidationError({	1✔
195	"stack_by": "stack_by requires a question.",
196	})
197
198	return data	1✔
199
200
201	class EscalationFilterSerializer(serializers.Serializer):	1✔
202	"""Validates query parameters for /visualization/escalation.
203
204	Criteria format: comma-separated, colon-delimited.
205	option_equals:{qid}:{value}
206	threshold_gt:{qid}:{value}
207	threshold_lt:{qid}:{value}
208	overdue:{completion_qid}:{deadline_qid}
209
210	Columns format: comma-separated, colon-delimited.
211	{key}:parent_name
212	{key}:administration
213	{key}:answer:{qid}
214	{key}:latest_date:{date_qid}
215	"""
216
217	monitoring_form_id = serializers.IntegerField(required=True)	1✔
218	criteria = serializers.CharField(required=True)	1✔
219	columns = serializers.CharField(required=True)	1✔
220	page = serializers.IntegerField(default=1, min_value=1)	1✔
221	page_size = serializers.IntegerField(	1✔
222	default=20, min_value=1, max_value=100,
223	)
224	from_date = serializers.DateField(required=False)	1✔
225	to_date = serializers.DateField(required=False)	1✔
226	date_question_name = serializers.CharField(required=False)	1✔
227	administration_id = serializers.IntegerField(required=False)	1✔
228	filter_criteria = serializers.CharField(required=False)	1✔
229
230	def validate_filter_criteria(self, value):	1✔
231	try:	1✔
232	return parse_criteria_string(	1✔
233	value, VALID_VALUES_CRITERIA_TYPES,
234	)
235	except ValueError as e:	1✔
236	raise serializers.ValidationError(str(e))	1✔
237
238	def validate_criteria(self, value):	1✔
239	"""Parse and validate criteria string."""
240	parsed = []	1✔
241	for item in value.split(","):	1✔
242	parts = item.strip().split(":")	1✔
243	if len(parts) < 3:	1!
244	raise serializers.ValidationError(	×
245	f"Invalid criteria format: '{item}'."
246	" Expected type:qid:value"
247	)
248	ctype = parts[0]	1✔
249	if ctype not in VALID_CRITERIA_TYPES:	1!
250	raise serializers.ValidationError(	×
251	f"Invalid criteria type: '{ctype}'."
252	f" Options: {VALID_CRITERIA_TYPES}"
253	)
254
255	try:	1✔
256	if ctype == "option_equals":	1✔
257	qname = validate_qname(parts[1])	1✔
258	normalized = [qname, parts[2]]	1✔
259	elif ctype in ("threshold_gt", "threshold_lt"):	1!
260	qname = validate_qname(parts[1])	1✔
261	threshold = float(parts[2])	1✔
262	normalized = [qname, threshold]	1✔
263	elif ctype == "overdue":	×
264	completion_qname = validate_qname(parts[1])	×
265	deadline_qname = validate_qname(parts[2])	×
266	normalized = [completion_qname, deadline_qname]	×
267	except ValueError:	1✔
268	raise serializers.ValidationError(	×
269	f"Invalid numeric value in criteria: '{item}'."
270	)
271
272	parsed.append({	1✔
273	"type": ctype,
274	"parts": normalized,
275	})
276	return parsed	1✔
277
278	def validate_columns(self, value):	1✔
279	"""Parse and validate columns string."""
280	qid_required_sources = {	1✔
281	"answer", "parent_answer", "latest_date",
282	}
283	parsed = []	1✔
284	for item in value.split(","):	1✔
285	parts = item.strip().split(":")	1✔
286	if len(parts) < 2:	1!
287	raise serializers.ValidationError(	×
288	f"Invalid column format: '{item}'."
289	" Expected key:source[:qid]"
290	)
291	key = parts[0]	1✔
292	source = parts[1]	1✔
293	if source not in VALID_COLUMN_SOURCES:	1✔
294	raise serializers.ValidationError(	1✔
295	f"Invalid column source: '{source}'."
296	f" Options: {VALID_COLUMN_SOURCES}"
297	)
298	col = {"key": key, "source": source}	1✔
299	if source in qid_required_sources and len(parts) < 3:	1!
300	raise serializers.ValidationError(	×
301	f"Column source '{source}' requires a"
302	f" question_name: '{item}'"
303	)
304	if len(parts) > 2:	1✔
305	col["question_name"] = validate_qname(parts[2])	1✔
306	parsed.append(col)	1✔
307	return parsed	1✔
308
309
310	class ProgressFilterSerializer(serializers.Serializer):	1✔
311	"""Validates query parameters for /visualization/progress.
312
313	Components format: comma-separated, colon-delimited.
314	{key}:{formula}:{qid1}:{qid2}:...:{total_items}
315
316	Example:
317	base:any_yes:111:222:333,tank:completed_binary:444,
318	pipes:ratio:555,security:multi_select_proportion:666:3
319	"""
320
321	monitoring_form_id = serializers.IntegerField(	1✔
322	required=True
323	)
324	components = serializers.CharField(required=True)	1✔
325	filter_question_name = serializers.CharField(	1✔
326	required=False
327	)
328	filter_option_value = serializers.CharField(	1✔
329	required=False
330	)
331	scope_question_name = serializers.CharField(	1✔
332	required=False
333	)
334	from_date = serializers.DateField(required=False)	1✔
335	to_date = serializers.DateField(required=False)	1✔
336	date_question_name = serializers.CharField(	1✔
337	required=False
338	)
339	administration_id = serializers.IntegerField(	1✔
340	required=False
341	)
342	criteria = serializers.CharField(required=False)	1✔
343
344	def validate_criteria(self, value):	1✔
345	try:	1✔
346	return parse_criteria_string(	1✔
347	value, VALID_VALUES_CRITERIA_TYPES,
348	)
349	except ValueError as e:	1✔
350	raise serializers.ValidationError(str(e))	1✔
351
352	def validate_components(self, value):	1✔
353	"""Parse and validate components string.
354
355	Format: key:formula:qid[:qid...][@type1\|type2\|...]
356	The optional @-suffix lists applicable project types.
357	"""
358	parsed = []	1✔
359	for item in value.split(","):	1✔
360	raw = item.strip()	1✔
361
362	# Split off optional @applicable_types suffix
363	applicable_types = None	1✔
364	if "@" in raw:	1!
365	raw, types_str = raw.split("@", 1)	×
366	applicable_types = [	×
367	t.strip() for t in types_str.split("\|")
368	if t.strip()
369	]
370
371	parts = raw.split(":")	1✔
372	if len(parts) < 3:	1!
373	raise serializers.ValidationError(	×
374	f"Invalid component format: '{item}'."
375	" Expected key:formula:qid[:qid...]"
376	)
377	key = parts[0]	1✔
378	formula = parts[1]	1✔
379	if formula not in VALID_PROGRESS_FORMULAS:	1✔
380	raise serializers.ValidationError(	1✔
381	f"Invalid formula: '{formula}'."
382	f" Options: {VALID_PROGRESS_FORMULAS}"
383	)
384
385	comp = {"key": key, "formula": formula}	1✔
386
387	if formula == "multi_select_proportion":	1✔
388	# Last segment is total_items
389	if len(parts) < 4:	1!
390	raise serializers.ValidationError(	×
391	f"{formula} requires total_items:"
392	f" '{item}'"
393	)
394	comp["question_names"] = [	1✔
395	validate_qname(q) for q in parts[2:-1]
396	]
397	try:	1✔
398	total_items = int(parts[-1])	1✔
399	except ValueError:	×
400	raise serializers.ValidationError(	×
401	f"Invalid total_items in component: '{item}'."
402	)
403	if total_items < 1:	1!
404	raise serializers.ValidationError(	×
405	f"total_items must be >= 1: '{item}'"
406	)
407	comp["total_items"] = total_items	1✔
408	elif formula == "ratio":	1!
409	# ratio requires implemented_qid:planned_qid
410	if len(parts) != 4:	1✔
411	raise serializers.ValidationError(	1✔
412	f"{formula} requires exactly two"
413	" question ids"
414	" (implemented:planned):"
415	f" '{item}'"
416	)
417	comp["question_names"] = [	1✔
418	validate_qname(parts[2]), validate_qname(parts[3]),
419	]
420	else:
421	comp["question_names"] = [	×
422	validate_qname(q) for q in parts[2:]
423	]
424
425	if applicable_types:	1!
426	comp["applicable_types"] = applicable_types	×
427
428	parsed.append(comp)	1✔
429	return parsed	1✔
430
431
432	# -- Response serializers (documentation only) --------------------
433	#
434	# These serializers describe the shape of JSON bodies returned by
435	# the dashboard endpoints. They are referenced from @extend_schema
436	# `responses=...` to replace Swagger's "No response body" with a
437	# concrete schema, and live alongside the request serializers for
438	# locality. None of them are used for actual DRF serialization —
439	# the views build plain dicts — so they only need field + type
440	# metadata that drf-spectacular can introspect.
441
442
443	class ValuesDataItemSerializer(serializers.Serializer):	1✔
444	"""One row in the /values `data` array.
445
446	Shape varies with `group_by`:
447	- none / stack_by: extra numeric columns are keyed dynamically,
448	so downstream readers should treat unknown keys as stack cells.
449	- option: `group` + `color` are populated.
450	- month / date: `group` is the machine-readable bucket key.
451	"""
452
453	value = serializers.FloatField(	1✔
454	required=False, allow_null=True,
455	help_text="Numeric aggregate (or percentage when requested).",
456	)
457	label = serializers.CharField(	1✔
458	required=False,
459	help_text="Human-readable label for this row.",
460	)
461	group = serializers.CharField(	1✔
462	required=False,
463	help_text=(
464	"Machine-readable key (option value, YYYY-MM, parent id,"
465	" …). Stable across translations."
466	),
467	)
468	color = serializers.CharField(	1✔
469	required=False,
470	help_text=(
471	"Hex color from QuestionOptions.color"
472	" (only when group_by=option)."
473	),
474	)
475
476
477	class ValuesResponseSerializer(serializers.Serializer):	1✔
478	"""/visualization/values response envelope.
479
480	For stacked responses (`stack_by=option\|parent_id`), each row in
481	`data` additionally carries one numeric column per stack — those
482	keys are dynamic and therefore not enumerable here. `stack_labels`
483	and `colors` are only present in that mode.
484	"""
485
486	data = ValuesDataItemSerializer(many=True)	1✔
487	labels = serializers.ListField(	1✔
488	child=serializers.CharField(),
489	help_text=(
490	"Ordered axis / legend labels — parallel to `data[].label`."
491	),
492	)
493	stack_labels = serializers.ListField(	1✔
494	child=serializers.CharField(), required=False,
495	help_text="Legend entries. Present only when stack_by is set.",
496	)
497	colors = serializers.ListField(	1✔
498	child=serializers.CharField(), required=False,
499	help_text="Per-stack colors when stack_by=option.",
500	)
501
502
503	class EscalationResultItemSerializer(serializers.Serializer):	1✔
504	"""One row from /escalation `results`.
505
506	Column keys are driven by the request's `columns=` param, so only
507	`id` is guaranteed. Other keys are documented per column in the
508	API spec; values are strings, numbers, or null.
509	"""
510
511	id = serializers.IntegerField()	1✔
512
513
514	class EscalationResponseSerializer(serializers.Serializer):	1✔
515	"""/visualization/escalation paginated envelope."""
516
517	count = serializers.IntegerField()	1✔
518	next = serializers.CharField(	1✔
519	allow_null=True, required=False,
520	)
521	previous = serializers.CharField(	1✔
522	allow_null=True, required=False,
523	)
524	results = EscalationResultItemSerializer(many=True)	1✔
525
526
527	class ProgressHistogramBucketSerializer(serializers.Serializer):	1✔
528	progress = serializers.CharField(	1✔
529	help_text="Bucket label, e.g. '0-10%', '11-20%'.",
530	)
531	count = serializers.IntegerField()	1✔
532
533
534	class ProgressDetailItemSerializer(serializers.Serializer):	1✔
535	"""One EPS / parent in the /progress `details` array."""
536
537	label = serializers.CharField()	1✔
538	group = serializers.CharField(	1✔
539	help_text="Parent FormData.id as string.",
540	)
541	components = serializers.DictField(	1✔
542	child=serializers.FloatField(),
543	help_text=(
544	"Per-component percent score, keyed by the component"
545	" `key` from the request."
546	),
547	)
548	overall = serializers.FloatField(	1✔
549	help_text="Average across components, 0-100.",
550	)
551
552
553	class ProgressResponseSerializer(serializers.Serializer):	1✔
554	"""/visualization/progress response envelope."""
555
556	histogram = ProgressHistogramBucketSerializer(many=True)	1✔
557	details = ProgressDetailItemSerializer(many=True)	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