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",)]