zkxw/backend/tests/test_oa_instance_history.py

"""OA list_instance_history：按 appId + 全员权限收窄。"""
from __future__ import annotations

import unittest
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.base import Base
from app.models.models import (
    OaApp,
    OaAppVersion,
    OaInstance,
    OaInstanceEvent,
    OaTask,
    SysDept,
    SysUser,
)
from app.services import oa_engine


class TestOaInstanceHistory(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(
            self.engine,
            tables=[
                OaApp.__table__,
                OaAppVersion.__table__,
                SysUser.__table__,
                SysDept.__table__,
                OaInstance.__table__,
                OaTask.__table__,
                OaInstanceEvent.__table__,
            ],
        )
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        t1 = datetime(2026, 8, 10, 9, 0, 0)
        t2 = datetime(2026, 8, 20, 15, 30, 0)
        t3 = datetime(2026, 8, 25, 11, 0, 0)
        self.db.add_all(
            [
                OaApp(
                    id="app-leave",
                    app_name="请假申请",
                    initiator_scope="all",
                    admin_scope="all_oa_admin",
                    status="1",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t1,
                    del_flag=0,
                ),
                OaAppVersion(
                    id="v-leave",
                    app_id="app-leave",
                    version_no=1,
                    status="published",
                    is_latest_published="1",
                    form_schema_json='{"rule":[{"type":"input","field":"leaveType","title":"请假类型"},{"type":"input","field":"days","title":"天数"},{"type":"upload","field":"voucher","title":"付款凭证"}]}',
                    flow_tree_json="{}",
                    advanced_settings_json="{}",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t1,
                    del_flag=0,
                ),
                SysUser(
                    id="u-a",
                    user_no="a",
                    nick_name="张三",
                    password="x",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t1,
                    del_flag=0,
                ),
                SysUser(
                    id="u-b",
                    user_no="b",
                    nick_name="李四",
                    password="x",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t1,
                    del_flag=0,
                ),
                SysDept(
                    id="dept-root",
                    parent_id="0",
                    dept_name="总部",
                    status="1",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t1,
                    del_flag=0,
                ),
                SysDept(
                    id="dept-sales",
                    parent_id="dept-root",
                    dept_name="销售部",
                    status="1",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t1,
                    del_flag=0,
                ),
                SysDept(
                    id="dept-sales-east",
                    parent_id="dept-sales",
                    dept_name="华东销售",
                    status="1",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t1,
                    del_flag=0,
                ),
                OaInstance(
                    id="inst-alice-done",
                    app_id="app-leave",
                    app_version_id="v-leave",
                    title="请假-张三",
                    initiator_user_id="u-a",
                    initiator_dept_id="dept-sales",
                    form_data_json='{"leaveType":"年假","days":"2","voucher":"f33a569a9dd64f2c853899498cd47498.png"}',
                    status="completed",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t1,
                    update_time=t2,
                    del_flag=0,
                ),
                OaInstance(
                    id="inst-bob-run",
                    app_id="app-leave",
                    app_version_id="v-leave",
                    title="请假-李四",
                    initiator_user_id="u-b",
                    initiator_dept_id="dept-sales-east",
                    form_data_json='{"leaveType":"事假","days":"1"}',
                    status="running",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t3,
                    update_time=t3,
                    del_flag=0,
                ),
            ]
        )
        self.db.commit()

    def tearDown(self):
        self.db.close()
        self.engine.dispose()

    def test_requires_app_id(self):
        with self.assertRaises(oa_engine.OaEngineError):
            oa_engine.list_instance_history(
                self.db,
                tenant_no="zkxw",
                user_id="u-a",
                can_view_all=False,
                app_id="",
            )

    def test_self_scope_hides_others(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=False,
            app_id="app-leave",
        )
        self.assertEqual(data["total"], 1)
        self.assertFalse(data["canViewAll"])
        self.assertEqual(data["list"][0]["id"], "inst-alice-done")

    def test_all_scope_returns_both(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=True,
            app_id="app-leave",
        )
        self.assertEqual(data["total"], 2)
        self.assertTrue(data["canViewAll"])
        ids = {r["id"] for r in data["list"]}
        self.assertEqual(ids, {"inst-alice-done", "inst-bob-run"})

    def test_initiator_name_ignored_without_all(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=False,
            app_id="app-leave",
            initiator_name="李四",
        )
        self.assertEqual(data["total"], 1)
        self.assertEqual(data["list"][0]["id"], "inst-alice-done")

    def test_finish_time_only_on_terminal(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=True,
            app_id="app-leave",
        )
        by_id = {r["id"]: r for r in data["list"]}
        self.assertTrue(by_id["inst-alice-done"].get("finishTime"))
        self.assertFalse(by_id["inst-bob-run"].get("finishTime"))

    def test_finish_time_filter(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=True,
            app_id="app-leave",
            finish_time_from="2026-08-20",
            finish_time_to="2026-08-20",
        )
        self.assertEqual(data["total"], 1)
        self.assertEqual(data["list"][0]["id"], "inst-alice-done")

    def test_preview_fields_limit_six(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=False,
            app_id="app-leave",
        )
        pf = data["list"][0].get("previewFields") or []
        self.assertIsInstance(pf, list)
        self.assertLessEqual(len(pf), 6)
        if pf:
            self.assertGreaterEqual(len(pf), 1)

    def test_preview_fields_keep_upload_files(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=False,
            app_id="app-leave",
        )
        pf = data["list"][0].get("previewFields") or []
        voucher = next((f for f in pf if f.get("label") == "付款凭证"), None)
        self.assertIsNotNone(voucher)
        self.assertEqual(voucher.get("fieldType"), "upload")
        self.assertEqual(
            voucher.get("files"),
            [
                {
                    "url": "f33a569a9dd64f2c853899498cd47498.png",
                    "name": "f33a569a9dd64f2c853899498cd47498.png",
                }
            ],
        )

    def test_export_history_rows(self):
        data = oa_engine.export_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=False,
            app_id="app-leave",
        )
        self.assertEqual(data["total"], 1)
        self.assertGreaterEqual(len(data["businessColumns"]), 1)
        self.assertEqual(len(data["fixedColumns"]), 13)
        row = data["rows"][0]
        self.assertEqual(row["id"], "inst-alice-done")
        self.assertEqual(row["business"].get("leaveType"), "年假")
        self.assertEqual(row["fixed"].get("initiatorName"), "张三")
        self.assertIn("approvalRecord", row["fixed"])

    def test_dept_filter_includes_descendants(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=True,
            app_id="app-leave",
            initiator_dept_id="dept-sales",
        )
        self.assertEqual(data["total"], 2)
        ids = {r["id"] for r in data["list"]}
        self.assertEqual(ids, {"inst-alice-done", "inst-bob-run"})

    def test_dept_filter_child_only(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=True,
            app_id="app-leave",
            initiator_dept_id="dept-sales-east",
        )
        self.assertEqual(data["total"], 1)
        self.assertEqual(data["list"][0]["id"], "inst-bob-run")

    def test_dept_filter_respects_self_scope(self):
        data = oa_engine.list_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=False,
            app_id="app-leave",
            initiator_dept_id="dept-sales-east",
        )
        self.assertEqual(data["total"], 0)

    def test_export_selected_ids_only(self):
        data = oa_engine.export_instance_history(
            self.db,
            tenant_no="zkxw",
            user_id="u-a",
            can_view_all=True,
            app_id="app-leave",
            instance_ids=["inst-bob-run"],
        )
        self.assertEqual(data["total"], 1)
        self.assertEqual(data["rows"][0]["id"], "inst-bob-run")


if __name__ == "__main__":
    unittest.main()