229 lines
7.3 KiB
Python
229 lines
7.3 KiB
Python
"""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()
|