"""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, OaInstanceEvent, OaTask self.engine = create_engine("sqlite:///:memory:") # 仅建引擎相关表 OaAppVersion.__table__.create(self.engine) OaInstance.__table__.create(self.engine) OaTask.__table__.create(self.engine) OaInstanceEvent.__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, round_no=1, 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, round_no=1, create_time=ts, update_time=ts, comment_text=None, resolve_source="users", ) inst = MagicMock( status="running", initiator_user_id="u0", current_round=1, 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, round_no=1, 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", current_round=1, 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") class TestReturnAndResubmit(unittest.TestCase): def setUp(self): from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.models.models import OaAppVersion, OaInstance, OaInstanceEvent, OaTask self.engine = create_engine("sqlite:///:memory:") OaAppVersion.__table__.create(self.engine) OaInstance.__table__.create(self.engine) OaTask.__table__.create(self.engine) OaInstanceEvent.__table__.create(self.engine) Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) self.db = Session() self.OaInstance = OaInstance self.OaTask = OaTask self.OaInstanceEvent = OaInstanceEvent flow = { "id": "s", "type": "start", "child": { "id": "a1", "type": "approval", "name": "审批人", "approverConfig": { "assigneeType": "users", "userIds": ["u1"], "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=json.dumps({"allowReturn": True}), 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", current_round=1, 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", round_no=1, 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_return_sets_returned_and_event(self): from app.services.oa_engine import return_task with patch( "app.services.oa_engine.get_instance_detail", return_value={"id": "i1", "status": "returned"}, ): return_task( self.db, "t1", tenant_no="zkxw", user_id="u1", comment="请补充材料" ) inst = self.db.query(self.OaInstance).filter_by(id="i1").first() task = self.db.query(self.OaTask).filter_by(id="t1").first() ev = ( self.db.query(self.OaInstanceEvent) .filter_by(instance_id="i1", event_type="return") .first() ) self.assertEqual(inst.status, "returned") self.assertEqual(task.status, "returned") self.assertIsNotNone(ev) self.assertEqual(ev.comment_text, "请补充材料") def test_resubmit_starts_new_round(self): from app.services.oa_engine import resubmit_instance, return_task with patch( "app.services.oa_engine.get_instance_detail", return_value={"id": "i1", "status": "returned"}, ): return_task( self.db, "t1", tenant_no="zkxw", user_id="u1", comment="请补充材料" ) fake_task = MagicMock() with patch( "app.services.oa_engine._create_tasks_for_node", return_value=[fake_task] ), patch( "app.services.oa_engine.get_instance_detail", return_value={"id": "i1", "status": "running"}, ): resubmit_instance( self.db, "i1", tenant_no="zkxw", user_id="u0", form_data={"x": 1}, ) inst = self.db.query(self.OaInstance).filter_by(id="i1").first() self.assertEqual(inst.status, "running") self.assertEqual(inst.current_round, 2) submit_ev = ( self.db.query(self.OaInstanceEvent) .filter_by(instance_id="i1", event_type="submit", round_no=2) .first() ) self.assertIsNotNone(submit_ev) if __name__ == "__main__": unittest.main()