feat(projectHistoryLogs): add endpoints for viewing project history logs TASK-973

kobo/apps/audit_log/permissions.py

@@ -1,12 +1,22 @@
 from rest_framework.permissions import IsAdminUser
 
-from kpi.mixins.validation_password_permission import (
-    ValidationPasswordPermissionMixin,
-)
+from kpi.constants import PERM_MANAGE_ASSET
+from kpi.mixins.validation_password_permission import ValidationPasswordPermissionMixin
+from kpi.permissions import IsAuthenticated
 
 
 class SuperUserPermission(ValidationPasswordPermissionMixin, IsAdminUser):
 
     def has_permission(self, request, view):
         self.validate_password(request)
         return bool(request.user and request.user.is_superuser)
+
+
+class ViewProjectHistoryLogsPermission(IsAuthenticated):
+
+    def has_permission(self, request, view):
+        has_asset_perm = bool(
+            request.user
+            and view.asset.has_perm(user_obj=request.user, perm=PERM_MANAGE_ASSET)
+        )
+        return has_asset_perm and super().has_permission(request, view)

kobo/apps/audit_log/serializers.py

@@ -2,7 +2,7 @@
 from rest_framework import serializers
 
 from kpi.fields import RelativePrefixHyperlinkedRelatedField
-from .models import AuditLog
+from .models import AuditLog, ProjectHistoryLog
 
 
 class AuditLogSerializer(serializers.ModelSerializer):
@@ -64,3 +64,39 @@ class AccessLogSerializer(serializers.Serializer):
 
     def get_date_created(self, audit_log):
         return audit_log['date_created'].strftime('%Y-%m-%dT%H:%M:%SZ')
+
+
+class ProjectHistoryLogSerializer(serializers.ModelSerializer):
+    user = serializers.HyperlinkedRelatedField(
+        queryset=get_user_model().objects.all(),
+        lookup_field='username',
+        view_name='user-kpi-detail',
+    )
+    date_created = serializers.SerializerMethodField()
+    username = serializers.SerializerMethodField()
+
+    class Meta:
+        model = ProjectHistoryLog
+        fields = (
+            'user',
+            'user_uid',
+            'username',
+            'action',
+            'metadata',
+            'date_created',
+        )
+
+        read_only_fields = (
+            'user',
+            'user_uid',
+            'username',
+            'action',
+            'metadata',
+            'date_created',
+        )
+
+    def get_date_created(self, audit_log):
+        return audit_log.date_created.strftime('%Y-%m-%dT%H:%M:%SZ')
+
+    def get_username(self, audit_log):
+        return audit_log.user.username

kobo/apps/audit_log/tests/api/v2/test_api_audit_log.py

@@ -6,13 +6,17 @@
 from rest_framework.reverse import reverse
 
 from kobo.apps.audit_log.audit_actions import AuditAction
-from kobo.apps.audit_log.models import AccessLog, AuditLog, AuditType
+from kobo.apps.audit_log.models import AccessLog, AuditLog, AuditType, ProjectHistoryLog
 from kobo.apps.audit_log.tests.test_signals import skip_login_access_log
 from kobo.apps.kobo_auth.shortcuts import User
 from kpi.constants import (
     ACCESS_LOG_SUBMISSION_AUTH_TYPE,
     ACCESS_LOG_SUBMISSION_GROUP_AUTH_TYPE,
+    PERM_MANAGE_ASSET,
+    PROJECT_HISTORY_LOG_PERMISSION_SUBTYPE,
+    PROJECT_HISTORY_LOG_PROJECT_SUBTYPE,
 )
+from kpi.models import Asset
 from kpi.models.import_export_task import AccessLogExportTask
 from kpi.tests.base_test_case import BaseTestCase
 from kpi.urls.router_api_v2 import URL_NAMESPACE as ROUTER_URL_NAMESPACE
@@ -41,6 +45,124 @@ def force_login_user(self, user):
             self.client.force_login(user)
 
 
