Auto-Finance/tests/test_mcp_analysis.py

"""MCP 分析层测试：金额匹配 runner 与 save_info / validate_info / rules_get

覆盖点：
- 一对多凑额（两张发票 233.00 对一笔 231.00，在 3% 容差内应配对）
- 违反业务约束时给出 review_required
- save_info 剥掉 can_submit/suggestion 并自动校验
- validate_info 的 draft 模式不落盘
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from src.core.extraction.cache import load_cache, save_extraction
from src.core.matching.runner import match_from_cache, split_cache
from src.mcp.tools import analysis as tools_analysis

TRAVEL_TICKET = {
    "invoice_type": "train",
    "invoice_number": "N1",
    "invoice_date": "2026-06-03",
    "ride_date": "2026-06-02",
    "departure": "阜阳西",
    "arrival": "合肥南",
    "seat_class": "二等座",
    "train_no": "G1967",
    "person_name": "示例人员甲",
    "total_amount": "115.50",
}
CARD = {
    "invoice_type": "payment",
    "card_date": "2026-06-01",
    "card_amount": "231.00",
    "card_no": "6282****1682",
    "merchant": "中国铁路网络有限公司",
}


def _seed(case_dir: Path) -> None:
    (case_dir / ".invoice_cache").mkdir(parents=True, exist_ok=True)
    (case_dir / "a.pdf").write_bytes(b"%PDF a")
    (case_dir / "b.pdf").write_bytes(b"%PDF b")
    (case_dir / "pay.jpg").write_bytes(b"img")
    save_extraction(case_dir, "a.pdf", dict(TRAVEL_TICKET))
    save_extraction(
        case_dir, "b.pdf", dict(TRAVEL_TICKET, invoice_number="N2", total_amount="117.50", train_no="G1318")
    )
    save_extraction(case_dir, "pay.jpg", dict(CARD))


def test_split_cache_buckets(tmp_path: Path) -> None:
    _seed(tmp_path)
    buckets = split_cache(load_cache(tmp_path))
    assert len(buckets["invoices"]) == 2
    assert len(buckets["cards"]) == 1


def test_match_from_cache_pairs_one_to_many(tmp_path: Path) -> None:
    _seed(tmp_path)
    result = match_from_cache(tmp_path)
    assert result["ok"] is True
    assert len(result["matched"]) == 1
    assert len(result["matched"][0]["invoices"]) == 2
    assert result["unmatched_invoices"] == []
    assert result["totals"]["invoice_total"] == 233.0
    assert (tmp_path / ".invoice_cache" / "match_result.json").exists()
    assert (tmp_path / "payment_records.csv").exists()
    assert (tmp_path / "invoice_groups.json").exists()
    assert result["suggested_invoice_type"] == "travel"


def test_match_from_cache_flags_short_invoices(tmp_path: Path) -> None:
    (tmp_path / ".invoice_cache").mkdir(parents=True)
    (tmp_path / "a.pdf").write_bytes(b"x")
    (tmp_path / "pay.jpg").write_bytes(b"x")
    save_extraction(tmp_path, "a.pdf", dict(TRAVEL_TICKET))
    save_extraction(tmp_path, "pay.jpg", dict(CARD))
    result = match_from_cache(tmp_path)
    assert any("违反" in line for line in result["review_required"])
    assert result["matched"] == []


def test_match_requires_invoices(tmp_path: Path) -> None:
    (tmp_path / ".invoice_cache").mkdir(parents=True)
    result = match_from_cache(tmp_path)
    assert result["ok"] is False


@pytest.fixture()
def case(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    from src.cases import CaseStore

    store = CaseStore()
    monkeypatch.setattr(type(store), "root", tmp_path)
    monkeypatch.setattr(tools_analysis._store, "root", tmp_path, raising=False)
    return store.create("单测会话")


def test_save_info_strips_self_declared_submit(case: Path) -> None:
    d = tools_analysis.save_info(
        case.name,
        "travel",
        {"basic_info": {"travel_purpose": "毕业采集"}, "can_submit": True, "suggestion": "可以交"},
    )
    assert d["ok"] is True
    assert d["validation"]["valid"] is False
    saved = json.loads(Path(d["saved"]).read_text(encoding="utf-8"))
    assert "can_submit" not in saved and "suggestion" not in saved


def test_validate_info_draft_does_not_persist(case: Path) -> None:
    r = tools_analysis.validate_info(case.name, "travel", draft={"basic_info": {}})
    assert r["valid"] is False
    assert not (case / ".invoice_cache" / "travel_info.json").exists()


def test_rules_get_shape(case: Path) -> None:
    r = tools_analysis.rules_get("normal")
    rules = r["rules"]
    assert {"fields", "arrays", "description"} <= set(rules)
    assert all(isinstance(f, dict) for f in rules["fields"])
    for arr in rules["arrays"]:
        assert isinstance(arr.get("min_items", 1), int)


def test_info_skeleton_shape_matches_rules(case: Path) -> None:
    for kind in ("travel", "normal"):
        sk = tools_analysis.info_skeleton(kind)
        assert sk["ok"] is True
        report = tools_analysis.validate_info(case.name, kind, draft=sk["skeleton"])
        assert report["valid"] is False
        assert len(report["missing_fields"]) > 0
    travel = tools_analysis.info_skeleton("travel")["skeleton"]
    assert travel["reimbursement_details"]["transport_fee"][0]["remark"] == ""
    assert travel["payment_methods"][0]["merchant"] == ""
    assert travel["attachments"] == []


def test_info_skeleton_filled_passes(case: Path) -> None:
    skeleton = tools_analysis.info_skeleton("normal")["skeleton"]
    skeleton["basic_info"]["reimbursement_description"] = "元器件采购"
    skeleton["reimbursement_details"]["total_invoices"] = 3
    skeleton["reimbursement_details"]["total_amount"] = 464.0
    pay = skeleton["payment_methods"][0]
    pay.update({"card_date": "2026-06-01", "card_amount": 231.0, "merchant": "中国铁路网络有限公司", "remark": ""})
    skeleton["attachments"] = [{"filename": "a.pdf", "attachment_type": "invoice"}]
    r = tools_analysis.save_info(case.name, "normal", skeleton)
    assert r["validation"]["valid"] is True, r["validation"]


def test_unknown_info_type(case: Path) -> None:
    assert tools_analysis.save_info(case.name, "mixed", {}).get("ok") is False
    assert tools_analysis.validate_info(case.name, "mixed").get("ok") is False