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