zlwuandCursor
a148961b55
Allow filtering OA history by initiator department including descendants, while keeping the existing all-history permission scope. Co-authored-by: Cursor <cursoragent@cursor.com>
341 lines
11 KiB
Python
341 lines
11 KiB
Python
"""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()
|