"""发票提取编排模块单元测试 覆盖范围： - extract_invoices：空目录、提取失败、正常流程、分类结果、申请单分离 """ from __future__ import annotations from pathlib import Path from typing import Any from src.doc.extractor import extract_invoices # 字段键名（与源码中的字符串字面量保持一致） K_INVOICE_TYPE = "invoice_type" K_INVOICE_NUMBER = "invoice_number" K_TOTAL_AMOUNT = "total_amount" K_ITEM_NAME = "item_name" K_PERSON_NAME = "person_name" K_CARD_DATE = "card_date" K_CARD_NO = "card_no" K_CARD_AMOUNT = "card_amount" K_MATCHED_INVOICES = "_matched_invoices" K_RELATIVE_INVOICE_COUNT = "relative_invoice_count" K_INVOICE_DETAIL = "invoice_detail" K_REMARK = "remark" # 发票类型 INVOICE_TYPE_TRAIN = "train" INVOICE_TYPE_HOTEL = "hotel" INVOICE_TYPE_GENERAL = "general" INVOICE_TYPE_PAYMENT = "payment" DOCUMENT_TYPE_APPLICATION = "application" # ------------------------------------------------------------------ # Fixture helpers # ------------------------------------------------------------------ def _make_invoice(number: str, amount: float, inv_type: str = INVOICE_TYPE_GENERAL) -> dict[str, Any]: inv: dict[str, Any] = { K_INVOICE_NUMBER: number, K_INVOICE_TYPE: inv_type, K_TOTAL_AMOUNT: str(amount), } if inv_type == INVOICE_TYPE_TRAIN: inv[K_PERSON_NAME] = f"person{number}" elif inv_type == INVOICE_TYPE_GENERAL: inv[K_ITEM_NAME] = f"item{number}" return inv def _make_application() -> dict[str, Any]: return { K_INVOICE_TYPE: DOCUMENT_TYPE_APPLICATION, "applicant": "张三", } def _make_card(amount: float) -> dict[str, Any]: return { K_INVOICE_TYPE: INVOICE_TYPE_PAYMENT, K_CARD_DATE: "2026-01-01", K_CARD_NO: "6228480000000000", K_CARD_AMOUNT: f"{amount:.2f}", } # ------------------------------------------------------------------ # extract_invoices # ------------------------------------------------------------------ class TestExtractInvoices: """extract_invoices 编排函数""" def test_empty_directory(self, tmp_path: Path, monkeypatch): monkeypatch.setattr("src.doc.extractor._find_all_files", lambda d: []) records, apps, groups = extract_invoices(str(tmp_path)) assert records == [] assert apps == [] assert groups == {"travel": [], "general": [], "application": []} def test_extraction_fails(self, tmp_path: Path, monkeypatch): pdf = tmp_path / "broken.pdf" pdf.touch() monkeypatch.setattr("src.doc.extractor._find_all_files", lambda d: [pdf]) monkeypatch.setattr("src.doc.extractor._extract_document", lambda p, c: None) records, apps, groups = extract_invoices(str(tmp_path)) assert records == [] assert apps == [] assert groups == {"travel": [], "general": [], "application": []} def test_normal_flow_general_invoices(self, tmp_path: Path, monkeypatch): pdf1 = tmp_path / "inv1.pdf" pdf2 = tmp_path / "inv2.pdf" pdf1.touch() pdf2.touch() inv1 = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL) inv2 = _make_invoice("INV002", 200.0, INVOICE_TYPE_GENERAL) monkeypatch.setattr("src.doc.extractor._find_all_files", lambda d: [pdf1, pdf2]) call_index = [0] def fake_extract(path, cache_dir): idx = call_index[0] call_index[0] += 1 return inv1 if idx == 0 else inv2 monkeypatch.setattr("src.doc.extractor._extract_document", fake_extract) def fake_match(invoices, cards): return [ { K_CARD_DATE: "2026-01-01", K_CARD_NO: "6228480000000000", K_CARD_AMOUNT: "500.00", K_RELATIVE_INVOICE_COUNT: "2", K_INVOICE_DETAIL: "", K_REMARK: "", K_MATCHED_INVOICES: [inv1, inv2], } ] monkeypatch.setattr("src.doc.extractor.match_invoices_to_cards", fake_match) records, apps, groups = extract_invoices(str(tmp_path)) assert len(records) == 1 assert records[0][K_RELATIVE_INVOICE_COUNT] == "2" assert len(groups["travel"]) == 0 assert len(groups["general"]) == 2 def test_normal_flow_mixed_invoices(self, tmp_path: Path, monkeypatch): pdf1 = tmp_path / "train.pdf" pdf2 = tmp_path / "hotel.pdf" pdf3 = tmp_path / "general.pdf" pdf1.touch() pdf2.touch() pdf3.touch() inv_train = _make_invoice("TRAIN001", 500.0, INVOICE_TYPE_TRAIN) inv_hotel = _make_invoice("HOTEL001", 800.0, INVOICE_TYPE_HOTEL) inv_general = _make_invoice("GEN001", 150.0, INVOICE_TYPE_GENERAL) monkeypatch.setattr( "src.doc.extractor._find_all_files", lambda d: [pdf1, pdf2, pdf3], ) invoices_list = [inv_train, inv_hotel, inv_general] call_index = [0] def fake_extract(path, cache_dir): idx = call_index[0] call_index[0] += 1 return invoices_list[idx] monkeypatch.setattr("src.doc.extractor._extract_document", fake_extract) def fake_match(invoices, cards): return [ { K_CARD_DATE: "2026-01-01", K_CARD_NO: "6228480000000000", K_CARD_AMOUNT: "500.00", K_RELATIVE_INVOICE_COUNT: "1", K_INVOICE_DETAIL: "", K_REMARK: "", K_MATCHED_INVOICES: [inv_train], }, { K_CARD_DATE: "2026-01-02", K_CARD_NO: "6228480000000000", K_CARD_AMOUNT: "800.00", K_RELATIVE_INVOICE_COUNT: "1", K_INVOICE_DETAIL: "", K_REMARK: "", K_MATCHED_INVOICES: [inv_hotel], }, { K_CARD_DATE: "", K_CARD_NO: "", K_CARD_AMOUNT: "", K_RELATIVE_INVOICE_COUNT: "1", K_INVOICE_DETAIL: "", K_REMARK: "unmatched", K_MATCHED_INVOICES: [inv_general], }, ] monkeypatch.setattr("src.doc.extractor.match_invoices_to_cards", fake_match) records, apps, groups = extract_invoices(str(tmp_path)) assert len(records) == 3 assert len(groups["travel"]) == 2 assert len(groups["general"]) == 1 travel_numbers = {inv[K_INVOICE_NUMBER] for inv in groups["travel"]} assert "TRAIN001" in travel_numbers assert "HOTEL001" in travel_numbers assert groups["general"][0][K_INVOICE_NUMBER] == "GEN001" def test_application_documents_separated(self, tmp_path: Path, monkeypatch): pdf1 = tmp_path / "invoice.pdf" pdf2 = tmp_path / "application.pdf" pdf1.touch() pdf2.touch() inv = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL) app = _make_application() monkeypatch.setattr( "src.doc.extractor._find_all_files", lambda d: [pdf1, pdf2], ) results = [inv, app] call_index = [0] def fake_extract(path, cache_dir): idx = call_index[0] call_index[0] += 1 return results[idx] monkeypatch.setattr("src.doc.extractor._extract_document", fake_extract) def fake_match(invoices, cards): return [ { K_CARD_DATE: "2026-01-01", K_CARD_NO: "6228480000000000", K_CARD_AMOUNT: "300.00", K_RELATIVE_INVOICE_COUNT: "1", K_INVOICE_DETAIL: "", K_REMARK: "", K_MATCHED_INVOICES: [inv], } ] monkeypatch.setattr("src.doc.extractor.match_invoices_to_cards", fake_match) records, apps, groups = extract_invoices(str(tmp_path)) assert len(apps) == 1 assert apps[0]["applicant"] == "张三" assert len(groups["application"]) == 1 def test_payment_records_included(self, tmp_path: Path, monkeypatch): pdf1 = tmp_path / "invoice.pdf" pdf2 = tmp_path / "card.png" pdf1.touch() pdf2.touch() inv = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL) card = _make_card(300.0) monkeypatch.setattr( "src.doc.extractor._find_all_files", lambda d: [pdf1, pdf2], ) results = [inv, card] call_index = [0] def fake_extract(path, cache_dir): idx = call_index[0] call_index[0] += 1 return results[idx] monkeypatch.setattr("src.doc.extractor._extract_document", fake_extract) def fake_match(invoices, cards): return [ { K_CARD_DATE: "2026-01-01", K_CARD_NO: "6228480000000000", K_CARD_AMOUNT: "300.00", K_RELATIVE_INVOICE_COUNT: "1", K_INVOICE_DETAIL: "", K_REMARK: "", K_MATCHED_INVOICES: [inv], } ] monkeypatch.setattr("src.doc.extractor.match_invoices_to_cards", fake_match) records, apps, groups = extract_invoices(str(tmp_path)) assert len(records) == 1 assert len(groups["general"]) == 1 def test_partial_extraction_failure(self, tmp_path: Path, monkeypatch): pdf1 = tmp_path / "good.pdf" pdf2 = tmp_path / "bad.pdf" pdf1.touch() pdf2.touch() inv = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL) monkeypatch.setattr( "src.doc.extractor._find_all_files", lambda d: [pdf1, pdf2], ) results = [inv, None] call_index = [0] def fake_extract(path, cache_dir): idx = call_index[0] call_index[0] += 1 return results[idx] monkeypatch.setattr("src.doc.extractor._extract_document", fake_extract) def fake_match(invoices, cards): return [ { K_CARD_DATE: "2026-01-01", K_CARD_NO: "6228480000000000", K_CARD_AMOUNT: "300.00", K_RELATIVE_INVOICE_COUNT: "1", K_INVOICE_DETAIL: "", K_REMARK: "", K_MATCHED_INVOICES: [inv], } ] monkeypatch.setattr("src.doc.extractor.match_invoices_to_cards", fake_match) records, apps, groups = extract_invoices(str(tmp_path)) assert len(records) == 1 assert len(groups["general"]) == 1