feat(projectHistoryLogs): add endpoints for viewing project history l…

…ogs TASK-973 (#5319)

### 📣 Summary
Add two new endpoints for viewing project history logs, one for all
logs, and one for logs for a given project.


### 📖 Description
Adds two new endpoints: `/api/v2/project-history-logs` (all project
history logs) and `/api/v2/assets/<uid>/history` (logs for the specific
project).
The former is only available to superusers, the latter is available to
anyone with the manage_asset permission for the project.
Both endpoints can be searched by a all fields, most importantly
`action`, `metadata__asset_uid`, and `username`. A full list of
available searchable fields is in the endpoint documentation.


### 👀 Preview steps


Feature/no-change template:
1. ℹ️ have two accounts, one super and one not. Have at least two
projects.
2. Change the name of both projects to generate some logs.
3. Log in as the superuser.
4. Go to `localhost/api/v2/project-history-logs`
5. 🟢 You should see logs for both projects
6. Grant the non-superuser the manage_project permission for one of the
projects (they will have this if they are the owner, or you can assign
it from Settings > Sharing)
7. Log out
8. 🟢 Go to `localhost/api/v2/project-history-logs`. You should get a
401.
9. Log in as the non-superuser.
10. 🟢 Go to `localhost/api/v2/project-history-logs`. You should get a
403.
11. Go to `localhost/api/v2/assets/<asset_uid>/history`
12. 🟢 You should see project history logs for just that project
13. Revoke the manage_project permission for the user. You can do this
by transferring ownership or just removing the singular permission (will
require you to log back in as the superuser).
14. 🟢 Reload `localhost/api/v2/assets/<asset_uid>/history`. You should
get a 403.

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'
 )