529 lines
17 KiB
Python
529 lines
17 KiB
Python
"""OA 引擎单元测试（纯逻辑 + 轻量 mock）"""
|
||
import json
|
||
import unittest
|
||
from datetime import datetime
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from app.services.oa_engine import (
|
||
OaEngineError,
|
||
_find_node,
|
||
_flow_node_key_order,
|
||
_match_condition_group,
|
||
_sort_tasks_for_detail,
|
||
advance_from,
|
||
)
|
||
|
||
|
||
class TestOaEngineHelpers(unittest.TestCase):
|
||
def test_find_node(self):
|
||
root = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
}
|
||
self.assertEqual(_find_node(root, "a1")["id"], "a1")
|
||
self.assertEqual(_find_node(root, "e1")["id"], "e1")
|
||
self.assertIsNone(_find_node(root, "x"))
|
||
|
||
def test_condition_match(self):
|
||
groups = [
|
||
{
|
||
"name": "gt3",
|
||
"when": "{days} > 3",
|
||
"child": {"id": "n1", "type": "end"},
|
||
},
|
||
{"name": "default", "rules": [], "child": {"id": "n2", "type": "end"}},
|
||
]
|
||
g = _match_condition_group(groups, {"days": 5})
|
||
self.assertEqual(g["name"], "gt3")
|
||
g2 = _match_condition_group(groups, {"days": 1})
|
||
self.assertEqual(g2["name"], "default")
|
||
|
||
def test_flow_node_key_order_cc_before_approval(self):
|
||
root = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "c1",
|
||
"type": "cc",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
},
|
||
}
|
||
order = _flow_node_key_order(root)
|
||
self.assertLess(order["c1"], order["a1"])
|
||
|
||
def test_sort_tasks_by_flow_when_same_create_time(self):
|
||
"""抄送先于审批设计时，即使 create_time / order_num 相同也应按树序展示。"""
|
||
root = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "c1",
|
||
"type": "cc",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
},
|
||
}
|
||
ts = datetime(2026, 7, 15, 14, 5, 7)
|
||
approval = MagicMock(node_key="a1", order_num=0, create_time=ts, id="AAAA")
|
||
cc = MagicMock(node_key="c1", order_num=0, create_time=ts, id="BBBB")
|
||
# 故意把审批放在列表前面，模拟 create_time 同秒时 PK 乱序
|
||
sorted_tasks = _sort_tasks_for_detail([approval, cc], flow_root=root)
|
||
self.assertEqual([t.node_key for t in sorted_tasks], ["c1", "a1"])
|
||
|
||
|
||
class TestAdvanceFrom(unittest.TestCase):
|
||
def test_linear_creates_approval_then_stops(self):
|
||
flow = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"name": "负责人",
|
||
"approverConfig": {"assigneeType": "dept_leader"},
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
}
|
||
db = MagicMock()
|
||
inst = MagicMock(
|
||
id="inst1",
|
||
tenant_no="zkxw",
|
||
initiator_user_id="u0",
|
||
initiator_dept_id="d1",
|
||
form_data_json="{}",
|
||
status="running",
|
||
)
|
||
ver = MagicMock(
|
||
flow_tree_json=json.dumps(flow),
|
||
advanced_settings_json=json.dumps({"supervisorFallbackToDeptLeader": True}),
|
||
)
|
||
fake_task = MagicMock()
|
||
with patch(
|
||
"app.services.oa_engine._next_task_order_num", return_value=1
|
||
), patch(
|
||
"app.services.oa_engine._create_tasks_for_node",
|
||
return_value=[fake_task],
|
||
) as mock_create:
|
||
advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
||
mock_create.assert_called_once()
|
||
self.assertEqual(inst.status, "running")
|
||
self.assertEqual(mock_create.call_args.kwargs.get("order_num_start"), 1)
|
||
|
||
def test_complete_path_to_end_via_cc(self):
|
||
flow = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "c1",
|
||
"type": "cc",
|
||
"name": "抄送",
|
||
"approverConfig": {"assigneeType": "users", "userIds": ["u9"]},
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
}
|
||
db = MagicMock()
|
||
inst = MagicMock(
|
||
id="inst1",
|
||
tenant_no="zkxw",
|
||
initiator_user_id="u0",
|
||
initiator_dept_id="d1",
|
||
form_data_json="{}",
|
||
status="running",
|
||
update_time=datetime.now(),
|
||
)
|
||
ver = MagicMock(
|
||
flow_tree_json=json.dumps(flow),
|
||
advanced_settings_json="{}",
|
||
)
|
||
with patch(
|
||
"app.services.oa_engine._next_task_order_num", return_value=1
|
||
), patch(
|
||
"app.services.oa_engine._create_tasks_for_node",
|
||
return_value=[MagicMock()],
|
||
):
|
||
advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
||
self.assertEqual(inst.status, "completed")
|
||
|
||
def test_empty_assignee_raises(self):
|
||
flow = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"approverConfig": {"assigneeType": "dept_leader"},
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
}
|
||
db = MagicMock()
|
||
inst = MagicMock(
|
||
id="inst1",
|
||
tenant_no="zkxw",
|
||
initiator_user_id="u0",
|
||
initiator_dept_id="d1",
|
||
form_data_json="{}",
|
||
status="running",
|
||
)
|
||
ver = MagicMock(flow_tree_json=json.dumps(flow), advanced_settings_json="{}")
|
||
with patch(
|
||
"app.services.oa_engine._next_task_order_num", return_value=1
|
||
), patch("app.services.oa_engine._create_tasks_for_node", return_value=[]):
|
||
with self.assertRaises(OaEngineError):
|
||
advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
||
|
||
|
||
class TestAddSign(unittest.TestCase):
|
||
def test_add_sign_rejects_when_disabled(self):
|
||
from app.services.oa_engine import add_sign
|
||
|
||
db = MagicMock()
|
||
task = MagicMock(
|
||
id="t1",
|
||
assignee_user_id="u1",
|
||
task_type="approval",
|
||
status="pending",
|
||
instance_id="i1",
|
||
node_key="a1",
|
||
node_name="审批",
|
||
sign_mode="or",
|
||
comment_text=None,
|
||
)
|
||
inst = MagicMock(id="i1", status="running", initiator_dept_id="d1")
|
||
ver = MagicMock(
|
||
id="v1",
|
||
advanced_settings_json=json.dumps({"allowAddSign": False}),
|
||
del_flag=0,
|
||
)
|
||
db.query.return_value.filter.return_value.first.side_effect = [task, inst, ver]
|
||
with self.assertRaises(OaEngineError) as ctx:
|
||
add_sign(
|
||
db,
|
||
"t1",
|
||
tenant_no="zkxw",
|
||
user_id="u1",
|
||
user_ids=["u2"],
|
||
)
|
||
self.assertIn("不允许加签", str(ctx.exception))
|
||
|
||
def test_add_sign_requires_users(self):
|
||
from app.services.oa_engine import add_sign
|
||
|
||
db = MagicMock()
|
||
task = MagicMock(
|
||
id="t1",
|
||
assignee_user_id="u1",
|
||
task_type="approval",
|
||
status="pending",
|
||
instance_id="i1",
|
||
)
|
||
inst = MagicMock(id="i1", status="running")
|
||
ver = MagicMock(advanced_settings_json="{}", del_flag=0)
|
||
db.query.return_value.filter.return_value.first.side_effect = [task, inst, ver]
|
||
with self.assertRaises(OaEngineError) as ctx:
|
||
add_sign(db, "t1", tenant_no="zkxw", user_id="u1", user_ids=[])
|
||
self.assertIn("加签人", str(ctx.exception))
|
||
|
||
|
||
class TestCompleteTaskOrSign(unittest.TestCase):
|
||
"""回归：Session autoflush=False 时，或签取消同节点待办不可覆盖本人已同意状态。"""
|
||
|
||
def setUp(self):
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import sessionmaker
|
||
from app.core.database import Base
|
||
from app.models.models import OaAppVersion, OaInstance, OaTask
|
||
|
||
self.engine = create_engine("sqlite:///:memory:")
|
||
# 仅建引擎相关表
|
||
OaAppVersion.__table__.create(self.engine)
|
||
OaInstance.__table__.create(self.engine)
|
||
OaTask.__table__.create(self.engine)
|
||
Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
|
||
self.db = Session()
|
||
self.Base = Base
|
||
self.OaAppVersion = OaAppVersion
|
||
self.OaInstance = OaInstance
|
||
self.OaTask = OaTask
|
||
|
||
flow = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"name": "审批人",
|
||
"approverConfig": {"signMode": "or"},
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
}
|
||
now = datetime.now()
|
||
self.db.add(
|
||
OaAppVersion(
|
||
id="v1",
|
||
app_id="app1",
|
||
version_no=1,
|
||
status="published",
|
||
is_latest_published="1",
|
||
form_schema_json="{}",
|
||
flow_tree_json=json.dumps(flow),
|
||
advanced_settings_json="{}",
|
||
tenant_no="zkxw",
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
)
|
||
self.db.add(
|
||
OaInstance(
|
||
id="i1",
|
||
app_id="app1",
|
||
app_version_id="v1",
|
||
title="单人审批",
|
||
initiator_user_id="u0",
|
||
initiator_dept_id="d1",
|
||
form_data_json="{}",
|
||
status="running",
|
||
tenant_no="zkxw",
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
)
|
||
self.db.add(
|
||
OaTask(
|
||
id="t1",
|
||
instance_id="i1",
|
||
node_key="a1",
|
||
node_name="审批人",
|
||
task_type="approval",
|
||
assignee_user_id="u1",
|
||
status="pending",
|
||
sign_mode="or",
|
||
tenant_no="zkxw",
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
)
|
||
self.db.commit()
|
||
|
||
def tearDown(self):
|
||
self.db.close()
|
||
self.engine.dispose()
|
||
|
||
def test_approve_keeps_approved_not_done(self):
|
||
from app.services.oa_engine import complete_task
|
||
|
||
with patch(
|
||
"app.services.oa_engine.get_instance_detail",
|
||
return_value={"id": "i1", "status": "completed"},
|
||
):
|
||
complete_task(
|
||
self.db,
|
||
"t1",
|
||
tenant_no="zkxw",
|
||
user_id="u1",
|
||
action="approve",
|
||
comment="同意",
|
||
)
|
||
|
||
task = self.db.query(self.OaTask).filter(self.OaTask.id == "t1").first()
|
||
self.assertEqual(task.status, "approved")
|
||
self.assertEqual(task.comment_text, "同意")
|
||
|
||
# 已处理列表应能查到
|
||
from app.services.oa_engine import list_tasks
|
||
|
||
done = list_tasks(self.db, tenant_no="zkxw", user_id="u1", kind="done")
|
||
self.assertEqual(len(done), 1)
|
||
self.assertEqual(done[0]["status"], "approved")
|
||
|
||
def test_or_sign_cancels_siblings_only(self):
|
||
from app.services.oa_engine import complete_task
|
||
|
||
now = datetime.now()
|
||
self.db.add(
|
||
self.OaTask(
|
||
id="t2",
|
||
instance_id="i1",
|
||
node_key="a1",
|
||
node_name="审批人",
|
||
task_type="approval",
|
||
assignee_user_id="u2",
|
||
status="pending",
|
||
sign_mode="or",
|
||
tenant_no="zkxw",
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
)
|
||
self.db.commit()
|
||
|
||
with patch(
|
||
"app.services.oa_engine.get_instance_detail",
|
||
return_value={"id": "i1", "status": "completed"},
|
||
):
|
||
complete_task(
|
||
self.db,
|
||
"t1",
|
||
tenant_no="zkxw",
|
||
user_id="u1",
|
||
action="approve",
|
||
comment="同意",
|
||
)
|
||
|
||
t1 = self.db.query(self.OaTask).filter(self.OaTask.id == "t1").first()
|
||
t2 = self.db.query(self.OaTask).filter(self.OaTask.id == "t2").first()
|
||
self.assertEqual(t1.status, "approved")
|
||
self.assertEqual(t2.status, "done")
|
||
|
||
|
||
class TestBuildProgress(unittest.TestCase):
|
||
def test_cc_before_approval_and_pending_ahead(self):
|
||
from app.services.oa_engine import build_progress_items
|
||
|
||
root = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "c1",
|
||
"type": "cc",
|
||
"name": "抄送人",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"name": "审批人",
|
||
"child": {
|
||
"id": "a2",
|
||
"type": "approval",
|
||
"name": "总经理",
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
},
|
||
},
|
||
}
|
||
ts = datetime(2026, 7, 15, 14, 5, 7)
|
||
# 审批任务 create 同秒但列表故意审批在前
|
||
t_approval = MagicMock(
|
||
id="tA",
|
||
node_key="a1",
|
||
node_name="审批人",
|
||
task_type="approval",
|
||
status="pending",
|
||
assignee_user_id="u2",
|
||
order_num=0,
|
||
create_time=ts,
|
||
update_time=ts,
|
||
comment_text=None,
|
||
resolve_source="users",
|
||
)
|
||
t_cc = MagicMock(
|
||
id="tC",
|
||
node_key="c1",
|
||
node_name="抄送人",
|
||
task_type="cc",
|
||
status="done",
|
||
assignee_user_id="u9",
|
||
order_num=0,
|
||
create_time=ts,
|
||
update_time=ts,
|
||
comment_text=None,
|
||
resolve_source="users",
|
||
)
|
||
inst = MagicMock(
|
||
status="running",
|
||
initiator_user_id="u0",
|
||
create_time=ts,
|
||
update_time=ts,
|
||
)
|
||
items = build_progress_items(
|
||
inst=inst,
|
||
flow_root=root,
|
||
tasks=[t_approval, t_cc],
|
||
name_map={"u0": "发起人甲", "u2": "审批乙", "u9": "抄送丙"},
|
||
events=[], # P0 可空
|
||
)
|
||
kinds = [x["kind"] for x in items]
|
||
self.assertEqual(kinds[0], "submit")
|
||
# 已产生：cc 再 approval
|
||
self.assertEqual(
|
||
[x["nodeKey"] for x in items if x["kind"] == "task"],
|
||
["c1", "a1"],
|
||
)
|
||
# running 时未到达 a2
|
||
pending = [x for x in items if x["kind"] == "pending_node"]
|
||
self.assertTrue(any(p["nodeKey"] == "a2" for p in pending))
|
||
|
||
def test_terminal_status_excludes_pending_nodes(self):
|
||
from app.services.oa_engine import build_progress_items
|
||
|
||
root = {
|
||
"id": "s",
|
||
"type": "start",
|
||
"child": {
|
||
"id": "a1",
|
||
"type": "approval",
|
||
"name": "审批人",
|
||
"child": {
|
||
"id": "a2",
|
||
"type": "approval",
|
||
"name": "总经理",
|
||
"child": {"id": "e1", "type": "end"},
|
||
},
|
||
},
|
||
}
|
||
ts = datetime(2026, 7, 15, 14, 5, 7)
|
||
t_approval = MagicMock(
|
||
id="tA",
|
||
node_key="a1",
|
||
node_name="审批人",
|
||
task_type="approval",
|
||
status="rejected",
|
||
assignee_user_id="u2",
|
||
order_num=0,
|
||
create_time=ts,
|
||
update_time=ts,
|
||
comment_text="不同意",
|
||
resolve_source="users",
|
||
)
|
||
for status in ("rejected", "completed", "cancelled"):
|
||
inst = MagicMock(
|
||
status=status,
|
||
initiator_user_id="u0",
|
||
create_time=ts,
|
||
update_time=ts,
|
||
)
|
||
items = build_progress_items(
|
||
inst=inst,
|
||
flow_root=root,
|
||
tasks=[t_approval],
|
||
name_map={"u0": "发起人甲", "u2": "审批乙"},
|
||
events=[],
|
||
)
|
||
pending = [x for x in items if x["kind"] == "pending_node"]
|
||
self.assertEqual(pending, [], f"status={status} should not have pending_node")
|
||
terminal = [x for x in items if x["kind"] == "terminal"]
|
||
self.assertEqual(len(terminal), 1, f"status={status} should have one terminal")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|