shanxia2.0/backend/tests/test_bx_reimburse_import.py

# backend/tests/test_bx_reimburse_import.py
from datetime import date
from io import BytesIO

from openpyxl import Workbook, load_workbook

from app.services.bx.reimburse_import import (
    HEADERS,
    build_applicant_label,
    build_expense_path_labels,
    build_import_template_bytes,
    is_example_row,
    iter_data_rows,
    partition_import_groups,
    validate_group_headers,
)
from app.utils.excel_list_validation import add_list_validation, write_option_sheet


def test_build_expense_path_labels_unique_and_enabled_only():
    nodes = [
        {"id": "1", "parent_id": "0", "item_name": "差旅费", "status": 1, "item_code": "CL"},
        {"id": "2", "parent_id": "1", "item_name": "市内交通", "status": 1, "item_code": "JT"},
        {"id": "3", "parent_id": "1", "item_name": "住宿费", "status": 0, "item_code": "ZS"},
        {"id": "4", "parent_id": "0", "item_name": "办公费", "status": 1, "item_code": "BG"},
    ]
    labels = build_expense_path_labels(nodes)
    assert labels["2"] == "差旅费 / 市内交通"
    assert labels["1"] == "差旅费"
    assert labels["4"] == "办公费"
    assert "3" not in labels


def test_build_expense_path_labels_exclude_orphan_when_parent_missing():
    """父级未传入（等同已删除）时，子节点不应进入下拉——与 Web 树一致。"""
    nodes = [
        {"id": "29", "parent_id": "28", "item_name": "办公用品", "status": 1, "item_code": "BGYP"},
        {"id": "30", "parent_id": "0", "item_name": "物流运输费用", "status": 1, "item_code": "YSF"},
    ]
    labels = build_expense_path_labels(nodes)
    assert "29" not in labels
    assert labels["30"] == "物流运输费用"


def test_build_expense_path_labels_disambiguate_collision():
    nodes = [
        {"id": "a", "parent_id": "0", "item_name": "同名", "status": 1, "item_code": "A1"},
        {"id": "b", "parent_id": "0", "item_name": "同名", "status": 1, "item_code": "B1"},
    ]
    labels = build_expense_path_labels(nodes)
    assert labels["a"] != labels["b"]
    assert "编码" in labels["a"] or "编码" in labels["b"] or labels["a"].endswith("）")


def test_cell_str_normalizes_excel_float_ids():
    from app.services.bx.reimburse_import import cell_str

    assert cell_str(29.0) == "29"
    assert cell_str(29) == "29"


def test_build_applicant_label():
    assert build_applicant_label("张三", "z001", "uid-abcdef") == "张三（z001）"
    assert build_applicant_label("李四", "", "abcdefghij") == "李四（efghij）"


def test_is_example_row():
    assert is_example_row({"报销事由": "【示例】地铁"}) is True
    assert is_example_row({"报销事由": "正常事由"}) is False


def test_partition_empty_group_is_solo():
    rows = [
        {"_row": 2, "分组号": "G1", "申请人": "A", "报销日期": "2026-08-18", "报销标题": "", "费用项目": "x", "金额（元）": 1, "报销事由": "r1", "发生时间": ""},
        {"_row": 3, "分组号": "G1", "申请人": "A", "报销日期": "2026-08-18", "报销标题": "", "费用项目": "y", "金额（元）": 2, "报销事由": "r2", "发生时间": ""},
        {"_row": 4, "分组号": "", "申请人": "B", "报销日期": "2026-08-18", "报销标题": "", "费用项目": "z", "金额（元）": 3, "报销事由": "r3", "发生时间": ""},
    ]
    groups = partition_import_groups(rows)
    assert len(groups) == 2
    g1 = next(g for g in groups if g["group_no"] == "G1")
    assert len(g1["rows"]) == 2
    solo = next(g for g in groups if g["group_no"] == "")
    assert len(solo["rows"]) == 1


def test_validate_group_headers_mismatch():
    rows = [
        {"_row": 2, "申请人": "A", "报销日期": "2026-08-18", "报销标题": ""},
        {"_row": 3, "申请人": "B", "报销日期": "2026-08-18", "报销标题": ""},
    ]
    err = validate_group_headers(rows)
    assert err and "申请人" in err


def test_validate_group_headers_title_partial_empty():
    rows = [
        {"_row": 2, "申请人": "A", "报销日期": "2026-08-18", "报销标题": "有标题"},
        {"_row": 3, "申请人": "A", "报销日期": "2026-08-18", "报销标题": ""},
    ]
    err = validate_group_headers(rows)
    assert err and "标题" in err


def test_write_option_sheet_and_validation():
    wb = Workbook()
    ws = wb.active
    ws.title = "报销导入"
    ws.append(["申请人"])
    write_option_sheet(wb, "_选项_申请人", ["张三（z001）"], ["u1"], hidden=True)
    add_list_validation(
        ws,
        option_sheet="_选项_申请人",
        option_rows=1,
        target_col=1,
        start_row=2,
        end_row=100,
    )
    assert wb["_选项_申请人"].sheet_state == "hidden"
    assert len(ws.data_validations.dataValidation) == 1


def test_build_import_template_has_guide_examples_and_dv():
    raw = build_import_template_bytes(
        expense_nodes=[
            {"id": "1", "parent_id": "0", "item_name": "差旅费", "status": 1, "item_code": "CL"},
            {"id": "2", "parent_id": "1", "item_name": "市内交通", "status": 1, "item_code": "JT"},
        ],
        applicants=[{"id": "u1", "nick_name": "张三", "user_no": "z001"}],
        current_applicant_label="张三（z001）",
        today=date(2026, 8, 18),
    )
    wb = load_workbook(BytesIO(raw))
    assert "填写说明" in wb.sheetnames
    assert "报销导入" in wb.sheetnames
    assert wb["_选项_费用项目"].sheet_state == "hidden"
    ws = wb["报销导入"]
    assert [c.value for c in ws[1]][:8] == list(HEADERS)
    assert str(ws["G2"].value).startswith("【示例】")
    assert len(ws.data_validations.dataValidation) >= 1


def test_iter_data_rows_skips_examples():
    wb = Workbook()
    ws = wb.active
    ws.title = "报销导入"
    ws.append(list(HEADERS))
    ws.append(
        ["G1", "张三（z001）", "2026-08-18", "", "差旅费", 50, "【示例】跳过", ""]
    )
    ws.append(
        ["G1", "张三（z001）", "2026-08-18", "", "差旅费", 80, "正式事由", ""]
    )
    rows = iter_data_rows(ws)
    assert len(rows) == 1
    assert rows[0]["报销事由"] == "正式事由"
    assert rows[0]["_row"] == 3


def test_build_annotated_result_bytes_appends_status_columns():
    from app.services.bx.reimburse_import import build_annotated_result_bytes

    wb = Workbook()
    ws = wb.active
    ws.title = "报销导入"
    ws.append(list(HEADERS))
    ws.append(["G1", "张三（z001）", "2026-08-18", "", "差旅费", 80, "正式事由", ""])
    buf = BytesIO()
    wb.save(buf)
    raw = build_annotated_result_bytes(
        buf.getvalue(),
        {2: {"status": "成功", "message": "成功，单号 BX001"}},
    )
    out = load_workbook(BytesIO(raw))["报销导入"]
    headers = [c.value for c in out[1]]
    assert "导入结果" in headers
    assert "失败原因" in headers
    status_i = headers.index("导入结果") + 1
    assert out.cell(row=2, column=status_i).value == "成功"