API endpoint for document audit log

src/documents/tests/test_api_documents.py

@@ -316,6 +316,29 @@ class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
         response = self.client.get(f"/api/documents/{doc.pk}/thumb/")
         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 
+    def test_document_audit_action(self):
+        doc = Document.objects.create(
+            title="First title",
+            checksum="123",
+            mime_type="application/pdf",
+        )
+        self.client.force_login(user=self.user)
+        self.client.patch(
+            f"/api/documents/{doc.pk}/",
+            {"title": "New title"},
+            format="json",
+        )
+
+        response = self.client.get(f"/api/documents/{doc.pk}/audit/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 2)
+        self.assertEqual(response.data[0]["actor"]["id"], self.user.id)
+        self.assertEqual(response.data[0]["action"], "update")
+        self.assertEqual(
+            response.data[0]["changes"],
+            {"title": ["First title", "New title"]},
+        )
+
     def test_document_filters(self):
         doc1 = Document.objects.create(
             title="none1",

src/documents/views.py

@@ -729,6 +729,39 @@ class DocumentViewSet(
             ]
             return Response(links)
 
+    @action(methods=["get"], detail=True, name="Audit Trail")
+    def audit(self, request, pk=None):
+        try:
+            doc = Document.objects.get(pk=pk)
+            if not request.user.has_perm("auditlog.view_logentry") or (
+                doc.owner is not None and doc.owner != request.user
+            ):
+                return HttpResponseForbidden(
+                    "Insufficient permissions",
+                )
+        except Document.DoesNotExist:
+            raise Http404
+
+        if request.method == "GET":
+            entries = [
+                {
+                    "id": entry.id,
+                    "timestamp": entry.timestamp,
+                    "action": entry.get_action_display(),
+                    "changes": json.loads(entry.changes),
+                    "remote_addr": entry.remote_addr,
+                    "actor": (
+                        {"id": entry.actor.id, "username": entry.actor.username}
+                        if entry.actor
+                        else None
+                    ),
+                }
+                for entry in LogEntry.objects.filter(object_pk=doc.pk).order_by(
+                    "-timestamp",
+                )
+            ]
+            return Response(entries)
+
 
 class SearchResultSerializer(DocumentSerializer, PassUserMixin):
     def to_representation(self, instance):