"""低值易耗品申购单模块单元测试 覆盖发票加载去重、规格推测、物品名称清洗。 """ 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] == ["热熔胶棒"]