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()