Auto-Finance/tests/test_requisition.py

"""低值易耗品申购单模块单元测试

覆盖发票加载去重、规格推测、物品名称清洗。
"""

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,
)


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) == "电子元器件及工具耗材采购"