From 503faac4f406aad5490cf4b4a8a9decaf0b15e46 Mon Sep 17 00:00:00 2001 From: Joe Heffer Date: Thu, 12 Feb 2026 13:22:32 +0000 Subject: [PATCH 1/2] feat: Add draft admin dashboard design --- .github/workflows/django-check.yaml | 2 +- Makefile | 7 +- SORT/admin.py | 3 + SORT/test/model_factory/user/superuser.py | 1 + home/admin.py | 62 +++- home/migrations/0013_adminauditlog.py | 85 +++++ home/models.py | 50 +++ home/services/__init__.py | 2 + home/services/audit.py | 60 ++++ home/templates/admin/dashboard.html | 119 +++++++ home/templates/admin/delete_with_reason.html | 91 ++++++ home/templates/admin/index.html | 16 + home/tests/test_admin_audit.py | 116 +++++++ home/tests/test_admin_views.py | 249 +++++++++++++++ home/urls.py | 5 + home/views_admin.py | 309 +++++++++++++++++++ scripts/deploy.sh | 2 +- survey/admin.py | 10 +- 18 files changed, 1176 insertions(+), 13 deletions(-) create mode 100644 SORT/admin.py create mode 100644 home/migrations/0013_adminauditlog.py create mode 100644 home/services/audit.py create mode 100644 home/templates/admin/dashboard.html create mode 100644 home/templates/admin/delete_with_reason.html create mode 100644 home/templates/admin/index.html create mode 100644 home/tests/test_admin_audit.py create mode 100644 home/tests/test_admin_views.py create mode 100644 home/views_admin.py diff --git a/.github/workflows/django-check.yaml b/.github/workflows/django-check.yaml index 52116a99..a6d523ee 100644 --- a/.github/workflows/django-check.yaml +++ b/.github/workflows/django-check.yaml @@ -45,7 +45,7 @@ jobs: - name: Run Django system checks run: | export DJANGO_SECRET_KEY="$(python -c "import secrets; print(secrets.token_urlsafe())")" - python manage.py check + python manage.py check --fail-level WARNING # https://docs.djangoproject.com/en/5.1/topics/testing/ - name: Run Django test suites run: | diff --git a/Makefile b/Makefile index fd817977..a0b5e24b 100644 --- a/Makefile +++ b/Makefile @@ -12,6 +12,7 @@ help: @echo " make runserver - Start Django development server" @echo " make migrations - Create new database migrations" @echo " make migrate - Apply database migrations" + @echo " make check - Run Django system checks (including migration check)" @echo " make superuser - Create a superuser account" @echo " make static - Collect static files" @echo " make shell - Open Django shell" @@ -31,6 +32,10 @@ migrations: migrate: $(MANAGE) migrate +# System checks +check: + $(MANAGE) check --fail-level WARNING + # User management superuser: $(MANAGE) createsuperuser @@ -63,4 +68,4 @@ lint: .DEFAULT_GOAL := help # Mark these targets as always needing to run (not files) -.PHONY: help runserver migrations migrate superuser static shell test clean requirements lint \ No newline at end of file +.PHONY: help runserver migrations migrate check superuser static shell test clean requirements lint \ No newline at end of file diff --git a/SORT/admin.py b/SORT/admin.py new file mode 100644 index 00000000..a9af02fb --- /dev/null +++ b/SORT/admin.py @@ -0,0 +1,3 @@ +"""Customizations for Django admin site""" +# This module provides hooks for customizing the Django admin interface +# The actual customization happens through custom views and templates diff --git a/SORT/test/model_factory/user/superuser.py b/SORT/test/model_factory/user/superuser.py index f604f231..c296fe4b 100644 --- a/SORT/test/model_factory/user/superuser.py +++ b/SORT/test/model_factory/user/superuser.py @@ -5,5 +5,6 @@ class SuperUserFactory(UserFactory): is_superuser = True + is_staff = True first_name = factory.Sequence(lambda n: f"Superuser{n}") email = factory.Sequence(lambda n: f"superuser{n}@sort.com") diff --git a/home/admin.py b/home/admin.py index 7f19fad1..587f7c03 100644 --- a/home/admin.py +++ b/home/admin.py @@ -1,21 +1,52 @@ from django.contrib import admin -from .models import Organisation, OrganisationMembership, Project, User +from .models import AdminAuditLog, Organisation, OrganisationMembership, Project, User @admin.register(User) class UserAdmin(admin.ModelAdmin): - list_display = ("email", "first_name", "last_name", "is_staff", "date_joined") + list_display = ("email", "first_name", "last_name", "is_staff", "is_superuser", + "organisation_count", "project_count", "date_joined") search_fields = ("email", "first_name", "last_name") - list_filter = ("is_staff", "is_superuser", "is_active") + list_filter = ("is_staff", "is_superuser", "is_active", "date_joined") ordering = ("-date_joined",) + readonly_fields = ("date_joined", "last_login") + + def organisation_count(self, obj): + """Number of organisations user belongs to""" + return obj.organisation_set.count() + organisation_count.short_description = "Orgs" + + def project_count(self, obj): + """Number of projects user has access to""" + return sum(1 for _ in obj.projects_iter()) + project_count.short_description = "Projects" + + def get_readonly_fields(self, request, obj=None): + if obj: # Editing existing user + return self.readonly_fields + ("email",) + return self.readonly_fields @admin.register(Organisation) class OrganisationAdmin(admin.ModelAdmin): - list_display = ("pk", "name", "created_at") - search_fields = ("name", "description",) - ordering = ("name",) + list_display = ("pk", "name", "member_count", "project_count", "survey_count", "created_at") + search_fields = ("name", "description") + ordering = ("-created_at",) + readonly_fields = ("created_at",) + + def member_count(self, obj): + return obj.members.count() + member_count.short_description = "Members" + + def project_count(self, obj): + return obj.projects.count() + project_count.short_description = "Projects" + + def survey_count(self, obj): + from survey.models import Survey + return Survey.objects.filter(project__organisation=obj).count() + survey_count.short_description = "Surveys" @admin.register(OrganisationMembership) @@ -30,3 +61,22 @@ class ProjectAdmin(admin.ModelAdmin): list_display = ("pk", "name", "created_by", "created_at", "organisation") search_fields = ("name", "description",) list_filter = ("organisation",) + + +@admin.register(AdminAuditLog) +class AdminAuditLogAdmin(admin.ModelAdmin): + list_display = ("timestamp", "performed_by", "action_type", "target_model", "target_representation") + list_filter = ("action_type", "target_model", "timestamp") + search_fields = ("target_representation", "reason", "performed_by__email") + readonly_fields = ("performed_by", "action_type", "timestamp", "target_model", + "target_id", "target_representation", "reason", "metadata") + date_hierarchy = "timestamp" + + def has_add_permission(self, request): + return False # Logs created programmatically only + + def has_delete_permission(self, request, obj=None): + return False # Immutable + + def has_change_permission(self, request, obj=None): + return False # View only diff --git a/home/migrations/0013_adminauditlog.py b/home/migrations/0013_adminauditlog.py new file mode 100644 index 00000000..c565b801 --- /dev/null +++ b/home/migrations/0013_adminauditlog.py @@ -0,0 +1,85 @@ +# Generated by Django 5.1.15 on 2026-02-12 12:57 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("home", "0012_alter_project_name"), + ] + + operations = [ + migrations.CreateModel( + name="AdminAuditLog", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "action_type", + models.CharField( + choices=[ + ("DELETE_USER", "Delete User"), + ("DELETE_ORG", "Delete Organisation"), + ("DELETE_PROJECT", "Delete Project"), + ("DELETE_SURVEY", "Delete Survey"), + ("EXPORT_CONSENTED", "Export Consented Data"), + ("BULK_DELETE", "Bulk Delete"), + ], + max_length=30, + ), + ), + ("timestamp", models.DateTimeField(auto_now_add=True)), + ("target_model", models.CharField(max_length=50)), + ("target_id", models.IntegerField(null=True)), + ("target_representation", models.CharField(max_length=200)), + ( + "reason", + models.TextField( + help_text="Explanation for why this action was taken" + ), + ), + ( + "metadata", + models.JSONField( + blank=True, + help_text="Additional action details (e.g., cascade counts, export stats)", + null=True, + ), + ), + ( + "performed_by", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="admin_actions", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "ordering": ["-timestamp"], + "indexes": [ + models.Index( + fields=["-timestamp"], name="home_admina_timesta_4adb99_idx" + ), + models.Index( + fields=["action_type"], name="home_admina_action__f02f66_idx" + ), + models.Index( + fields=["target_model", "target_id"], + name="home_admina_target__b98dbf_idx", + ), + ], + }, + ), + ] diff --git a/home/models.py b/home/models.py index 2f554a9e..0c80f106 100644 --- a/home/models.py +++ b/home/models.py @@ -136,3 +136,53 @@ def is_active(self) -> bool: @property def surveys(self): return self.survey.all() + + +class AdminAuditLog(models.Model): + """ + Audit log for superuser administrative actions. + Tracks deletions, exports, and other sensitive operations. + """ + + class ActionType(models.TextChoices): + DELETE_USER = "DELETE_USER", "Delete User" + DELETE_ORGANISATION = "DELETE_ORG", "Delete Organisation" + DELETE_PROJECT = "DELETE_PROJECT", "Delete Project" + DELETE_SURVEY = "DELETE_SURVEY", "Delete Survey" + EXPORT_CONSENTED = "EXPORT_CONSENTED", "Export Consented Data" + BULK_DELETE = "BULK_DELETE", "Bulk Delete" + + performed_by = models.ForeignKey( + User, + on_delete=models.SET_NULL, + null=True, + related_name="admin_actions" + ) + action_type = models.CharField(max_length=30, choices=ActionType.choices) + timestamp = models.DateTimeField(auto_now_add=True) + + # Target object details + target_model = models.CharField(max_length=50) # "User", "Organisation", etc. + target_id = models.IntegerField(null=True) + target_representation = models.CharField(max_length=200) # str(obj) before deletion + + # Justification + reason = models.TextField(help_text="Explanation for why this action was taken") + + # Additional context (JSON) + metadata = models.JSONField( + null=True, + blank=True, + help_text="Additional action details (e.g., cascade counts, export stats)" + ) + + class Meta: + ordering = ["-timestamp"] + indexes = [ + models.Index(fields=["-timestamp"]), + models.Index(fields=["action_type"]), + models.Index(fields=["target_model", "target_id"]), + ] + + def __str__(self): + return f"{self.performed_by} - {self.action_type} - {self.timestamp}" diff --git a/home/services/__init__.py b/home/services/__init__.py index f64d339d..05e83a79 100644 --- a/home/services/__init__.py +++ b/home/services/__init__.py @@ -1,3 +1,4 @@ +from .audit import audit_service from .base import BasePermissionService from .organisation import OrganisationService, organisation_service from .project import ProjectService, project_service @@ -6,6 +7,7 @@ __all__ = [ + "audit_service", "BasePermissionService", "ProjectService", "OrganisationService", diff --git a/home/services/audit.py b/home/services/audit.py new file mode 100644 index 00000000..87bb59c2 --- /dev/null +++ b/home/services/audit.py @@ -0,0 +1,60 @@ +"""Audit logging service for administrative actions""" + +from typing import Any, Dict, Optional +from django.db import models +from ..models import AdminAuditLog, User + + +class AuditService: + """Service for creating audit log entries""" + + @staticmethod + def log_deletion( + user: User, + target: models.Model, + reason: str, + cascade_info: Optional[Dict[str, int]] = None + ) -> AdminAuditLog: + """Log a deletion action with cascade impact details""" + action_type_map = { + "User": AdminAuditLog.ActionType.DELETE_USER, + "Organisation": AdminAuditLog.ActionType.DELETE_ORGANISATION, + "Project": AdminAuditLog.ActionType.DELETE_PROJECT, + "Survey": AdminAuditLog.ActionType.DELETE_SURVEY, + } + + model_name = target.__class__.__name__ + action_type = action_type_map.get(model_name, AdminAuditLog.ActionType.BULK_DELETE) + + return AdminAuditLog.objects.create( + performed_by=user, + action_type=action_type, + target_model=model_name, + target_id=target.pk, + target_representation=str(target), + reason=reason, + metadata={ + "cascade_deletes": cascade_info or {}, + "object_pk": target.pk, + } + ) + + @staticmethod + def log_export(user: User, export_type: str, survey_count: int, response_count: int) -> AdminAuditLog: + """Log data export action""" + return AdminAuditLog.objects.create( + performed_by=user, + action_type=AdminAuditLog.ActionType.EXPORT_CONSENTED, + target_model="Survey", + target_id=None, + target_representation=f"{survey_count} surveys exported", + reason=f"Exported consented data for research ({export_type})", + metadata={ + "export_type": export_type, + "survey_count": survey_count, + "response_count": response_count, + } + ) + + +audit_service = AuditService() diff --git a/home/templates/admin/dashboard.html b/home/templates/admin/dashboard.html new file mode 100644 index 00000000..cf6ff094 --- /dev/null +++ b/home/templates/admin/dashboard.html @@ -0,0 +1,119 @@ +{% extends "admin/base_site.html" %} +{% load static %} + +{% block title %}Admin Dashboard - SORT{% endblock %} + +{% block breadcrumbs %} +
+ Home + › Dashboard +
+{% endblock %} + +{% block content %} +

