341 lines
11 KiB
Python
341 lines
11 KiB
Python
"""发票提取编排模块单元测试
|
||
|
||
覆盖范围：
|
||
- 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
|