"""HTTP integration tests for the hardened import API (B-42). Covers streaming multipart limits, per-worksheet parse results with the four diagnostic evidence items, the confirm/ignore review lifecycle with tenant scoping and idempotency, unconfirmed-sheet export gating, and concurrent duplicate uploads. Uses a real ``ThreadingHTTPServer`` like ``test_server_auth``. """ from __future__ import annotations import io import itertools import json import os from pathlib import Path import tempfile import threading import unittest from openpyxl import Workbook from bank_importer import auth from bank_importer.db import connect, migrate import server from test_server_auth import Client, as_json ROOT = Path(__file__).resolve().parents[1] SAMPLES = ROOT / "流水模板" CCB_SAMPLE = SAMPLES / "中国建设银行账户流水.xls" CITIC_SAMPLE = SAMPLES / "中信银行账户流水.xlsx" BOOTSTRAP_PASSWORD = "BootAdmin123" ADMIN_PASSWORD = "AdminPass123" # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- CCB_HEADER = [ "客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）", "余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注", ] CCB_DATA = [ ["6228480000000000", "测试公司", "2026-01-05 10:00:00", "100.00", "", "99900.00", "RMB", "供应商", "1002003004", "某银行", "货款", ""], ["6228480000000000", "测试公司", "2026-01-06 11:00:00", "", "200.00", "100100.00", "RMB", "客户", "2003004005", "某银行", "收款", ""], ] def workbook_bytes(sheets): """Build an xlsx in memory. ``sheets`` is a list of ``(name, rows)``; an empty ``rows`` list means an empty worksheet. """ workbook = Workbook() workbook.remove(workbook.active) for name, rows in sheets: worksheet = workbook.create_sheet(name) for row in rows: worksheet.append(row) buffer = io.BytesIO() workbook.save(buffer) return buffer.getvalue() # Workbook fixtures are content-hashed for dedupe, so every helper produces # distinct bytes per call (each test uploads a genuinely new file). _counter = itertools.count() def multi_sheet_bytes(): """Valid CCB sheet + unknown-header sheet + empty sheet in one workbook.""" return workbook_bytes( [ ("正常流水", [CCB_HEADER, *CCB_DATA]), ("未知模板", [["日期", "金额", "备注"], ["2026-01-01", next(_counter), "x"]]), ("空表", []), ] ) def empty_workbook_bytes(): return workbook_bytes([("Sheet1", [])]) def single_cell_bytes(): return workbook_bytes([("候选", [[f"只有一个单元格 {next(_counter)}"]])]) def unknown_header_bytes(): return workbook_bytes([("流水", [["日期", "金额", "备注"], ["2026-01-01", next(_counter), "x"]])]) # --------------------------------------------------------------------------- # Test server # --------------------------------------------------------------------------- class ImportApiServerTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.temp_dir = tempfile.TemporaryDirectory() root = Path(cls.temp_dir.name) cls.db_path = root / "app.db" cls.storage = root / "files" cls._old_db_path = server.DB_PATH cls._old_storage = server.STORAGE_DIR server.DB_PATH = cls.db_path server.STORAGE_DIR = cls.storage os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD connection = connect(cls.db_path) migrate(connection) generated = server.ensure_bootstrap_admin(connection) assert generated is None connection.close() class QuietHandler(server.AppHandler): def log_message(self, *args) -> None: pass cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler) cls.port = cls.httpd.server_address[1] cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) cls.thread.start() cls.admin = Client("127.0.0.1", cls.port) status, _, data = cls.admin.post_json( "/api/login", {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"}, ) assert status == 200, data status, _, data = cls.admin.post_json( "/api/password/change", {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD}, ) assert status == 200, data status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "甲公司"}) assert status == 200, data cls.company_a = as_json(data)["company_id"] status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "乙公司"}) assert status == 200, data cls.company_b = as_json(data)["company_id"] cls.cashier_a, cls.cashier_a_password = cls.create_company_user( cls.admin, cls.port, "cashier-a" ) cls.cashier_b, cls.cashier_b_password = cls.create_company_user( cls.admin, cls.port, "cashier-b" ) @classmethod def create_company_user(cls, admin, port, username): status, _, data = admin.post_json( "/api/admin/users", {"username": username, "company_id": cls.company_a} ) assert status == 200, data initial = as_json(data)["initial_password"] client = Client("127.0.0.1", port) status, _, data = client.post_json( "/api/login", {"username": username, "password": initial, "portal": "company"}, ) assert status == 200, data new_password = "Changed456" status, _, data = client.post_json( "/api/password/change", {"old_password": initial, "new_password": new_password}, ) assert status == 200, data return client, new_password @classmethod def tearDownClass(cls) -> None: cls.httpd.shutdown() cls.httpd.server_close() server.DB_PATH = cls._old_db_path server.STORAGE_DIR = cls._old_storage os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None) cls.temp_dir.cleanup() def fresh_client(self) -> Client: return Client("127.0.0.1", self.port) def upload(self, client, content: bytes, filename: str = "语句.xlsx", fields=None): return client.post_multipart( "/api/parse", fields or {}, filename, content ) def upload_for_company(self, content: bytes, company_id: int, filename: str = "语句.xlsx"): return self.admin.post_multipart( "/api/parse", {"company_id": str(company_id)}, filename, content ) def upload_for_a(self, content: bytes, filename: str = "语句.xlsx"): return self.cashier_a.post_multipart("/api/parse", {}, filename, content) def batch_sheets(self, client, batch_id: int): status, _, data = client.get(f"/api/batches/{batch_id}/sheets") self.assertEqual(200, status, data) return as_json(data)["sheets"] def confirm(self, client, batch_id: int, sheets, reason=None): payload = {"sheets": sheets} if reason is not None: payload["reason"] = reason return client.post_json(f"/api/batches/{batch_id}/confirm", payload) def ignore(self, client, batch_id: int, sheets, reason): return client.post_json( f"/api/batches/{batch_id}/ignore", {"sheets": sheets, "reason": reason} ) # ------------------------------------------------------------------ # Multipart validation # ------------------------------------------------------------------ def test_oversized_body_returns_stable_error(self) -> None: content = b"x" * (server.MAX_UPLOAD_BYTES + 1) status, _, data = self.upload_for_a(content) self.assertEqual(422, status) payload = as_json(data) self.assertIn("20 MB", payload["message"]) self.assertNotIn(server.STORAGE_DIR.as_posix(), payload["message"]) self.assertNotIn(str(self.storage), payload["message"]) def test_missing_boundary_returns_stable_error(self) -> None: status, _, data = self.cashier_a.request( "POST", "/api/parse", body=b"whatever", headers={"Content-Type": "multipart/form-data"}, ) self.assertEqual(422, status) self.assertIn("边界", as_json(data)["message"]) def test_wrong_content_type_returns_stable_error(self) -> None: status, _, data = self.cashier_a.request( "POST", "/api/parse", body=CCB_SAMPLE.read_bytes(), headers={"Content-Type": "application/octet-stream"}, ) self.assertEqual(422, status) def test_missing_content_length_returns_stable_error(self) -> None: boundary = "----missingcl" body = f"--{boundary}--\r\n".encode() status, _, data = self.cashier_a.request( "POST", "/api/parse", body=body, headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, ) self.assertEqual(422, status) def test_bad_extension_returns_stable_error(self) -> None: status, _, data = self.upload_for_a(CCB_SAMPLE.read_bytes(), filename="statement.pdf") self.assertEqual(422, status) self.assertIn(".xls", as_json(data)["message"]) def test_signature_mismatch_returns_stable_error(self) -> None: # Valid .xls content renamed with a .xlsx extension must be rejected. status, _, data = self.upload_for_a(CCB_SAMPLE.read_bytes(), filename="fake.xlsx") self.assertEqual(422, status) self.assertIn("损坏", as_json(data)["message"]) def test_corrupt_file_returns_stable_error_without_temp_path(self) -> None: status, _, data = self.upload_for_a(b"PK\x03\x04 not a real zip at all") self.assertEqual(422, status) payload = as_json(data) text = json.dumps(payload, ensure_ascii=False) self.assertNotIn(str(self.storage), text) self.assertNotIn("temp", text.lower()) self.assertNotIn("data/files", text) def test_two_file_parts_rejected(self) -> None: boundary = "----twofilesboundary" content = CCB_SAMPLE.read_bytes() crlf = b"\r\n" body = ( b"--" + boundary.encode() + crlf + b'Content-Disposition: form-data; name="file"; filename="a.xls"' + crlf + b"Content-Type: application/octet-stream" + crlf + crlf + content + crlf + b"--" + boundary.encode() + crlf + b'Content-Disposition: form-data; name="file"; filename="b.xls"' + crlf + b"Content-Type: application/octet-stream" + crlf + crlf + content + crlf + b"--" + boundary.encode() + b"--" + crlf ) status, _, data = self.cashier_a.request( "POST", "/api/parse", body=body, headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, ) self.assertEqual(422, status) def test_no_file_part_rejected(self) -> None: boundary = "----nofile" crlf = b"\r\n" body = ( b"--" + boundary.encode() + crlf + b'Content-Disposition: form-data; name="company_id"' + crlf + crlf + b"1" + crlf + b"--" + boundary.encode() + b"--" + crlf ) status, _, data = self.cashier_a.request( "POST", "/api/parse", body=body, headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, ) self.assertEqual(422, status) self.assertIn("文件", as_json(data)["message"]) def test_upload_trailing_bytes_preserved_exactly(self) -> None: # The stored source file must be byte-identical to what was uploaded # even when it ends with CRLF/LF; the parser must never trim tail # bytes, and the upload temp must not appear anywhere in the payload. content = CCB_SAMPLE.read_bytes() uploaded = content + b"\r\n\r\n\x00\x01" status, _, data = self.upload_for_a(uploaded, filename="尾部字节.xls") self.assertEqual(200, status, data) batch_id = as_json(data)["batch_id"] connection = connect(self.db_path) try: row = connection.execute( """ SELECT f.storage_path FROM source_files f JOIN import_batches b ON b.source_file_id = f.id WHERE b.id = ? """, (batch_id,), ).fetchone() finally: connection.close() self.assertIsNotNone(row) self.assertEqual(uploaded, Path(row["storage_path"]).read_bytes()) self.assertNotIn("upload-", json.dumps(as_json(data), ensure_ascii=False)) # ------------------------------------------------------------------ # Per-sheet results and four-item evidence # ------------------------------------------------------------------ def test_multi_sheet_workbook_returns_every_sheet_result(self) -> None: status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="多工作表.xlsx") self.assertEqual(200, status, data) payload = as_json(data) self.assertEqual("parsed", payload["status"]) self.assertEqual("多工作表.xlsx", payload.get("original_filename")) sheets = payload["sheets"] self.assertEqual(3, len(sheets)) by_name = {sheet["sheet_name"]: sheet for sheet in sheets} self.assertEqual("parsed", by_name["正常流水"]["outcome"]) self.assertEqual("pending", by_name["正常流水"]["review_status"]) self.assertEqual("中国建设银行", by_name["正常流水"]["bank"]) self.assertEqual(2, by_name["正常流水"]["transactions"]) unknown = by_name["未知模板"] self.assertEqual("exception", unknown["outcome"]) self.assertIn("未识别到受支持的银行表头", unknown["message"]) self.assertGreaterEqual(unknown["scanned_rows"], 2) self.assertTrue(unknown["candidate_headers"]) self.assertIn("日期、金额、备注", "；".join(unknown["candidate_headers"])) empty = by_name["空表"] self.assertEqual("ignored", empty["outcome"]) self.assertIn("为空", empty["message"]) self.assertEqual(0, empty["scanned_rows"]) def test_batch_detail_matches_parse_response(self) -> None: status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="详情.xlsx") batch_id = as_json(data)["batch_id"] sheets = self.batch_sheets(self.cashier_a, batch_id) self.assertEqual(3, len(sheets)) status, _, data = self.cashier_a.get(f"/api/batches/{batch_id}/sheets") self.assertEqual(200, status) detail = as_json(data) self.assertEqual("详情.xlsx", detail["original_filename"]) def test_unknown_header_file_returns_four_evidence_items(self) -> None: status, _, data = self.upload_for_a(unknown_header_bytes(), filename="未知银行.xlsx") self.assertEqual(422, status, data) payload = as_json(data) self.assertEqual("exception", payload["status"]) self.assertEqual("未知银行.xlsx", payload.get("original_filename")) sheet = payload["sheets"][0] self.assertEqual("exception", sheet["outcome"]) self.assertEqual("流水", sheet["sheet_name"]) self.assertGreaterEqual(sheet["scanned_rows"], 1) self.assertTrue(sheet["candidate_headers"]) text = json.dumps(payload, ensure_ascii=False) for evidence in (str(self.storage), "data/files", ".uploads", "upload-"): self.assertNotIn(evidence, text) def test_empty_workbook_returns_evidence(self) -> None: status, _, data = self.upload_for_a(empty_workbook_bytes(), filename="空工作簿.xlsx") self.assertEqual(422, status, data) payload = as_json(data) self.assertEqual("空工作簿.xlsx", payload.get("original_filename")) sheet = payload["sheets"][0] self.assertEqual("ignored", sheet["outcome"]) self.assertEqual("Sheet1", sheet["sheet_name"]) self.assertEqual(0, sheet["scanned_rows"]) def test_single_cell_sheet_returns_candidate_evidence(self) -> None: status, _, data = self.upload_for_a(single_cell_bytes(), filename="单格.xlsx") self.assertEqual(422, status, data) sheet = as_json(data)["sheets"][0] self.assertEqual("exception", sheet["outcome"]) self.assertEqual("候选", sheet["sheet_name"]) self.assertEqual(1, sheet["scanned_rows"]) self.assertTrue(sheet["candidate_headers"]) self.assertIn("只有一个单元格", "；".join(sheet["candidate_headers"])) # ------------------------------------------------------------------ # Confirm / ignore lifecycle # ------------------------------------------------------------------ def test_confirm_and_ignore_lifecycle(self) -> None: status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="确认.xlsx") payload = as_json(data) batch_id = payload["batch_id"] names = [s["sheet_name"] for s in payload["sheets"]] self.assertIn("正常流水", names) # Confirm the parsed sheet; exception/ignored cannot be confirmed. status, _, data = self.confirm(self.cashier_a, batch_id, ["正常流水"]) self.assertEqual(200, status, data) result = as_json(data) self.assertEqual(["正常流水"], result["updated"]) sheet = next(s for s in result["sheets"] if s["sheet_name"] == "正常流水") self.assertEqual("confirmed", sheet["review_status"]) # Confirming an exception sheet is a conflict. status, _, data = self.confirm(self.cashier_a, batch_id, ["未知模板"]) self.assertEqual(409, status, data) # Ignore the exception sheet with a reason. status, _, data = self.ignore(self.cashier_a, batch_id, ["未知模板"], "模板待补充") self.assertEqual(200, status, data) sheet = next(s for s in as_json(data)["sheets"] if s["sheet_name"] == "未知模板") self.assertEqual("ignored", sheet["review_status"]) self.assertEqual("模板待补充", sheet["review_reason"]) # Idempotent repeat of an already-applied decision. status, _, data = self.confirm(self.cashier_a, batch_id, ["正常流水"]) self.assertEqual(200, status, data) result = as_json(data) self.assertEqual([], result["updated"]) self.assertEqual(["正常流水"], result["already"]) # Changing a settled decision is a conflict. status, _, data = self.ignore(self.cashier_a, batch_id, ["正常流水"], "改主意") self.assertEqual(409, status, data) def test_ignore_requires_reason(self) -> None: status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="忽略原因.xlsx") batch_id = as_json(data)["batch_id"] status, _, data = self.cashier_a.post_json( f"/api/batches/{batch_id}/ignore", {"sheets": ["未知模板"], "reason": " "} ) self.assertEqual(400, status, data) self.assertIn("原因", as_json(data)["message"]) def test_unknown_sheet_name_is_rejected_and_atomic(self) -> None: status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="原子.xlsx") batch_id = as_json(data)["batch_id"] # One valid + one unknown sheet in the same request must not leave a # half-applied confirmation behind. status, _, data = self.confirm(self.cashier_a, batch_id, ["正常流水", "不存在"]) self.assertEqual(400, status, data) sheet = next( s for s in self.batch_sheets(self.cashier_a, batch_id) if s["sheet_name"] == "正常流水" ) self.assertEqual("pending", sheet["review_status"]) def test_company_cannot_confirm_another_companys_batch(self) -> None: status, _, data = self.upload_for_company(multi_sheet_bytes(), self.company_b) batch_b = as_json(data)["batch_id"] status, _, data = self.confirm(self.cashier_a, batch_b, ["正常流水"]) self.assertEqual(404, status, data) status, _, data = self.cashier_a.get(f"/api/batches/{batch_b}/sheets") self.assertEqual(404, status, data) def test_export_excludes_unconfirmed_and_ignore_keeps_excluded(self) -> None: def export_line_count() -> int: status, _, data = self.cashier_a.get("/api/export.csv") self.assertEqual(200, status) return len([line for line in data.decode("utf-8-sig").splitlines() if line]) baseline = export_line_count() # Unconfirmed upload: export stays unchanged (no downstream leak). status, _, data = self.upload_for_a(CCB_SAMPLE.read_bytes(), filename="导出.xls") batch_id = as_json(data)["batch_id"] self.assertEqual("parsed", as_json(data)["status"]) self.assertEqual(baseline, export_line_count()) # Confirm → the sheet's rows enter the export. sheets = self.batch_sheets(self.cashier_a, batch_id) name = sheets[0]["sheet_name"] status, _, data = self.confirm(self.cashier_a, batch_id, [name]) self.assertEqual(200, status, data) self.assertGreater(export_line_count(), baseline) # A confirmed sheet cannot be silently revoked: the settled decision # is a conflict, not a silent data removal. status, _, data = self.ignore(self.cashier_a, batch_id, [name], "取消") self.assertEqual(409, status, data) self.assertGreater(export_line_count(), baseline) # A sheet that was only ever ignored never enters the export. status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="忽略.xlsx") batch2 = as_json(data)["batch_id"] parsed = [ sheet for sheet in self.batch_sheets(self.cashier_a, batch2) if sheet["outcome"] == "parsed" ] self.assertTrue(parsed) before_ignore = export_line_count() status, _, data = self.ignore( self.cashier_a, batch2, [parsed[0]["sheet_name"]], "该表复核后不采用" ) self.assertEqual(200, status, data) self.assertEqual(before_ignore, export_line_count()) # ------------------------------------------------------------------ # Concurrency # ------------------------------------------------------------------ def test_concurrent_duplicate_upload_keeps_one_fact_set(self) -> None: content = CITIC_SAMPLE.read_bytes() results: list[tuple[int, dict]] = [] errors: list[Exception] = [] barrier = threading.Barrier(2) def upload() -> None: try: client = self.fresh_client() client.cookies.update(self.cashier_a.cookies) barrier.wait(timeout=10) status, _, data = client.post_multipart( "/api/parse", {}, "并发.xlsx", content ) results.append((status, as_json(data))) except Exception as exc: # pragma: no cover errors.append(exc) threads = [threading.Thread(target=upload) for _ in range(2)] for thread in threads: thread.start() for thread in threads: thread.join(timeout=30) self.assertEqual([], errors) self.assertEqual(2, len(results)) statuses = sorted(payload["status"] for _, payload in results) self.assertEqual(["duplicate", "parsed"], statuses) connection = connect(self.db_path) try: # The concurrent upload adds exactly one source file: identical # bytes never create a second fact set. count = connection.execute( "SELECT COUNT(*) AS n FROM source_files WHERE original_filename = '并发.xlsx'" ).fetchone()["n"] self.assertEqual(1, count) duplicates = connection.execute( "SELECT COUNT(*) AS n FROM import_batches WHERE status = 'duplicate'" ).fetchone()["n"] self.assertGreaterEqual(duplicates, 1) finally: connection.close() if __name__ == "__main__": unittest.main()