223 lines
8.3 KiB
Python
223 lines
8.3 KiB
Python
from decimal import Decimal
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
from django.contrib.auth.models import User
|
|
from django.core.files.base import ContentFile
|
|
from django.test import Client
|
|
from openpyxl import load_workbook
|
|
|
|
from finance.analysis import analyze, context, default_preferences, metrics
|
|
from finance.importing import decimal, period_value, process_dataset, validate_mapping
|
|
from finance.models import Fact, Membership
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_money_and_period_parsing_reject_invalid_values():
|
|
assert decimal("1,234.50") == Decimal("1234.50")
|
|
assert decimal("", required=False) is None
|
|
assert period_value({"period": "2026/8/1"}) == "2026-08"
|
|
assert period_value({"year": "2026年", "month": "8月"}) == "2026-08"
|
|
for value in ("NaN", "1e16", "0.0000000000000000000000001"):
|
|
with pytest.raises(ValueError):
|
|
decimal(value)
|
|
with pytest.raises(ValueError, match="必填"):
|
|
decimal(None, required=True)
|
|
with pytest.raises(ValueError, match="期间"):
|
|
period_value({"period": "2026-13"})
|
|
|
|
|
|
def test_mapping_and_context_enforce_ranges_and_dimensions(workspace):
|
|
ds, *_ = workspace
|
|
valid = validate_mapping(ds.mapping, {"sheets": [{"name": "CSV"}]})
|
|
assert valid["roi_formula"] == "profit_over_cost"
|
|
with pytest.raises(ValueError, match="ROI"):
|
|
validate_mapping({**ds.mapping, "roi_formula": "=1+1"}, {"sheets": [{"name": "CSV"}]})
|
|
reserved = {**ds.mapping, "columns": [dict(column) for column in ds.mapping["columns"]]}
|
|
reserved["columns"][1]["key"] = "revenue"
|
|
with pytest.raises(ValueError, match="保留列名冲突"):
|
|
validate_mapping(reserved, {"sheets": [{"name": "CSV"}]})
|
|
with pytest.raises(ValueError, match="维度"):
|
|
context(ds, {"filters": {"unknown": "x"}})
|
|
with pytest.raises(ValueError, match="重叠"):
|
|
context(ds, {"comparison": "custom", "start": "2026-08", "end": "2026-08", "baseline_end": "2026-08"})
|
|
with pytest.raises(ValueError, match="60"):
|
|
context(ds, {"start": "2019-01", "end": "2026-08"})
|
|
|
|
|
|
def test_metrics_and_analysis_preserve_roi_refunds_and_distinct_customers(workspace):
|
|
ds, member, *_ = workspace
|
|
prefs = default_preferences(ds)
|
|
result = analyze(
|
|
ds,
|
|
member,
|
|
{"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
|
|
prefs,
|
|
)
|
|
assert result["summary"]["revenue"]["current"] == Decimal("90")
|
|
assert result["summary"]["customers"]["current"] == 3
|
|
assert result["summary"]["roi"]["current"] == Decimal("1")
|
|
assert result["summary"]["discount_rate"]["current"] == Decimal("90") / Decimal("125")
|
|
assert result["context"]["missing"] == []
|
|
assert {row["label"] for row in result["rows"]} == {"儿科", "外科"}
|
|
|
|
unavailable = metrics(
|
|
{
|
|
"rows": 1,
|
|
"revenue_sum": Decimal(5),
|
|
"cost_sum": None,
|
|
"gross_sum": None,
|
|
"cost_n": 0,
|
|
"gross_n": 0,
|
|
"customer_n": 0,
|
|
"customers": 0,
|
|
"list_sum": None,
|
|
"list_n": 0,
|
|
"self_sum": None,
|
|
"upgraded": 0,
|
|
},
|
|
"gross_less_cost_over_cost",
|
|
["医疗服务"],
|
|
)
|
|
assert unavailable["roi"] is None
|
|
assert unavailable["customers"] is None
|
|
assert unavailable["self_revenue"] == 0
|
|
|
|
|
|
def test_dataset_and_evidence_are_organization_and_department_scoped(workspace):
|
|
ds, _, _, _, other_client = workspace
|
|
url = f"/api/datasets/{ds.pk}"
|
|
assert other_client.get(url).status_code == 404
|
|
|
|
dept_user = User.objects.create_user("finance-dept", password="Finance-TestPassword991!")
|
|
Membership.objects.create(
|
|
user=dept_user,
|
|
organization=ds.organization,
|
|
role="analyst",
|
|
departments=["儿科"],
|
|
)
|
|
scoped = Client()
|
|
assert scoped.login(username=dept_user.username, password="Finance-TestPassword991!")
|
|
assert scoped.get(url).json()["row_count"] is None
|
|
evidence = scoped.post(
|
|
f"{url}/evidence",
|
|
data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
|
|
content_type="application/json",
|
|
)
|
|
assert evidence.status_code == 200
|
|
assert evidence.json()["total"] == 2
|
|
assert all("raw" not in row for row in evidence.json()["rows"])
|
|
|
|
|
|
def test_evidence_pagination_returns_stable_pages_and_rejects_invalid_bounds(workspace):
|
|
ds, _, client, *_ = workspace
|
|
url = f"/api/datasets/{ds.pk}/evidence"
|
|
payload = {
|
|
"start": "2026-08",
|
|
"end": "2026-08",
|
|
"comparison": "yoy",
|
|
"group": "department",
|
|
"side": "current",
|
|
"offset": 1,
|
|
"limit": 1,
|
|
}
|
|
page = client.post(url, data=payload, content_type="application/json")
|
|
assert page.status_code == 200
|
|
assert page.json()["total"] == 3
|
|
assert len(page.json()["rows"]) == 1
|
|
assert page.json()["rows"][0]["source_row"] == 5
|
|
|
|
for invalid in ({"offset": -1}, {"limit": 201}, {"offset": True}):
|
|
response = client.post(url, data={**payload, **invalid}, content_type="application/json")
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_settings_revision_and_invalid_metric_are_rejected(workspace):
|
|
ds, _, client, *_ = workspace
|
|
url = f"/api/datasets/{ds.pk}/settings"
|
|
initial = client.get(url).json()
|
|
value = initial["value"]
|
|
value["min_samples"] = 1
|
|
saved = client.put(
|
|
url,
|
|
data={"value": value, "revision": initial["revision"]},
|
|
content_type="application/json",
|
|
)
|
|
assert saved.status_code == 200
|
|
assert saved.json()["revision"] == 1
|
|
conflict = client.put(
|
|
url,
|
|
data={"value": value, "revision": 0},
|
|
content_type="application/json",
|
|
)
|
|
assert conflict.status_code == 409
|
|
value["metric"] = "unknown"
|
|
invalid = client.put(url, data={"value": value, "revision": 1}, content_type="application/json")
|
|
assert invalid.status_code == 400
|
|
|
|
|
|
def test_csv_export_neutralizes_formula_injection_and_preserves_exact_values(workspace):
|
|
ds, _, client, *_ = workspace
|
|
Fact.objects.filter(dataset=ds).update(dimensions={"department": "=1+1", "product": "检查"})
|
|
response = client.post(
|
|
f"/api/datasets/{ds.pk}/export",
|
|
data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
|
|
content_type="application/json",
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.streaming
|
|
content = b"".join(response.streaming_content).decode("utf-8-sig")
|
|
assert "'=1+1" in content
|
|
assert ",-10.000000000000000000000000," in content
|
|
assert "'-10.000000000000000000000000" not in content
|
|
assert "traceiq-evidence.csv" in response["Content-Disposition"]
|
|
|
|
xlsx = client.post(
|
|
f"/api/datasets/{ds.pk}/export",
|
|
data={
|
|
"start": "2026-08",
|
|
"end": "2026-08",
|
|
"comparison": "yoy",
|
|
"group": "department",
|
|
"format": "xlsx",
|
|
},
|
|
content_type="application/json",
|
|
)
|
|
assert xlsx.streaming
|
|
workbook = load_workbook(BytesIO(b"".join(xlsx.streaming_content)), data_only=False)
|
|
cell = workbook["证据明细"]["D6"]
|
|
assert cell.value == "=1+1"
|
|
assert cell.data_type == "s"
|
|
|
|
|
|
def test_invalid_import_rolls_back_facts_and_records_failure(workspace, tmp_path):
|
|
ds, *_ = workspace
|
|
ds.facts.all().delete()
|
|
ds.status = "queued"
|
|
ds.mapping = {
|
|
"header": 1,
|
|
"sheet": "CSV",
|
|
"columns": [
|
|
{"index": 0, "name": "期间", "role": "period"},
|
|
{"index": 1, "name": "科室", "role": "dimension", "key": "department"},
|
|
{"index": 2, "name": "实际收入", "role": "revenue"},
|
|
],
|
|
"roi_formula": "profit_over_cost",
|
|
}
|
|
ds.save()
|
|
csv_path = tmp_path / "invalid.csv"
|
|
csv_path.write_text("期间,科室,实际收入\n2026-08,儿科,10\n2026-13,儿科,20\n", encoding="utf-8")
|
|
ds.file.name = str(csv_path)
|
|
ds.save(update_fields=["file"])
|
|
result = process_dataset(ds.pk)
|
|
assert result.status == "failed"
|
|
assert result.facts.count() == 0
|
|
assert "第 3 行" in result.error
|
|
|
|
|
|
def test_upload_requires_role_and_rejects_unsupported_file(workspace):
|
|
_, _, client, _, _ = workspace
|
|
response = client.post("/api/datasets", data={"file": ContentFile(b"not a workbook", name="bad.txt")})
|
|
assert response.status_code == 403 or response.status_code == 400
|