chore: import existing codebase as baseline

tests/test_importing.py

@@ -0,0 +1,125 @@
+import hashlib
+import zipfile
+from datetime import date
+from decimal import Decimal
+from io import BytesIO
+
+import pytest
+from django.contrib.auth.models import User
+
+from finance.importing import (
+    decimal,
+    file_digest,
+    period_value,
+    process_dataset,
+    profile_file,
+    rows_for,
+    safe_text,
+    suggest_mapping,
+    validate_archive,
+    validate_mapping,
+)
+from finance.models import Dataset, Membership, Organization
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def import_dataset(tmp_path, settings):
+    settings.MEDIA_ROOT = tmp_path
+    org = Organization.objects.create(name="Import tests")
+    user = User.objects.create_user("importer", password="Import-TestPassword991!")
+    Membership.objects.create(user=user, organization=org, role="admin")
+    ds = Dataset.objects.create(
+        organization=org,
+        created_by=user,
+        name="sample.csv",
+        sha256="c" * 64,
+        status="queued",
+        mapping={
+            "sheet": "CSV",
+            "header": 1,
+            "columns": [
+                {"index": 0, "name": "期间", "role": "period"},
+                {"index": 1, "name": "科室", "role": "dimension", "key": "department"},
+                {"index": 2, "name": "产品一级分类", "role": "dimension", "key": "product"},
+                {"index": 3, "name": "实际收入", "role": "revenue"},
+                {"index": 4, "name": "毛利贡献", "role": "gross"},
+                {"index": 5, "name": "客户编码", "role": "customer"},
+            ],
+            "roi_formula": "profit_over_cost",
+            "margin_rates": {"医疗服务": "0.6"},
+        },
+    )
+    source = tmp_path / "sample.csv"
+    source.write_text(
+        "期间,科室,产品一级分类,实际收入,毛利贡献,客户编码\n"
+        "2026-08,儿科,医疗服务,100,60,C1\n"
+        "2026-08,儿科,未知产品,25,,\n"
+        "\n",
+        encoding="utf-8",
+    )
+    ds.file.name = str(source)
+    ds.save(update_fields=["file"])
+    return ds
+
+
+def test_csv_import_commits_rows_quality_and_derived_gross(import_dataset):
+    ds = process_dataset(import_dataset.pk)
+    assert ds.status == "ready"
+    assert ds.row_count == 2
+    assert ds.periods == ["2026-08"]
+    assert ds.quality["missing_cost"] == 2
+    assert ds.quality["missing_gross"] == 1
+    assert ds.quality["missing_customer"] == 1
+    assert ds.quality["blank_rows"] == 1
+    assert list(ds.facts.order_by("source_row").values_list("gross", flat=True)) == [
+        Decimal("60.000000000000000000000000"),
+        None,
+    ]
+    assert ds.facts.get(source_row=2).dimensions["department"] == "儿科"
+
+
+def test_profile_csv_mapping_and_file_digest(tmp_path):
+    source = tmp_path / "input.csv"
+    source.write_text("期间,科室,实际收入\n2026-08,儿科,12\n", encoding="utf-8")
+    profile = profile_file(source)
+    mapping = suggest_mapping(profile)
+    assert mapping["columns"][0]["role"] == "period"
+    assert mapping["columns"][2]["role"] == "revenue"
+    assert validate_mapping(mapping, profile) == mapping
+    from django.core.files.uploadedfile import SimpleUploadedFile
+
+    upload = SimpleUploadedFile("input.csv", source.read_bytes(), content_type="text/csv")
+    assert file_digest(upload) == hashlib.sha256(source.read_bytes()).hexdigest()
+
+
+def test_safe_text_and_period_inputs():
+    assert safe_text(date(2026, 8, 3)) == "2026-08-03"
+    assert safe_text("x" * 5000) == "x" * 4000
+    assert period_value({"period": date(2026, 8, 3)}) == "2026-08"
+    assert period_value({"period": "2026-8-19T12:00:00"}) == "2026-08"
+    with pytest.raises(ValueError, match="精度"):
+        decimal("0.0000000000000000000000001")
+
+
+def test_archive_rejects_macro_and_external_links():
+    for member in ("xl/vbaProject.bin", "xl/externalLinks/externalLink1.xml"):
+        archive_file = BytesIO()
+        with zipfile.ZipFile(archive_file, "w") as archive:
+            archive.writestr(member, "blocked")
+        archive_file.seek(0)
+        with pytest.raises(ValueError, match="宏或外部链接"):
+            validate_archive(archive_file)
+
+
+def test_workbook_rows_can_read_formulas_from_selected_sheet(tmp_path):
+    from openpyxl import Workbook
+
+    path = tmp_path / "book.xlsx"
+    workbook = Workbook()
+    workbook.active.title = "Data"
+    workbook.active.append(["value"])
+    workbook.active.append(["=1+1"])
+    workbook.save(path)
+    assert list(rows_for(path, "Data", formulas=True)) == [("value",), ("=1+1",)]