feat: validate survey response answers against JSON Schema

.flake8

@@ -6,6 +6,7 @@ exclude =
     migrations,
     settings.py,
     venv,
+    .venv,
     env,
     node_modules
 max-line-length = 120

Makefile

@@ -0,0 +1,60 @@
+# Custom Django Development Makefile
+# This Makefile contains common Django management commands for easier development
+
+# Variables
+PYTHON = .venv/bin/python
+MANAGE = $(PYTHON) manage.py
+PROJECT_NAME = SORT
+
+# Development server
+runserver:
+	$(MANAGE) runserver
+
+# Database operations
+migrations:
+	$(MANAGE) makemigrations
+
+migrate:
+	$(MANAGE) migrate
+
+# System checks
+check:
+	$(MANAGE) check --fail-level WARNING
+	$(MANAGE) makemigrations --check --dry-run
+
+# User management
+superuser:
+	$(MANAGE) createsuperuser
+
+# Static files
+static:
+	$(MANAGE) collectstatic --noinput
+
+# Development tools
+shell:
+	$(MANAGE) shell
+
+test:
+	$(MANAGE) test --parallel=auto --failfast
+	npm test
+
+clean:
+	find . -type f -name "*.pyc" -delete
+	find . -type d -name "__pycache__" -delete
+
+# Dependencies
+requirements:
+	pip install -r requirements.txt
+
+# Code quality - only check project source files
+lint:
+	flake8
+
+format:
+	black $(PROJECT_NAME) --exclude="migrations|settings.py"
+
+# Default target when just running 'make'
+.DEFAULT_GOAL := help
+
+# Mark these targets as always needing to run (not files)
+.PHONY: help runserver migrations migrate check superuser static shell test clean requirements lint

SORT/settings.py

@@ -280,6 +280,12 @@ def cast_to_boolean(obj: Any) -> bool:
             "level": "WARNING",
             "propagate": False,
         },
+        # Suppress 4xx WARNING tracebacks in test output (e.g. expected 403s from permission tests)
+        "django.request": {
+            "handlers": ["console"],
+            "level": "ERROR",
+            "propagate": False,
+        },
     },
 }
 

SORT/test/test_case/view.py

@@ -42,11 +42,11 @@ def login_superuser(self):
         )
 
     def get(
-            self,
-            view_name: str,
-            expected_status_code: int = HTTPStatus.OK,
-            login: bool = True,
-            **kwargs
+        self,
+        view_name: str,
+        expected_status_code: int = HTTPStatus.OK,
+        login: bool = True,
+        **kwargs
     ):
         """
         Helper method to make a GET request to one of the views in this app.
@@ -63,12 +63,12 @@ def get(
         return response
 
     def post(
-            self,
-            view_name: str,
-            expected_status_code: int = HTTPStatus.OK,
-            login: bool = True,
-            data: dict = None,
-            **kwargs
+        self,
+        view_name: str,
+        expected_status_code: int = HTTPStatus.OK,
+        login: bool = True,
+        data: dict = None,
+        **kwargs
     ):
         """
         Helper method to make a POST request to one of the views in this app.

requirements.txt

@@ -9,6 +9,7 @@ django-allauth==65.14.1
 psycopg[binary]==3.2.*
 python-dotenv==1.0.1
 gunicorn==23.*
+jsonschema==4.*
 strenum==0.4.15
 xlsxwriter==3.2.5
 sqlparse==0.5.4

survey/management/commands/validate_responses.py

@@ -0,0 +1,25 @@
+from django.core.exceptions import ValidationError
+from django.core.management import BaseCommand
+
+from survey.models import Survey
+
+
+class Command(BaseCommand):
+    help = "Validate all survey response answers against their survey's JSON Schema"
+
+    def handle(self, *args, **options):
+        errors = 0
+        total = 0
+        for survey in Survey.objects.prefetch_related("survey_response").iterator(chunk_size=100):
+            for response in survey.survey_response.all():
+                total += 1
+                try:
+                    response.validate()
+                except ValidationError as exc:
+                    errors += 1
+                    self.stderr.write(
+                        f"Survey {survey.pk} / Response {response.pk}: {exc.message}"
+                    )
+        self.stdout.write(f"Validated {total} responses - {errors} error(s)")
+        if errors:
+            exit(1)

survey/models.py

@@ -10,15 +10,18 @@
 from typing import Generator, ContextManager
 from contextlib import contextmanager
 
