提交
This commit is contained in:
1 parent
c3bd2c00ae
commit
a52fe77e34
39 files changed
+609647
-9238
No files matched your search
@@ -0,0 +1,178 @@
|
||||
# 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 == "成功"
|
||||
Reference in new issue
Block a user