qlockify-backend-deployment/apps/reports/tests/test_tasks.py

from datetime import timedelta
from decimal import Decimal
from io import BytesIO

import pytest
from django.core.files.base import ContentFile
from django.core.files.storage import default_storage
from django.utils import timezone
from openpyxl import load_workbook

from apps.notifications.services import store as notification_store
from apps.reports.models import ReportExportJob
from apps.reports.tasks import cleanup_expired_report_exports_task, generate_report_export_task
from apps.time_entries.models import TimeEntry
from apps.users.models import User
from apps.workspaces.models import Workspace


class FakeRedis:
    def pipeline(self):
        return self

    def zadd(self, *args, **kwargs):
        return self

    def hset(self, *args, **kwargs):
        return self

    def sadd(self, *args, **kwargs):
        return self

    def execute(self):
        return []

    def publish(self, *args, **kwargs):
        return None

    def zrevrange(self, *args, **kwargs):
        return []

    def hget(self, *args, **kwargs):
        return None

    def zrem(self, *args, **kwargs):
        return 1

    def hdel(self, *args, **kwargs):
        return 1

    def zcard(self, *args, **kwargs):
        return 0

    def smembers(self, *args, **kwargs):
        return set()

    def srem(self, *args, **kwargs):
        return 1


@pytest.fixture()
def fake_redis(monkeypatch):
    redis = FakeRedis()
    monkeypatch.setattr(notification_store, "redis_client", redis)
    return redis


@pytest.fixture()
def owner(db):
    return User.objects.create_user(mobile="09129990001", password="secret123", first_name="Owner", last_name="User")


@pytest.fixture()
def teammate(db):
    return User.objects.create_user(mobile="09129990002", password="secret123", first_name="Team", last_name="Mate")


@pytest.fixture()
def workspace(owner, teammate):
    workspace = Workspace.objects.create(name="Exports", owner=owner)
    workspace.memberships.create(user=teammate, role="member", is_active=True)
    return workspace


@pytest.fixture()
def time_entry(workspace, owner):
    return TimeEntry.objects.create(
        workspace=workspace,
        user=owner,
        description="Export row",
        start_time="2026-04-12T08:00:00+03:30",
        end_time="2026-04-12T10:00:00+03:30",
        duration=timedelta(hours=2),
        is_billable=True,
        hourly_rate=Decimal("15.00"),
        currency="USD",
    )


@pytest.fixture()
def teammate_entry(workspace, teammate):
    return TimeEntry.objects.create(
        workspace=workspace,
        user=teammate,
        description="Team row",
        start_time="2026-04-13T08:00:00+03:30",
        end_time="2026-04-13T09:00:00+03:30",
        duration=timedelta(hours=1),
        is_billable=False,
        currency="USD",
    )


def test_generate_export_creates_file_and_marks_job_complete(fake_redis, workspace, owner, time_entry):
    job = ReportExportJob.objects.create(
        requesting_user=owner,
        workspace=workspace,
        export_type=ReportExportJob.ExportType.EXCEL,
        filters={
            "workspace": str(workspace.id),
            "period": "this_month",
            "from_date": "2026-04-01",
            "to_date": "2026-04-30",
            "user": str(owner.id),
            "client": None,
            "project": None,
            "tags": [],
            "language": "en",
        },
    )

    generate_report_export_task(str(job.id))
    job.refresh_from_db()

    assert job.status == ReportExportJob.Status.COMPLETED
    assert bool(job.file)
    assert default_storage.exists(job.file.name)


def test_generate_excel_export_adds_per_user_sheets_for_all_users_scope(
    fake_redis,
    workspace,
    owner,
    teammate,
    time_entry,
    teammate_entry,
):
    job = ReportExportJob.objects.create(
        requesting_user=owner,
        workspace=workspace,
        export_type=ReportExportJob.ExportType.EXCEL,
        filters={
            "workspace": str(workspace.id),
            "period": "this_month",
            "from_date": "2026-04-01",
            "to_date": "2026-04-30",
            "user": None,
            "client": None,
            "project": None,
            "tags": [],
            "language": "en",
        },
    )

    generate_report_export_task(str(job.id))
    job.refresh_from_db()

    workbook = load_workbook(BytesIO(job.file.read()))
    assert workbook.sheetnames[0] == "Overall Report"
    assert any("Owner User" in sheet for sheet in workbook.sheetnames[1:])
    assert any("Team Mate" in sheet for sheet in workbook.sheetnames[1:])
    assert len(workbook.sheetnames) == 3


def test_generate_pdf_export_supports_persian_locale(fake_redis, workspace, owner, time_entry):
    job = ReportExportJob.objects.create(
        requesting_user=owner,
        workspace=workspace,
        export_type=ReportExportJob.ExportType.PDF,
        filters={
            "workspace": str(workspace.id),
            "period": "this_month",
            "from_date": "2026-04-01",
            "to_date": "2026-04-30",
            "user": str(owner.id),
            "client": None,
            "project": None,
            "tags": [],
            "language": "fa",
        },
    )

    generate_report_export_task(str(job.id))
    job.refresh_from_db()

    assert job.status == ReportExportJob.Status.COMPLETED
    assert job.file.read(4) == b"%PDF"


def test_cleanup_expires_and_removes_files(fake_redis, workspace, owner):
    job = ReportExportJob.objects.create(
        requesting_user=owner,
        workspace=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()

    assert removed == 1
    assert job.status == ReportExportJob.Status.EXPIRED
    assert not default_storage.exists(file_name)