wandering
7e900ebdea
历史清洗删掉了根目录两份 低值易耗品申购*.xls——它们其实是**已填写的真实申购单** （半导体型号、单价、项目名都在），不是模板；同时旧实现「项目根 glob 取名字最大者」会把 上一次生成的产物当模板，所以一并修掉： - 新增 resolve_requisition_template()：REQUISITION_TEMPLATE 环境变量 → templates/ → 项目根（兼容旧位置）； 环境变量指向不存在的文件时告警并继续回退，不静默失败 - 入库 templates/低值易耗品申购.xls 空白件（由旧文件删掉第 3 行起的全部内容得到，UsedRange 只剩 2 行： 标题 + 表头；真实采购明细不进仓库） - generate_requisition 默认输出改为 <会话>/artifacts/低值易耗品申购<ts>.xls，不再写模板旁边 - mcp/tools/documents.py 的 _find_template 改为复用同一解析函数，避免两处规则漂移 - 模块 docstring 与 CLI help 里 src/web/uploads 的过时示例改为 cases/ - tests：5 个纯单测覆盖解析优先级（env 优先 / templates 胜过根目录 / 根目录取最新 / env 指向不存在的文件时回退 / 全都没有时返回 None），不依赖 COM 验证：真实 Excel COM 端到端 2 项 → 产物 UsedRange 5 行（标题/表头/2 明细/合计），第 6 行起为空， 无旧明细残留；fill_requisition_excel 原本就会先删第 3 行起内容，这条也实测确认过。 门禁：ruff / format --check / mypy src / deptry 全绿，pytest 112 passed，mcp_smoke 17 工具。
214 lines
8.6 KiB
Python
214 lines
8.6 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
|