SORT Admin Dashboard

+ +
+

Platform Statistics

+
+
+

{{ stats.total_users }}

+

Total Users

+
+
+

{{ stats.total_organisations }}

+

Organisations

+
+
+

{{ stats.total_projects }}

+

Projects

+
+
+

{{ stats.total_surveys }}

+

Surveys

+ {{ stats.active_surveys }} active +
+
+

{{ stats.consented_surveys }}

+

Consented Surveys

+ is_shared=True +
+
+

{{ stats.total_responses }}

+

Total Responses

+ {{ stats.consented_responses }} consented +
+
+
+ +
+

Export Consented Research Data

+
+

Export all survey data where participants have consented to share their responses for research purposes (is_shared=True).

+
+ Export CSV + Export Excel +
+

+ Available data: {{ stats.consented_surveys }} surveys with {{ stats.consented_responses }} responses +

+
+
+ +
+

Top Organisations by Survey Activity

+ + + + + + + + + {% for org in top_organisations %} + + + + + {% empty %} + + + + {% endfor %} + +
OrganisationSurvey Count
{{ org.name }}{{ org.survey_count }}
No organisations yet
+
+ +
+

Recent Admin Actions

+ + + + + + + + + + + {% for log in recent_audit_logs %} + + + + + + + {% empty %} + + + + {% endfor %} + +
TimestampAdminActionTarget
{{ log.timestamp|date:"Y-m-d H:i" }}{{ log.performed_by }}{{ log.get_action_type_display }}{{ log.target_representation }}
No audit logs yet
+ {% if recent_audit_logs %} +
+ View all audit logs » +
+ {% endif %} +
+{% endblock %} diff --git a/home/templates/admin/delete_with_reason.html b/home/templates/admin/delete_with_reason.html new file mode 100644 index 00000000..37baf36b --- /dev/null +++ b/home/templates/admin/delete_with_reason.html @@ -0,0 +1,91 @@ +{% extends "admin/base_site.html" %} +{% load static %} + +{% block title %}Delete {{ model_name|title }}(s) - SORT Admin{% endblock %} + +{% block breadcrumbs %} +
+ Home + › {{ model_name|title }}s + › Delete with audit log +
+{% endblock %} + +{% block content %} +
+

