B-42: 导入功能加固——逐工作表确认、流式上传与诊断证据
This commit is contained in:
+67
-1
@@ -1,10 +1,19 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
import io
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from bank_importer.parser import UnknownTemplateError, detect_header, parse_directory
|
||||
from openpyxl import Workbook
|
||||
|
||||
from bank_importer.parser import (
|
||||
UnknownTemplateError,
|
||||
analyze_workbook,
|
||||
detect_header,
|
||||
parse_directory,
|
||||
)
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
@@ -83,5 +92,62 @@ class StatementParserTests(unittest.TestCase):
|
||||
self.assertIn("日期、金额、备注", message)
|
||||
|
||||
|
||||
class SheetResultTests(unittest.TestCase):
|
||||
"""Per-worksheet outcomes: parsed / exception / ignored with evidence."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp_dir.cleanup)
|
||||
|
||||
def _write(self, sheets) -> Path:
|
||||
workbook = Workbook()
|
||||
workbook.remove(workbook.active)
|
||||
for name, rows in sheets.items():
|
||||
worksheet = workbook.create_sheet(name)
|
||||
for row in rows:
|
||||
worksheet.append(row)
|
||||
path = Path(self.temp_dir.name) / "workbook.xlsx"
|
||||
workbook.save(path)
|
||||
return path
|
||||
|
||||
def test_every_sheet_has_an_independent_result(self) -> None:
|
||||
path = self._write(
|
||||
{
|
||||
"正常": [
|
||||
["客户账号", "账户名称", "交易时间", "借方发生额(支取)", "贷方发生额(收入)", "余额", "对方账号"],
|
||||
["6228480000000000", "测试", "2026-01-05 10:00:00", "100.00", "", "99900.00", "1002003004"],
|
||||
],
|
||||
"未知": [["日期", "金额", "备注"], ["2026-01-01", "100", "x"]],
|
||||
"空表": [],
|
||||
}
|
||||
)
|
||||
results = analyze_workbook(path)
|
||||
by_name = {result.sheet_name: result for result in results}
|
||||
self.assertEqual({"正常", "未知", "空表"}, set(by_name))
|
||||
self.assertEqual("parsed", by_name["正常"].outcome)
|
||||
self.assertIsNotNone(by_name["正常"].batch)
|
||||
self.assertEqual("exception", by_name["未知"].outcome)
|
||||
self.assertIn("未识别到受支持的银行表头", by_name["未知"].message)
|
||||
self.assertGreaterEqual(by_name["未知"].scanned_rows, 2)
|
||||
self.assertTrue(any("日期、金额、备注" in row for row in by_name["未知"].candidate_headers))
|
||||
self.assertEqual("ignored", by_name["空表"].outcome)
|
||||
self.assertEqual(0, by_name["空表"].scanned_rows)
|
||||
|
||||
def test_single_cell_rows_are_candidate_evidence(self) -> None:
|
||||
path = self._write({"候选": [["只有一个单元格"], ["又一格"]]})
|
||||
(result,) = analyze_workbook(path)
|
||||
self.assertEqual("exception", result.outcome)
|
||||
self.assertEqual(2, result.scanned_rows)
|
||||
self.assertTrue(result.candidate_headers)
|
||||
self.assertIn("只有一个单元格", ";".join(result.candidate_headers))
|
||||
|
||||
def test_messages_never_contain_stored_filename(self) -> None:
|
||||
path = self._write({"流水": [["日期", "金额", "备注"]]})
|
||||
(result,) = analyze_workbook(path)
|
||||
self.assertEqual("exception", result.outcome)
|
||||
self.assertNotIn(path.name, result.message)
|
||||
self.assertNotIn(str(path), result.message)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user