fa-agent/tests/test_operations.py

from datetime import timedelta

import pytest
from django.contrib.auth.models import User
from django.core.management import call_command
from django.core.management.base import CommandError
from django.utils import timezone

from finance.models import Audit, FormulaExport

pytestmark = pytest.mark.django_db


def test_bootstrap_creates_admin_once_and_rejects_duplicate(workspace, monkeypatch):
    monkeypatch.setenv("BOOTSTRAP_PASSWORD", "Bootstrap-TestPassword991!")
    with pytest.raises(CommandError, match="already exists"):
        call_command("bootstrap", username="finance-admin", organization="Duplicate")
    user_count = User.objects.count()
    call_command("bootstrap", username="new-admin", organization="New Organization")
    user = User.objects.get(username="new-admin")
    assert user.check_password("Bootstrap-TestPassword991!")
    assert user.membership.role == "admin"
    assert User.objects.count() == user_count + 1


def test_cli_import_previews_then_queues_and_deduplicates(workspace, tmp_path):
    ds, _, _, _, _ = workspace
    source = tmp_path / "cli.csv"
    source.write_text("期间,科室,实际收入\n2026-08,儿科,10\n", encoding="utf-8")
    call_command("import_file", str(source), username=ds.created_by.username)
    assert not Audit.objects.filter(action="dataset.cli_import").exists()
    call_command("import_file", str(source), username=ds.created_by.username, confirm_mapping=True)
    assert Audit.objects.filter(action="dataset.cli_import").count() == 1
    call_command("import_file", str(source), username=ds.created_by.username, confirm_mapping=True)
    assert Audit.objects.filter(action="dataset.cli_import").count() == 1


def test_worker_processes_queued_import_and_formula_export(workspace):
    ds, _, _, *_ = workspace
    job = FormulaExport.objects.create(dataset=ds, user=ds.created_by, context={"custom_columns": []})
    from unittest.mock import patch

    with patch("finance.management.commands.import_worker.process_formula_export", return_value={"rows": 5}):
        call_command("import_worker", once=True)
    job.refresh_from_db()
    assert job.status == "processing"


def test_worker_reclaims_processing_formula_export_after_restart(workspace):
    ds, _, _, *_ = workspace
    job = FormulaExport.objects.create(
        dataset=ds,
        user=ds.created_by,
        context={"custom_columns": []},
        status="processing",
    )
    from unittest.mock import patch

    with patch(
        "finance.management.commands.import_worker.process_formula_export", return_value={"rows": 5}
    ) as process:
        call_command("import_worker", once=True)
    process.assert_called_once_with(job.pk)


def test_formula_export_status_download_and_access_controls(workspace, settings, tmp_path):
    ds, _, client, _, other_client = workspace
    url = f"/api/datasets/{ds.pk}"
    job = FormulaExport.objects.create(dataset=ds, user=ds.created_by, context={}, status="queued")
    status_url = f"{url}/formula-exports/{job.pk}"
    assert client.get(status_url).json()["status"] == "queued"
    assert client.get(f"{status_url}/download").status_code == 409
    assert other_client.get(status_url).status_code == 404

    job.status = "succeeded"
    job.expires_at = timezone.now() + timedelta(hours=1)
    target = tmp_path / "formula-exports" / str(ds.created_by_id) / f"{job.pk}.xlsx"
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(b"xlsx")
    job.path = str(target)
    job.save()
    response = client.get(f"{status_url}/download")
    assert response.status_code == 200
    assert response["Content-Disposition"].endswith('"traceiq-formula-analysis.xlsx"')

    other_target = target.with_name("another-task.xlsx")
    other_target.write_bytes(b"other")
    job.path = str(other_target)
    job.save(update_fields=["path"])
    assert client.get(f"{status_url}/download").status_code == 410
    job.path = str(target)
    job.save(update_fields=["path"])

    target.unlink()
    target.symlink_to(other_target)
    assert client.get(f"{status_url}/download").status_code == 410
    target.unlink()
    target.write_bytes(b"xlsx")

    job.departments = ["儿科"]
    job.save(update_fields=["departments"])
    assert client.get(status_url).status_code == 403


def test_formula_export_rechecks_role_when_worker_runs_and_file_is_downloaded(workspace, settings, tmp_path):
    from finance.exporting import process_formula_export

    ds, member, client, _, _ = workspace
    job = FormulaExport.objects.create(dataset=ds, user=ds.created_by, context={}, status="processing")
    member.role = "viewer"
    member.save(update_fields=["role"])
    with pytest.raises(PermissionError, match="撤权"):
        process_formula_export(job.pk)
    job.refresh_from_db()
    assert job.status == "failed"

    job.status = "succeeded"
    job.expires_at = timezone.now() + timedelta(hours=1)
    target = tmp_path / "formula-exports" / str(ds.created_by_id) / f"{job.pk}.xlsx"
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(b"xlsx")
    job.path = str(target)
    job.save(update_fields=["status", "expires_at", "path"])
    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}").status_code == 403
    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == 403


def test_formula_export_create_validates_definition_and_records_job(workspace):
    ds, _, client, *_ = workspace
    url = f"/api/datasets/{ds.pk}/formula-exports"
    ctx = {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"}
    response = client.post(url, data=ctx, content_type="application/json")
    assert response.status_code == 202
    assert FormulaExport.objects.filter(dataset=ds, status="queued").exists()