309 lines
11 KiB
Python
309 lines
11 KiB
Python
"""OA：审批人为发起人时自动通过"""
|
|
import json
|
|
import unittest
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
|
|
class TestAutoPassInitiator(unittest.TestCase):
|
|
def setUp(self):
|
|
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.OaAppVersion = OaAppVersion
|
|
self.OaInstance = OaInstance
|
|
self.OaTask = OaTask
|
|
self.now = datetime.now()
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
|
|
def _seed(self, *, flow, advanced, initiator="u0", tasks=None):
|
|
self.db.add(
|
|
self.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=json.dumps(advanced),
|
|
tenant_no="zkxw",
|
|
create_dept_id="",
|
|
create_time=self.now,
|
|
update_time=self.now,
|
|
del_flag=0,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.OaInstance(
|
|
id="i1",
|
|
app_id="app1",
|
|
app_version_id="v1",
|
|
title="t",
|
|
initiator_user_id=initiator,
|
|
initiator_dept_id="d1",
|
|
form_data_json="{}",
|
|
status="running",
|
|
tenant_no="zkxw",
|
|
create_dept_id="",
|
|
create_time=self.now,
|
|
update_time=self.now,
|
|
del_flag=0,
|
|
)
|
|
)
|
|
for t in tasks or []:
|
|
self.db.add(t)
|
|
self.db.commit()
|
|
|
|
def test_or_sign_sole_initiator_auto_passes_to_end(self):
|
|
"""开关开：或签唯一审批人=发起人 → 提交推进后自动通过到结束。"""
|
|
from app.services.oa_engine import advance_from
|
|
|
|
flow = {
|
|
"id": "s",
|
|
"type": "start",
|
|
"child": {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "审批人",
|
|
"approverConfig": {
|
|
"assigneeType": "users",
|
|
"userIds": ["u0"],
|
|
"signMode": "or",
|
|
},
|
|
"child": {"id": "e1", "type": "end"},
|
|
},
|
|
}
|
|
self._seed(
|
|
flow=flow,
|
|
advanced={"autoPassWhenApproverIsInitiator": True},
|
|
)
|
|
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
|
|
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
|
|
|
|
with patch(
|
|
"app.services.oa_engine.resolve_assignees",
|
|
return_value=(["u0"], {"resolve_source": "users"}),
|
|
), patch(
|
|
"app.services.oa_engine._next_task_order_num",
|
|
return_value=1,
|
|
):
|
|
advance_from(
|
|
self.db, inst=inst, ver=ver, from_node=None, user_id="u0"
|
|
)
|
|
|
|
# Session autoflush=False：未 flush 时 query 看不到刚 add 的任务
|
|
self.db.flush()
|
|
tasks = self.db.query(self.OaTask).filter_by(instance_id="i1").all()
|
|
self.assertEqual(len(tasks), 1)
|
|
self.assertEqual(tasks[0].status, "approved")
|
|
self.assertEqual(tasks[0].comment_text, "系统自动通过")
|
|
self.assertEqual(inst.status, "completed")
|
|
|
|
def test_switch_off_keeps_pending(self):
|
|
from app.services.oa_engine import advance_from
|
|
|
|
flow = {
|
|
"id": "s",
|
|
"type": "start",
|
|
"child": {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "审批人",
|
|
"approverConfig": {
|
|
"assigneeType": "users",
|
|
"userIds": ["u0"],
|
|
"signMode": "or",
|
|
},
|
|
"child": {"id": "e1", "type": "end"},
|
|
},
|
|
}
|
|
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": False})
|
|
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
|
|
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
|
|
with patch(
|
|
"app.services.oa_engine.resolve_assignees",
|
|
return_value=(["u0"], {"resolve_source": "users"}),
|
|
), patch(
|
|
"app.services.oa_engine._next_task_order_num", return_value=1
|
|
):
|
|
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
|
self.db.flush()
|
|
t = self.db.query(self.OaTask).filter_by(instance_id="i1").one()
|
|
self.assertEqual(t.status, "pending")
|
|
self.assertEqual(inst.status, "running")
|
|
|
|
def test_or_sign_multi_cancels_siblings(self):
|
|
from app.services.oa_engine import advance_from
|
|
|
|
flow = {
|
|
"id": "s",
|
|
"type": "start",
|
|
"child": {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "审批人",
|
|
"approverConfig": {
|
|
"assigneeType": "users",
|
|
"userIds": ["u0", "u2"],
|
|
"signMode": "or",
|
|
},
|
|
"child": {"id": "e1", "type": "end"},
|
|
},
|
|
}
|
|
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
|
|
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
|
|
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
|
|
with patch(
|
|
"app.services.oa_engine.resolve_assignees",
|
|
return_value=(["u0", "u2"], {"resolve_source": "users"}),
|
|
), patch(
|
|
"app.services.oa_engine._next_task_order_num", return_value=1
|
|
):
|
|
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
|
self.db.flush()
|
|
by_user = {
|
|
t.assignee_user_id: t
|
|
for t in self.db.query(self.OaTask).filter_by(instance_id="i1").all()
|
|
}
|
|
self.assertEqual(by_user["u0"].status, "approved")
|
|
self.assertEqual(by_user["u0"].comment_text, "系统自动通过")
|
|
self.assertEqual(by_user["u2"].status, "done")
|
|
self.assertEqual(inst.status, "completed")
|
|
|
|
def test_and_sign_only_completes_self(self):
|
|
from app.services.oa_engine import advance_from
|
|
|
|
flow = {
|
|
"id": "s",
|
|
"type": "start",
|
|
"child": {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "审批人",
|
|
"approverConfig": {
|
|
"assigneeType": "users",
|
|
"userIds": ["u0", "u2"],
|
|
"signMode": "and",
|
|
},
|
|
"child": {"id": "e1", "type": "end"},
|
|
},
|
|
}
|
|
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
|
|
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
|
|
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
|
|
with patch(
|
|
"app.services.oa_engine.resolve_assignees",
|
|
return_value=(["u0", "u2"], {"resolve_source": "users"}),
|
|
), patch(
|
|
"app.services.oa_engine._next_task_order_num", return_value=1
|
|
):
|
|
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
|
self.db.flush()
|
|
by_user = {
|
|
t.assignee_user_id: t
|
|
for t in self.db.query(self.OaTask).filter_by(instance_id="i1").all()
|
|
}
|
|
self.assertEqual(by_user["u0"].status, "approved")
|
|
self.assertEqual(by_user["u2"].status, "pending")
|
|
self.assertEqual(inst.status, "running")
|
|
|
|
def test_consecutive_nodes_both_auto_pass(self):
|
|
from app.services.oa_engine import advance_from
|
|
|
|
flow = {
|
|
"id": "s",
|
|
"type": "start",
|
|
"child": {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "一审",
|
|
"approverConfig": {
|
|
"assigneeType": "users",
|
|
"userIds": ["u0"],
|
|
"signMode": "or",
|
|
},
|
|
"child": {
|
|
"id": "a2",
|
|
"type": "approval",
|
|
"name": "二审",
|
|
"approverConfig": {
|
|
"assigneeType": "users",
|
|
"userIds": ["u0"],
|
|
"signMode": "or",
|
|
},
|
|
"child": {"id": "e1", "type": "end"},
|
|
},
|
|
},
|
|
}
|
|
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
|
|
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
|
|
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
|
|
with patch(
|
|
"app.services.oa_engine.resolve_assignees",
|
|
return_value=(["u0"], {"resolve_source": "users"}),
|
|
), patch(
|
|
"app.services.oa_engine._next_task_order_num", return_value=1
|
|
):
|
|
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
|
self.db.flush()
|
|
tasks = self.db.query(self.OaTask).filter_by(instance_id="i1").all()
|
|
self.assertEqual(len(tasks), 2)
|
|
self.assertTrue(all(t.status == "approved" for t in tasks))
|
|
self.assertTrue(all(t.comment_text == "系统自动通过" for t in tasks))
|
|
self.assertEqual(inst.status, "completed")
|
|
|
|
def test_comment_required_does_not_block_auto_pass(self):
|
|
from app.services.oa_engine import advance_from
|
|
|
|
flow = {
|
|
"id": "s",
|
|
"type": "start",
|
|
"child": {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "审批人",
|
|
"approverConfig": {
|
|
"assigneeType": "users",
|
|
"userIds": ["u0"],
|
|
"signMode": "or",
|
|
},
|
|
"child": {"id": "e1", "type": "end"},
|
|
},
|
|
}
|
|
self._seed(
|
|
flow=flow,
|
|
advanced={
|
|
"autoPassWhenApproverIsInitiator": True,
|
|
"commentRequired": True,
|
|
},
|
|
)
|
|
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
|
|
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
|
|
with patch(
|
|
"app.services.oa_engine.resolve_assignees",
|
|
return_value=(["u0"], {"resolve_source": "users"}),
|
|
), patch(
|
|
"app.services.oa_engine._next_task_order_num", return_value=1
|
|
):
|
|
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
|
|
self.db.flush()
|
|
t = self.db.query(self.OaTask).filter_by(instance_id="i1").one()
|
|
self.assertEqual(t.status, "approved")
|
|
self.assertEqual(inst.status, "completed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|