+import jsonschema
 import xlsxwriter
 
 from django.conf import settings
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.http import HttpRequest
 from django.urls import reverse
 import django.core.validators
 
 from home.models import Project
+from survey.schema import field_schema
 
 logger = logging.getLogger(__name__)
 
@@ -184,6 +187,34 @@ def sections(self) -> tuple[dict]:
         # survey_config field
         return tuple(self.survey_config["sections"])
 
+    @property
+    def response_schema(self) -> dict:
+        """
+        Generate a JSON Schema that validates the structure of response answers.
+        """
+
+        section_schemas = list()
+        for section in self.sections:
+            fields = section.get("fields", [])
+            field_schemas = [field_schema(f) for f in fields]
+            section_schemas.append(
+                {
+                    "type": "array",
+                    "prefixItems": field_schemas,
+                    "items": False,
+                    "minItems": len(fields),
+                    "maxItems": len(fields),
+                }
+            )
+
+        return {
+            "type": "array",
+            "prefixItems": section_schemas,
+            "items": False,
+            "minItems": len(section_schemas),
+            "maxItems": len(section_schemas),
+        }
+
     @classmethod
     def _generate_random_field_value(cls, field_config):
         field_type = field_config["type"]
@@ -243,7 +274,8 @@ def accept_response(self, answers: list):
         """
         Enter a new survey submission.
         """
-        survey_response = SurveyResponse.objects.create(survey=self, answers=answers)
+        survey_response = SurveyResponse(survey=self, answers=answers)
+        survey_response.full_clean()
         survey_response.save()
         return survey_response
 
@@ -468,12 +500,27 @@ def __str__(self):
     def get_absolute_url(self, token):
         return reverse("survey", kwargs={"pk": self.survey.pk})
 
+    def validate(self) -> None:
+        """
+        Validate response answers against this survey's JSON Schema.
+
+        Raises django.core.exceptions.ValidationError if the answers do not match.
+        """
+
+        try:
+            jsonschema.validate(self.answers, self.survey.response_schema)
+        except jsonschema.ValidationError as exc:
+            raise ValidationError(exc.message) from exc
+
     def clean(self):
         super().clean()
 
         # Paused survey
         if not self.survey.is_active:
-            raise ValueError("Cannot submit response to an inactive survey")
+            raise ValidationError("Cannot submit response to an inactive survey")
+
+        # Validate response structure against survey config
+        self.validate()
 
     @property
     def answers_values(self) -> Generator[str, None, None]:

survey/schema.py

@@ -0,0 +1,63 @@
+"""
+Generate JSON Schema from survey configuration for validating response answers.
+"""
+
+
+def field_schema(field_config: dict) -> dict:
+    """Return a JSON Schema for one field's answer value."""
+    field_type = field_config["type"]
+
+    if field_type == "likert":
+        return _likert_schema(field_config)
+    elif field_type in ("radio", "select"):
+        return _radio_schema(field_config)
+    elif field_type == "checkbox":
+        return _checkbox_schema(field_config)
+    elif field_type in ("text", "textarea"):
+        return _text_schema(field_config)
+    else:
+        # Empty schema: any value is valid (no constraints imposed)
+        return {}
+
+
+def _likert_schema(field_config: dict) -> dict:
+    """Likert answer: array of strings from options, one per sublabel."""
+    num_sublabels = len(field_config.get("sublabels", []))
+    options = field_config.get("options", [])
+    schema = {
+        "type": "array",
+        "items": {"type": "string", "enum": options},
+        "minItems": num_sublabels,
+        "maxItems": num_sublabels,
+    }
+    return schema
+
+
+def _radio_schema(field_config: dict) -> dict:
+    """Radio/select answer: single string from options."""
+    options = field_config.get("options", [])
+    schema = {"type": "string"}
+    if options:
+        schema["enum"] = options
+    if field_config.get("required"):
+        schema["minLength"] = 1
+    return schema
+
+
+def _checkbox_schema(field_config: dict) -> dict:
+    """Checkbox answer: array of strings from options."""
+    options = field_config.get("options", [])
+    schema = {"type": "array", "items": {"type": "string"}}
+    if options:
+        schema["items"]["enum"] = options
+    if field_config.get("required"):
+        schema["minItems"] = 1
+    return schema
+
+
+def _text_schema(field_config: dict) -> dict:
+    """Text/textarea answer: string."""
+    schema = {"type": "string"}
+    if field_config.get("required"):
+        schema["minLength"] = 1
+    return schema