162 lines
5.5 KiB
Python
162 lines
5.5 KiB
Python
from auditlog.models import LogEntry
|
|
from rest_framework_simplejwt.tokens import AccessToken
|
|
from rest_framework.test import APITestCase
|
|
|
|
from apps.reports.models import ReportExportJob
|
|
from apps.users.models import User
|
|
from apps.workspaces.models import Workspace, WorkspaceMembership
|
|
|
|
|
|
class WorkspaceLogViewTests(APITestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.owner = cls._user(1)
|
|
cls.admin = cls._user(2)
|
|
cls.member = cls._user(3)
|
|
cls.outsider = cls._user(4)
|
|
|
|
cls.workspace = Workspace.objects.create(
|
|
name="Logs WS",
|
|
description="",
|
|
owner=cls.owner,
|
|
)
|
|
WorkspaceMembership.objects.create(
|
|
workspace=cls.workspace,
|
|
user=cls.admin,
|
|
role=WorkspaceMembership.Role.ADMIN,
|
|
is_active=True,
|
|
)
|
|
WorkspaceMembership.objects.create(
|
|
workspace=cls.workspace,
|
|
user=cls.member,
|
|
role=WorkspaceMembership.Role.MEMBER,
|
|
is_active=True,
|
|
)
|
|
|
|
@staticmethod
|
|
def _user(index):
|
|
return User.objects.create_user(
|
|
mobile=f"093355500{index:02d}",
|
|
password="secret123",
|
|
first_name=f"Log{index}",
|
|
last_name="User",
|
|
)
|
|
|
|
@staticmethod
|
|
def _auth_headers(user):
|
|
token = str(AccessToken.for_user(user))
|
|
return {"HTTP_AUTHORIZATION": f"Bearer {token}"}
|
|
|
|
def _create_tag(self, user, *, name="Audit Tag"):
|
|
return self.client.post(
|
|
"/api/tags/",
|
|
{
|
|
"workspace_id": str(self.workspace.id),
|
|
"name": name,
|
|
"color": "#123456",
|
|
},
|
|
format="json",
|
|
**self._auth_headers(user),
|
|
)
|
|
|
|
def test_owner_and_admin_can_list_workspace_logs(self):
|
|
create_response = self._create_tag(self.owner)
|
|
self.assertEqual(create_response.status_code, 201)
|
|
|
|
owner_response = self.client.get(
|
|
f"/api/logs/?workspace={self.workspace.id}",
|
|
**self._auth_headers(self.owner),
|
|
)
|
|
admin_response = self.client.get(
|
|
f"/api/logs/?workspace={self.workspace.id}",
|
|
**self._auth_headers(self.admin),
|
|
)
|
|
|
|
self.assertEqual(owner_response.status_code, 200)
|
|
self.assertEqual(admin_response.status_code, 200)
|
|
self.assertEqual(owner_response.data["items"][0]["section"], "tags")
|
|
|
|
def test_member_and_non_member_cannot_list_workspace_logs(self):
|
|
self._create_tag(self.owner)
|
|
|
|
member_response = self.client.get(
|
|
f"/api/logs/?workspace={self.workspace.id}",
|
|
**self._auth_headers(self.member),
|
|
)
|
|
outsider_response = self.client.get(
|
|
f"/api/logs/?workspace={self.workspace.id}",
|
|
**self._auth_headers(self.outsider),
|
|
)
|
|
|
|
self.assertEqual(member_response.status_code, 403)
|
|
self.assertEqual(outsider_response.status_code, 403)
|
|
|
|
def test_jwt_authenticated_writes_capture_actor_and_workspace_metadata(self):
|
|
response = self._create_tag(self.owner, name="JWT Tag")
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
log_entry = LogEntry.objects.filter(content_type__app_label="tags").latest(
|
|
"timestamp"
|
|
)
|
|
|
|
self.assertEqual(log_entry.actor_id, self.owner.id)
|
|
self.assertEqual(log_entry.additional_data["workspace_id"], str(self.workspace.id))
|
|
self.assertEqual(log_entry.additional_data["section"], "tags")
|
|
|
|
def test_logs_support_section_filter_and_detail(self):
|
|
tag_response = self._create_tag(self.owner, name="Filtered Tag")
|
|
self.assertEqual(tag_response.status_code, 201)
|
|
|
|
list_response = self.client.get(
|
|
f"/api/logs/?workspace={self.workspace.id}§ion=tags",
|
|
**self._auth_headers(self.owner),
|
|
)
|
|
|
|
self.assertEqual(list_response.status_code, 200)
|
|
self.assertTrue(list_response.data["items"])
|
|
log_id = list_response.data["items"][0]["id"]
|
|
|
|
detail_response = self.client.get(
|
|
f"/api/logs/{log_id}/",
|
|
**self._auth_headers(self.owner),
|
|
)
|
|
|
|
self.assertEqual(detail_response.status_code, 200)
|
|
self.assertEqual(detail_response.data["target"]["name"], "Filtered Tag")
|
|
self.assertTrue(detail_response.data["changes"])
|
|
|
|
def test_soft_delete_and_actorless_background_logs_are_filtered(self):
|
|
create_response = self._create_tag(self.owner, name="Delete Me")
|
|
self.assertEqual(create_response.status_code, 201)
|
|
tag_id = create_response.data["id"]
|
|
|
|
delete_response = self.client.delete(
|
|
f"/api/tags/{tag_id}/",
|
|
**self._auth_headers(self.owner),
|
|
)
|
|
self.assertEqual(delete_response.status_code, 204)
|
|
|
|
ReportExportJob.objects.create(
|
|
requesting_user=self.owner,
|
|
workspace=self.workspace,
|
|
export_type=ReportExportJob.ExportType.PDF,
|
|
filters={"workspace": str(self.workspace.id)},
|
|
status=ReportExportJob.Status.PENDING,
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"/api/logs/?workspace={self.workspace.id}&event=delete",
|
|
**self._auth_headers(self.owner),
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(
|
|
any(
|
|
item["event"] == "delete" and item["section"] == "tags"
|
|
for item in response.data["items"]
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(item["section"] != "report_exports" for item in response.data["items"])
|
|
)
|