Initial project
This commit is contained in:
1
surveys/__init__.py
Normal file
1
surveys/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
|
||||
186
surveys/admin.py
Normal file
186
surveys/admin.py
Normal file
@@ -0,0 +1,186 @@
|
||||
from django.contrib import admin
|
||||
from django.urls import reverse
|
||||
from django.utils.html import format_html
|
||||
|
||||
from .models import Survey, SurveyAnswer, SurveyChoice, SurveyQuestion, SurveyResponse
|
||||
|
||||
|
||||
class SurveyAdminThemeMixin:
|
||||
class Media:
|
||||
css = {"all": ("surveys/admin.css",)}
|
||||
|
||||
|
||||
class SurveyQuestionInline(SurveyAdminThemeMixin, admin.StackedInline):
|
||||
model = SurveyQuestion
|
||||
extra = 1
|
||||
show_change_link = True
|
||||
prepopulated_fields = {"slug": ("prompt",)}
|
||||
fields = (
|
||||
("position", "type", "is_required"),
|
||||
"prompt",
|
||||
"slug",
|
||||
"help_text",
|
||||
("min_value", "max_value"),
|
||||
)
|
||||
|
||||
|
||||
class SurveyChoiceInline(SurveyAdminThemeMixin, admin.TabularInline):
|
||||
model = SurveyChoice
|
||||
extra = 2
|
||||
fields = ("position", "label", "value")
|
||||
prepopulated_fields = {"value": ("label",)}
|
||||
|
||||
|
||||
class SurveyAnswerInline(admin.TabularInline):
|
||||
model = SurveyAnswer
|
||||
extra = 0
|
||||
can_delete = False
|
||||
fields = ("question", "value", "selected_choices_display")
|
||||
readonly_fields = fields
|
||||
|
||||
def has_add_permission(self, request, obj=None):
|
||||
return False
|
||||
|
||||
@admin.display(description="Selected choices")
|
||||
def selected_choices_display(self, obj):
|
||||
labels = [choice.label for choice in obj.selected_choices.all()]
|
||||
return ", ".join(labels) or "-"
|
||||
|
||||
|
||||
@admin.register(Survey)
|
||||
class SurveyAdmin(SurveyAdminThemeMixin, admin.ModelAdmin):
|
||||
list_display = (
|
||||
"title",
|
||||
"status",
|
||||
"starts_at",
|
||||
"ends_at",
|
||||
"response_count",
|
||||
"api_link",
|
||||
)
|
||||
list_filter = ("status", "starts_at", "ends_at")
|
||||
search_fields = ("title", "description", "slug")
|
||||
prepopulated_fields = {"slug": ("title",)}
|
||||
readonly_fields = ("created_at", "updated_at", "response_count", "api_link")
|
||||
inlines = [SurveyQuestionInline]
|
||||
actions = ("publish_surveys", "close_surveys")
|
||||
fieldsets = (
|
||||
(
|
||||
"Survey",
|
||||
{
|
||||
"fields": (
|
||||
"title",
|
||||
"slug",
|
||||
"description",
|
||||
"thank_you_message",
|
||||
)
|
||||
},
|
||||
),
|
||||
(
|
||||
"Publishing",
|
||||
{
|
||||
"fields": (
|
||||
"status",
|
||||
"starts_at",
|
||||
"ends_at",
|
||||
"allow_multiple_responses",
|
||||
"api_link",
|
||||
)
|
||||
},
|
||||
),
|
||||
(
|
||||
"System",
|
||||
{
|
||||
"fields": ("response_count", "created_at", "updated_at"),
|
||||
"classes": ("collapse",),
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
@admin.display(description="Responses")
|
||||
def response_count(self, obj):
|
||||
if not obj.pk:
|
||||
return 0
|
||||
return obj.responses.count()
|
||||
|
||||
@admin.display(description="Public API")
|
||||
def api_link(self, obj):
|
||||
if not obj.pk:
|
||||
return "-"
|
||||
url = reverse("survey-detail", args=[obj.slug])
|
||||
return format_html('<a class="button" href="{}" target="_blank">Open API</a>', url)
|
||||
|
||||
@admin.action(description="Publish selected surveys")
|
||||
def publish_surveys(self, request, queryset):
|
||||
queryset.update(status=Survey.Status.PUBLISHED)
|
||||
|
||||
@admin.action(description="Close selected surveys")
|
||||
def close_surveys(self, request, queryset):
|
||||
queryset.update(status=Survey.Status.CLOSED)
|
||||
|
||||
|
||||
@admin.register(SurveyQuestion)
|
||||
class SurveyQuestionAdmin(SurveyAdminThemeMixin, admin.ModelAdmin):
|
||||
list_display = ("prompt", "survey", "type", "is_required", "position")
|
||||
list_filter = ("type", "is_required", "survey__status")
|
||||
search_fields = ("prompt", "slug", "survey__title")
|
||||
prepopulated_fields = {"slug": ("prompt",)}
|
||||
inlines = [SurveyChoiceInline]
|
||||
fieldsets = (
|
||||
(
|
||||
"Question",
|
||||
{
|
||||
"fields": (
|
||||
"survey",
|
||||
"prompt",
|
||||
"slug",
|
||||
"type",
|
||||
"help_text",
|
||||
"is_required",
|
||||
"position",
|
||||
)
|
||||
},
|
||||
),
|
||||
(
|
||||
"Rating Options",
|
||||
{
|
||||
"fields": ("min_value", "max_value"),
|
||||
"description": "Used only for rating questions.",
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@admin.register(SurveyChoice)
|
||||
class SurveyChoiceAdmin(SurveyAdminThemeMixin, admin.ModelAdmin):
|
||||
list_display = ("label", "question", "survey_title", "value", "position")
|
||||
list_filter = ("question__survey",)
|
||||
search_fields = ("label", "value", "question__prompt", "question__survey__title")
|
||||
prepopulated_fields = {"value": ("label",)}
|
||||
|
||||
@admin.display(description="Survey")
|
||||
def survey_title(self, obj):
|
||||
return obj.question.survey.title
|
||||
|
||||
|
||||
@admin.register(SurveyResponse)
|
||||
class SurveyResponseAdmin(SurveyAdminThemeMixin, admin.ModelAdmin):
|
||||
list_display = ("survey", "respondent_email", "submitted_at", "ip_address")
|
||||
list_filter = ("survey", "submitted_at")
|
||||
search_fields = ("survey__title", "respondent_email", "ip_address")
|
||||
readonly_fields = (
|
||||
"survey",
|
||||
"respondent_email",
|
||||
"metadata",
|
||||
"ip_address",
|
||||
"user_agent",
|
||||
"submitted_at",
|
||||
)
|
||||
inlines = [SurveyAnswerInline]
|
||||
|
||||
def has_add_permission(self, request):
|
||||
return False
|
||||
|
||||
|
||||
admin.site.site_header = "Survey Studio"
|
||||
admin.site.site_title = "Survey Studio"
|
||||
admin.site.index_title = "Survey Management"
|
||||
7
surveys/apps.py
Normal file
7
surveys/apps.py
Normal file
@@ -0,0 +1,7 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class SurveysConfig(AppConfig):
|
||||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "surveys"
|
||||
verbose_name = "Surveys"
|
||||
36
surveys/middleware.py
Normal file
36
surveys/middleware.py
Normal file
@@ -0,0 +1,36 @@
|
||||
from django.conf import settings
|
||||
from django.http import HttpResponse
|
||||
from django.utils.cache import patch_vary_headers
|
||||
|
||||
|
||||
class PublicApiCorsMiddleware:
|
||||
"""Add simple CORS support for the public survey API."""
|
||||
|
||||
def __init__(self, get_response):
|
||||
self.get_response = get_response
|
||||
|
||||
def __call__(self, request):
|
||||
if request.path.startswith("/api/") and request.method == "OPTIONS":
|
||||
response = HttpResponse(status=204)
|
||||
else:
|
||||
response = self.get_response(request)
|
||||
|
||||
if request.path.startswith("/api/"):
|
||||
self._set_cors_headers(request, response)
|
||||
|
||||
return response
|
||||
|
||||
def _set_cors_headers(self, request, response):
|
||||
configured_origins = getattr(settings, "SURVEY_API_CORS_ORIGINS", "*")
|
||||
origins = [origin.strip() for origin in configured_origins.split(",") if origin.strip()]
|
||||
request_origin = request.headers.get("Origin")
|
||||
|
||||
if "*" in origins:
|
||||
response["Access-Control-Allow-Origin"] = "*"
|
||||
elif request_origin in origins:
|
||||
response["Access-Control-Allow-Origin"] = request_origin
|
||||
patch_vary_headers(response, ("Origin",))
|
||||
|
||||
response["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
|
||||
response["Access-Control-Allow-Headers"] = "Content-Type"
|
||||
response["Access-Control-Max-Age"] = "86400"
|
||||
213
surveys/migrations/0001_initial.py
Normal file
213
surveys/migrations/0001_initial.py
Normal file
@@ -0,0 +1,213 @@
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
initial = True
|
||||
|
||||
dependencies = []
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name="Survey",
|
||||
fields=[
|
||||
(
|
||||
"id",
|
||||
models.BigAutoField(
|
||||
auto_created=True,
|
||||
primary_key=True,
|
||||
serialize=False,
|
||||
verbose_name="ID",
|
||||
),
|
||||
),
|
||||
("title", models.CharField(max_length=180)),
|
||||
(
|
||||
"slug",
|
||||
models.SlugField(
|
||||
help_text="Stable public identifier used in API URLs.",
|
||||
max_length=140,
|
||||
unique=True,
|
||||
),
|
||||
),
|
||||
("description", models.TextField(blank=True)),
|
||||
(
|
||||
"thank_you_message",
|
||||
models.CharField(
|
||||
blank=True,
|
||||
default="Thanks for your response.",
|
||||
max_length=240,
|
||||
),
|
||||
),
|
||||
(
|
||||
"status",
|
||||
models.CharField(
|
||||
choices=[
|
||||
("draft", "Draft"),
|
||||
("published", "Published"),
|
||||
("closed", "Closed"),
|
||||
],
|
||||
default="draft",
|
||||
max_length=20,
|
||||
),
|
||||
),
|
||||
("starts_at", models.DateTimeField(blank=True, null=True)),
|
||||
("ends_at", models.DateTimeField(blank=True, null=True)),
|
||||
("allow_multiple_responses", models.BooleanField(default=True)),
|
||||
("created_at", models.DateTimeField(auto_now_add=True)),
|
||||
("updated_at", models.DateTimeField(auto_now=True)),
|
||||
],
|
||||
options={
|
||||
"ordering": ["-updated_at", "title"],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name="SurveyQuestion",
|
||||
fields=[
|
||||
(
|
||||
"id",
|
||||
models.BigAutoField(
|
||||
auto_created=True,
|
||||
primary_key=True,
|
||||
serialize=False,
|
||||
verbose_name="ID",
|
||||
),
|
||||
),
|
||||
("prompt", models.CharField(max_length=280)),
|
||||
("slug", models.SlugField(blank=True, max_length=140)),
|
||||
("help_text", models.CharField(blank=True, max_length=240)),
|
||||
(
|
||||
"type",
|
||||
models.CharField(
|
||||
choices=[
|
||||
("short_text", "Short text"),
|
||||
("long_text", "Long text"),
|
||||
("single_choice", "Single choice"),
|
||||
("multiple_choice", "Multiple choice"),
|
||||
("rating", "Rating"),
|
||||
("yes_no", "Yes/No"),
|
||||
],
|
||||
default="short_text",
|
||||
max_length=30,
|
||||
),
|
||||
),
|
||||
("is_required", models.BooleanField(default=True)),
|
||||
("position", models.PositiveIntegerField(default=0)),
|
||||
("min_value", models.IntegerField(blank=True, default=1, null=True)),
|
||||
("max_value", models.IntegerField(blank=True, default=5, null=True)),
|
||||
(
|
||||
"survey",
|
||||
models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="questions",
|
||||
to="surveys.survey",
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
"ordering": ["position", "id"],
|
||||
"unique_together": {("survey", "slug")},
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name="SurveyChoice",
|
||||
fields=[
|
||||
(
|
||||
"id",
|
||||
models.BigAutoField(
|
||||
auto_created=True,
|
||||
primary_key=True,
|
||||
serialize=False,
|
||||
verbose_name="ID",
|
||||
),
|
||||
),
|
||||
("label", models.CharField(max_length=180)),
|
||||
("value", models.SlugField(blank=True, max_length=140)),
|
||||
("position", models.PositiveIntegerField(default=0)),
|
||||
(
|
||||
"question",
|
||||
models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="choices",
|
||||
to="surveys.surveyquestion",
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
"ordering": ["position", "id"],
|
||||
"unique_together": {("question", "value")},
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name="SurveyResponse",
|
||||
fields=[
|
||||
(
|
||||
"id",
|
||||
models.BigAutoField(
|
||||
auto_created=True,
|
||||
primary_key=True,
|
||||
serialize=False,
|
||||
verbose_name="ID",
|
||||
),
|
||||
),
|
||||
("respondent_email", models.EmailField(blank=True, max_length=254)),
|
||||
("metadata", models.JSONField(blank=True, default=dict)),
|
||||
("ip_address", models.GenericIPAddressField(blank=True, null=True)),
|
||||
("user_agent", models.CharField(blank=True, max_length=255)),
|
||||
("submitted_at", models.DateTimeField(auto_now_add=True)),
|
||||
(
|
||||
"survey",
|
||||
models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="responses",
|
||||
to="surveys.survey",
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
"ordering": ["-submitted_at"],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name="SurveyAnswer",
|
||||
fields=[
|
||||
(
|
||||
"id",
|
||||
models.BigAutoField(
|
||||
auto_created=True,
|
||||
primary_key=True,
|
||||
serialize=False,
|
||||
verbose_name="ID",
|
||||
),
|
||||
),
|
||||
("value", models.JSONField(blank=True, default=dict)),
|
||||
(
|
||||
"question",
|
||||
models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.PROTECT,
|
||||
related_name="answers",
|
||||
to="surveys.surveyquestion",
|
||||
),
|
||||
),
|
||||
(
|
||||
"response",
|
||||
models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="answers",
|
||||
to="surveys.surveyresponse",
|
||||
),
|
||||
),
|
||||
(
|
||||
"selected_choices",
|
||||
models.ManyToManyField(
|
||||
blank=True,
|
||||
related_name="answers",
|
||||
to="surveys.surveychoice",
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
"ordering": ["question__position", "id"],
|
||||
"unique_together": {("response", "question")},
|
||||
},
|
||||
),
|
||||
]
|
||||
1
surveys/migrations/__init__.py
Normal file
1
surveys/migrations/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
|
||||
200
surveys/models.py
Normal file
200
surveys/models.py
Normal file
@@ -0,0 +1,200 @@
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.db import models
|
||||
from django.db.models import Q
|
||||
from django.utils import timezone
|
||||
from django.utils.text import slugify
|
||||
|
||||
|
||||
class SurveyQuerySet(models.QuerySet):
|
||||
def public(self):
|
||||
now = timezone.now()
|
||||
return (
|
||||
self.filter(status=Survey.Status.PUBLISHED)
|
||||
.filter(Q(starts_at__isnull=True) | Q(starts_at__lte=now))
|
||||
.filter(Q(ends_at__isnull=True) | Q(ends_at__gte=now))
|
||||
)
|
||||
|
||||
|
||||
class Survey(models.Model):
|
||||
class Status(models.TextChoices):
|
||||
DRAFT = "draft", "Draft"
|
||||
PUBLISHED = "published", "Published"
|
||||
CLOSED = "closed", "Closed"
|
||||
|
||||
title = models.CharField(max_length=180)
|
||||
slug = models.SlugField(
|
||||
max_length=140,
|
||||
unique=True,
|
||||
help_text="Stable public identifier used in API URLs.",
|
||||
)
|
||||
description = models.TextField(blank=True)
|
||||
thank_you_message = models.CharField(
|
||||
max_length=240,
|
||||
blank=True,
|
||||
default="Thanks for your response.",
|
||||
)
|
||||
status = models.CharField(
|
||||
max_length=20,
|
||||
choices=Status.choices,
|
||||
default=Status.DRAFT,
|
||||
)
|
||||
starts_at = models.DateTimeField(blank=True, null=True)
|
||||
ends_at = models.DateTimeField(blank=True, null=True)
|
||||
allow_multiple_responses = models.BooleanField(default=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
objects = SurveyQuerySet.as_manager()
|
||||
|
||||
class Meta:
|
||||
ordering = ["-updated_at", "title"]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.title
|
||||
|
||||
def clean(self):
|
||||
if self.starts_at and self.ends_at and self.starts_at >= self.ends_at:
|
||||
raise ValidationError("The survey start time must be before the end time.")
|
||||
|
||||
def accepts_responses(self):
|
||||
now = timezone.now()
|
||||
if self.status != self.Status.PUBLISHED:
|
||||
return False, "Survey is not published."
|
||||
if self.starts_at and self.starts_at > now:
|
||||
return False, "Survey has not started yet."
|
||||
if self.ends_at and self.ends_at < now:
|
||||
return False, "Survey has ended."
|
||||
return True, ""
|
||||
|
||||
|
||||
class SurveyQuestion(models.Model):
|
||||
class Type(models.TextChoices):
|
||||
SHORT_TEXT = "short_text", "Short text"
|
||||
LONG_TEXT = "long_text", "Long text"
|
||||
SINGLE_CHOICE = "single_choice", "Single choice"
|
||||
MULTIPLE_CHOICE = "multiple_choice", "Multiple choice"
|
||||
RATING = "rating", "Rating"
|
||||
YES_NO = "yes_no", "Yes/No"
|
||||
|
||||
survey = models.ForeignKey(Survey, related_name="questions", on_delete=models.CASCADE)
|
||||
prompt = models.CharField(max_length=280)
|
||||
slug = models.SlugField(max_length=140, blank=True)
|
||||
help_text = models.CharField(max_length=240, blank=True)
|
||||
type = models.CharField(
|
||||
max_length=30,
|
||||
choices=Type.choices,
|
||||
default=Type.SHORT_TEXT,
|
||||
)
|
||||
is_required = models.BooleanField(default=True)
|
||||
position = models.PositiveIntegerField(default=0)
|
||||
min_value = models.IntegerField(default=1, blank=True, null=True)
|
||||
max_value = models.IntegerField(default=5, blank=True, null=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ["position", "id"]
|
||||
unique_together = ("survey", "slug")
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.prompt
|
||||
|
||||
def clean(self):
|
||||
if self.type == self.Type.RATING:
|
||||
if self.min_value is None or self.max_value is None:
|
||||
raise ValidationError("Rating questions need minimum and maximum values.")
|
||||
if self.min_value >= self.max_value:
|
||||
raise ValidationError("The rating minimum must be lower than the maximum.")
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
if not self.slug:
|
||||
self.slug = self._build_unique_slug()
|
||||
super().save(*args, **kwargs)
|
||||
|
||||
def _build_unique_slug(self):
|
||||
base = slugify(self.prompt)[:120] or "question"
|
||||
candidate = base
|
||||
suffix = 2
|
||||
while (
|
||||
SurveyQuestion.objects.filter(survey_id=self.survey_id, slug=candidate)
|
||||
.exclude(pk=self.pk)
|
||||
.exists()
|
||||
):
|
||||
candidate = f"{base[:110]}-{suffix}"
|
||||
suffix += 1
|
||||
return candidate
|
||||
|
||||
|
||||
class SurveyChoice(models.Model):
|
||||
question = models.ForeignKey(
|
||||
SurveyQuestion,
|
||||
related_name="choices",
|
||||
on_delete=models.CASCADE,
|
||||
)
|
||||
label = models.CharField(max_length=180)
|
||||
value = models.SlugField(max_length=140, blank=True)
|
||||
position = models.PositiveIntegerField(default=0)
|
||||
|
||||
class Meta:
|
||||
ordering = ["position", "id"]
|
||||
unique_together = ("question", "value")
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.label
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
if not self.value:
|
||||
self.value = self._build_unique_value()
|
||||
super().save(*args, **kwargs)
|
||||
|
||||
def _build_unique_value(self):
|
||||
base = slugify(self.label)[:120] or "choice"
|
||||
candidate = base
|
||||
suffix = 2
|
||||
while (
|
||||
SurveyChoice.objects.filter(question_id=self.question_id, value=candidate)
|
||||
.exclude(pk=self.pk)
|
||||
.exists()
|
||||
):
|
||||
candidate = f"{base[:110]}-{suffix}"
|
||||
suffix += 1
|
||||
return candidate
|
||||
|
||||
|
||||
class SurveyResponse(models.Model):
|
||||
survey = models.ForeignKey(Survey, related_name="responses", on_delete=models.CASCADE)
|
||||
respondent_email = models.EmailField(blank=True)
|
||||
metadata = models.JSONField(default=dict, blank=True)
|
||||
ip_address = models.GenericIPAddressField(blank=True, null=True)
|
||||
user_agent = models.CharField(max_length=255, blank=True)
|
||||
submitted_at = models.DateTimeField(auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ["-submitted_at"]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.survey} response #{self.pk}"
|
||||
|
||||
|
||||
class SurveyAnswer(models.Model):
|
||||
response = models.ForeignKey(
|
||||
SurveyResponse,
|
||||
related_name="answers",
|
||||
on_delete=models.CASCADE,
|
||||
)
|
||||
question = models.ForeignKey(
|
||||
SurveyQuestion,
|
||||
related_name="answers",
|
||||
on_delete=models.PROTECT,
|
||||
)
|
||||
value = models.JSONField(default=dict, blank=True)
|
||||
selected_choices = models.ManyToManyField(
|
||||
SurveyChoice,
|
||||
related_name="answers",
|
||||
blank=True,
|
||||
)
|
||||
|
||||
class Meta:
|
||||
ordering = ["question__position", "id"]
|
||||
unique_together = ("response", "question")
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.question}: {self.value}"
|
||||
137
surveys/tests.py
Normal file
137
surveys/tests.py
Normal file
@@ -0,0 +1,137 @@
|
||||
from django.contrib.staticfiles import finders
|
||||
from django.test import Client, TestCase
|
||||
|
||||
from .models import Survey, SurveyAnswer, SurveyChoice, SurveyQuestion, SurveyResponse
|
||||
|
||||
|
||||
class SurveyApiTests(TestCase):
|
||||
def setUp(self):
|
||||
self.client = Client()
|
||||
self.survey = Survey.objects.create(
|
||||
title="Customer Feedback",
|
||||
slug="customer-feedback",
|
||||
status=Survey.Status.PUBLISHED,
|
||||
thank_you_message="Thanks for helping us improve.",
|
||||
)
|
||||
self.name_question = SurveyQuestion.objects.create(
|
||||
survey=self.survey,
|
||||
prompt="What is your name?",
|
||||
slug="name",
|
||||
type=SurveyQuestion.Type.SHORT_TEXT,
|
||||
position=1,
|
||||
)
|
||||
self.mood_question = SurveyQuestion.objects.create(
|
||||
survey=self.survey,
|
||||
prompt="How was your experience?",
|
||||
slug="experience",
|
||||
type=SurveyQuestion.Type.SINGLE_CHOICE,
|
||||
position=2,
|
||||
)
|
||||
self.good_choice = SurveyChoice.objects.create(
|
||||
question=self.mood_question,
|
||||
label="Good",
|
||||
value="good",
|
||||
position=1,
|
||||
)
|
||||
SurveyChoice.objects.create(
|
||||
question=self.mood_question,
|
||||
label="Bad",
|
||||
value="bad",
|
||||
position=2,
|
||||
)
|
||||
|
||||
def test_lists_public_surveys(self):
|
||||
Survey.objects.create(title="Draft Survey", slug="draft", status=Survey.Status.DRAFT)
|
||||
|
||||
response = self.client.get("/api/surveys/")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(len(payload["results"]), 1)
|
||||
self.assertEqual(payload["results"][0]["slug"], "customer-feedback")
|
||||
|
||||
def test_returns_survey_detail(self):
|
||||
response = self.client.get("/api/surveys/customer-feedback/")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["title"], "Customer Feedback")
|
||||
self.assertEqual(len(payload["questions"]), 2)
|
||||
self.assertEqual(payload["questions"][1]["choices"][0]["value"], "good")
|
||||
|
||||
def test_accepts_anonymous_response(self):
|
||||
response = self.client.post(
|
||||
"/api/surveys/customer-feedback/responses/",
|
||||
data={
|
||||
"respondent_email": "person@example.com",
|
||||
"metadata": {"source": "test"},
|
||||
"answers": [
|
||||
{"question": "name", "value": "Alex"},
|
||||
{"question": "experience", "choice": "good"},
|
||||
],
|
||||
},
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 201)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["thank_you_message"], "Thanks for helping us improve.")
|
||||
self.assertEqual(SurveyResponse.objects.count(), 1)
|
||||
self.assertEqual(SurveyAnswer.objects.count(), 2)
|
||||
choice_answer = SurveyAnswer.objects.get(question=self.mood_question)
|
||||
self.assertEqual(list(choice_answer.selected_choices.all()), [self.good_choice])
|
||||
|
||||
def test_rejects_missing_required_answer(self):
|
||||
response = self.client.post(
|
||||
"/api/surveys/customer-feedback/responses/",
|
||||
data={"answers": [{"question": "name", "value": "Alex"}]},
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn("experience", response.json()["errors"])
|
||||
|
||||
def test_rejects_closed_survey_response(self):
|
||||
self.survey.status = Survey.Status.CLOSED
|
||||
self.survey.save(update_fields=["status"])
|
||||
|
||||
response = self.client.post(
|
||||
"/api/surveys/customer-feedback/responses/",
|
||||
data={"answers": []},
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 403)
|
||||
|
||||
|
||||
class SitePageTests(TestCase):
|
||||
def setUp(self):
|
||||
self.client = Client()
|
||||
|
||||
def test_welcome_page_renders(self):
|
||||
response = self.client.get("/")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, "Survey Studio")
|
||||
self.assertContains(response, "/api/surveys/")
|
||||
|
||||
def test_docs_page_renders(self):
|
||||
response = self.client.get("/docs/")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, "Survey Studio Docs")
|
||||
self.assertContains(response, "POST")
|
||||
|
||||
def test_public_static_assets_are_discoverable(self):
|
||||
self.assertIsNotNone(finders.find("surveys/admin.css"))
|
||||
self.assertIsNotNone(finders.find("surveys/site.css"))
|
||||
self.assertIsNotNone(finders.find("surveys/survey-flow.svg"))
|
||||
|
||||
def test_admin_theme_css_loads_after_page_css(self):
|
||||
response = self.client.get("/admin/login/")
|
||||
content = response.content.decode()
|
||||
|
||||
self.assertLess(
|
||||
content.index("/static/admin/css/login.css"),
|
||||
content.index("/static/surveys/admin.css"),
|
||||
)
|
||||
14
surveys/urls.py
Normal file
14
surveys/urls.py
Normal file
@@ -0,0 +1,14 @@
|
||||
from django.urls import path
|
||||
|
||||
from . import views
|
||||
|
||||
|
||||
urlpatterns = [
|
||||
path("surveys/", views.survey_list, name="survey-list"),
|
||||
path("surveys/<slug:slug>/", views.survey_detail, name="survey-detail"),
|
||||
path(
|
||||
"surveys/<slug:slug>/responses/",
|
||||
views.submit_survey_response,
|
||||
name="survey-response-submit",
|
||||
),
|
||||
]
|
||||
370
surveys/views.py
Normal file
370
surveys/views.py
Normal file
@@ -0,0 +1,370 @@
|
||||
import json
|
||||
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.validators import validate_email, validate_ipv46_address
|
||||
from django.db import transaction
|
||||
from django.http import JsonResponse
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django.urls import reverse
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
from django.views.decorators.http import require_GET, require_http_methods
|
||||
|
||||
from .models import Survey, SurveyAnswer, SurveyQuestion, SurveyResponse
|
||||
|
||||
|
||||
@require_GET
|
||||
def survey_list(request):
|
||||
surveys = Survey.objects.public()
|
||||
return JsonResponse(
|
||||
{
|
||||
"results": [
|
||||
serialize_survey(request, survey, include_questions=False)
|
||||
for survey in surveys
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@require_GET
|
||||
def survey_detail(request, slug):
|
||||
survey = get_object_or_404(
|
||||
Survey.objects.public().prefetch_related("questions__choices"),
|
||||
slug=slug,
|
||||
)
|
||||
return JsonResponse(serialize_survey(request, survey, include_questions=True))
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
@require_http_methods(["POST"])
|
||||
def submit_survey_response(request, slug):
|
||||
survey = get_object_or_404(
|
||||
Survey.objects.prefetch_related("questions__choices"),
|
||||
slug=slug,
|
||||
)
|
||||
accepts_responses, reason = survey.accepts_responses()
|
||||
if not accepts_responses:
|
||||
return JsonResponse({"detail": reason}, status=403)
|
||||
|
||||
payload, error = parse_json_body(request)
|
||||
if error:
|
||||
return JsonResponse({"detail": error}, status=400)
|
||||
|
||||
respondent_email = str(payload.get("respondent_email", "")).strip()
|
||||
email_error = validate_optional_email(respondent_email)
|
||||
if email_error:
|
||||
return JsonResponse({"errors": {"respondent_email": email_error}}, status=400)
|
||||
|
||||
metadata = payload.get("metadata") or {}
|
||||
if not isinstance(metadata, dict):
|
||||
return JsonResponse({"errors": {"metadata": "Metadata must be an object."}}, status=400)
|
||||
|
||||
if not survey.allow_multiple_responses and respondent_email:
|
||||
already_responded = survey.responses.filter(respondent_email=respondent_email).exists()
|
||||
if already_responded:
|
||||
return JsonResponse(
|
||||
{"detail": "This email has already submitted a response."},
|
||||
status=409,
|
||||
)
|
||||
|
||||
answers, errors = validate_answers(survey, payload.get("answers"))
|
||||
if errors:
|
||||
return JsonResponse({"errors": errors}, status=400)
|
||||
|
||||
with transaction.atomic():
|
||||
response = SurveyResponse.objects.create(
|
||||
survey=survey,
|
||||
respondent_email=respondent_email,
|
||||
metadata=metadata,
|
||||
ip_address=get_client_ip(request),
|
||||
user_agent=request.META.get("HTTP_USER_AGENT", "")[:255],
|
||||
)
|
||||
for answer_data in answers:
|
||||
answer = SurveyAnswer.objects.create(
|
||||
response=response,
|
||||
question=answer_data["question"],
|
||||
value=answer_data["value"],
|
||||
)
|
||||
if answer_data["choices"]:
|
||||
answer.selected_choices.set(answer_data["choices"])
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"id": response.pk,
|
||||
"survey": survey.slug,
|
||||
"submitted_at": response.submitted_at.isoformat(),
|
||||
"thank_you_message": survey.thank_you_message,
|
||||
},
|
||||
status=201,
|
||||
)
|
||||
|
||||
|
||||
def serialize_survey(request, survey, include_questions):
|
||||
response_url = reverse("survey-response-submit", args=[survey.slug])
|
||||
detail_url = reverse("survey-detail", args=[survey.slug])
|
||||
data = {
|
||||
"id": survey.pk,
|
||||
"title": survey.title,
|
||||
"slug": survey.slug,
|
||||
"description": survey.description,
|
||||
"thank_you_message": survey.thank_you_message,
|
||||
"starts_at": serialize_datetime(survey.starts_at),
|
||||
"ends_at": serialize_datetime(survey.ends_at),
|
||||
"url": request.build_absolute_uri(detail_url),
|
||||
"response_url": request.build_absolute_uri(response_url),
|
||||
}
|
||||
if include_questions:
|
||||
data["questions"] = [serialize_question(question) for question in survey.questions.all()]
|
||||
return data
|
||||
|
||||
|
||||
def serialize_question(question):
|
||||
data = {
|
||||
"id": question.pk,
|
||||
"slug": question.slug,
|
||||
"prompt": question.prompt,
|
||||
"help_text": question.help_text,
|
||||
"type": question.type,
|
||||
"required": question.is_required,
|
||||
"position": question.position,
|
||||
}
|
||||
if question.type == SurveyQuestion.Type.RATING:
|
||||
data["min_value"] = question.min_value
|
||||
data["max_value"] = question.max_value
|
||||
if question.type in {
|
||||
SurveyQuestion.Type.SINGLE_CHOICE,
|
||||
SurveyQuestion.Type.MULTIPLE_CHOICE,
|
||||
}:
|
||||
data["choices"] = [serialize_choice(choice) for choice in question.choices.all()]
|
||||
else:
|
||||
data["choices"] = []
|
||||
return data
|
||||
|
||||
|
||||
def serialize_choice(choice):
|
||||
return {
|
||||
"id": choice.pk,
|
||||
"label": choice.label,
|
||||
"value": choice.value,
|
||||
"position": choice.position,
|
||||
}
|
||||
|
||||
|
||||
def serialize_datetime(value):
|
||||
if value is None:
|
||||
return None
|
||||
return value.isoformat()
|
||||
|
||||
|
||||
def parse_json_body(request):
|
||||
try:
|
||||
payload = json.loads(request.body.decode("utf-8") or "{}")
|
||||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||||
return None, "Request body must be valid JSON."
|
||||
if not isinstance(payload, dict):
|
||||
return None, "Request body must be a JSON object."
|
||||
return payload, None
|
||||
|
||||
|
||||
def validate_optional_email(email):
|
||||
if not email:
|
||||
return None
|
||||
try:
|
||||
validate_email(email)
|
||||
except ValidationError:
|
||||
return "Enter a valid email address."
|
||||
return None
|
||||
|
||||
|
||||
def validate_answers(survey, raw_answers):
|
||||
normalized_answers, normalize_error = normalize_answers(raw_answers)
|
||||
if normalize_error:
|
||||
return [], {"answers": normalize_error}
|
||||
|
||||
questions = list(survey.questions.all())
|
||||
question_by_lookup = {}
|
||||
for question in questions:
|
||||
question_by_lookup[str(question.pk)] = question
|
||||
question_by_lookup[question.slug] = question
|
||||
|
||||
validated = []
|
||||
errors = {}
|
||||
answered_question_ids = set()
|
||||
|
||||
for index, answer_payload in enumerate(normalized_answers):
|
||||
if not isinstance(answer_payload, dict):
|
||||
errors[f"answers[{index}]"] = "Each answer must be an object."
|
||||
continue
|
||||
|
||||
question_lookup = answer_payload.get("question", answer_payload.get("question_id"))
|
||||
question = question_by_lookup.get(str(question_lookup))
|
||||
if question is None:
|
||||
errors[f"answers[{index}].question"] = "Unknown question."
|
||||
continue
|
||||
if question.pk in answered_question_ids:
|
||||
errors[question.slug] = "Question answered more than once."
|
||||
continue
|
||||
|
||||
answer_value, selected_choices, error = validate_answer_value(question, answer_payload)
|
||||
if error:
|
||||
errors[question.slug] = error
|
||||
continue
|
||||
if answer_value is None:
|
||||
continue
|
||||
|
||||
answered_question_ids.add(question.pk)
|
||||
validated.append(
|
||||
{
|
||||
"question": question,
|
||||
"value": answer_value,
|
||||
"choices": selected_choices,
|
||||
}
|
||||
)
|
||||
|
||||
for question in questions:
|
||||
if question.is_required and question.pk not in answered_question_ids:
|
||||
errors.setdefault(question.slug, "This question is required.")
|
||||
|
||||
return validated, errors
|
||||
|
||||
|
||||
def normalize_answers(raw_answers):
|
||||
if raw_answers is None:
|
||||
return [], None
|
||||
if isinstance(raw_answers, dict):
|
||||
return [
|
||||
{"question": question, "value": value}
|
||||
for question, value in raw_answers.items()
|
||||
], None
|
||||
if isinstance(raw_answers, list):
|
||||
return raw_answers, None
|
||||
return None, "Answers must be a list or an object keyed by question slug."
|
||||
|
||||
|
||||
def validate_answer_value(question, answer_payload):
|
||||
if question.type in {SurveyQuestion.Type.SHORT_TEXT, SurveyQuestion.Type.LONG_TEXT}:
|
||||
return validate_text_answer(question, answer_payload)
|
||||
if question.type == SurveyQuestion.Type.SINGLE_CHOICE:
|
||||
return validate_single_choice_answer(question, answer_payload)
|
||||
if question.type == SurveyQuestion.Type.MULTIPLE_CHOICE:
|
||||
return validate_multiple_choice_answer(question, answer_payload)
|
||||
if question.type == SurveyQuestion.Type.RATING:
|
||||
return validate_rating_answer(question, answer_payload)
|
||||
if question.type == SurveyQuestion.Type.YES_NO:
|
||||
return validate_yes_no_answer(question, answer_payload)
|
||||
return None, [], "Unsupported question type."
|
||||
|
||||
|
||||
def validate_text_answer(question, answer_payload):
|
||||
raw_value = answer_payload.get("value", answer_payload.get("text"))
|
||||
if is_empty(raw_value):
|
||||
if question.is_required:
|
||||
return None, [], "This question is required."
|
||||
return None, [], None
|
||||
if not isinstance(raw_value, str):
|
||||
return None, [], "Answer must be text."
|
||||
return {"text": raw_value.strip()}, [], None
|
||||
|
||||
|
||||
def validate_single_choice_answer(question, answer_payload):
|
||||
raw_choice = answer_payload.get("choice", answer_payload.get("value"))
|
||||
if is_empty(raw_choice):
|
||||
if question.is_required:
|
||||
return None, [], "Select one choice."
|
||||
return None, [], None
|
||||
|
||||
choice, error = resolve_choice(question, raw_choice)
|
||||
if error:
|
||||
return None, [], error
|
||||
return {"choice": choice.value}, [choice], None
|
||||
|
||||
|
||||
def validate_multiple_choice_answer(question, answer_payload):
|
||||
raw_choices = answer_payload.get("choices", answer_payload.get("value"))
|
||||
if raw_choices in (None, "") or raw_choices == []:
|
||||
if question.is_required:
|
||||
return None, [], "Select at least one choice."
|
||||
return None, [], None
|
||||
if not isinstance(raw_choices, list):
|
||||
return None, [], "Answer must be a list of choices."
|
||||
|
||||
selected_choices = []
|
||||
seen_choice_ids = set()
|
||||
for raw_choice in raw_choices:
|
||||
choice, error = resolve_choice(question, raw_choice)
|
||||
if error:
|
||||
return None, [], error
|
||||
if choice.pk not in seen_choice_ids:
|
||||
selected_choices.append(choice)
|
||||
seen_choice_ids.add(choice.pk)
|
||||
|
||||
if question.is_required and not selected_choices:
|
||||
return None, [], "Select at least one choice."
|
||||
|
||||
return {"choices": [choice.value for choice in selected_choices]}, selected_choices, None
|
||||
|
||||
|
||||
def validate_rating_answer(question, answer_payload):
|
||||
raw_value = answer_payload.get("value", answer_payload.get("rating"))
|
||||
if is_empty(raw_value):
|
||||
if question.is_required:
|
||||
return None, [], "Rating is required."
|
||||
return None, [], None
|
||||
if isinstance(raw_value, bool):
|
||||
return None, [], "Rating must be a number."
|
||||
try:
|
||||
rating = int(raw_value)
|
||||
except (TypeError, ValueError):
|
||||
return None, [], "Rating must be a number."
|
||||
|
||||
min_value = question.min_value if question.min_value is not None else 1
|
||||
max_value = question.max_value if question.max_value is not None else 5
|
||||
if rating < min_value or rating > max_value:
|
||||
return None, [], f"Rating must be between {min_value} and {max_value}."
|
||||
return {"rating": rating}, [], None
|
||||
|
||||
|
||||
def validate_yes_no_answer(question, answer_payload):
|
||||
raw_value = answer_payload.get("value", answer_payload.get("answer"))
|
||||
if is_empty(raw_value):
|
||||
if question.is_required:
|
||||
return None, [], "Answer yes or no."
|
||||
return None, [], None
|
||||
if isinstance(raw_value, bool):
|
||||
return {"answer": raw_value}, [], None
|
||||
if isinstance(raw_value, str):
|
||||
normalized_value = raw_value.strip().lower()
|
||||
if normalized_value in {"true", "yes", "1"}:
|
||||
return {"answer": True}, [], None
|
||||
if normalized_value in {"false", "no", "0"}:
|
||||
return {"answer": False}, [], None
|
||||
return None, [], "Answer must be true or false."
|
||||
|
||||
|
||||
def resolve_choice(question, raw_choice):
|
||||
choices = list(question.choices.all())
|
||||
if not choices:
|
||||
return None, "This question has no choices configured."
|
||||
|
||||
lookup = str(raw_choice)
|
||||
for choice in choices:
|
||||
if lookup in {str(choice.pk), choice.value}:
|
||||
return choice, None
|
||||
|
||||
allowed_values = ", ".join(choice.value for choice in choices)
|
||||
return None, f"Select one of: {allowed_values}."
|
||||
|
||||
|
||||
def is_empty(value):
|
||||
return value is None or (isinstance(value, str) and not value.strip())
|
||||
|
||||
|
||||
def get_client_ip(request):
|
||||
raw_ip = request.META.get("HTTP_X_FORWARDED_FOR", "").split(",")[0].strip()
|
||||
ip_address = raw_ip or request.META.get("REMOTE_ADDR")
|
||||
if not ip_address:
|
||||
return None
|
||||
try:
|
||||
validate_ipv46_address(ip_address)
|
||||
except (TypeError, ValidationError):
|
||||
return None
|
||||
return ip_address
|
||||
Reference in New Issue
Block a user