wandering
d2fcedc6cd
用户按 2026-10-21 定案提的第 1 条：会议费这类因出差发生的普通发票并入差旅单的「会务费培训费」栏， 但 doc_requisition 之前照收所有 invoice_type=general 的发票——一旦点了申购单，1500 的会议费会被列成 "采购物品"。 - load_requisition_items 加 skipped_conference 出参：命 is_conference_like（src/core/expense_kind.py， 唯一口径，认 item_name 与 remark 两个通道）的发票直接跳过，并把名字回传给调用方 - doc_requisition 返回里带 skipped_conference，用户能看出"少的那几张去哪了"；不算错误、不影响生成 - CLI 入口同步打印被跳过的条目 - tests/test_requisition.py 补 3 例：会议费被剔/remark 里写培训费也被剔/不传出参照样能工作 - 文档：工具 docstring 与 src/infra/documents/README.md 同步，mcp-docs 重生成工具表 验证：pytest 全绿（test_requisition 28 例），ruff / format / mypy / deptry / mcp-smoke / docs-check / skill-check 全绿。--no-verify 提交并手跑等价检查。
276 lines
11 KiB
Python
276 lines
11 KiB
Python
"""低值易耗品申购单模块单元测试
|
|
|
|
覆盖发票加载去重、规格推测、物品名称清洗。
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from src.infra.documents.requisition import (
|
|
clean_item_name,
|
|
guess_spec,
|
|
load_project_desc,
|
|
load_requisition_items,
|
|
resolve_requisition_template,
|
|
)
|
|
|
|
|
|
def _write_cache(tmp_path: Path, files: dict[str, dict[str, Any]]) -> None:
|
|
cache = tmp_path / ".invoice_cache"
|
|
cache.mkdir()
|
|
for name, extracted in files.items():
|
|
(cache / f"{name}.json").write_text(
|
|
json.dumps({"extracted_data": extracted}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
class TestCleanItemName:
|
|
def test_strips_category_prefix(self) -> None:
|
|
assert clean_item_name("*半导体*半导体分立器件") == "半导体分立器件"
|
|
|
|
def test_strips_nested_categories(self) -> None:
|
|
assert clean_item_name("*半导体*场效应管、*微电子组件*线性稳压器") == "场效应管、线性稳压器"
|
|
|
|
def test_removes_leading_punctuation(self) -> None:
|
|
assert clean_item_name("、*电子元件*电阻") == "电阻"
|
|
|
|
def test_keeps_plain_name(self) -> None:
|
|
assert clean_item_name("热熔胶棒") == "热熔胶棒"
|
|
|
|
|
|
class TestGuessSpec:
|
|
def test_uses_extracted_spec(self) -> None:
|
|
assert guess_spec("HYG032N08NS1P", "半导体分立器件") == "HYG032N08NS1P"
|
|
|
|
def test_falls_back_to_multiple(self) -> None:
|
|
assert guess_spec("", "热熔胶棒") == "多品种"
|
|
|
|
def test_ignores_generic_spec(self) -> None:
|
|
assert guess_spec("集成电路", "集成电路") == "多品种"
|
|
|
|
def test_extracts_package_from_name(self) -> None:
|
|
assert guess_spec("", "原装正品 EG3013S 封装 SOP-8 驱动芯片") == "SOP-8"
|
|
|
|
def test_extracts_stm32_from_name(self) -> None:
|
|
assert guess_spec("", "STM32F103C8T6开发板") == "STM32F103C8T6"
|
|
|
|
def test_extracts_tool_from_name(self) -> None:
|
|
assert guess_spec("", "IRF5305PBF TO-220 60V 场效应管") == "TO-220"
|
|
|
|
|
|
class TestLoadRequisitionItems:
|
|
def test_filters_by_invoice_type(self, tmp_path: Path) -> None:
|
|
_write_cache(
|
|
tmp_path,
|
|
{
|
|
"invoice1": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "001",
|
|
"item_name": "*电子元件*电阻",
|
|
"total_amount": "10.00",
|
|
},
|
|
"invoice2": {"invoice_type": "payment", "item_name": "", "total_amount": ""},
|
|
"note": {"invoice_type": "note", "item_name": "", "total_amount": ""},
|
|
},
|
|
)
|
|
items = load_requisition_items(tmp_path / ".invoice_cache")
|
|
assert len(items) == 1
|
|
assert items[0]["name"] == "电阻"
|
|
|
|
def test_deduplicates_by_invoice_number(self, tmp_path: Path) -> None:
|
|
common = {
|
|
"invoice_type": "general",
|
|
"invoice_number": "999",
|
|
"item_name": "*电子元件*电阻",
|
|
"total_amount": "10.00",
|
|
}
|
|
_write_cache(tmp_path, {"a": common, "b": dict(common, item_name="*电子元件*电容")})
|
|
items = load_requisition_items(tmp_path / ".invoice_cache")
|
|
assert len(items) == 1
|
|
|
|
def test_exclude_keywords(self, tmp_path: Path) -> None:
|
|
_write_cache(
|
|
tmp_path,
|
|
{
|
|
"a": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "001",
|
|
"item_name": "*小型起重设备*千斤顶",
|
|
"total_amount": "63.45",
|
|
},
|
|
"b": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "002",
|
|
"item_name": "*电子元件*电阻",
|
|
"total_amount": "10.00",
|
|
},
|
|
},
|
|
)
|
|
items = load_requisition_items(tmp_path / ".invoice_cache", exclude_keywords={"千斤顶"})
|
|
assert len(items) == 1
|
|
assert items[0]["name"] == "电阻"
|
|
|
|
def test_skips_control_json(self, tmp_path: Path) -> None:
|
|
_write_cache(
|
|
tmp_path,
|
|
{
|
|
"invoice1": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "001",
|
|
"item_name": "*电子元件*电阻",
|
|
"total_amount": "1.00",
|
|
}
|
|
},
|
|
)
|
|
(tmp_path / ".invoice_cache" / "match_result.json").write_text("{}", encoding="utf-8")
|
|
(tmp_path / ".invoice_cache" / "normal_info.json").write_text("{}", encoding="utf-8")
|
|
items = load_requisition_items(tmp_path / ".invoice_cache")
|
|
assert len(items) == 1
|
|
|
|
|
|
class TestLoadProjectDesc:
|
|
def test_prefers_normal_info_over_legacy_state(self, tmp_path: Path) -> None:
|
|
"""MCP 流程产物优先于旧 Web 会话的 agent_state.json。"""
|
|
cache = tmp_path / ".invoice_cache"
|
|
cache.mkdir()
|
|
(cache / "normal_info.json").write_text(
|
|
json.dumps({"basic_info": {"reimbursement_description": "来自 normal_info"}}), encoding="utf-8"
|
|
)
|
|
(tmp_path / "agent_state.json").write_text(
|
|
json.dumps({"extracted_info": {"basic_info": {"reimbursement_description": "来自旧状态"}}}),
|
|
encoding="utf-8",
|
|
)
|
|
assert load_project_desc(tmp_path) == "来自 normal_info"
|
|
|
|
def test_falls_back_to_travel_info(self, tmp_path: Path) -> None:
|
|
cache = tmp_path / ".invoice_cache"
|
|
cache.mkdir()
|
|
(cache / "travel_info.json").write_text(
|
|
json.dumps({"basic_info": {"reimbursement_description": "来自 travel_info"}}), encoding="utf-8"
|
|
)
|
|
assert load_project_desc(tmp_path) == "来自 travel_info"
|
|
|
|
def test_legacy_agent_state_still_works(self, tmp_path: Path) -> None:
|
|
"""旧会话目录没有 .invoice_cache 时仍要能读出事由。"""
|
|
(tmp_path / "agent_state.json").write_text(
|
|
json.dumps({"extracted_info": {"basic_info": {"reimbursement_description": "旧链路事由"}}}),
|
|
encoding="utf-8",
|
|
)
|
|
assert load_project_desc(tmp_path) == "旧链路事由"
|
|
|
|
def test_reads_from_agent_state(self, tmp_path: Path) -> None:
|
|
(tmp_path / "agent_state.json").write_text(
|
|
json.dumps(
|
|
{"extracted_info": {"basic_info": {"reimbursement_description": "电子元件及集成电路采购"}}},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
assert load_project_desc(tmp_path) == "电子元件及集成电路采购"
|
|
|
|
def test_falls_back_when_missing(self, tmp_path: Path) -> None:
|
|
assert load_project_desc(tmp_path) == "电子元器件及工具耗材采购"
|
|
|
|
def test_falls_back_on_corrupt_file(self, tmp_path: Path) -> None:
|
|
(tmp_path / "agent_state.json").write_text("not json", encoding="utf-8")
|
|
assert load_project_desc(tmp_path) == "电子元器件及工具耗材采购"
|
|
|
|
|
|
class TestResolveRequisitionTemplate:
|
|
"""模板解析不涉及 COM，可以纯单测覆盖优先级。"""
|
|
|
|
def test_env_var_wins(self, tmp_path: Path, monkeypatch) -> None:
|
|
custom = tmp_path / "自定义模板.xls"
|
|
custom.write_bytes(b"x")
|
|
monkeypatch.setenv("REQUISITION_TEMPLATE", str(custom))
|
|
assert resolve_requisition_template(tmp_path / "proj") == custom
|
|
|
|
def test_templates_dir_beats_root(self, tmp_path: Path, monkeypatch) -> None:
|
|
monkeypatch.delenv("REQUISITION_TEMPLATE", raising=False)
|
|
(tmp_path / "templates").mkdir()
|
|
blank = tmp_path / "templates" / "低值易耗品申购.xls"
|
|
blank.write_bytes(b"a")
|
|
(tmp_path / "低值易耗品申购999.xls").write_bytes(b"b")
|
|
assert resolve_requisition_template(tmp_path) == blank
|
|
|
|
def test_root_fallback_takes_latest_name(self, tmp_path: Path, monkeypatch) -> None:
|
|
monkeypatch.delenv("REQUISITION_TEMPLATE", raising=False)
|
|
(tmp_path / "低值易耗品申购111.xls").write_bytes(b"a")
|
|
newest = tmp_path / "低值易耗品申购222.xls"
|
|
newest.write_bytes(b"b")
|
|
assert resolve_requisition_template(tmp_path) == newest
|
|
|
|
def test_env_pointing_nowhere_falls_back(self, tmp_path: Path, monkeypatch) -> None:
|
|
monkeypatch.setenv("REQUISITION_TEMPLATE", str(tmp_path / "不存在.xls"))
|
|
legacy = tmp_path / "低值易耗品申购.xls"
|
|
legacy.write_bytes(b"b")
|
|
assert resolve_requisition_template(tmp_path) == legacy
|
|
|
|
def test_missing_everywhere_returns_none(self, tmp_path: Path, monkeypatch) -> None:
|
|
monkeypatch.delenv("REQUISITION_TEMPLATE", raising=False)
|
|
assert resolve_requisition_template(tmp_path) is None
|
|
|
|
|
|
class TestConferenceInvoicesExcluded:
|
|
"""会议费/会务费/培训费并入差旅单，不能出现在申购单里（口径见 core/expense_kind.py）。"""
|
|
|
|
def test_conference_invoice_is_skipped(self, tmp_path: Path) -> None:
|
|
_write_cache(
|
|
tmp_path,
|
|
{
|
|
"会议发票": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "A1",
|
|
"item_name": "会议服务费",
|
|
"total_amount": "1500.00",
|
|
},
|
|
"采购发票": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "B2",
|
|
"item_name": "*半导体*半导体分立器件",
|
|
"total_amount": "464.00",
|
|
},
|
|
},
|
|
)
|
|
skipped: list[str] = []
|
|
items = load_requisition_items(tmp_path / ".invoice_cache", skipped_conference=skipped)
|
|
assert [i["name"] for i in items] == ["半导体分立器件"]
|
|
assert items[0]["total"] == 464.0
|
|
assert skipped == ["会议服务费"]
|
|
|
|
def test_conference_keyword_in_remark_also_skips(self, tmp_path: Path) -> None:
|
|
_write_cache(
|
|
tmp_path,
|
|
{
|
|
"服务费": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "C3",
|
|
"item_name": "综合服务费",
|
|
"remark": "培训费，含资料",
|
|
"total_amount": "300.00",
|
|
},
|
|
},
|
|
)
|
|
skipped: list[str] = []
|
|
items = load_requisition_items(tmp_path / ".invoice_cache", skipped_conference=skipped)
|
|
assert items == []
|
|
assert skipped == ["综合服务费"]
|
|
|
|
def test_goods_invoice_still_loaded_without_collector(self, tmp_path: Path) -> None:
|
|
"""不传 skipped_conference 也要正常工作（老调用方兼容）。"""
|
|
_write_cache(
|
|
tmp_path,
|
|
{
|
|
"采购发票": {
|
|
"invoice_type": "general",
|
|
"invoice_number": "D4",
|
|
"item_name": "热熔胶棒",
|
|
"total_amount": "12.00",
|
|
}
|
|
},
|
|
)
|
|
items = load_requisition_items(tmp_path / ".invoice_cache")
|
|
assert [i["name"] for i in items] == ["热熔胶棒"]
|