"""OA：审批人去重 — 同一实例内已同意的人，后续节点系统自动通过。""" import json import unittest from datetime import datetime from unittest.mock import patch from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker def _resolve_from_config(_db, *, assignee_config, **_kwargs): ids = list(assignee_config.get("userIds") or []) return ids, {"resolve_source": "users"} class TestApproverDedup(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"): 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, ) ) self.db.commit() def _load(self): inst = self.db.query(self.OaInstance).filter_by(id="i1").first() ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first() return inst, ver def _tasks(self): self.db.flush() return self.db.query(self.OaTask).filter_by(instance_id="i1").all() def _pending(self, user_id): return next( t for t in self._tasks() if t.assignee_user_id == user_id and t.status == "pending" ) def _approve(self, task): from app.services.oa_engine import _approve_task_core inst, ver = self._load() _approve_task_core( self.db, task=task, inst=inst, ver=ver, user_id=task.assignee_user_id, comment="同意", ) def _chain_flow(self): """第 1 步 A，第 2 步 B，第 3、4 步再是 A。""" def approval(node_id, name, user_ids, child): return { "id": node_id, "type": "approval", "name": name, "approverConfig": { "assigneeType": "users", "userIds": user_ids, "signMode": "or", }, "child": child, } return { "id": "s", "type": "start", "child": approval( "a1", "第一步", ["A"], approval( "a2", "第二步", ["B"], approval( "a3", "第三步", ["A"], approval("a4", "第四步", ["A"], {"id": "e1", "type": "end"}), ), ), ), } def test_later_nodes_auto_pass_after_same_person_approved(self): """A 在第 1 步同意后，第 3、4 步不再待办，系统记为自动通过。""" from app.services.oa_engine import advance_from self._seed(flow=self._chain_flow(), advanced={"approverDedup": True}) inst, ver = self._load() with patch( "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config ), 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._approve(self._pending("A")) self._approve(self._pending("B")) by_node = {t.node_key: t for t in self._tasks()} self.assertEqual(by_node["a1"].status, "approved") self.assertEqual(by_node["a1"].comment_text, "同意") self.assertEqual(by_node["a3"].status, "approved") self.assertEqual(by_node["a3"].comment_text, "审批人去重，系统自动通过") self.assertEqual(by_node["a4"].status, "approved") self.assertEqual(by_node["a4"].comment_text, "审批人去重，系统自动通过") self.assertEqual(inst.status, "completed") self.assertFalse(any(t.status == "pending" for t in by_node.values())) def test_switch_off_keeps_repeat_approver_pending(self): from app.services.oa_engine import advance_from self._seed(flow=self._chain_flow(), advanced={"approverDedup": False}) inst, ver = self._load() with patch( "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config ), 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._approve(self._pending("A")) self._approve(self._pending("B")) step3 = next(t for t in self._tasks() if t.node_key == "a3") self.assertEqual(step3.assignee_user_id, "A") self.assertEqual(step3.status, "pending") self.assertEqual(inst.status, "running") self.assertFalse(any(t.node_key == "a4" for t in self._tasks())) def test_or_sign_auto_pass_closes_whole_node(self): from app.services.oa_engine import advance_from flow = { "id": "s", "type": "start", "child": { "id": "a1", "type": "approval", "name": "第一步", "approverConfig": { "assigneeType": "users", "userIds": ["A"], "signMode": "or", }, "child": { "id": "a2", "type": "approval", "name": "或签", "approverConfig": { "assigneeType": "users", "userIds": ["A", "C"], "signMode": "or", }, "child": {"id": "e1", "type": "end"}, }, }, } self._seed(flow=flow, advanced={"approverDedup": True}) inst, ver = self._load() with patch( "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config ), 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._approve(self._pending("A")) by_key = {(t.node_key, t.assignee_user_id): t for t in self._tasks()} self.assertEqual(by_key[("a2", "A")].status, "approved") self.assertEqual(by_key[("a2", "A")].comment_text, "审批人去重，系统自动通过") self.assertEqual(by_key[("a2", "C")].status, "done") self.assertEqual(inst.status, "completed") def test_and_sign_only_auto_passes_repeat_person(self): from app.services.oa_engine import advance_from flow = { "id": "s", "type": "start", "child": { "id": "a1", "type": "approval", "name": "第一步", "approverConfig": { "assigneeType": "users", "userIds": ["A"], "signMode": "or", }, "child": { "id": "a2", "type": "approval", "name": "会签", "approverConfig": { "assigneeType": "users", "userIds": ["A", "C"], "signMode": "and", }, "child": {"id": "e1", "type": "end"}, }, }, } self._seed(flow=flow, advanced={"approverDedup": True}) inst, ver = self._load() with patch( "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config ), 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._approve(self._pending("A")) by_user = { t.assignee_user_id: t for t in self._tasks() if t.node_key == "a2" } self.assertEqual(by_user["A"].status, "approved") self.assertEqual(by_user["A"].comment_text, "审批人去重，系统自动通过") self.assertEqual(by_user["C"].status, "pending") self.assertEqual(inst.status, "running") def test_non_approved_history_does_not_dedup(self): """仅统计已同意；拒绝等状态不触发后续自动通过。""" from app.services.oa_engine import advance_from flow = { "id": "s", "type": "start", "child": { "id": "a1", "type": "approval", "name": "当前", "approverConfig": { "assigneeType": "users", "userIds": ["A"], "signMode": "or", }, "child": {"id": "e1", "type": "end"}, }, } self._seed(flow=flow, advanced={"approverDedup": True}) self.db.add( self.OaTask( id="old-reject", instance_id="i1", node_key="old", node_name="历史", task_type="approval", assignee_user_id="A", status="rejected", sign_mode="or", round_no=1, is_read=0, tenant_no="zkxw", order_num=0, create_dept_id="", create_time=self.now, update_time=self.now, del_flag=0, ) ) self.db.commit() inst, ver = self._load() with patch( "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config ), 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") current = next(t for t in self._tasks() if t.node_key == "a1") self.assertEqual(current.status, "pending") self.assertEqual(inst.status, "running")