+class ProjectHistoryLogTestCaseMixin:
+    """
+    Common tests for /project-history-logs and asset/<uid>/history
+    """
+
+    def test_results_have_expected_fields(self):
+        now = timezone.now()
+        metadata_dict = {
+            'asset_uid': self.asset.uid,
+            'ip_address': '1.2.3.4',
+            'source': 'source',
+            'log_subtype': 'project',
+            'some': 'thing',
+        }
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=self.asset.id,
+            action=AuditAction.DELETE,
+            metadata=metadata_dict,
+            date_created=now,
+        )
+        response = self.client.get(self.url)
+        self.assertEqual(response.data['count'], 1)
+        ph_log = response.data['results'][0]
+        self.assertListEqual(
+            sorted(list(ph_log.keys())),
+            ['action', 'date_created', 'metadata', 'user', 'user_uid', 'username'],
+        )
+        self.assertEqual(ph_log['action'], AuditAction.DELETE),
+        self.assertEqual(ph_log['date_created'], now.strftime('%Y-%m-%dT%H:%M:%SZ'))
+        self.assertEqual(
+            ph_log['user'],
+            reverse(
+                'api_v2:user-kpi-detail',
+                kwargs={'username': self.user.username},
+                request=response.wsgi_request,
+            ),
+        )
+        self.assertEqual(ph_log['user_uid'], self.user.extra_details.uid)
+        self.assertEqual(ph_log['username'], self.user.username)
+        self.assertDictEqual(ph_log['metadata'], metadata_dict)
+
+    def test_results_are_sorted_by_date_descending(self):
+        now = timezone.now()
+        yesterday = now - timedelta(days=1)
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=self.asset.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': self.asset.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': 'project',
+            },
+            date_created=yesterday,
+        )
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=self.asset.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': self.asset.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': 'project',
+            },
+            date_created=now,
+        )
+        response = self.client.get(self.url)
+        self.assertEqual(response.data['count'], 2)
+        self.assertEqual(
+            response.data['results'][0]['date_created'],
+            now.strftime('%Y-%m-%dT%H:%M:%SZ'),
+        )
+        self.assertEqual(
+            response.data['results'][1]['date_created'],
+            yesterday.strftime('%Y-%m-%dT%H:%M:%SZ'),
+        )
+
+    def test_results_can_be_searched_by_subtype(self):
+        now = timezone.now()
+        yesterday = now - timedelta(days=1)
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=self.asset.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': self.asset.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': PROJECT_HISTORY_LOG_PROJECT_SUBTYPE,
+            },
+            date_created=now,
+        )
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=self.asset.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': self.asset.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': PROJECT_HISTORY_LOG_PERMISSION_SUBTYPE,
+            },
+            date_created=yesterday,
+        )
+        response = self.client.get(
+            f'{self.url}?q=metadata__log_subtype:'
+            f'{PROJECT_HISTORY_LOG_PERMISSION_SUBTYPE}'
+        )
+        self.assertEqual(response.data['count'], 1)
+        self.assertEqual(
+            response.data['results'][0]['metadata']['log_subtype'],
+            PROJECT_HISTORY_LOG_PERMISSION_SUBTYPE,
+        )
+
+
 class ApiAuditLogTestCase(BaseAuditLogTestCase):
 
     def get_endpoint_basename(self):
@@ -429,6 +551,109 @@ def test_can_search_access_logs_by_date_including_submission_groups(self):
         )
 
 
+class ApiProjectHistoryLogsTestCase(BaseTestCase, ProjectHistoryLogTestCaseMixin):
+
+    fixtures = ['test_data']
+
+    def setUp(self):
+        super().setUp()
+        self.asset = Asset.objects.get(pk=1)
+        self.url = reverse(
+            'api_v2:history-list', kwargs={'parent_lookup_asset': self.asset.uid}
+        )
+        self.user = User.objects.get(username='someuser')
+        self.asset.assign_perm(user_obj=self.user, perm=PERM_MANAGE_ASSET)
+        self.client.force_login(self.user)
+
+    def test_list_without_permissions_returns_forbidden(self):
+        user2 = User.objects.get(username='anotheruser')
+        self.client.force_login(user2)
+        response = self.client.get(self.url)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.asset.assign_perm(user_obj=user2, perm=PERM_MANAGE_ASSET)
+        response = self.client.get(self.url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+    def test_show_project_history_logs_filters_to_project(self):
+        asset2 = Asset.objects.get(pk=2)
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=self.asset.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': self.asset.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': 'project',
+            },
+        )
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=asset2.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': asset2.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': 'project',
+            },
+        )
+        response = self.client.get(self.url)
+        self.assertEqual(response.data['count'], 1)
+        self.assertEqual(
+            response.data['results'][0]['metadata']['asset_uid'], self.asset.uid
+        )
+
+
+class ApiAllProjectHistoryLogsTestCase(
+    BaseAuditLogTestCase, ProjectHistoryLogTestCaseMixin
+):
+
+    def get_endpoint_basename(self):
+        return 'all-project-history-logs-list'
+
+    def setUp(self):
+        super().setUp()
+        self.user = User.objects.get(username='admin')
+        self.asset = Asset.objects.get(pk=1)
+        self.force_login_user(self.user)
+
+    def test_show_all_project_history_logs(self):
+        asset1 = Asset.objects.get(pk=1)
+        asset2 = Asset.objects.get(pk=2)
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=asset1.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': asset1.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': 'project',
+            },
+        )
+        ProjectHistoryLog.objects.create(
+            user=self.user,
+            object_id=asset2.id,
+            action=AuditAction.DELETE,
+            metadata={
+                'asset_uid': asset2.uid,
+                'ip_address': '1.2.3.4',
+                'source': 'source',
+                'log_subtype': 'project',
+            },
+        )
+        response = self.client.get(self.url)
+        self.assertEqual(response.data['count'], 2)
+        self.assertEqual(
+            response.data['results'][0]['metadata']['asset_uid'], asset2.uid
+        )
+        self.assertEqual(
+            response.data['results'][1]['metadata']['asset_uid'], asset1.uid
+
+        )
+
+
 class ApiAccessLogsExportTestCase(BaseAuditLogTestCase):
 
     def get_endpoint_basename(self):

kobo/apps/audit_log/urls.py

@@ -5,13 +5,20 @@
     AccessLogViewSet,
     AllAccessLogsExportViewSet,
     AllAccessLogViewSet,
+    AllProjectHistoryLogViewSet,
     AuditLogViewSet,
 )
 
 router = DefaultRouter()
 router.register(r'audit-logs', AuditLogViewSet, basename='audit-log')
 router.register(r'access-logs', AllAccessLogViewSet, basename='all-access-logs')
 router.register(r'access-logs/me', AccessLogViewSet, basename='access-log')
+# routes for PH logs for individual assets are registered in router_api_v2.py
+router.register(
+    r'project-history-logs',
+    AllProjectHistoryLogViewSet,
+    basename='all-project-history-logs',
+)
 router.register(
     r'access-logs/export', AllAccessLogsExportViewSet, basename='all-access-logs-export'
 )