112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
from unittest.mock import patch
|
|
|
|
from django.core.files.base import ContentFile
|
|
from django.core.files.storage import default_storage
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
|
|
from apps.reports.models import ReportExportJob
|
|
from apps.reports.tasks import (
|
|
cleanup_expired_report_exports_task,
|
|
generate_report_export_task,
|
|
)
|
|
from apps.users.models import User
|
|
from apps.workspaces.models import Workspace
|
|
|
|
|
|
class ReportTaskTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.owner = User.objects.create_user(
|
|
mobile="09129990001",
|
|
password="secret123",
|
|
first_name="Owner",
|
|
last_name="User",
|
|
)
|
|
cls.workspace = Workspace.objects.create(name="Exports", owner=cls.owner)
|
|
|
|
def test_generate_excel_export_marks_job_complete_and_sends_notification(self):
|
|
job = ReportExportJob.objects.create(
|
|
requesting_user=self.owner,
|
|
workspace=self.workspace,
|
|
export_type=ReportExportJob.ExportType.EXCEL,
|
|
filters={
|
|
"workspace": str(self.workspace.id),
|
|
"period": "this_month",
|
|
"from_date": "2026-04-01",
|
|
"to_date": "2026-04-30",
|
|
"user": str(self.owner.id),
|
|
"client": None,
|
|
"project": None,
|
|
"tags": [],
|
|
"language": "en",
|
|
},
|
|
)
|
|
|
|
with patch("apps.reports.tasks.build_table_report", return_value={"scope": {}, "summary": {}, "days": [], "clients": [], "projects": [], "tags": []}) as build_table_report:
|
|
with patch("apps.reports.tasks.build_user_scoped_table_reports", return_value=[]) as build_user_reports:
|
|
with patch("apps.reports.tasks.build_excel_report", return_value=b"excel-content") as build_excel_report:
|
|
with patch("apps.reports.tasks.RedisNotificationStore.add") as notify:
|
|
generate_report_export_task(str(job.id))
|
|
|
|
job.refresh_from_db()
|
|
self.assertEqual(job.status, ReportExportJob.Status.COMPLETED)
|
|
self.assertTrue(bool(job.file))
|
|
self.assertTrue(default_storage.exists(job.file.name))
|
|
build_table_report.assert_called_once()
|
|
build_user_reports.assert_called_once()
|
|
build_excel_report.assert_called_once()
|
|
notify.assert_called_once()
|
|
self.assertEqual(notify.call_args.args[0], str(self.owner.id))
|
|
self.assertEqual(notify.call_args.args[1]["type"], "report_export_ready")
|
|
|
|
def test_generate_pdf_export_failure_marks_job_failed_and_notifies(self):
|
|
job = ReportExportJob.objects.create(
|
|
requesting_user=self.owner,
|
|
workspace=self.workspace,
|
|
export_type=ReportExportJob.ExportType.PDF,
|
|
filters={
|
|
"workspace": str(self.workspace.id),
|
|
"period": "this_month",
|
|
"from_date": "2026-04-01",
|
|
"to_date": "2026-04-30",
|
|
"user": str(self.owner.id),
|
|
"client": None,
|
|
"project": None,
|
|
"tags": [],
|
|
"language": "fa",
|
|
},
|
|
)
|
|
|
|
with patch("apps.reports.tasks.build_table_report", return_value={"scope": {}, "summary": {}, "days": [], "clients": [], "projects": [], "tags": []}):
|
|
with patch("apps.reports.tasks.build_pdf_report", side_effect=RuntimeError("boom")):
|
|
with patch("apps.reports.tasks.RedisNotificationStore.add") as notify:
|
|
with self.assertRaises(RuntimeError):
|
|
generate_report_export_task(str(job.id))
|
|
|
|
job.refresh_from_db()
|
|
self.assertEqual(job.status, ReportExportJob.Status.FAILED)
|
|
self.assertEqual(job.error_message, "boom")
|
|
notify.assert_called_once()
|
|
self.assertEqual(notify.call_args.args[1]["type"], "report_export_failed")
|
|
|
|
def test_cleanup_expires_and_removes_files(self):
|
|
job = ReportExportJob.objects.create(
|
|
requesting_user=self.owner,
|
|
workspace=self.workspace,
|
|
export_type=ReportExportJob.ExportType.EXCEL,
|
|
status=ReportExportJob.Status.COMPLETED,
|
|
filters={},
|
|
expires_at=timezone.now() - timezone.timedelta(days=1),
|
|
)
|
|
file_name = f"reports/exports/{job.id}-old.xlsx"
|
|
job.file.save(file_name, ContentFile(b"old-data"), save=False)
|
|
job.save(update_fields=["file", "updated_at"])
|
|
|
|
removed = cleanup_expired_report_exports_task()
|
|
job.refresh_from_db()
|
|
|
|
self.assertEqual(removed, 1)
|
|
self.assertEqual(job.status, ReportExportJob.Status.EXPIRED)
|
|
self.assertFalse(default_storage.exists(file_name))
|