zkxw/backend/tests/test_oa_tasks_pagination.py

"""OA 待办/已办/我发起/抄送列表分页"""
import unittest
from datetime import datetime, timedelta

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


class TestListTasksPagination(unittest.TestCase):
    def setUp(self):
        from app.models.models import OaApp, OaAppVersion, OaInstance, OaTask, SysUser

        self.engine = create_engine("sqlite:///:memory:")
        OaApp.__table__.create(self.engine)
        OaAppVersion.__table__.create(self.engine)
        OaInstance.__table__.create(self.engine)
        OaTask.__table__.create(self.engine)
        SysUser.__table__.create(self.engine)
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        self.OaTask = OaTask
        self.OaInstance = OaInstance
        now = datetime(2026, 8, 11, 10, 0, 0)
        self.db.add(
            OaApp(
                id="app1",
                app_name="请假审批",
                initiator_scope="all",
                admin_scope="all_oa_admin",
                status="1",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaAppVersion(
                id="v1",
                app_id="app1",
                version_no=1,
                status="published",
                is_latest_published="1",
                form_schema_json="{}",
                flow_tree_json="{}",
                advanced_settings_json="{}",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            SysUser(
                id="u0",
                user_no="initiator",
                nick_name="发起人",
                password="x",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            SysUser(
                id="u1",
                user_no="approver",
                nick_name="审批人",
                password="x",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        # 25 条 pending 给 u1；各绑独立实例便于 enrich
        for i in range(25):
            iid = f"i-p-{i:02d}"
            t = now + timedelta(minutes=i)
            self.db.add(
                OaInstance(
                    id=iid,
                    app_id="app1",
                    app_version_id="v1",
                    title=f"单-{i}",
                    initiator_user_id="u0",
                    initiator_dept_id="d1",
                    form_data_json="{}",
                    status="running",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t,
                    update_time=t,
                    del_flag=0,
                )
            )
            self.db.add(
                OaTask(
                    id=f"t-p-{i:02d}",
                    instance_id=iid,
                    node_key="a1",
                    node_name="审批",
                    task_type="approval",
                    assignee_user_id="u1",
                    status="pending",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t,
                    update_time=t,
                    del_flag=0,
                )
            )
        # 3 条 mine 实例
        for i in range(3):
            t = now + timedelta(hours=i)
            self.db.add(
                OaInstance(
                    id=f"i-m-{i}",
                    app_id="app1",
                    app_version_id="v1",
                    title=f"我发-{i}",
                    initiator_user_id="u0",
                    initiator_dept_id="d1",
                    form_data_json="{}",
                    status="running",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t,
                    update_time=t,
                    del_flag=0,
                )
            )
        # 5 条 cc，其中 2 未读
        for i in range(5):
            iid = f"i-c-{i}"
            t = now + timedelta(seconds=i)
            self.db.add(
                OaInstance(
                    id=iid,
                    app_id="app1",
                    app_version_id="v1",
                    title=f"抄送-{i}",
                    initiator_user_id="u0",
                    initiator_dept_id="d1",
                    form_data_json="{}",
                    status="running",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t,
                    update_time=t,
                    del_flag=0,
                )
            )
            self.db.add(
                OaTask(
                    id=f"t-c-{i}",
                    instance_id=iid,
                    node_key="cc1",
                    node_name="抄送",
                    task_type="cc",
                    assignee_user_id="u1",
                    status="done",
                    is_read=0 if i < 2 else 1,
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t,
                    update_time=t,
                    del_flag=0,
                )
            )
        self.db.commit()

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

    def test_pending_paginates(self):
        from app.services.oa_engine import list_tasks

        page1 = list_tasks(
            self.db, tenant_no="zkxw", user_id="u1", kind="pending", page=1, page_size=20
        )
        self.assertEqual(page1["total"], 25)
        self.assertEqual(len(page1["list"]), 20)

        page2 = list_tasks(
            self.db, tenant_no="zkxw", user_id="u1", kind="pending", page=2, page_size=20
        )
        self.assertEqual(page2["total"], 25)
        self.assertEqual(len(page2["list"]), 5)
        ids1 = {r["id"] for r in page1["list"]}
        ids2 = {r["id"] for r in page2["list"]}
        self.assertFalse(ids1 & ids2)

    def test_mine_paginates(self):
        from app.services.oa_engine import list_tasks

        # u0 发起：25 pending 实例 + 3 mine + 5 cc 实例 = 33
        res = list_tasks(
            self.db, tenant_no="zkxw", user_id="u0", kind="mine", page=1, page_size=10
        )
        self.assertEqual(res["total"], 33)
        self.assertEqual(len(res["list"]), 10)

    def test_cc_unread_filter_then_paginate(self):
        from app.services.oa_engine import list_tasks

        res = list_tasks(
            self.db,
            tenant_no="zkxw",
            user_id="u1",
            kind="cc",
            read_status="unread",
            page=1,
            page_size=20,
        )
        self.assertEqual(res["total"], 2)
        self.assertEqual(len(res["list"]), 2)
        self.assertTrue(all(not r.get("isRead") for r in res["list"]))


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