95 lines
4.1 KiB
Python
95 lines
4.1 KiB
Python
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_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)
|
|
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_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()
|