zkxw/backend/tests/test_oa_related_approval.py

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