309 lines
12 KiB
Python
309 lines
12 KiB
Python
import unittest
|
||
|
||
from app.services.oa_related_approval import (
|
||
extract_related_approval_links,
|
||
flatten_form_fields_for_summary,
|
||
merge_relation_targets,
|
||
)
|
||
|
||
|
||
class TestFlatten(unittest.TestCase):
|
||
def test_skips_layout_and_summarizes_upload(self):
|
||
schema = {
|
||
"rule": [
|
||
{"type": "html", "title": "说明"},
|
||
{"type": "input", "field": "reason", "title": "事由"},
|
||
{"type": "upload", "field": "files", "title": "附件"},
|
||
{"type": "relatedApproval", "field": "prev", "title": "关联"},
|
||
{"type": "textarea", "field": "long", "title": "详细说明"},
|
||
]
|
||
}
|
||
data = {
|
||
"reason": "出差",
|
||
"files": ["https://cdn.example.com/a.png", "https://cdn.example.com/b.png"],
|
||
"prev": ["id1", "id2"],
|
||
"long": "很长很长的说明文字" * 5,
|
||
}
|
||
fields = flatten_form_fields_for_summary(schema, data)
|
||
labels = [f["label"] for f in fields]
|
||
self.assertNotIn("说明", labels)
|
||
self.assertEqual(fields[0], {"fieldKey": "reason", "label": "事由", "value": "出差"})
|
||
self.assertEqual(
|
||
fields[1]["value"],
|
||
"a.png、b.png",
|
||
)
|
||
self.assertEqual(fields[1].get("fieldType"), "upload")
|
||
self.assertEqual(
|
||
fields[1].get("files"),
|
||
[
|
||
{"url": "https://cdn.example.com/a.png", "name": "a.png"},
|
||
{"url": "https://cdn.example.com/b.png", "name": "b.png"},
|
||
],
|
||
)
|
||
self.assertTrue(fields[1].get("fullWidth"))
|
||
self.assertEqual(fields[2]["value"], "已关联 2 条")
|
||
self.assertEqual(fields[2].get("relatedIds"), ["id1", "id2"])
|
||
self.assertTrue(fields[3].get("fullWidth"))
|
||
|
||
def test_dept_formats_path_names(self):
|
||
schema = {
|
||
"rule": [
|
||
{"type": "dept", "field": "dept", "title": "部门"},
|
||
{
|
||
"type": "dept",
|
||
"field": "depts",
|
||
"title": "多部门",
|
||
"props": {"multiple": True},
|
||
},
|
||
]
|
||
}
|
||
data = {
|
||
"dept": {"id": "d1", "name": "中科兴皖 / 研发"},
|
||
"depts": [
|
||
{"id": "d1", "name": "中科兴皖 / 研发"},
|
||
{"id": "d2", "name": "中科兴皖 / 质量"},
|
||
],
|
||
}
|
||
fields = flatten_form_fields_for_summary(schema, data)
|
||
by_label = {f["label"]: f["value"] for f in fields}
|
||
self.assertEqual(by_label["部门"], "中科兴皖 / 研发")
|
||
self.assertEqual(by_label["多部门"], "中科兴皖 / 研发、中科兴皖 / 质量")
|
||
|
||
def test_person_formats_names(self):
|
||
schema = {
|
||
"rule": [
|
||
{"type": "person", "field": "p1", "title": "协同人"},
|
||
{
|
||
"type": "person",
|
||
"field": "p2",
|
||
"title": "多人",
|
||
"props": {"multiple": True},
|
||
},
|
||
]
|
||
}
|
||
data = {
|
||
"p1": {"id": "u1", "name": "张三"},
|
||
"p2": [
|
||
{"id": "u1", "name": "张三"},
|
||
{"id": "u2", "name": "李四"},
|
||
],
|
||
}
|
||
fields = flatten_form_fields_for_summary(schema, data)
|
||
by_label = {f["label"]: f["value"] for f in fields}
|
||
self.assertEqual(by_label["协同人"], "张三")
|
||
self.assertEqual(by_label["多人"], "张三、李四")
|
||
|
||
def test_related_approval_field_includes_related_ids(self):
|
||
schema = {
|
||
"rule": [
|
||
{"type": "input", "field": "reason", "title": "事由"},
|
||
{"type": "relatedApproval", "field": "prev", "title": "关联审批单"},
|
||
]
|
||
}
|
||
data = {"reason": "拜访", "prev": ["id1", "", "id2", "id1"]}
|
||
fields = flatten_form_fields_for_summary(schema, data)
|
||
rel = next(f for f in fields if f["label"] == "关联审批单")
|
||
self.assertEqual(rel["value"], "已关联 2 条")
|
||
self.assertEqual(rel["relatedIds"], ["id1", "id2"])
|
||
|
||
def test_related_approval_empty_ids(self):
|
||
schema = {"rule": [{"type": "relatedApproval", "field": "prev", "title": "关联"}]}
|
||
fields = flatten_form_fields_for_summary(schema, {"prev": []})
|
||
self.assertEqual(fields[0]["value"], "已关联 0 条")
|
||
self.assertEqual(fields[0]["relatedIds"], [])
|
||
|
||
def test_upload_object_list_extracts_url(self):
|
||
schema = {"rule": [{"type": "upload", "field": "files", "title": "附件"}]}
|
||
data = {
|
||
"files": [
|
||
{"name": "a.pdf", "url": "https://cdn.example.com/a.pdf"},
|
||
{"name": "b.pdf", "response": {"data": {"fileUrl": "/mini/b.pdf"}}},
|
||
]
|
||
}
|
||
fields = flatten_form_fields_for_summary(schema, data)
|
||
self.assertEqual(
|
||
fields[0]["value"],
|
||
"a.pdf、b.pdf",
|
||
)
|
||
self.assertEqual(
|
||
fields[0].get("files"),
|
||
[
|
||
{"url": "https://cdn.example.com/a.pdf", "name": "a.pdf"},
|
||
{"url": "/mini/b.pdf", "name": "b.pdf"},
|
||
],
|
||
)
|
||
|
||
def test_extract_links_with_field_key_and_dedupe(self):
|
||
schema = {
|
||
"rule": [
|
||
{"type": "relatedApproval", "field": "relA"},
|
||
{
|
||
"type": "fcRow",
|
||
"children": [{"type": "relatedApproval", "field": "relB"}],
|
||
},
|
||
]
|
||
}
|
||
data = {"relA": ["x", "x", ""], "relB": "solo"}
|
||
links = extract_related_approval_links(schema, data)
|
||
self.assertEqual(links, [("relA", "x"), ("relB", "solo")])
|
||
|
||
|
||
def test_trip_itinerary_formats_legs(self):
|
||
schema = {
|
||
"rule": [
|
||
{"type": "tripItinerary", "field": "legs", "title": "行程"},
|
||
]
|
||
}
|
||
data = {
|
||
"legs": [
|
||
{
|
||
"transport": "汽车",
|
||
"tripType": "往返",
|
||
"fromCity": "合肥",
|
||
"toCity": "上海",
|
||
"startDate": "2026-08-10",
|
||
"startHalf": "上午",
|
||
"endDate": "2026-08-13",
|
||
"endHalf": "下午",
|
||
"duration": 4.0,
|
||
"durationManual": False,
|
||
}
|
||
]
|
||
}
|
||
fields = flatten_form_fields_for_summary(schema, data)
|
||
self.assertEqual(len(fields), 1)
|
||
self.assertIn("合肥 → 上海", fields[0]["value"])
|
||
self.assertIn("汽车", fields[0]["value"])
|
||
self.assertNotIn("durationManual", fields[0]["value"])
|
||
self.assertTrue(fields[0].get("fullWidth"))
|
||
|
||
def test_region_and_cascader_path(self):
|
||
schema = {
|
||
"rule": [
|
||
{"type": "region", "field": "r1", "title": "省市区"},
|
||
{"type": "cascader", "field": "c1", "title": "级联"},
|
||
{"type": "dateRange", "field": "d1", "title": "日期区间"},
|
||
]
|
||
}
|
||
data = {
|
||
"r1": ["安徽省", "合肥市", "蜀山区"],
|
||
"c1": {"province": "安徽省", "city": "合肥市", "district": "蜀山区"},
|
||
"d1": ["2026-08-10", "2026-08-13"],
|
||
}
|
||
fields = flatten_form_fields_for_summary(schema, data)
|
||
by = {f["label"]: f["value"] for f in fields}
|
||
self.assertEqual(by["省市区"], "安徽省 / 合肥市 / 蜀山区")
|
||
self.assertEqual(by["级联"], "安徽省 / 合肥市 / 蜀山区")
|
||
self.assertEqual(by["日期区间"], "2026-08-10 ～ 2026-08-13")
|
||
|
||
def test_unknown_object_does_not_dump_json(self):
|
||
schema = {"rule": [{"type": "input", "field": "x", "title": "怪字段"}]}
|
||
# 模拟误写入对象：不应出现整段 dict 字面量
|
||
fields = flatten_form_fields_for_summary(
|
||
schema, {"x": {"foo": 1, "bar": "baz", "id": "hidden"}}
|
||
)
|
||
self.assertNotIn("{", fields[0]["value"])
|
||
self.assertNotIn("hidden", fields[0]["value"])
|
||
self.assertIn("baz", fields[0]["value"])
|
||
|
||
def test_kit_containers_skipped(self):
|
||
schema = {
|
||
"rule": [
|
||
{
|
||
"type": "tripKit",
|
||
"field": "kit1",
|
||
"title": "出差套件",
|
||
"children": [
|
||
{"type": "input", "field": "reason", "title": "出差事由"},
|
||
],
|
||
}
|
||
]
|
||
}
|
||
fields = flatten_form_fields_for_summary(schema, {"kit1": {"x": 1}, "reason": "学习"})
|
||
labels = [f["label"] for f in fields]
|
||
self.assertNotIn("出差套件", labels)
|
||
self.assertEqual(fields[0], {"fieldKey": "reason", "label": "出差事由", "value": "学习"})
|
||
|
||
|
||
class TestValidateLinkable(unittest.TestCase):
|
||
def _inst(self, **kw):
|
||
from unittest.mock import MagicMock
|
||
|
||
defaults = dict(
|
||
status="running",
|
||
initiator_user_id="u1",
|
||
tenant_no="t",
|
||
del_flag=0,
|
||
id="i1",
|
||
)
|
||
defaults.update(kw)
|
||
return MagicMock(**defaults)
|
||
|
||
def test_initiator_any_status_ok(self):
|
||
from app.services.oa_related_approval import assert_linkable_instance
|
||
|
||
for st in ("running", "completed", "rejected", "returned", "cancelled"):
|
||
assert_linkable_instance(
|
||
self._inst(status=st), user_id="u1", tenant_no="t", db=None
|
||
)
|
||
|
||
def test_rejects_unrelated_user_without_db_tasks(self):
|
||
from app.services.oa_related_approval import assert_linkable_instance
|
||
from app.services.oa_engine import OaEngineError
|
||
from unittest.mock import MagicMock
|
||
|
||
db = MagicMock()
|
||
# no tasks
|
||
db.query.return_value.filter.return_value.first.return_value = None
|
||
with self.assertRaises(OaEngineError):
|
||
assert_linkable_instance(
|
||
self._inst(initiator_user_id="other"),
|
||
user_id="u1",
|
||
tenant_no="t",
|
||
db=db,
|
||
)
|
||
|
||
def test_done_approver_ok(self):
|
||
from app.services.oa_related_approval import assert_linkable_instance
|
||
from unittest.mock import MagicMock
|
||
|
||
db = MagicMock()
|
||
db.query.return_value.filter.return_value.first.return_value = MagicMock()
|
||
assert_linkable_instance(
|
||
self._inst(initiator_user_id="other", status="running"),
|
||
user_id="u1",
|
||
tenant_no="t",
|
||
db=db,
|
||
)
|
||
|
||
|
||
class TestCandidateCardHelpers(unittest.TestCase):
|
||
def test_status_text(self):
|
||
from app.services.oa_related_approval import related_status_text
|
||
|
||
self.assertEqual(related_status_text("running"), "处理中")
|
||
self.assertEqual(related_status_text("completed"), "已通过")
|
||
self.assertEqual(related_status_text("rejected"), "已拒绝")
|
||
self.assertEqual(related_status_text("returned"), "已打回")
|
||
self.assertEqual(related_status_text("cancelled"), "已撤销")
|
||
|
||
def test_display_title(self):
|
||
from app.services.oa_related_approval import related_display_title
|
||
|
||
self.assertEqual(
|
||
related_display_title("吴自龙", "人事审批", ""),
|
||
"吴自龙提交的人事审批",
|
||
)
|
||
self.assertEqual(related_display_title("", "请假", "自定义标题"), "请假")
|
||
|
||
|
||
class TestMergeRelationTargets(unittest.TestCase):
|
||
def test_merge_prefers_form_fields(self):
|
||
schema = {"rule": [{"type": "relatedApproval", "field": "r1"}]}
|
||
m = merge_relation_targets(schema, {"r1": ["a"]}, ["a", "b"])
|
||
self.assertEqual(m, [("r1", "a"), (None, "b")])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|