# 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 == "成功"