wandering1/Auto-Finance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
44496a54e62bc0d64d6c91c9a3eb3e4899edea23
Auto-Finance/tests/test_extractor.py
T
wandering 44496a54e6 fix: 补充 __init__.py 导出符号，修复测试导入错误 
- src/core/extraction/__init__.py: 导出 _image_to_base64
- src/core/matching/__init__.py: 导出所有内部函数（_build_invoice_summary, _build_payment_records, _invoices_to_records, _match, _match_by_filename, _match_one_to_many, _match_one_to_one, _relative_tolerance, _safe_float）
- src/infra/documents/__init__.py: 导出 INVOICE_LEVEL_COLUMNS, PAYMENT_RECORD_COLUMNS
- tests/test_extractor.py: 修正 monkeypatch 路径为模块级引用
2026-07-02 18:51:59 +08:00

333 lines
11 KiB
Python
Raw Blame History

"""发票提取编排模块单元测试
覆盖范围：
- extract_invoices：空目录、提取失败、正常流程、分类结果、申请单分离
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import pytest
from src import exceptions
from src.core.extraction 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.core.extraction.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.core.extraction.extractor._find_all_files", lambda d: [pdf])
monkeypatch.setattr("src.core.extraction.extractor._extract_document", lambda p, c, s: (None, "parse error"))
with pytest.raises(exceptions.ExtractionError) as exc_info:
extract_invoices(str(tmp_path))
assert "broken.pdf" in exc_info.value.failed_files
assert exc_info.value.details["broken.pdf"] == "parse error"
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.core.extraction.extractor._find_all_files", lambda d: [pdf1, pdf2])
path_to_result = {pdf1: inv1, pdf2: inv2}
def fake_extract(path, cache_dir, source_dir):
return (path_to_result[path], None)
monkeypatch.setattr("src.core.extraction.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.core.extraction.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.core.extraction.extractor._find_all_files",
lambda d: [pdf1, pdf2, pdf3],
)
invoices_list = [inv_train, inv_hotel, inv_general]
path_to_result = dict(zip([pdf1, pdf2, pdf3], invoices_list, strict=True))
def fake_extract(path, cache_dir, source_dir):
return (path_to_result[path], None)
monkeypatch.setattr("src.core.extraction.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.core.extraction.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.core.extraction.extractor._find_all_files",
lambda d: [pdf1, pdf2],
)
path_to_result = {pdf1: inv, pdf2: app}
def fake_extract(path, cache_dir, source_dir):
return (path_to_result[path], None)
monkeypatch.setattr("src.core.extraction.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.core.extraction.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.core.extraction.extractor._find_all_files",
lambda d: [pdf1, pdf2],
)
path_to_result = {pdf1: inv, pdf2: card}
def fake_extract(path, cache_dir, source_dir):
return (path_to_result[path], None)
monkeypatch.setattr("src.core.extraction.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.core.extraction.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.core.extraction.extractor._find_all_files",
lambda d: [pdf1, pdf2],
)
path_to_result = {pdf1: inv, pdf2: None}
def fake_extract(path, cache_dir, source_dir):
result = path_to_result[path]
return (result, "parse error") if result is None else (result, None)
monkeypatch.setattr("src.core.extraction.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.core.extraction.extractor.match_invoices_to_cards", fake_match)
records, apps, groups = extract_invoices(str(tmp_path))
assert len(records) == 1
assert len(groups["general"]) == 1
Reference in New Issue View Git Blame Copy Permalink