⚠️ Delete {{ model_name|title }}(s)

+

This action cannot be undone. All related data will also be deleted.

+
+ +
+

Objects to delete

+ +
+ +{% if cascade_info %} +
+

Cascade Deletion Impact

+
+ {% for obj in objects %} +

{{ obj }}

+ {% if cascade_info %} +
    + {% for key, impact in cascade_info.items %} + {% if key == obj.pk %} + {% for item_key, item_count in impact.items %} + {% if item_count > 0 %} +
  • {{ item_count }} {{ item_key }}
    • + {% endif %} + {% endfor %} + {% endif %} + {% endfor %} +
+ {% endif %} + {% endfor %} +

+ Warning: All of the above will be permanently deleted along with the selected {{ model_name }}(s). +

+
+
+{% endif %} + +
+

Audit Log Required

+
+ {% csrf_token %} + +
+ + +
+ +
+ +
+ +
+ + + Cancel + +
+
+
+{% endblock %} diff --git a/home/templates/admin/index.html b/home/templates/admin/index.html new file mode 100644 index 00000000..ff33c60b --- /dev/null +++ b/home/templates/admin/index.html @@ -0,0 +1,16 @@ +{% extends "admin/index.html" %} +{% load i18n static %} + +{% block content %} +{% if user.is_superuser %} +
+

