from datetime import date from types import SimpleNamespace from unittest.mock import patch from django.core.files.base import ContentFile from django.test import TestCase from rest_framework.test import APIClient from apps.reports.models import ReportExportJob from apps.users.models import User from apps.workspaces.models import Workspace, WorkspaceMembership class ReportExportApiTests(TestCase): @classmethod def setUpTestData(cls): cls.owner = User.objects.create_user( mobile="09126660001", password="secret123", first_name="Owner", ) cls.admin = User.objects.create_user( mobile="09126660002", password="secret123", first_name="Admin", ) cls.workspace = Workspace.objects.create(name="Exports", owner=cls.owner) WorkspaceMembership.objects.create( workspace=cls.workspace, user=cls.admin, role=WorkspaceMembership.Role.ADMIN, is_active=True, ) def setUp(self): self.client = APIClient() def test_create_export_job_enqueues_background_task(self): self.client.force_authenticate(user=self.owner) filters = SimpleNamespace( workspace=self.workspace, period="this_month", from_date=date(2026, 4, 1), to_date=date(2026, 4, 30), user_id=None, client_id=None, project_id=None, tag_ids=[], ) with patch("apps.reports.api.views.load_report_filters", return_value=filters): with patch("apps.reports.api.views.generate_report_export_task.delay") as delay: response = self.client.post( "/api/reports/exports/", { "workspace": str(self.workspace.id), "period": "this_month", "export_type": "excel", "language": "en", }, format="json", ) self.assertEqual(response.status_code, 202) self.assertEqual(ReportExportJob.objects.count(), 1) delay.assert_called_once() def test_list_only_returns_requesting_users_jobs(self): own_job = ReportExportJob.objects.create( requesting_user=self.owner, workspace=self.workspace, export_type=ReportExportJob.ExportType.EXCEL, filters={"workspace": str(self.workspace.id)}, ) ReportExportJob.objects.create( requesting_user=self.admin, workspace=self.workspace, export_type=ReportExportJob.ExportType.PDF, filters={"workspace": str(self.workspace.id)}, ) self.client.force_authenticate(user=self.owner) response = self.client.get("/api/reports/exports/") self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 1) self.assertEqual(response.data[0]["id"], str(own_job.id)) def test_download_returns_completed_file(self): job = ReportExportJob.objects.create( requesting_user=self.owner, workspace=self.workspace, export_type=ReportExportJob.ExportType.EXCEL, status=ReportExportJob.Status.COMPLETED, filters={"workspace": str(self.workspace.id)}, file_name="report.xlsx", ) job.file.save("reports/exports/report.xlsx", ContentFile(b"content"), save=False) job.save(update_fields=["file", "updated_at"]) self.client.force_authenticate(user=self.owner) response = self.client.get(f"/api/reports/exports/{job.id}/download/") self.assertEqual(response.status_code, 200) self.assertIn("attachment; filename=", response["Content-Disposition"])