From 80ef84e68f9ffb65b9bbd0e488634fdea62bd528 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Tue, 11 Aug 2026 10:47:56 +0800 Subject: [PATCH] test: add failing OA task list filter coverage Co-authored-by: Cursor --- backend/tests/test_oa_tasks_filters.py | 308 +++++++++++++++++++++++++ 1 file changed, 308 insertions(+) create mode 100644 backend/tests/test_oa_tasks_filters.py diff --git a/backend/tests/test_oa_tasks_filters.py b/backend/tests/test_oa_tasks_filters.py new file mode 100644 index 00000000..e00be5b7 --- /dev/null +++ b/backend/tests/test_oa_tasks_filters.py @@ -0,0 +1,308 @@ +"""OA list_tasks 检索过滤。""" +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, OaTask, SysUser + + +class TestOaTasksFilters(unittest.TestCase): + def setUp(self): + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + OaApp.__table__, + OaAppVersion.__table__, + SysUser.__table__, + OaInstance.__table__, + OaTask.__table__, + ], + ) + Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) + self.db = Session() + t1 = datetime(2026, 7, 10, 9, 0, 0) + t2 = datetime(2026, 7, 20, 15, 30, 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, + ), + OaApp( + id="app-ot", + 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="{}", + flow_tree_json="{}", + advanced_settings_json="{}", + tenant_no="zkxw", + create_dept_id="", + create_time=t1, + update_time=t1, + del_flag=0, + ), + OaAppVersion( + id="v-ot", + app_id="app-ot", + 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=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, + ), + SysUser( + id="u-approver", + user_no="ap", + nick_name="审批人", + password="x", + tenant_no="zkxw", + create_dept_id="", + create_time=t1, + update_time=t1, + del_flag=0, + ), + OaInstance( + id="i-leave", + app_id="app-leave", + app_version_id="v-leave", + title="请假-忽略标题", + initiator_user_id="u-a", + form_data_json="{}", + status="running", + tenant_no="zkxw", + create_dept_id="", + create_time=t1, + update_time=t1, + del_flag=0, + ), + OaInstance( + id="i-ot", + app_id="app-ot", + app_version_id="v-ot", + title="加班标题含报销", + initiator_user_id="u-b", + form_data_json="{}", + status="completed", + tenant_no="zkxw", + create_dept_id="", + create_time=t2, + update_time=t2, + del_flag=0, + ), + OaTask( + id="t-leave-pending", + instance_id="i-leave", + node_key="n1", + node_name="审批", + task_type="approval", + assignee_user_id="u-approver", + status="pending", + tenant_no="zkxw", + create_dept_id="", + create_time=t1, + update_time=t1, + del_flag=0, + ), + OaTask( + id="t-ot-done", + instance_id="i-ot", + node_key="n1", + node_name="审批", + task_type="approval", + assignee_user_id="u-approver", + status="approved", + tenant_no="zkxw", + create_dept_id="", + create_time=t2, + update_time=t2, + del_flag=0, + ), + OaTask( + id="t-leave-cc-unread", + instance_id="i-leave", + node_key="cc1", + node_name="抄送", + task_type="cc", + assignee_user_id="u-approver", + status="done", + is_read=0, + tenant_no="zkxw", + create_dept_id="", + create_time=t1, + update_time=t1, + del_flag=0, + ), + OaTask( + id="t-ot-cc-read", + instance_id="i-ot", + node_key="cc1", + node_name="抄送", + task_type="cc", + assignee_user_id="u-approver", + status="done", + is_read=1, + tenant_no="zkxw", + create_dept_id="", + create_time=t2, + update_time=t2, + del_flag=0, + ), + ] + ) + self.db.commit() + + def tearDown(self): + self.db.close() + self.engine.dispose() + + def test_mine_filter_by_app_name(self): + from app.services.oa_engine import list_tasks + + res = list_tasks( + self.db, + tenant_no="zkxw", + user_id="u-a", + kind="mine", + app_name="请假", + ) + self.assertEqual(res["total"], 1) + self.assertEqual(res["list"][0]["id"], "i-leave") + + def test_mine_filter_by_app_name_matches_title(self): + from app.services.oa_engine import list_tasks + + res = list_tasks( + self.db, + tenant_no="zkxw", + user_id="u-b", + kind="mine", + app_name="报销", + ) + self.assertEqual(res["total"], 1) + self.assertEqual(res["list"][0]["id"], "i-ot") + + def test_pending_filter_by_initiator_name(self): + from app.services.oa_engine import list_tasks + + res = list_tasks( + self.db, + tenant_no="zkxw", + user_id="u-approver", + kind="pending", + initiator_name="张", + ) + self.assertEqual(res["total"], 1) + self.assertEqual(res["list"][0]["instanceId"], "i-leave") + + def test_done_filter_by_apply_time_range(self): + from app.services.oa_engine import list_tasks + + res = list_tasks( + self.db, + tenant_no="zkxw", + user_id="u-approver", + kind="done", + apply_time_from="2026-07-15", + apply_time_to="2026-07-25", + ) + self.assertEqual(res["total"], 1) + self.assertEqual(res["list"][0]["instanceId"], "i-ot") + + def test_mine_filter_by_instance_status(self): + from app.services.oa_engine import list_tasks + + res = list_tasks( + self.db, + tenant_no="zkxw", + user_id="u-b", + kind="mine", + instance_status="completed", + ) + self.assertEqual(res["total"], 1) + self.assertEqual(res["list"][0]["instanceStatus"], "completed") + + def test_invalid_instance_status_ignored(self): + from app.services.oa_engine import list_tasks + + res = list_tasks( + self.db, + tenant_no="zkxw", + user_id="u-a", + kind="mine", + instance_status="not-a-real-status", + ) + self.assertEqual(res["total"], 1) + + def test_cc_combine_read_status_and_app_name(self): + from app.services.oa_engine import list_tasks + + res = list_tasks( + self.db, + tenant_no="zkxw", + user_id="u-approver", + kind="cc", + read_status="unread", + app_name="请假", + ) + self.assertEqual(res["total"], 1) + self.assertEqual(res["list"][0]["id"], "t-leave-cc-unread") + + +if __name__ == "__main__": + unittest.main()