Admin Portal

+

Access enhanced admin features for managing the SORT platform:

+ + Go to Admin Dashboard → + +
+{% endif %} + +{{ block.super }} +{% endblock %} diff --git a/home/tests/test_admin_audit.py b/home/tests/test_admin_audit.py new file mode 100644 index 00000000..a0ac5fdf --- /dev/null +++ b/home/tests/test_admin_audit.py @@ -0,0 +1,116 @@ +"""Tests for admin audit logging service""" +from django.test import TestCase + +from home.models import AdminAuditLog +from home.services import audit_service +from SORT.test.model_factory import SuperUserFactory, OrganisationFactory, UserFactory + + +class AuditServiceTestCase(TestCase): + """Test cases for audit logging service""" + + def setUp(self): + self.superuser = SuperUserFactory() + self.organisation = OrganisationFactory() + self.user = UserFactory() + + def test_log_deletion_organisation(self): + """Test audit log creation for organisation deletion""" + log = audit_service.log_deletion( + user=self.superuser, + target=self.organisation, + reason="Test deletion - spurious organisation", + cascade_info={"projects": 3, "surveys": 5} + ) + + self.assertEqual(log.action_type, AdminAuditLog.ActionType.DELETE_ORGANISATION) + self.assertEqual(log.performed_by, self.superuser) + self.assertEqual(log.target_model, "Organisation") + self.assertEqual(log.target_id, self.organisation.pk) + self.assertEqual(log.reason, "Test deletion - spurious organisation") + self.assertEqual(log.metadata["cascade_deletes"]["projects"], 3) + self.assertEqual(log.metadata["cascade_deletes"]["surveys"], 5) + + def test_log_deletion_user(self): + """Test audit log creation for user deletion""" + log = audit_service.log_deletion( + user=self.superuser, + target=self.user, + reason="User requested account deletion", + cascade_info={"organisations": 1} + ) + + self.assertEqual(log.action_type, AdminAuditLog.ActionType.DELETE_USER) + self.assertEqual(log.performed_by, self.superuser) + self.assertEqual(log.target_model, "User") + self.assertEqual(log.target_representation, str(self.user)) + + def test_log_export(self): + """Test audit log creation for data export""" + log = audit_service.log_export( + user=self.superuser, + export_type="CSV", + survey_count=10, + response_count=50 + ) + + self.assertEqual(log.action_type, AdminAuditLog.ActionType.EXPORT_CONSENTED) + self.assertEqual(log.performed_by, self.superuser) + self.assertEqual(log.target_model, "Survey") + self.assertIn("10 surveys exported", log.target_representation) + self.assertEqual(log.metadata["export_type"], "CSV") + self.assertEqual(log.metadata["survey_count"], 10) + self.assertEqual(log.metadata["response_count"], 50) + + def test_audit_log_ordering(self): + """Test that audit logs are ordered by timestamp descending""" + log1 = audit_service.log_deletion( + user=self.superuser, + target=self.organisation, + reason="First deletion" + ) + log2 = audit_service.log_deletion( + user=self.superuser, + target=self.user, + reason="Second deletion" + ) + + logs = AdminAuditLog.objects.all() + self.assertEqual(logs[0].pk, log2.pk) # Most recent first + self.assertEqual(logs[1].pk, log1.pk) + + +class AdminAuditLogAdminTestCase(TestCase): + """Test cases for AdminAuditLog admin interface""" + + def setUp(self): + self.superuser = SuperUserFactory() + self.organisation = OrganisationFactory() + + def test_audit_log_immutable(self): + """Verify audit logs cannot be modified via admin""" + from home.admin import AdminAuditLogAdmin + from django.contrib.admin.sites import site + + admin_instance = AdminAuditLogAdmin(AdminAuditLog, site) + + # Cannot add new logs via admin + self.assertFalse(admin_instance.has_add_permission(None)) + + # Cannot delete logs + self.assertFalse(admin_instance.has_delete_permission(None)) + + # Cannot edit logs + self.assertFalse(admin_instance.has_change_permission(None)) + + def test_audit_log_str_representation(self): + """Test string representation of audit log""" + log = audit_service.log_deletion( + user=self.superuser, + target=self.organisation, + reason="Test" + ) + + log_str = str(log) + self.assertIn(str(self.superuser), log_str) + self.assertIn("DELETE_ORG", log_str) diff --git a/home/tests/test_admin_views.py b/home/tests/test_admin_views.py new file mode 100644 index 00000000..be67cd7b --- /dev/null +++ b/home/tests/test_admin_views.py @@ -0,0 +1,249 @@ +"""Tests for custom admin views""" +from django.test import TestCase, Client +from django.urls import reverse + +from home.models import AdminAuditLog +from SORT.test.model_factory import ( + SuperUserFactory, + UserFactory, + OrganisationFactory, + ProjectFactory, + SurveyFactory, +) + + +class AdminDashboardTestCase(TestCase): + """Test cases for admin dashboard view""" + + def setUp(self): + self.client = Client() + self.superuser = SuperUserFactory() + self.normal_user = UserFactory() + + def test_dashboard_access_superuser(self): + """Superuser can access admin dashboard""" + self.client.force_login(self.superuser) + response = self.client.get(reverse("admin_dashboard")) + + self.assertEqual(response.status_code, 200) + self.assertContains(response, "Admin Dashboard") + self.assertContains(response, "Platform Statistics") + + def test_dashboard_denied_normal_user(self): + """Normal users denied access to dashboard""" + self.client.force_login(self.normal_user) + response = self.client.get(reverse("admin_dashboard")) + + # @staff_member_required redirects non-staff users + self.assertEqual(response.status_code, 302) + self.assertIn("/admin/", response.url) + + def test_dashboard_denied_anonymous(self): + """Anonymous users redirected to login""" + response = self.client.get(reverse("admin_dashboard")) + + self.assertEqual(response.status_code, 302) + self.assertIn("/login/", response.url) + + def test_dashboard_statistics(self): + """Dashboard displays correct statistics""" + # Create test data + org = OrganisationFactory() + project = ProjectFactory(organisation=org) + survey = SurveyFactory(project=project, is_shared=True) + + self.client.force_login(self.superuser) + response = self.client.get(reverse("admin_dashboard")) + + self.assertIn("stats", response.context) + stats = response.context["stats"] + + # Verify key stats are present and reasonable + self.assertGreaterEqual(stats["total_organisations"], 1) + self.assertGreaterEqual(stats["total_projects"], 1) + self.assertGreaterEqual(stats["total_surveys"], 1) + self.assertGreaterEqual(stats["consented_surveys"], 1) + + +class ConsentedDataExportTestCase(TestCase): + """Test cases for consented data export functionality""" + + def setUp(self): + self.client = Client() + self.superuser = SuperUserFactory() + self.normal_user = UserFactory() + + # Create test surveys + org = OrganisationFactory() + project = ProjectFactory(organisation=org) + self.consented_survey = SurveyFactory(project=project, is_shared=True) + self.non_consented_survey = SurveyFactory(project=project, is_shared=False) + + def test_export_csv_superuser_only(self): + """Only superusers can export consented data""" + self.client.force_login(self.normal_user) + response = self.client.get(reverse("admin_export_consented_data") + "?format=csv") + + # @staff_member_required redirects non-staff users + self.assertEqual(response.status_code, 302) + + def test_export_csv_only_consented(self): + """Export includes only consented surveys""" + self.client.force_login(self.superuser) + response = self.client.get(reverse("admin_export_consented_data") + "?format=csv") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response["Content-Type"], "text/csv") + self.assertIn("attachment", response["Content-Disposition"]) + self.assertIn(".csv", response["Content-Disposition"]) + + # Verify audit log created + audit_log = AdminAuditLog.objects.filter( + action_type=AdminAuditLog.ActionType.EXPORT_CONSENTED + ).first() + self.assertIsNotNone(audit_log) + self.assertEqual(audit_log.performed_by, self.superuser) + self.assertEqual(audit_log.metadata["export_type"], "CSV") + + def test_export_excel_format(self): + """Export in Excel format works""" + self.client.force_login(self.superuser) + response = self.client.get(reverse("admin_export_consented_data") + "?format=excel") + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response["Content-Type"], + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ) + self.assertIn(".xlsx", response["Content-Disposition"]) + + def test_export_no_consented_surveys(self): + """Export handles case with no consented surveys""" + # Mark all surveys as non-consented + self.consented_survey.is_shared = False + self.consented_survey.save() + + self.client.force_login(self.superuser) + response = self.client.get(reverse("admin_export_consented_data") + "?format=csv") + + # Should redirect with warning message + self.assertEqual(response.status_code, 302) + self.assertIn("/admin/survey/survey/", response.url) + + def test_export_invalid_format(self): + """Export handles invalid format parameter""" + self.client.force_login(self.superuser) + response = self.client.get(reverse("admin_export_consented_data") + "?format=invalid") + + self.assertEqual(response.status_code, 302) + + +class DeleteWithReasonTestCase(TestCase): + """Test cases for safe deletion with audit logging""" + + def setUp(self): + self.client = Client() + self.superuser = SuperUserFactory() + self.normal_user = UserFactory() + self.org = OrganisationFactory() + + def test_delete_view_access_superuser_only(self): + """Only superusers can access delete with reason view""" + self.client.force_login(self.normal_user) + response = self.client.get( + reverse("admin_delete_with_reason", kwargs={"model_name": "organisation"}) + + f"?ids={self.org.pk}" + ) + + # @staff_member_required redirects non-staff users + self.assertEqual(response.status_code, 302) + + def test_delete_view_get_shows_confirmation(self): + """GET request shows confirmation page""" + self.client.force_login(self.superuser) + response = self.client.get( + reverse("admin_delete_with_reason", kwargs={"model_name": "organisation"}) + + f"?ids={self.org.pk}" + ) + + self.assertEqual(response.status_code, 200) + self.assertContains(response, "Delete Organisation") + self.assertContains(response, str(self.org)) + self.assertContains(response, "Reason for deletion") + + def test_delete_view_post_requires_reason(self): + """POST without reason shows error""" + self.client.force_login(self.superuser) + response = self.client.post( + reverse("admin_delete_with_reason", kwargs={"model_name": "organisation"}) + + f"?ids={self.org.pk}", + data={"reason": ""} + ) + + self.assertEqual(response.status_code, 200) + self.assertContains(response, "Reason is required") + + def test_delete_view_post_with_reason_deletes(self): + """POST with reason deletes and creates audit log""" + org_pk = self.org.pk + org_name = str(self.org) + + self.client.force_login(self.superuser) + response = self.client.post( + reverse("admin_delete_with_reason", kwargs={"model_name": "organisation"}) + + f"?ids={org_pk}", + data={"reason": "Test deletion - spurious organisation"} + ) + + # Should redirect to changelist + self.assertEqual(response.status_code, 302) + + # Organisation should be deleted + from home.models import Organisation + self.assertFalse(Organisation.objects.filter(pk=org_pk).exists()) + + # Audit log should exist + audit_log = AdminAuditLog.objects.filter( + action_type=AdminAuditLog.ActionType.DELETE_ORGANISATION, + target_id=org_pk + ).first() + self.assertIsNotNone(audit_log) + self.assertEqual(audit_log.performed_by, self.superuser) + self.assertEqual(audit_log.reason, "Test deletion - spurious organisation") + self.assertEqual(audit_log.target_representation, org_name) + + def test_delete_view_no_ids_parameter(self): + """View handles missing ids parameter""" + self.client.force_login(self.superuser) + response = self.client.get( + reverse("admin_delete_with_reason", kwargs={"model_name": "organisation"}) + ) + + self.assertEqual(response.status_code, 302) + + def test_delete_view_invalid_model_name(self): + """View handles invalid model name""" + self.client.force_login(self.superuser) + response = self.client.get( + reverse("admin_delete_with_reason", kwargs={"model_name": "invalid"}) + + "?ids=1" + ) + + self.assertEqual(response.status_code, 302) + + def test_delete_cascade_impact_calculated(self): + """Cascade impact is calculated and displayed""" + # Create related objects + project = ProjectFactory(organisation=self.org) + survey = SurveyFactory(project=project) + + self.client.force_login(self.superuser) + response = self.client.get( + reverse("admin_delete_with_reason", kwargs={"model_name": "organisation"}) + + f"?ids={self.org.pk}" + ) + + self.assertEqual(response.status_code, 200) + self.assertContains(response, "Cascade Deletion Impact") + # Should show project and survey counts + self.assertIn("cascade_info", response.context) diff --git a/home/urls.py b/home/urls.py index 25a3b6aa..96e5d0ea 100644 --- a/home/urls.py +++ b/home/urls.py @@ -4,6 +4,7 @@ from django.urls import path, re_path from . import views +from . import views_admin urlpatterns = [ path("", views.LandingView.as_view(), name="landing"), @@ -119,4 +120,8 @@ views.ParticipantInformationView.as_view(), name="participant_information", ), + # Admin portal URLs (superuser only) + path("admin-portal/dashboard/", views_admin.admin_dashboard, name="admin_dashboard"), + path("admin-portal/export-consented/", views_admin.export_consented_data, name="admin_export_consented_data"), + path("admin-portal/delete//", views_admin.delete_with_reason_view, name="admin_delete_with_reason"), ] diff --git a/home/views_admin.py b/home/views_admin.py new file mode 100644 index 00000000..fdb03377 --- /dev/null +++ b/home/views_admin.py @@ -0,0 +1,309 @@ +"""Custom admin views for superuser portal""" + +import csv +import io +from django.contrib import messages +from django.contrib.admin.views.decorators import staff_member_required +from django.core.exceptions import PermissionDenied +from django.db.models import Count +from django.http import HttpResponse +from django.shortcuts import redirect, render +from django.utils import timezone +from datetime import timedelta + +from home.models import AdminAuditLog, Organisation, Project, User +from survey.models import Survey, SurveyResponse + + +@staff_member_required +def admin_dashboard(request): + """Custom admin dashboard with platform statistics""" + if not request.user.is_superuser: + raise PermissionDenied("Only superusers can access this dashboard") + + now = timezone.now() + last_30_days = now - timedelta(days=30) + + stats = { + "total_users": User.objects.filter(is_superuser=False).count(), + "total_organisations": Organisation.objects.count(), + "total_projects": Project.objects.count(), + "total_surveys": Survey.objects.count(), + "total_responses": SurveyResponse.objects.count(), + "consented_surveys": Survey.objects.filter(is_shared=True).count(), + "consented_responses": SurveyResponse.objects.filter(survey__is_shared=True).count(), + "new_surveys_30d": Survey.objects.filter(created_at__gte=last_30_days).count(), + "active_surveys": Survey.objects.filter(is_active=True).count(), + } + + top_organisations = Organisation.objects.annotate( + survey_count=Count("projects__survey") + ).order_by("-survey_count")[:10] + + recent_audit_logs = AdminAuditLog.objects.select_related("performed_by")[:20] + + context = { + "stats": stats, + "top_organisations": top_organisations, + "recent_audit_logs": recent_audit_logs, + } + + return render(request, "admin/dashboard.html", context) + + +@staff_member_required +def export_consented_data(request): + """Export all consented survey data (is_shared=True)""" + if not request.user.is_superuser: + raise PermissionDenied() + + export_format = request.GET.get("format", "csv") + consented_surveys = Survey.objects.filter(is_shared=True).select_related("project__organisation") + + if not consented_surveys.exists(): + messages.warning(request, "No consented surveys available for export") + return redirect("admin:survey_survey_changelist") + + total_responses = sum(survey.responses_count for survey in consented_surveys) + + if export_format == "csv": + csv_data = _generate_consented_csv(consented_surveys) + + # Log export + from home.services import audit_service + audit_service.log_export( + user=request.user, + export_type="CSV", + survey_count=consented_surveys.count(), + response_count=total_responses + ) + + response = HttpResponse(csv_data, content_type="text/csv") + filename = f"sort_consented_data_{timezone.now().strftime('%Y%m%d_%H%M%S')}.csv" + response["Content-Disposition"] = f'attachment; filename="{filename}"' + return response + + elif export_format == "excel": + excel_data = _generate_consented_excel(consented_surveys) + + # Log export + from home.services import audit_service + audit_service.log_export( + user=request.user, + export_type="Excel", + survey_count=consented_surveys.count(), + response_count=total_responses + ) + + response = HttpResponse( + excel_data, + content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ) + filename = f"sort_consented_data_{timezone.now().strftime('%Y%m%d_%H%M%S')}.xlsx" + response["Content-Disposition"] = f'attachment; filename="{filename}"' + return response + + messages.error(request, f"Unsupported format: {export_format}") + return redirect("admin:survey_survey_changelist") + + +def _generate_consented_csv(surveys): + """Generate combined CSV using existing Survey.to_csv() pattern""" + buffer = io.StringIO() + + # Metadata columns + survey fields + metadata_fields = ["survey_id", "survey_name", "organisation", "project", "response_created_at"] + all_fields = set() + for survey in surveys: + # Skip surveys without valid config + if survey.survey_config is None: + continue + try: + all_fields.update(survey.fields) + except (KeyError, TypeError, AttributeError): + # Survey config is malformed, skip + continue + + fieldnames = metadata_fields + sorted(all_fields) + writer = csv.DictWriter(buffer, fieldnames=fieldnames) + writer.writeheader() + + for survey in surveys: + # Skip surveys without valid config + if survey.survey_config is None: + continue + + for response in survey.survey_response.all(): + row_data = { + "survey_id": survey.pk, + "survey_name": survey.name, + "organisation": survey.organisation.name, + "project": survey.project.name, + "response_created_at": response.created_at.isoformat(), + } + try: + row_data.update(dict(zip(survey.fields, response.answers_values))) + except (KeyError, TypeError, AttributeError): + # Skip malformed responses + continue + writer.writerow(row_data) + + return buffer.getvalue() + + +def _generate_consented_excel(surveys): + """Generate combined Excel file using existing Survey.to_excel() pattern""" + from openpyxl import Workbook + from openpyxl.styles import Font + + wb = Workbook() + ws = wb.active + ws.title = "Consented Survey Data" + + # Metadata columns + survey fields + metadata_fields = ["survey_id", "survey_name", "organisation", "project", "response_created_at"] + all_fields = set() + for survey in surveys: + # Skip surveys without valid config + if survey.survey_config is None: + continue + try: + all_fields.update(survey.fields) + except (KeyError, TypeError, AttributeError): + continue + + headers = metadata_fields + sorted(all_fields) + + # Write headers with bold font + for col_num, header in enumerate(headers, 1): + cell = ws.cell(row=1, column=col_num, value=header) + cell.font = Font(bold=True) + + # Write data rows + row_num = 2 + for survey in surveys: + # Skip surveys without valid config + if survey.survey_config is None: + continue + + for response in survey.survey_response.all(): + row_data = { + "survey_id": survey.pk, + "survey_name": survey.name, + "organisation": survey.organisation.name, + "project": survey.project.name, + "response_created_at": response.created_at.isoformat(), + } + try: + row_data.update(dict(zip(survey.fields, response.answers_values))) + except (KeyError, TypeError, AttributeError): + continue + + for col_num, header in enumerate(headers, 1): + ws.cell(row=row_num, column=col_num, value=row_data.get(header, "")) + + row_num += 1 + + # Save to buffer + buffer = io.BytesIO() + wb.save(buffer) + buffer.seek(0) + return buffer.getvalue() + + +@staff_member_required +def delete_with_reason_view(request, model_name): + """Intermediate view for deleting objects with audit logging""" + if not request.user.is_superuser: + raise PermissionDenied() + + ids_str = request.GET.get("ids", "") + if not ids_str: + messages.error(request, "No objects selected") + return redirect("admin:index") + + ids = [int(id_) for id_ in ids_str.split(",")] + + model_map = { + "user": User, + "organisation": Organisation, + "project": Project, + "survey": Survey, + } + + model = model_map.get(model_name) + if not model: + messages.error(request, f"Unknown model: {model_name}") + return redirect("admin:index") + + objects = model.objects.filter(pk__in=ids) + + if request.method == "POST": + reason = request.POST.get("reason", "").strip() + + if not reason: + messages.error(request, "Reason is required for deletion") + return render(request, "admin/delete_with_reason.html", { + "objects": objects, + "model_name": model_name, + }) + + cascade_info = _calculate_cascade_impact(objects, model) + + # Log each deletion + from home.services import audit_service + for obj in objects: + audit_service.log_deletion( + user=request.user, + target=obj, + reason=reason, + cascade_info=cascade_info.get(obj.pk, {}) + ) + + count = objects.count() + objects.delete() + + messages.success(request, f"Successfully deleted {count} {model_name}(s)") + return redirect(f"admin:home_{model_name}_changelist") + + # GET - show confirmation + cascade_info = _calculate_cascade_impact(objects, model) + context = { + "objects": objects, + "model_name": model_name, + "cascade_info": cascade_info, + } + + return render(request, "admin/delete_with_reason.html", context) + + +def _calculate_cascade_impact(objects, model): + """Calculate what will be cascade deleted""" + cascade_map = {} + + if model == Organisation: + for org in objects: + cascade_map[org.pk] = { + "projects": org.projects.count(), + "surveys": Survey.objects.filter(project__organisation=org).count(), + "responses": SurveyResponse.objects.filter(survey__project__organisation=org).count(), + } + elif model == Project: + for project in objects: + cascade_map[project.pk] = { + "surveys": project.survey.count(), + "responses": SurveyResponse.objects.filter(survey__project=project).count(), + } + elif model == Survey: + for survey in objects: + cascade_map[survey.pk] = { + "responses": survey.survey_response.count(), + } + elif model == User: + for user in objects: + cascade_map[user.pk] = { + "organisations": user.organisation_set.count(), + "projects_created": Project.objects.filter(created_by=user).count(), + } + + return cascade_map diff --git a/scripts/deploy.sh b/scripts/deploy.sh index d769b69a..6da26058 100644 --- a/scripts/deploy.sh +++ b/scripts/deploy.sh @@ -228,7 +228,7 @@ echo "Database connection successful" echo "Checking Django system..." # https://docs.djangoproject.com/en/5.1/howto/deployment/checklist/ # shellcheck source=/opt/sort/.env -(cd "$sort_dir" && set -a && source "$env_file" && set +a && exec $python manage.py check --deploy) +(cd "$sort_dir" && set -a && source "$env_file" && set +a && exec $python manage.py check --deploy --fail-level WARNING) # Migrate database changes # https://docs.djangoproject.com/en/5.1/topics/migrations/ diff --git a/survey/admin.py b/survey/admin.py index 4e2c64e2..d06b9422 100644 --- a/survey/admin.py +++ b/survey/admin.py @@ -19,10 +19,12 @@ class InvitationAdmin(admin.ModelAdmin): @admin.register(Survey) class SurveyAdmin(admin.ModelAdmin): - list_display = ("pk", "name", "created_at", "project", "project__organisation", "responses_count") - search_fields = ("name", "description",) - ordering = ("created_at",) - list_filter = ("project", "project__organisation",) + list_display = ("pk", "name", "is_shared", "is_active", "responses_count", + "created_at", "project", "project__organisation") + search_fields = ("name", "description", "project__name", "project__organisation__name") + ordering = ("-created_at",) + list_filter = ("is_shared", "is_active", "survey_body_path", "project__organisation", "created_at") + readonly_fields = ("created_at", "survey_config") @admin.register(SurveyResponse) From 2562903400d81a8fc8cd79fad63d1e2c9713cff9 Mon Sep 17 00:00:00 2001 From: Joe Heffer Date: Thu, 12 Feb 2026 13:35:15 +0000 Subject: [PATCH 2/2] style: Apply code formatting --- home/services/audit.py | 2 +- home/tests/test_admin_views.py | 4 ++-- .../migrations/0021_remove_survey_consent_config_and_more.py | 3 +-- survey/migrations/0022_survey_is_shared.py | 1 - 4 files changed, 4 insertions(+), 6 deletions(-) diff --git a/home/services/audit.py b/home/services/audit.py index 87bb59c2..8b03e887 100644 --- a/home/services/audit.py +++ b/home/services/audit.py @@ -1,6 +1,6 @@ """Audit logging service for administrative actions""" -from typing import Any, Dict, Optional +from typing import Dict, Optional from django.db import models from ..models import AdminAuditLog, User diff --git a/home/tests/test_admin_views.py b/home/tests/test_admin_views.py index be67cd7b..32fab9c5 100644 --- a/home/tests/test_admin_views.py +++ b/home/tests/test_admin_views.py @@ -50,7 +50,7 @@ def test_dashboard_statistics(self): # Create test data org = OrganisationFactory() project = ProjectFactory(organisation=org) - survey = SurveyFactory(project=project, is_shared=True) + SurveyFactory(project=project, is_shared=True) self.client.force_login(self.superuser) response = self.client.get(reverse("admin_dashboard")) @@ -235,7 +235,7 @@ def test_delete_cascade_impact_calculated(self): """Cascade impact is calculated and displayed""" # Create related objects project = ProjectFactory(organisation=self.org) - survey = SurveyFactory(project=project) + SurveyFactory(project=project) self.client.force_login(self.superuser) response = self.client.get( diff --git a/survey/migrations/0021_remove_survey_consent_config_and_more.py b/survey/migrations/0021_remove_survey_consent_config_and_more.py index bee0274e..04873f85 100644 --- a/survey/migrations/0021_remove_survey_consent_config_and_more.py +++ b/survey/migrations/0021_remove_survey_consent_config_and_more.py @@ -1,7 +1,6 @@ # Generated by Django 5.1.4 on 2025-09-19 08:19 -import django.db.models.deletion -from django.db import migrations, models +from django.db import migrations class Migration(migrations.Migration): diff --git a/survey/migrations/0022_survey_is_shared.py b/survey/migrations/0022_survey_is_shared.py index 81fa77c0..67d175fc 100644 --- a/survey/migrations/0022_survey_is_shared.py +++ b/survey/migrations/0022_survey_is_shared.py @@ -1,6 +1,5 @@ # Generated by Django 5.1.4 on 2025-11-07 15:40 -import django.db.models.deletion from django.db import migrations, models