opensource/bisheng
3
0
Fork 0
mirror of https://github.com/dataelement/bisheng.git synced 2026-09-01 15:32:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(approval): write submit audit log when instance lands in exception

Browse Source 
Route-missing and approver-empty paths created an ApprovalInstance but
never invoked AuditLogDao.ainsert_v2, so audit_log lost the submission
trace whenever the gate fell into _create_exception_result. Emit
approval.request.submit alongside the existing exception record, with
exception_type / instance_status / flow_version_id / route_rule_id /
current_node_name in metadata for forensic context.
This commit is contained in:
GuoQing Zhang
2026-05-27 17:45:39 +08:00
parent a76306c93e
commit a55d4951af
3 changed files with 544 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/bisheng/approval/domain/services/approval_gate.py
+86 -52
View File
@@ -41,31 +41,34 @@ async def _get_user_role_labels(user_id: int, tenant_id: int) -> frozenset[str]:
PRD §4.3: "同一申请人可能同时具备多个身份标签，条件匹配采用'包含即命中'"
"""
labels: set[str] = {'regular_user'}
labels: set[str] = {"regular_user"}
try:
from bisheng.database.constants import AdminRole
from bisheng.user.domain.models.user_role import UserRoleDao
user_roles = await UserRoleDao.aget_user_roles(user_id)
for ur in user_roles:
if ur.role_id == AdminRole:
labels.add('admin')
labels.add("admin")
else:
labels.add(f'role_{ur.role_id}')
labels.add(f"role_{ur.role_id}")
except Exception:
pass
try:
from bisheng.database.models.department import DepartmentDao
dept_admins = await DepartmentDao.aget_user_admin_departments(user_id)
if dept_admins:
labels.add('dept_admin')
labels.add("dept_admin")
except Exception:
pass
try:
from bisheng.tenant.domain.services.tenant_service import TenantService
if await TenantService._is_tenant_admin(user_id, tenant_id):
labels.add('tenant_admin')
labels.add("tenant_admin")
except Exception:
pass
@@ -119,7 +122,7 @@ class ApprovalGate:
exception_type=ApprovalExceptionType.ROUTE_MISSING,
)
if matched_route.route_type == 'pass':
if matched_route.route_type == "pass":
instance = await self.instance_repository.create_instance(
ApprovalInstance(
tenant_id=req.tenant_id,
@@ -137,7 +140,7 @@ class ApprovalGate:
reason=req.reason,
payload_snapshot=req.payload_snapshot,
detail_snapshot=detail_snapshot,
route_rule_id=getattr(matched_route, 'id', None),
route_rule_id=getattr(matched_route, "id", None),
)
)
# PASS route still needs to execute the business handler via outbox
@@ -155,16 +158,16 @@ class ApprovalGate:
tenant_id=req.tenant_id,
operator_id=0,
operator_tenant_id=req.tenant_id,
action='approval.route.pass',
target_type='approval_instance',
action="approval.route.pass",
target_type="approval_instance",
target_id=str(instance.id),
reason=getattr(matched_route, 'route_name', None),
reason=getattr(matched_route, "route_name", None),
metadata={
'instance_id': instance.id,
'scenario_code': req.scenario_code,
'route_id': getattr(matched_route, 'id', None),
'route_name': getattr(matched_route, 'route_name', None),
'payload_snapshot': req.payload_snapshot,
"instance_id": instance.id,
"scenario_code": req.scenario_code,
"route_id": getattr(matched_route, "id", None),
"route_name": getattr(matched_route, "route_name", None),
"payload_snapshot": req.payload_snapshot,
},
object_name=business_name,
ip_address=req.ip_address,
@@ -199,8 +202,8 @@ class ApprovalGate:
detail_snapshot=detail_snapshot,
exception_type=ApprovalExceptionType.APPROVER_EMPTY,
flow_version_id=flow_version.id,
route_rule_id=getattr(matched_route, 'id', None),
current_node_name=getattr(first_node, 'node_name', None),
route_rule_id=getattr(matched_route, "id", None),
current_node_name=getattr(first_node, "node_name", None),
node=first_node,
)
@@ -218,7 +221,7 @@ class ApprovalGate:
applicant_user_name=req.applicant_user_name,
applicant_department_id=req.applicant_department_id,
flow_version_id=flow_version.id,
route_rule_id=getattr(matched_route, 'id', None),
route_rule_id=getattr(matched_route, "id", None),
status=ApprovalInstanceStatus.PENDING,
reason=req.reason,
payload_snapshot=req.payload_snapshot,
@@ -237,7 +240,7 @@ class ApprovalGate:
node_name=first_node.node_name,
node_order=first_node.node_order,
approver_user_id=approver_user_id,
approver_source_type='resolved',
approver_source_type="resolved",
node_mode=first_node.node_mode,
status=ApprovalTaskStatus.PENDING,
)
@@ -247,7 +250,7 @@ class ApprovalGate:
ApprovalActionLog(
tenant_id=req.tenant_id,
instance_id=instance.id,
action='submitted',
action="submitted",
operator_user_id=req.applicant_user_id,
operator_user_name=req.applicant_user_name,
detail={},
@@ -257,17 +260,17 @@ class ApprovalGate:
tenant_id=req.tenant_id,
operator_id=req.applicant_user_id,
operator_tenant_id=req.tenant_id,
action='approval.request.submit',
target_type='approval_instance',
action="approval.request.submit",
target_type="approval_instance",
target_id=str(instance.id),
reason=req.reason,
metadata={
'instance_id': instance.id,
'scenario_code': req.scenario_code,
'handler': req.scenario_code,
'payload_snapshot': req.payload_snapshot,
'business_resource_type': req.business_resource_type,
'business_resource_id': req.business_resource_id,
"instance_id": instance.id,
"scenario_code": req.scenario_code,
"handler": req.scenario_code,
"payload_snapshot": req.payload_snapshot,
"business_resource_type": req.business_resource_type,
"business_resource_id": req.business_resource_id,
},
operator_name=req.applicant_user_name,
object_name=business_name,
@@ -319,17 +322,19 @@ class ApprovalGate:
)
)
exception_detail: dict[str, Any] = {
'scenario_code': req.scenario_code,
'business_key': req.business_key,
'current_node_name': current_node_name,
"scenario_code": req.scenario_code,
"business_key": req.business_key,
"current_node_name": current_node_name,
}
if node is not None:
exception_detail.update({
'node_code': getattr(node, 'node_code', None),
'node_name': getattr(node, 'node_name', None),
'node_order': getattr(node, 'node_order', None),
'node_mode': getattr(node, 'node_mode', None),
})
exception_detail.update(
{
"node_code": getattr(node, "node_code", None),
"node_name": getattr(node, "node_name", None),
"node_order": getattr(node, "node_order", None),
"node_mode": getattr(node, "node_mode", None),
}
)
await self.instance_repository.create_exception(
ApprovalException(
tenant_id=req.tenant_id,
@@ -338,6 +343,33 @@ class ApprovalGate:
detail=exception_detail,
)
)
# Audit the submission even when the instance lands in exception state — every
# instance creation must leave a trace per the approval module compliance rule.
await AuditLogDao.ainsert_v2(
tenant_id=req.tenant_id,
operator_id=req.applicant_user_id,
operator_tenant_id=req.tenant_id,
action="approval.request.submit",
target_type="approval_instance",
target_id=str(instance.id),
reason=req.reason,
metadata={
"instance_id": instance.id,
"scenario_code": req.scenario_code,
"handler": handler_key,
"payload_snapshot": req.payload_snapshot,
"business_resource_type": req.business_resource_type,
"business_resource_id": req.business_resource_id,
"instance_status": status,
"exception_type": exception_type,
"flow_version_id": flow_version_id,
"route_rule_id": route_rule_id,
"current_node_name": current_node_name,
},
operator_name=req.applicant_user_name,
object_name=business_name,
ip_address=req.ip_address,
)
# Notify tenant admins so they can handle the exception
await self._notify_admins_of_exception(
tenant_id=req.tenant_id,
@@ -356,12 +388,12 @@ class ApprovalGate:
def _dispatch_outbox_task(outbox_id: int) -> None:
try:
from bisheng.worker.approval.tasks import execute_approval_outbox
execute_approval_outbox.delay(outbox_id)
except Exception:
import logging
logging.getLogger(__name__).exception(
'failed to dispatch approval outbox task: outbox_id=%s', outbox_id
)
logging.getLogger(__name__).exception("failed to dispatch approval outbox task: outbox_id=%s", outbox_id)
@staticmethod
async def _notify_admins_of_exception(
@@ -377,6 +409,7 @@ class ApprovalGate:
from bisheng.database.constants import AdminRole
from bisheng.message.api.dependencies import get_message_service as _get_message_service
from bisheng.user.domain.models.user_role import UserRoleDao
admin_rows = await UserRoleDao.aget_roles_user([AdminRole])
admin_ids = [int(r.user_id) for r in admin_rows if r.user_id != applicant_user_id]
if not admin_ids:
@@ -384,13 +417,13 @@ class ApprovalGate:
async with get_async_db_session() as session:
message_service = await _get_message_service(session)
content = [
{'type': 'system_text', 'content': f'approval_exception_{exception_type}'},
{"type": "system_text", "content": f"approval_exception_{exception_type}"},
{
'type': 'business_url',
'content': f'--{business_name}',
'metadata': {
'business_type': 'approval_instance_id',
'data': {'approval_instance_id': str(instance_id)},
"type": "business_url",
"content": f"--{business_name}",
"metadata": {
"business_type": "approval_instance_id",
"data": {"approval_instance_id": str(instance_id)},
},
},
]
@@ -401,8 +434,9 @@ class ApprovalGate:
)
except Exception:
import logging
logging.getLogger(__name__).exception(
'failed to notify admin of approval exception: instance_id=%s', instance_id
"failed to notify admin of approval exception: instance_id=%s", instance_id
)
async def _match_first_route(self, route_rules: list[Any], req: ApprovalGateRequest) -> Any | None:
@@ -428,16 +462,16 @@ class ApprovalGate:
user_labels: frozenset[str] | None = None # lazily resolved; shared across routes
for route in route_rules:
if not getattr(route, 'enabled', True):
if not getattr(route, "enabled", True):
continue
match_config = getattr(route, 'match_config', {}) or {}
field = match_config.get('field', '')
match_config = getattr(route, "match_config", {}) or {}
field = match_config.get("field", "")
if not field:
return route # catch-all branch
expected = str(match_config.get('value', ''))
expected = str(match_config.get("value", ""))
if field == 'applicant_role':
if field == "applicant_role":
if user_labels is None:
user_labels = await _get_user_role_labels(req.applicant_user_id, req.tenant_id)
if expected in user_labels:
src/backend/test/approval/test_approval_audit_log.py
+229 -77
View File
@@ -4,13 +4,21 @@ These cover the audit writes that previous releases either skipped or
mislabelled. Each test patches only the DAO entry points so the production
service code runs end-to-end inside the test.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from bisheng.approval.domain.models.approval_instance import (
ApprovalExceptionType,
ApprovalInstanceStatus,
)
from bisheng.approval.domain.schemas.approval_center_schema import (
ApprovalGateDecision,
ApprovalGateRequest,
)
from bisheng.approval.domain.services.approval_gate import ApprovalGate
from bisheng.approval.domain.services.approval_outbox_service import ApprovalOutboxService
from bisheng.approval.domain.services.approval_scenario_admin_service import (
ApprovalScenarioAdminService,
@@ -19,21 +27,21 @@ from bisheng.approval.domain.services.approval_scenario_admin_service import (
class _FakeScenario:
def __init__(self, **kw):
self.id = kw.get('id', 42)
self.tenant_id = kw.get('tenant_id', 1)
self.scenario_code = kw.get('scenario_code', 'menu_access_request')
self.scenario_name = kw.get('scenario_name', '菜单权限申请')
self.enabled = kw.get('enabled', True)
self.display_name = kw.get('display_name')
self.id = kw.get("id", 42)
self.tenant_id = kw.get("tenant_id", 1)
self.scenario_code = kw.get("scenario_code", "menu_access_request")
self.scenario_name = kw.get("scenario_name", "菜单权限申请")
self.enabled = kw.get("enabled", True)
self.display_name = kw.get("display_name")
def model_dump(self):
return {
'id': self.id,
'tenant_id': self.tenant_id,
'scenario_code': self.scenario_code,
'scenario_name': self.scenario_name,
'enabled': self.enabled,
'display_name': self.display_name,
"id": self.id,
"tenant_id": self.tenant_id,
"scenario_code": self.scenario_code,
"scenario_name": self.scenario_name,
"enabled": self.enabled,
"display_name": self.display_name,
}
@@ -45,67 +53,71 @@ async def test_update_scenario_writes_toggle_audit_when_enabled_flips():
captured.append(kwargs)
return SimpleNamespace(id=999)
with patch(
'bisheng.approval.domain.services.approval_scenario_admin_service.'
'ApprovalScenarioRepository.get_scenario',
new=AsyncMock(return_value=scenario),
), patch(
'bisheng.approval.domain.services.approval_scenario_admin_service.'
'ApprovalScenarioRepository.update_scenario',
new=AsyncMock(return_value=scenario),
), patch(
'bisheng.approval.domain.services.approval_scenario_admin_service.'
'AuditLogDao.ainsert_v2',
new=AsyncMock(side_effect=fake_ainsert_v2),
with (
patch(
"bisheng.approval.domain.services.approval_scenario_admin_service.ApprovalScenarioRepository.get_scenario",
new=AsyncMock(return_value=scenario),
),
patch(
"bisheng.approval.domain.services.approval_scenario_admin_service."
"ApprovalScenarioRepository.update_scenario",
new=AsyncMock(return_value=scenario),
),
patch(
"bisheng.approval.domain.services.approval_scenario_admin_service.AuditLogDao.ainsert_v2",
new=AsyncMock(side_effect=fake_ainsert_v2),
),
):
await ApprovalScenarioAdminService.update_scenario(
tenant_id=1,
scenario_id=42,
payload={'enabled': False},
payload={"enabled": False},
operator_user_id=7,
operator_user_name='admin',
ip_address='10.0.0.1',
operator_user_name="admin",
ip_address="10.0.0.1",
)
assert len(captured) == 1, 'audit log not written on enabled flip'
assert len(captured) == 1, "audit log not written on enabled flip"
row = captured[0]
assert row['action'] == 'approval.scenario.toggle'
assert row['target_type'] == 'approval_scenario'
assert row['target_id'] == '42'
assert row['operator_id'] == 7
assert row['ip_address'] == '10.0.0.1'
assert row['metadata']['scenario_code'] == 'menu_access_request'
assert row['metadata']['before_enabled'] is True
assert row['metadata']['after_enabled'] is False
assert row["action"] == "approval.scenario.toggle"
assert row["target_type"] == "approval_scenario"
assert row["target_id"] == "42"
assert row["operator_id"] == 7
assert row["ip_address"] == "10.0.0.1"
assert row["metadata"]["scenario_code"] == "menu_access_request"
assert row["metadata"]["before_enabled"] is True
assert row["metadata"]["after_enabled"] is False
async def test_update_scenario_skips_toggle_audit_when_enabled_unchanged():
scenario = _FakeScenario(enabled=True)
captured: list[dict] = []
with patch(
'bisheng.approval.domain.services.approval_scenario_admin_service.'
'ApprovalScenarioRepository.get_scenario',
new=AsyncMock(return_value=scenario),
), patch(
'bisheng.approval.domain.services.approval_scenario_admin_service.'
'ApprovalScenarioRepository.update_scenario',
new=AsyncMock(return_value=scenario),
), patch(
'bisheng.approval.domain.services.approval_scenario_admin_service.'
'AuditLogDao.ainsert_v2',
new=AsyncMock(side_effect=lambda **kw: captured.append(kw)),
with (
patch(
"bisheng.approval.domain.services.approval_scenario_admin_service.ApprovalScenarioRepository.get_scenario",
new=AsyncMock(return_value=scenario),
),
patch(
"bisheng.approval.domain.services.approval_scenario_admin_service."
"ApprovalScenarioRepository.update_scenario",
new=AsyncMock(return_value=scenario),
),
patch(
"bisheng.approval.domain.services.approval_scenario_admin_service.AuditLogDao.ainsert_v2",
new=AsyncMock(side_effect=lambda **kw: captured.append(kw)),
),
):
# Same enabled value: rename only, no toggle.
await ApprovalScenarioAdminService.update_scenario(
tenant_id=1,
scenario_id=42,
payload={'enabled': True, 'scenario_name': 'renamed'},
payload={"enabled": True, "scenario_name": "renamed"},
operator_user_id=7,
operator_user_name='admin',
operator_user_name="admin",
)
assert captured == [], 'toggle audit must not fire when enabled unchanged'
assert captured == [], "toggle audit must not fire when enabled unchanged"
async def test_outbox_success_emits_handler_success_audit():
@@ -114,18 +126,18 @@ async def test_outbox_success_emits_handler_success_audit():
instance = SimpleNamespace(
id=11,
tenant_id=1,
scenario_code='menu_access_request',
handler_key='menu_access_request',
business_name='菜单权限申请',
status='approved',
scenario_code="menu_access_request",
handler_key="menu_access_request",
business_name="菜单权限申请",
status="approved",
)
outbox = SimpleNamespace(
id=77,
instance_id=11,
status='pending',
status="pending",
retry_count=0,
error_summary=None,
payload_snapshot={'menu_key': 'flow'},
payload_snapshot={"menu_key": "flow"},
)
repo = SimpleNamespace(
@@ -137,7 +149,7 @@ async def test_outbox_success_emits_handler_success_audit():
)
with patch(
'bisheng.approval.domain.services.approval_outbox_service.AuditLogDao.ainsert_v2',
"bisheng.approval.domain.services.approval_outbox_service.AuditLogDao.ainsert_v2",
new=AsyncMock(side_effect=lambda **kw: captured.append(kw)),
):
service = ApprovalOutboxService(instance_repository=repo)
@@ -146,11 +158,11 @@ async def test_outbox_success_emits_handler_success_audit():
assert result is True
assert len(captured) == 1
row = captured[0]
assert row['action'] == 'approval.handler.success'
assert row['operator_id'] == 0 # system
assert row['metadata']['outbox_id'] == 77
assert row['metadata']['handler'] == 'menu_access_request'
assert row['metadata']['business_result'] == 'success'
assert row["action"] == "approval.handler.success"
assert row["operator_id"] == 0 # system
assert row["metadata"]["outbox_id"] == 77
assert row["metadata"]["handler"] == "menu_access_request"
assert row["metadata"]["business_result"] == "success"
async def test_outbox_failure_emits_handler_failed_audit_with_error_summary():
@@ -158,18 +170,18 @@ async def test_outbox_failure_emits_handler_failed_audit_with_error_summary():
instance = SimpleNamespace(
id=11,
tenant_id=1,
scenario_code='menu_access_request',
handler_key='menu_access_request',
business_name='菜单权限申请',
status='approved',
scenario_code="menu_access_request",
handler_key="menu_access_request",
business_name="菜单权限申请",
status="approved",
)
outbox = SimpleNamespace(
id=77,
instance_id=11,
status='pending',
status="pending",
retry_count=0,
error_summary=None,
payload_snapshot={'menu_key': 'flow'},
payload_snapshot={"menu_key": "flow"},
)
repo = SimpleNamespace(
get_outbox=AsyncMock(return_value=outbox),
@@ -180,19 +192,159 @@ async def test_outbox_failure_emits_handler_failed_audit_with_error_summary():
)
with patch(
'bisheng.approval.domain.services.approval_outbox_service.AuditLogDao.ainsert_v2',
"bisheng.approval.domain.services.approval_outbox_service.AuditLogDao.ainsert_v2",
new=AsyncMock(side_effect=lambda **kw: captured.append(kw)),
):
service = ApprovalOutboxService(instance_repository=repo)
result = await service.execute_outbox(
outbox_id=77,
executor=lambda _outbox: (False, 'boom: downstream 500'),
executor=lambda _outbox: (False, "boom: downstream 500"),
)
assert result is False
assert len(captured) == 1
row = captured[0]
assert row['action'] == 'approval.handler.failed'
assert row['reason'] == 'boom: downstream 500'
assert row['metadata']['error_stack_summary'] == 'boom: downstream 500'
assert row['metadata']['payload_snapshot'] == {'menu_key': 'flow'}
assert row["action"] == "approval.handler.failed"
assert row["reason"] == "boom: downstream 500"
assert row["metadata"]["error_stack_summary"] == "boom: downstream 500"
assert row["metadata"]["payload_snapshot"] == {"menu_key": "flow"}
async def _run_gate_for_exception(*, exception_type: str):
"""Drive ApprovalGate.request_or_pass into an exception branch.
Patches AuditLogDao.ainsert_v2 to capture audit kwargs without touching the
real DB, and tailors the scenario/route stubs to reproduce either the
route-missing or approver-empty exception.
"""
captured: list[dict] = []
handler = SimpleNamespace(
build_detail=AsyncMock(return_value={"menu_name": "知识管理"}),
build_title=AsyncMock(return_value="知识管理"),
resolve_approvers=AsyncMock(
return_value=[] if exception_type == ApprovalExceptionType.APPROVER_EMPTY else None
),
)
registry = SimpleNamespace(get_handler=AsyncMock(return_value=handler))
if exception_type == ApprovalExceptionType.ROUTE_MISSING:
scenario_repository = SimpleNamespace(
get_scenario_by_code=AsyncMock(
return_value=SimpleNamespace(
id=1,
scenario_code="menu_access_request",
scenario_name="菜单权限申请",
enabled=True,
)
),
list_route_rules=AsyncMock(return_value=[]),
get_active_flow_version=AsyncMock(),
list_node_definitions=AsyncMock(),
)
route_matcher = AsyncMock(return_value=None)
else: # approver_empty: route hits a flow but the first node yields no approvers
node = SimpleNamespace(
node_code="first_node",
node_name="一级审批",
node_order=1,
node_mode="or",
approver_config={"type": "department_admin"},
)
scenario_repository = SimpleNamespace(
get_scenario_by_code=AsyncMock(
return_value=SimpleNamespace(
id=2,
scenario_code="knowledge_space_subscribe_request",
scenario_name="知识空间加入审批",
enabled=True,
)
),
list_route_rules=AsyncMock(return_value=[SimpleNamespace(id=31, route_type="flow", flow_definition_id=9)]),
get_active_flow_version=AsyncMock(return_value=SimpleNamespace(id=21)),
list_node_definitions=AsyncMock(return_value=[node]),
)
route_matcher = AsyncMock(return_value=SimpleNamespace(id=31, route_type="flow", flow_definition_id=9))
instance_repository = SimpleNamespace(
find_duplicate_active_instance=AsyncMock(return_value=None),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={"id": 901})),
create_exception=AsyncMock(),
create_task=AsyncMock(),
)
gate = ApprovalGate(
registry=registry,
scenario_repository=scenario_repository,
instance_repository=instance_repository,
route_matcher=route_matcher,
)
request_kwargs = {
"tenant_id": 1,
"business_key": "menu:knowledge:user:7",
"business_resource_type": "web_menu",
"business_resource_id": "knowledge",
"business_name": "知识管理",
"applicant_user_id": 7,
"applicant_user_name": "alice",
"reason": "需要权限",
"ip_address": "10.0.0.42",
}
if exception_type == ApprovalExceptionType.ROUTE_MISSING:
request_kwargs["scenario_code"] = "menu_access_request"
request_kwargs["payload_snapshot"] = {"menu_key": "knowledge"}
else:
request_kwargs["scenario_code"] = "knowledge_space_subscribe_request"
request_kwargs["business_key"] = "space:12:user:7"
request_kwargs["business_resource_type"] = "knowledge_space"
request_kwargs["business_resource_id"] = "12"
request_kwargs["business_name"] = "研发知识空间"
request_kwargs["payload_snapshot"] = {"space_id": 12}
with (
patch(
"bisheng.approval.domain.services.approval_gate.AuditLogDao.ainsert_v2",
new=AsyncMock(side_effect=lambda **kw: captured.append(kw)),
),
patch.object(ApprovalGate, "_notify_admins_of_exception", new=AsyncMock()),
):
result = await gate.request_or_pass(ApprovalGateRequest(**request_kwargs))
assert result.decision == ApprovalGateDecision.EXCEPTION
assert result.exception_type == exception_type
return captured, result
async def test_gate_emits_submit_audit_when_route_missing_exception():
captured, result = await _run_gate_for_exception(
exception_type=ApprovalExceptionType.ROUTE_MISSING,
)
assert len(captured) == 1, "submit audit log must be written even when instance hits route_missing exception"
row = captured[0]
assert row["action"] == "approval.request.submit"
assert row["target_type"] == "approval_instance"
assert row["target_id"] == str(result.instance_id)
assert row["operator_id"] == 7
assert row["operator_name"] == "alice"
assert row["ip_address"] == "10.0.0.42"
assert row["reason"] == "需要权限"
assert row["metadata"]["scenario_code"] == "menu_access_request"
assert row["metadata"]["exception_type"] == ApprovalExceptionType.ROUTE_MISSING
assert row["metadata"]["instance_status"] == ApprovalInstanceStatus.EXCEPTION
async def test_gate_emits_submit_audit_when_approver_empty_exception():
captured, result = await _run_gate_for_exception(
exception_type=ApprovalExceptionType.APPROVER_EMPTY,
)
assert len(captured) == 1, "submit audit log must be written even when first node has no approvers"
row = captured[0]
assert row["action"] == "approval.request.submit"
assert row["target_id"] == str(result.instance_id)
assert row["metadata"]["exception_type"] == ApprovalExceptionType.APPROVER_EMPTY
assert row["metadata"]["flow_version_id"] == 21
assert row["metadata"]["route_rule_id"] == 31
assert row["metadata"]["current_node_name"] == "一级审批"
src/backend/test/approval/test_approval_gate.py
+229 -204
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from contextlib import asynccontextmanager
from types import SimpleNamespace
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, patch
import pytest
import pytest_asyncio
@@ -18,9 +18,8 @@ from bisheng.approval.domain.models.approval_instance import (
ApprovalInstanceStatus,
ApprovalTask,
)
from bisheng.common.errcode.approval import ApprovalScenarioDisabledError
from bisheng.approval.domain.models.approval_scenario import ApprovalScenario
from bisheng.approval.domain.models.user_menu_access import UserMenuAccess, UserMenuAccessStatus
from bisheng.approval.domain.models.user_menu_access import UserMenuAccess
from bisheng.approval.domain.repositories.approval_instance_repository import ApprovalInstanceRepository
from bisheng.approval.domain.repositories.approval_query_repository import ApprovalQueryRepository
from bisheng.approval.domain.repositories.approval_scenario_repository import ApprovalScenarioRepository
@@ -28,13 +27,14 @@ from bisheng.approval.domain.repositories.user_menu_access_repository import Use
from bisheng.approval.domain.schemas.approval_center_schema import ApprovalGateDecision, ApprovalGateRequest
from bisheng.approval.domain.services.approval_gate import ApprovalGate
from bisheng.approval.domain.services.approval_registry import ApprovalRegistry
from bisheng.common.errcode.approval import ApprovalScenarioDisabledError
@pytest_asyncio.fixture
async def approval_db_engine():
engine = create_async_engine(
'sqlite+aiosqlite://',
connect_args={'check_same_thread': False},
"sqlite+aiosqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
tables = [
@@ -58,19 +58,19 @@ async def patched_approval_repositories(approval_db_engine, monkeypatch):
yield session
monkeypatch.setattr(
'bisheng.approval.domain.repositories.approval_scenario_repository.get_async_db_session',
"bisheng.approval.domain.repositories.approval_scenario_repository.get_async_db_session",
_factory,
)
monkeypatch.setattr(
'bisheng.approval.domain.repositories.approval_instance_repository.get_async_db_session',
"bisheng.approval.domain.repositories.approval_instance_repository.get_async_db_session",
_factory,
)
monkeypatch.setattr(
'bisheng.approval.domain.repositories.approval_query_repository.get_async_db_session',
"bisheng.approval.domain.repositories.approval_query_repository.get_async_db_session",
_factory,
)
monkeypatch.setattr(
'bisheng.approval.domain.repositories.user_menu_access_repository.get_async_db_session',
"bisheng.approval.domain.repositories.user_menu_access_repository.get_async_db_session",
_factory,
)
yield
@@ -81,8 +81,8 @@ async def test_repository_flow_creates_route_missing_exception(patched_approval_
scenario = await ApprovalScenarioRepository.create_scenario(
ApprovalScenario(
tenant_id=1,
scenario_code='menu_access_request',
scenario_name='菜单权限申请',
scenario_code="menu_access_request",
scenario_name="菜单权限申请",
enabled=True,
)
)
@@ -91,16 +91,16 @@ async def test_repository_flow_creates_route_missing_exception(patched_approval_
tenant_id=1,
scenario_code=scenario.scenario_code,
scenario_name=scenario.scenario_name,
handler_key='menu_access',
business_key='menu:knowledge',
business_resource_type='web_menu',
business_resource_id='knowledge',
business_name='知识管理',
handler_key="menu_access",
business_key="menu:knowledge",
business_resource_type="web_menu",
business_resource_id="knowledge",
business_name="知识管理",
applicant_user_id=7,
applicant_user_name='alice',
applicant_user_name="alice",
status=ApprovalInstanceStatus.EXCEPTION,
payload_snapshot={'menu_key': 'knowledge'},
detail_snapshot={'menu_name': '知识管理'},
payload_snapshot={"menu_key": "knowledge"},
detail_snapshot={"menu_name": "知识管理"},
)
)
await ApprovalInstanceRepository.create_exception(
@@ -108,7 +108,7 @@ async def test_repository_flow_creates_route_missing_exception(patched_approval_
tenant_id=1,
instance_id=instance.id,
exception_type=ApprovalExceptionType.ROUTE_MISSING,
detail={'scenario_code': scenario.scenario_code},
detail={"scenario_code": scenario.scenario_code},
)
)
@@ -127,25 +127,25 @@ async def test_repository_returns_existing_duplicate_pending_instance(patched_ap
await ApprovalInstanceRepository.create_instance(
ApprovalInstance(
tenant_id=1,
scenario_code='channel_subscribe_request',
scenario_name='频道订阅审批',
handler_key='channel_subscribe',
business_key='channel:88:user:7',
business_resource_type='channel',
business_resource_id='88',
business_name='资讯频道',
scenario_code="channel_subscribe_request",
scenario_name="频道订阅审批",
handler_key="channel_subscribe",
business_key="channel:88:user:7",
business_resource_type="channel",
business_resource_id="88",
business_name="资讯频道",
applicant_user_id=7,
applicant_user_name='alice',
applicant_user_name="alice",
status=ApprovalInstanceStatus.PENDING,
payload_snapshot={'channel_id': 88},
detail_snapshot={'channel_name': '资讯频道'},
payload_snapshot={"channel_id": 88},
detail_snapshot={"channel_name": "资讯频道"},
)
)
duplicate = await ApprovalInstanceRepository.find_duplicate_active_instance(
tenant_id=1,
scenario_code='channel_subscribe_request',
business_key='channel:88:user:7',
scenario_code="channel_subscribe_request",
business_key="channel:88:user:7",
applicant_user_id=7,
)
@@ -158,43 +158,43 @@ async def test_repository_allows_new_application_after_closed_instance(patched_a
await ApprovalInstanceRepository.create_instance(
ApprovalInstance(
tenant_id=1,
scenario_code='channel_subscribe_request',
scenario_name='频道订阅审批',
handler_key='channel_subscribe',
business_key='channel:88:user:7',
business_resource_type='channel',
business_resource_id='88',
business_name='资讯频道',
scenario_code="channel_subscribe_request",
scenario_name="频道订阅审批",
handler_key="channel_subscribe",
business_key="channel:88:user:7",
business_resource_type="channel",
business_resource_id="88",
business_name="资讯频道",
applicant_user_id=7,
applicant_user_name='alice',
applicant_user_name="alice",
status=ApprovalInstanceStatus.APPROVED,
payload_snapshot={'channel_id': 88},
detail_snapshot={'channel_name': '资讯频道'},
payload_snapshot={"channel_id": 88},
detail_snapshot={"channel_name": "资讯频道"},
)
)
created = await ApprovalInstanceRepository.create_instance(
ApprovalInstance(
tenant_id=1,
scenario_code='channel_subscribe_request',
scenario_name='频道订阅审批',
handler_key='channel_subscribe',
business_key='channel:88:user:7',
business_resource_type='channel',
business_resource_id='88',
business_name='资讯频道',
scenario_code="channel_subscribe_request",
scenario_name="频道订阅审批",
handler_key="channel_subscribe",
business_key="channel:88:user:7",
business_resource_type="channel",
business_resource_id="88",
business_name="资讯频道",
applicant_user_id=7,
applicant_user_name='alice',
applicant_user_name="alice",
status=ApprovalInstanceStatus.PENDING,
payload_snapshot={'channel_id': 88},
detail_snapshot={'channel_name': '资讯频道'},
payload_snapshot={"channel_id": 88},
detail_snapshot={"channel_name": "资讯频道"},
)
)
duplicate = await ApprovalInstanceRepository.find_duplicate_active_instance(
tenant_id=1,
scenario_code='channel_subscribe_request',
business_key='channel:88:user:7',
scenario_code="channel_subscribe_request",
business_key="channel:88:user:7",
applicant_user_id=7,
)
@@ -208,42 +208,42 @@ async def test_user_menu_access_repository_lists_active_grants(patched_approval_
await UserMenuAccessRepository.upsert_active_grant(
tenant_id=1,
user_id=9,
menu_key='workstation',
menu_name='工作台',
grant_source='approval_instance',
menu_key="workstation",
menu_name="工作台",
grant_source="approval_instance",
grant_instance_id=11,
)
await UserMenuAccessRepository.upsert_active_grant(
tenant_id=1,
user_id=9,
menu_key='home',
menu_name='首页',
grant_source='approval_instance',
menu_key="home",
menu_name="首页",
grant_source="approval_instance",
grant_instance_id=11,
)
await UserMenuAccessRepository.upsert_active_grant(
tenant_id=1,
user_id=10,
menu_key='apps',
menu_name='应用中心',
grant_source='approval_instance',
menu_key="apps",
menu_name="应用中心",
grant_source="approval_instance",
grant_instance_id=12,
)
await UserMenuAccessRepository.revoke_grant(
tenant_id=1,
user_id=9,
menu_key='workstation',
grant_source='approval_instance',
menu_key="workstation",
grant_source="approval_instance",
revoked_by_user_id=1,
revoked_reason='manual revoke',
revoked_reason="manual revoke",
)
user9_keys = await UserMenuAccessRepository.list_active_menu_keys(1, 9)
user10_keys = await UserMenuAccessRepository.list_active_menu_keys(1, 10)
revoked = await UserMenuAccessRepository.get_active_grant(1, 9, 'workstation')
revoked = await UserMenuAccessRepository.get_active_grant(1, 9, "workstation")
assert user9_keys == ['home']
assert user10_keys == ['apps']
assert user9_keys == ["home"]
assert user10_keys == ["apps"]
assert revoked is None
@@ -251,7 +251,7 @@ async def test_user_menu_access_repository_lists_active_grants(patched_approval_
async def test_gate_raises_disabled_error_when_scenario_not_configured():
handler = SimpleNamespace(
build_detail=AsyncMock(return_value={}),
build_title=AsyncMock(return_value='知识管理'),
build_title=AsyncMock(return_value="知识管理"),
)
registry = SimpleNamespace(get_handler=AsyncMock(return_value=handler))
scenario_repository = SimpleNamespace(
@@ -273,14 +273,14 @@ async def test_gate_raises_disabled_error_when_scenario_not_configured():
await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code='menu_access_request',
business_key='menu:knowledge:user:7',
business_resource_type='web_menu',
business_resource_id='knowledge',
business_name='知识管理',
scenario_code="menu_access_request",
business_key="menu:knowledge:user:7",
business_resource_type="web_menu",
business_resource_id="knowledge",
business_name="知识管理",
applicant_user_id=7,
applicant_user_name='alice',
payload_snapshot={'menu_key': 'knowledge'},
applicant_user_name="alice",
payload_snapshot={"menu_key": "knowledge"},
)
)
instance_repository.create_instance.assert_not_awaited()
@@ -290,15 +290,15 @@ async def test_gate_raises_disabled_error_when_scenario_not_configured():
async def test_gate_raises_disabled_error_when_scenario_disabled():
handler = SimpleNamespace(
build_detail=AsyncMock(return_value={}),
build_title=AsyncMock(return_value='知识管理'),
build_title=AsyncMock(return_value="知识管理"),
)
registry = SimpleNamespace(get_handler=AsyncMock(return_value=handler))
scenario_repository = SimpleNamespace(
get_scenario_by_code=AsyncMock(
return_value=SimpleNamespace(
id=1,
scenario_code='menu_access_request',
scenario_name='菜单权限申请',
scenario_code="menu_access_request",
scenario_name="菜单权限申请",
enabled=False,
)
),
@@ -319,14 +319,14 @@ async def test_gate_raises_disabled_error_when_scenario_disabled():
await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code='menu_access_request',
business_key='menu:knowledge:user:7',
business_resource_type='web_menu',
business_resource_id='knowledge',
business_name='知识管理',
scenario_code="menu_access_request",
business_key="menu:knowledge:user:7",
business_resource_type="web_menu",
business_resource_id="knowledge",
business_name="知识管理",
applicant_user_id=7,
applicant_user_name='alice',
payload_snapshot={'menu_key': 'knowledge'},
applicant_user_name="alice",
payload_snapshot={"menu_key": "knowledge"},
)
)
instance_repository.create_instance.assert_not_awaited()
@@ -335,8 +335,8 @@ async def test_gate_raises_disabled_error_when_scenario_disabled():
@pytest.mark.asyncio
async def test_gate_pass_when_route_direct_approve():
handler = SimpleNamespace(
build_detail=AsyncMock(return_value={'menu_name': '知识管理'}),
build_title=AsyncMock(return_value='知识管理'),
build_detail=AsyncMock(return_value={"menu_name": "知识管理"}),
build_title=AsyncMock(return_value="知识管理"),
resolve_approvers=AsyncMock(),
)
registry = SimpleNamespace(get_handler=AsyncMock(return_value=handler))
@@ -344,18 +344,18 @@ async def test_gate_pass_when_route_direct_approve():
get_scenario_by_code=AsyncMock(
return_value=SimpleNamespace(
id=1,
scenario_code='menu_access_request',
scenario_name='菜单权限申请',
scenario_code="menu_access_request",
scenario_name="菜单权限申请",
enabled=True,
)
),
list_route_rules=AsyncMock(return_value=[SimpleNamespace(id=11, route_type='pass')]),
list_route_rules=AsyncMock(return_value=[SimpleNamespace(id=11, route_type="pass")]),
get_active_flow_version=AsyncMock(),
list_node_definitions=AsyncMock(),
)
instance_repository = SimpleNamespace(
find_duplicate_active_instance=AsyncMock(return_value=None),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={'id': 201})),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={"id": 201})),
create_exception=AsyncMock(),
create_task=AsyncMock(),
)
@@ -363,20 +363,20 @@ async def test_gate_pass_when_route_direct_approve():
registry=registry,
scenario_repository=scenario_repository,
instance_repository=instance_repository,
route_matcher=AsyncMock(return_value=SimpleNamespace(id=11, route_type='pass')),
route_matcher=AsyncMock(return_value=SimpleNamespace(id=11, route_type="pass")),
)
result = await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code='menu_access_request',
business_key='menu:knowledge:user:7',
business_resource_type='web_menu',
business_resource_id='knowledge',
business_name='知识管理',
scenario_code="menu_access_request",
business_key="menu:knowledge:user:7",
business_resource_type="web_menu",
business_resource_id="knowledge",
business_name="知识管理",
applicant_user_id=7,
applicant_user_name='alice',
payload_snapshot={'menu_key': 'knowledge'},
applicant_user_name="alice",
payload_snapshot={"menu_key": "knowledge"},
)
)
@@ -390,29 +390,29 @@ async def test_gate_pass_when_route_direct_approve():
@pytest.mark.asyncio
async def test_gate_pending_when_route_hits_flow():
handler = SimpleNamespace(
build_detail=AsyncMock(return_value={'space_name': '研发知识空间'}),
build_title=AsyncMock(return_value='研发知识空间'),
build_detail=AsyncMock(return_value={"space_name": "研发知识空间"}),
build_title=AsyncMock(return_value="研发知识空间"),
resolve_approvers=AsyncMock(return_value=[1001, 1002]),
)
registry = SimpleNamespace(get_handler=AsyncMock(return_value=handler))
flow_version = SimpleNamespace(id=21)
node = SimpleNamespace(
node_code='first_node',
node_name='一级审批',
node_code="first_node",
node_name="一级审批",
node_order=1,
node_mode='or',
approver_config={'type': 'direct'},
node_mode="or",
approver_config={"type": "direct"},
)
scenario_repository = SimpleNamespace(
get_scenario_by_code=AsyncMock(
return_value=SimpleNamespace(
id=2,
scenario_code='knowledge_space_subscribe_request',
scenario_name='知识空间加入审批',
scenario_code="knowledge_space_subscribe_request",
scenario_name="知识空间加入审批",
enabled=True,
)
),
list_route_rules=AsyncMock(return_value=[SimpleNamespace(id=31, route_type='flow', flow_definition_id=9)]),
list_route_rules=AsyncMock(return_value=[SimpleNamespace(id=31, route_type="flow", flow_definition_id=9)]),
get_active_flow_version=AsyncMock(return_value=flow_version),
list_node_definitions=AsyncMock(return_value=[node]),
)
@@ -421,28 +421,28 @@ async def test_gate_pending_when_route_hits_flow():
instance_repository = SimpleNamespace(
find_duplicate_active_instance=AsyncMock(return_value=None),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={'id': 300})),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={"id": 300})),
create_exception=AsyncMock(),
create_task=AsyncMock(side_effect=lambda row: row.model_copy(update={'id': next(task_ids)})),
create_task=AsyncMock(side_effect=lambda row: row.model_copy(update={"id": next(task_ids)})),
)
gate = ApprovalGate(
registry=registry,
scenario_repository=scenario_repository,
instance_repository=instance_repository,
route_matcher=AsyncMock(return_value=SimpleNamespace(id=31, route_type='flow', flow_definition_id=9)),
route_matcher=AsyncMock(return_value=SimpleNamespace(id=31, route_type="flow", flow_definition_id=9)),
)
result = await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code='knowledge_space_subscribe_request',
business_key='space:12:user:7',
business_resource_type='knowledge_space',
business_resource_id='12',
business_name='研发知识空间',
scenario_code="knowledge_space_subscribe_request",
business_key="space:12:user:7",
business_resource_type="knowledge_space",
business_resource_id="12",
business_name="研发知识空间",
applicant_user_id=7,
applicant_user_name='alice',
payload_snapshot={'space_id': 12},
applicant_user_name="alice",
payload_snapshot={"space_id": 12},
)
)
@@ -470,14 +470,14 @@ async def test_gate_returns_existing_instance_for_duplicate_business_key():
result = await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code='channel_subscribe_request',
business_key='channel:88:user:7',
business_resource_type='channel',
business_resource_id='88',
business_name='资讯频道',
scenario_code="channel_subscribe_request",
business_key="channel:88:user:7",
business_resource_type="channel",
business_resource_id="88",
business_name="资讯频道",
applicant_user_id=7,
applicant_user_name='alice',
payload_snapshot={'channel_id': 88},
applicant_user_name="alice",
payload_snapshot={"channel_id": 88},
)
)
@@ -487,11 +487,17 @@ async def test_gate_returns_existing_instance_for_duplicate_business_key():
@pytest.mark.asyncio
async def test_gate_creates_route_missing_exception():
handler = SimpleNamespace(build_detail=AsyncMock(return_value={}), build_title=AsyncMock(return_value='频道'), resolve_approvers=AsyncMock())
handler = SimpleNamespace(
build_detail=AsyncMock(return_value={}),
build_title=AsyncMock(return_value="频道"),
resolve_approvers=AsyncMock(),
)
registry = SimpleNamespace(get_handler=AsyncMock(return_value=handler))
scenario_repository = SimpleNamespace(
get_scenario_by_code=AsyncMock(
return_value=SimpleNamespace(id=1, scenario_code='channel_subscribe_request', scenario_name='频道订阅审批', enabled=True)
return_value=SimpleNamespace(
id=1, scenario_code="channel_subscribe_request", scenario_name="频道订阅审批", enabled=True
)
),
list_route_rules=AsyncMock(return_value=[]),
get_active_flow_version=AsyncMock(),
@@ -499,7 +505,7 @@ async def test_gate_creates_route_missing_exception():
)
instance_repository = SimpleNamespace(
find_duplicate_active_instance=AsyncMock(return_value=None),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={'id': 501})),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={"id": 501})),
create_exception=AsyncMock(),
create_task=AsyncMock(),
)
@@ -510,19 +516,23 @@ async def test_gate_creates_route_missing_exception():
route_matcher=AsyncMock(return_value=None),
)
result = await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code='channel_subscribe_request',
business_key='channel:88:user:7',
business_resource_type='channel',
business_resource_id='88',
business_name='资讯频道',
applicant_user_id=7,
applicant_user_name='alice',
payload_snapshot={'channel_id': 88},
with patch(
"bisheng.approval.domain.services.approval_gate.AuditLogDao.ainsert_v2",
new=AsyncMock(),
):
result = await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code="channel_subscribe_request",
business_key="channel:88:user:7",
business_resource_type="channel",
business_resource_id="88",
business_name="资讯频道",
applicant_user_id=7,
applicant_user_name="alice",
payload_snapshot={"channel_id": 88},
)
)
)
assert result.decision == ApprovalGateDecision.EXCEPTION
assert result.exception_type == ApprovalExceptionType.ROUTE_MISSING
@@ -531,29 +541,31 @@ async def test_gate_creates_route_missing_exception():
@pytest.mark.asyncio
async def test_gate_creates_approver_empty_exception():
handler = SimpleNamespace(
build_detail=AsyncMock(return_value={'space_name': '研发知识空间'}),
build_title=AsyncMock(return_value='研发知识空间'),
build_detail=AsyncMock(return_value={"space_name": "研发知识空间"}),
build_title=AsyncMock(return_value="研发知识空间"),
resolve_approvers=AsyncMock(return_value=[]),
)
registry = SimpleNamespace(get_handler=AsyncMock(return_value=handler))
node = SimpleNamespace(
node_code='first_node',
node_name='一级审批',
node_code="first_node",
node_name="一级审批",
node_order=1,
node_mode='or',
approver_config={'type': 'department_admin'},
node_mode="or",
approver_config={"type": "department_admin"},
)
scenario_repository = SimpleNamespace(
get_scenario_by_code=AsyncMock(
return_value=SimpleNamespace(id=2, scenario_code='knowledge_space_subscribe_request', scenario_name='知识空间加入审批', enabled=True)
return_value=SimpleNamespace(
id=2, scenario_code="knowledge_space_subscribe_request", scenario_name="知识空间加入审批", enabled=True
)
),
list_route_rules=AsyncMock(return_value=[SimpleNamespace(id=31, route_type='flow', flow_definition_id=9)]),
list_route_rules=AsyncMock(return_value=[SimpleNamespace(id=31, route_type="flow", flow_definition_id=9)]),
get_active_flow_version=AsyncMock(return_value=SimpleNamespace(id=21)),
list_node_definitions=AsyncMock(return_value=[node]),
)
instance_repository = SimpleNamespace(
find_duplicate_active_instance=AsyncMock(return_value=None),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={'id': 601})),
create_instance=AsyncMock(side_effect=lambda row: row.model_copy(update={"id": 601})),
create_exception=AsyncMock(),
create_task=AsyncMock(),
)
@@ -561,22 +573,26 @@ async def test_gate_creates_approver_empty_exception():
registry=registry,
scenario_repository=scenario_repository,
instance_repository=instance_repository,
route_matcher=AsyncMock(return_value=SimpleNamespace(id=31, route_type='flow', flow_definition_id=9)),
route_matcher=AsyncMock(return_value=SimpleNamespace(id=31, route_type="flow", flow_definition_id=9)),
)
result = await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code='knowledge_space_subscribe_request',
business_key='space:12:user:7',
business_resource_type='knowledge_space',
business_resource_id='12',
business_name='研发知识空间',
applicant_user_id=7,
applicant_user_name='alice',
payload_snapshot={'space_id': 12},
with patch(
"bisheng.approval.domain.services.approval_gate.AuditLogDao.ainsert_v2",
new=AsyncMock(),
):
result = await gate.request_or_pass(
ApprovalGateRequest(
tenant_id=1,
scenario_code="knowledge_space_subscribe_request",
business_key="space:12:user:7",
business_resource_type="knowledge_space",
business_resource_id="12",
business_name="研发知识空间",
applicant_user_id=7,
applicant_user_name="alice",
payload_snapshot={"space_id": 12},
)
)
)
assert result.decision == ApprovalGateDecision.EXCEPTION
assert result.exception_type == ApprovalExceptionType.APPROVER_EMPTY
@@ -585,6 +601,7 @@ async def test_gate_creates_approver_empty_exception():
# ── _match_first_route unit tests ────────────────────────────────────────────
def _make_gate() -> ApprovalGate:
registry = SimpleNamespace(get_handler=AsyncMock())
return ApprovalGate(registry=registry)
@@ -593,18 +610,18 @@ def _make_gate() -> ApprovalGate:
def _req(user_id: int = 7, tenant_id: int = 1, payload: dict | None = None) -> ApprovalGateRequest:
return ApprovalGateRequest(
tenant_id=tenant_id,
scenario_code='menu_access_request',
business_key=f'menu:k:user:{user_id}',
business_resource_type='web_menu',
business_resource_id='k',
business_name='知识管理',
scenario_code="menu_access_request",
business_key=f"menu:k:user:{user_id}",
business_resource_type="web_menu",
business_resource_id="k",
business_name="知识管理",
applicant_user_id=user_id,
applicant_user_name='alice',
applicant_user_name="alice",
payload_snapshot=payload or {},
)
def _route(route_type: str = 'pass', match_config: dict | None = None, enabled: bool = True):
def _route(route_type: str = "pass", match_config: dict | None = None, enabled: bool = True):
return SimpleNamespace(
id=99,
route_type=route_type,
@@ -618,75 +635,81 @@ def _route(route_type: str = 'pass', match_config: dict | None = None, enabled:
async def test_match_first_route_catchall_returns_first_enabled(monkeypatch):
"""Empty match_config → always matches first enabled route."""
import bisheng.approval.domain.services.approval_gate as gm
monkeypatch.setattr(gm, '_get_user_role_labels', AsyncMock(return_value=frozenset({'regular_user'})))
monkeypatch.setattr(gm, "_get_user_role_labels", AsyncMock(return_value=frozenset({"regular_user"})))
gate = _make_gate()
result = await gate._match_first_route([_route()], _req())
assert result is not None
assert result.route_type == 'pass'
assert result.route_type == "pass"
@pytest.mark.asyncio
async def test_match_first_route_skips_disabled_routes(monkeypatch):
"""Disabled routes are never selected, even if they would otherwise match."""
import bisheng.approval.domain.services.approval_gate as gm
monkeypatch.setattr(gm, '_get_user_role_labels', AsyncMock(return_value=frozenset({'regular_user'})))
monkeypatch.setattr(gm, "_get_user_role_labels", AsyncMock(return_value=frozenset({"regular_user"})))
gate = _make_gate()
disabled = _route(enabled=False)
enabled_fallback = _route(route_type='flow', match_config={})
enabled_fallback = _route(route_type="flow", match_config={})
result = await gate._match_first_route([disabled, enabled_fallback], _req())
assert result is not None
assert result.route_type == 'flow'
assert result.route_type == "flow"
@pytest.mark.asyncio
async def test_match_first_route_applicant_role_admin_matches(monkeypatch):
"""applicant_role=admin condition matches when user has admin label."""
import bisheng.approval.domain.services.approval_gate as gm
monkeypatch.setattr(gm, '_get_user_role_labels', AsyncMock(return_value=frozenset({'admin', 'regular_user'})))
monkeypatch.setattr(gm, "_get_user_role_labels", AsyncMock(return_value=frozenset({"admin", "regular_user"})))
gate = _make_gate()
admin_route = _route(route_type='pass', match_config={'field': 'applicant_role', 'value': 'admin'})
regular_route = _route(route_type='flow', match_config={'field': 'applicant_role', 'value': 'regular_user'})
admin_route = _route(route_type="pass", match_config={"field": "applicant_role", "value": "admin"})
regular_route = _route(route_type="flow", match_config={"field": "applicant_role", "value": "regular_user"})
result = await gate._match_first_route([admin_route, regular_route], _req())
assert result is not None
assert result.route_type == 'pass'
assert result.route_type == "pass"
@pytest.mark.asyncio
async def test_match_first_route_applicant_role_regular_skips_admin_route(monkeypatch):
"""regular_user should skip admin-only route and fall through to regular route."""
import bisheng.approval.domain.services.approval_gate as gm
monkeypatch.setattr(gm, '_get_user_role_labels', AsyncMock(return_value=frozenset({'regular_user'})))
monkeypatch.setattr(gm, "_get_user_role_labels", AsyncMock(return_value=frozenset({"regular_user"})))
gate = _make_gate()
admin_route = _route(route_type='pass', match_config={'field': 'applicant_role', 'value': 'admin'})
regular_route = _route(route_type='flow', match_config={'field': 'applicant_role', 'value': 'regular_user'})
admin_route = _route(route_type="pass", match_config={"field": "applicant_role", "value": "admin"})
regular_route = _route(route_type="flow", match_config={"field": "applicant_role", "value": "regular_user"})
result = await gate._match_first_route([admin_route, regular_route], _req())
assert result is not None
assert result.route_type == 'flow'
assert result.route_type == "flow"
@pytest.mark.asyncio
async def test_match_first_route_dept_admin_matches(monkeypatch):
"""dept_admin user should match dept_admin route, not admin route."""
import bisheng.approval.domain.services.approval_gate as gm
monkeypatch.setattr(gm, '_get_user_role_labels', AsyncMock(return_value=frozenset({'dept_admin', 'regular_user'})))
monkeypatch.setattr(gm, "_get_user_role_labels", AsyncMock(return_value=frozenset({"dept_admin", "regular_user"})))
gate = _make_gate()
admin_route = _route(route_type='pass', match_config={'field': 'applicant_role', 'value': 'admin'})
dept_route = _route(route_type='flow', match_config={'field': 'applicant_role', 'value': 'dept_admin'})
admin_route = _route(route_type="pass", match_config={"field": "applicant_role", "value": "admin"})
dept_route = _route(route_type="flow", match_config={"field": "applicant_role", "value": "dept_admin"})
result = await gate._match_first_route([admin_route, dept_route], _req())
assert result is not None
assert result.route_type == 'flow'
assert result.route_type == "flow"
@pytest.mark.asyncio
async def test_match_first_route_user_labels_queried_once(monkeypatch):
"""_get_user_role_labels is called at most once per match, even with multiple role conditions."""
import bisheng.approval.domain.services.approval_gate as gm
mock_labels = AsyncMock(return_value=frozenset({'regular_user'}))
monkeypatch.setattr(gm, '_get_user_role_labels', mock_labels)
mock_labels = AsyncMock(return_value=frozenset({"regular_user"}))
monkeypatch.setattr(gm, "_get_user_role_labels", mock_labels)
gate = _make_gate()
routes = [
_route(match_config={'field': 'applicant_role', 'value': 'admin'}),
_route(match_config={'field': 'applicant_role', 'value': 'dept_admin'}),
_route(match_config={"field": "applicant_role", "value": "admin"}),
_route(match_config={"field": "applicant_role", "value": "dept_admin"}),
_route(match_config={}), # catch-all
]
result = await gate._match_first_route(routes, _req())
@@ -699,25 +722,27 @@ async def test_match_first_route_user_labels_queried_once(monkeypatch):
async def test_match_first_route_payload_condition(monkeypatch):
"""Payload-based condition matches when payload_snapshot contains the expected value."""
import bisheng.approval.domain.services.approval_gate as gm
monkeypatch.setattr(gm, '_get_user_role_labels', AsyncMock(return_value=frozenset({'regular_user'})))
monkeypatch.setattr(gm, "_get_user_role_labels", AsyncMock(return_value=frozenset({"regular_user"})))
gate = _make_gate()
menu_route = _route(route_type='pass', match_config={'field': 'menu_key', 'value': 'approval_manage'})
other_route = _route(route_type='flow', match_config={})
req = _req(payload={'menu_key': 'approval_manage'})
menu_route = _route(route_type="pass", match_config={"field": "menu_key", "value": "approval_manage"})
other_route = _route(route_type="flow", match_config={})
req = _req(payload={"menu_key": "approval_manage"})
result = await gate._match_first_route([menu_route, other_route], req)
assert result is not None
assert result.route_type == 'pass'
assert result.route_type == "pass"
@pytest.mark.asyncio
async def test_match_first_route_returns_none_when_no_match(monkeypatch):
"""Returns None when all routes have conditions that don't match."""
import bisheng.approval.domain.services.approval_gate as gm
monkeypatch.setattr(gm, '_get_user_role_labels', AsyncMock(return_value=frozenset({'regular_user'})))
monkeypatch.setattr(gm, "_get_user_role_labels", AsyncMock(return_value=frozenset({"regular_user"})))
gate = _make_gate()
routes = [
_route(match_config={'field': 'applicant_role', 'value': 'admin'}),
_route(match_config={'field': 'applicant_role', 'value': 'dept_admin'}),
_route(match_config={"field": "applicant_role", "value": "admin"}),
_route(match_config={"field": "applicant_role", "value": "dept_admin"}),
]
result = await gate._match_first_route(routes, _req())
assert result is None
@@ -729,8 +754,8 @@ def test_registry_exposes_default_presets():
presets = {preset.scenario_code: preset for preset in registry.list_presets()}
assert set(presets) == {
'menu_access_request',
'channel_subscribe_request',
'knowledge_space_subscribe_request',
"menu_access_request",
"channel_subscribe_request",
"knowledge_space_subscribe_request",
}
assert presets['menu_access_request'].handler_key == 'menu_access_request'
assert presets["menu_access_request"].handler_key == "menu_access_request"