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/data/survey_config/sort_only_config_ahp.json b/data/survey_config/sort_only_config_ahp.json index aa041615..8b3f9615 100644 --- a/data/survey_config/sort_only_config_ahp.json +++ b/data/survey_config/sort_only_config_ahp.json @@ -17,7 +17,7 @@ "A3. Our organisation identifies and celebrates success in the research related activity", "A4. Our organisation provides research advice sessions where AHPs can explore ideas for project development", "A5. Our organisation provides help to AHPs to navigate research funding submissions, ethics and governance systems", - "A6. Our organisation has a finance department that can cost research project involvement for external funding applications", + "A6. Our organisation has a finance department that can cost research project involvement for external funding applications", "A7. Our organisation has an active research-related mentorship programme", "A8. Our organisation provides mentorship to AHPs to successfully apply for internships and fellowship opportunities", "A9. Our organisation enables the use of awarded grant funding in the manner intended (for example, protected time and spending decisions)", @@ -173,7 +173,7 @@ "sublabels": [ "E1. Our organisation provides training for AHPs to enable them to practise effectively in a digitally enabled environment", "E2. Our organisation trains AHPs to use and interpret data to make improvements to care (using audit, service evaluation or research)", - "E3. Our organisation has digital AHP leaders in place who can provide advice and guidance in the use of digital technology in service development", + "E3. Our organisation has digital AHP leaders in place who can provide advice and guidance in the use of digital technology in service development", "E4. Our organisation has digital AHP leaders in place who can provide advice and guidance in the use of digital technology in research", "E5. Our organisation has the infrastructure to support visualisation of data using business intelligence tools", "E6. Our organisation has the internal structures that facilitate, support and enable AHP-led digital innovation", diff --git a/data/survey_config/sort_only_config_midwives.json b/data/survey_config/sort_only_config_midwives.json index 9aa2a8f1..2a9c436d 100644 --- a/data/survey_config/sort_only_config_midwives.json +++ b/data/survey_config/sort_only_config_midwives.json @@ -17,7 +17,7 @@ "A3. Our organisation identifies and celebrates success in the midwifery research related activity", "A4. Our organisation provides research advice sessions where midwives can explore ideas for project development", "A5. Our organisation provides help to midwives to navigate research funding submissions, ethics and governance systems", - "A6. Our organisation has a finance department that can cost research project involvement for external funding applications", + "A6. Our organisation has a finance department that can cost research project involvement for external funding applications", "A7. Our organisation has an active research-related mentorship programme", "A8. Our organisation provides mentorship to midwives to successfully apply for internships and fellowship opportunities", "A9. Our organisation enables the use of awarded grant funding in the manner intended (for example, protected time and spending decisions)", 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..8b03e887 --- /dev/null +++ b/home/services/audit.py @@ -0,0 +1,60 @@ +"""Audit logging service for administrative actions""" + +from typing import 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 %} + + {% 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..32fab9c5 --- /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) + 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) + 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/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) 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