fix: 完成系统向智能体数据链路切换
- 切换工作台、聊天历史、资源候选与公共调用到 Agent - 加固资源绑定、删除保护及发布运行并发控制 - 隔离旧 Bot 专属服务和组件并保留兼容入口
This commit is contained in:
@@ -113,6 +113,34 @@ public class AgentRuntimeCommandConsumerTest {
|
||||
Assert.assertEquals("cmd-expire", resultRegistry.lastSuccessCommandId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 Agent 集群取消命令只取消目标节点的对应 Agent 运行。
|
||||
*
|
||||
* @throws Exception 消息序列化异常
|
||||
*/
|
||||
@Test
|
||||
public void consumerShouldHandleCancelAgentCommand() throws Exception {
|
||||
AgentRuntimeProperties properties = new AgentRuntimeProperties();
|
||||
properties.setInstanceId("node-a");
|
||||
RecordingAgentRunService service = new RecordingAgentRunService();
|
||||
RecordingCommandResultRegistry resultRegistry = new RecordingCommandResultRegistry();
|
||||
AgentRuntimeCommandConsumer consumer = new AgentRuntimeCommandConsumer(
|
||||
new ObjectMapper(),
|
||||
properties,
|
||||
new MQProperties(),
|
||||
service,
|
||||
resultRegistry
|
||||
);
|
||||
AgentRuntimeCommandMessage command = command("cmd-cancel", "node-a");
|
||||
command.setAction(AgentRuntimeCommandAction.CANCEL_AGENT);
|
||||
command.setAgentId("1001");
|
||||
|
||||
consumer.handle(List.of(message(command)));
|
||||
|
||||
Assert.assertEquals("1001", service.lastCancelledAgentId);
|
||||
Assert.assertEquals("cmd-cancel", resultRegistry.lastSuccessCommandId);
|
||||
}
|
||||
|
||||
private AgentRuntimeCommandMessage command(String commandId, String targetNodeId) {
|
||||
AgentRuntimeCommandMessage command = new AgentRuntimeCommandMessage();
|
||||
command.setCommandId(commandId);
|
||||
@@ -138,6 +166,7 @@ public class AgentRuntimeCommandConsumerTest {
|
||||
private int expireCount;
|
||||
private String lastRequestId;
|
||||
private String lastReason;
|
||||
private String lastCancelledAgentId;
|
||||
|
||||
@Override
|
||||
public void approveRuntimeLocal(String requestId, String resumeToken, BigInteger operatorId, String userId) {
|
||||
@@ -151,6 +180,11 @@ public class AgentRuntimeCommandConsumerTest {
|
||||
lastRequestId = requestId;
|
||||
lastReason = reason;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancelAgentLocal(String agentId) {
|
||||
lastCancelledAgentId = agentId;
|
||||
}
|
||||
}
|
||||
|
||||
private static class RecordingCommandResultRegistry extends AgentRuntimeCommandResultRegistry {
|
||||
|
||||
@@ -6,11 +6,13 @@ import org.junit.Test;
|
||||
import org.mockito.ArgumentMatchers;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.core.SetOperations;
|
||||
import org.springframework.data.redis.core.ValueOperations;
|
||||
import tech.easyflow.agent.config.AgentRuntimeProperties;
|
||||
import tech.easyflow.agent.distributed.AgentRuntimeRouteRegistry;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* {@link AgentRuntimeRouteRegistry} 回归测试。
|
||||
@@ -40,12 +42,49 @@ public class AgentRuntimeRouteRegistryTest {
|
||||
"easyflow:agent:runtime:resume-token:token-1", "request-1", Duration.ofHours(24));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证正式运行会写入 Agent 反向索引，并可解析全部 owner 节点。
|
||||
*/
|
||||
@Test
|
||||
public void agentRunIndexShouldTrackOwnerNodes() {
|
||||
StringRedisTemplate redisTemplate = Mockito.mock(StringRedisTemplate.class);
|
||||
@SuppressWarnings("unchecked")
|
||||
ValueOperations<String, String> valueOperations = Mockito.mock(ValueOperations.class);
|
||||
@SuppressWarnings("unchecked")
|
||||
SetOperations<String, String> setOperations = Mockito.mock(SetOperations.class);
|
||||
Mockito.when(redisTemplate.opsForValue()).thenReturn(valueOperations);
|
||||
Mockito.when(redisTemplate.opsForSet()).thenReturn(setOperations);
|
||||
Mockito.when(setOperations.members("easyflow:agent:runtime:agent:1001"))
|
||||
.thenReturn(Set.of("request-1", "request-2"));
|
||||
Mockito.when(valueOperations.get("easyflow:agent:runtime:request:request-1"))
|
||||
.thenReturn("{\"nodeId\":\"node-a\",\"bootId\":\"boot-a\",\"agentId\":\"1001\"}");
|
||||
Mockito.when(valueOperations.get("easyflow:agent:runtime:request:request-2"))
|
||||
.thenReturn("{\"nodeId\":\"node-b\",\"bootId\":\"boot-b\",\"agentId\":\"1001\"}");
|
||||
Mockito.when(valueOperations.get("easyflow:agent:runtime:node:node-a")).thenReturn("boot-a");
|
||||
Mockito.when(valueOperations.get("easyflow:agent:runtime:node:node-b")).thenReturn("boot-b");
|
||||
AgentRuntimeRouteRegistry registry = registry(redisTemplate, properties("node-a"));
|
||||
|
||||
registry.registerRun("request-1", "1001");
|
||||
|
||||
Mockito.verify(valueOperations).set(
|
||||
ArgumentMatchers.eq("easyflow:agent:runtime:request:request-1"),
|
||||
ArgumentMatchers.contains("\"agentId\":\"1001\""),
|
||||
ArgumentMatchers.eq(Duration.ofHours(24))
|
||||
);
|
||||
Mockito.verify(setOperations).add("easyflow:agent:runtime:agent:1001", "request-1");
|
||||
Mockito.verify(redisTemplate).expire("easyflow:agent:runtime:agent:1001", Duration.ofHours(24));
|
||||
Assert.assertEquals(Set.of("node-a", "node-b"), registry.findOwnerNodesByAgent("1001"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证运行结束时清理 Redis 路由。
|
||||
*/
|
||||
@Test
|
||||
public void removeShouldDeleteRunAndTokenRoutes() {
|
||||
StringRedisTemplate redisTemplate = Mockito.mock(StringRedisTemplate.class);
|
||||
@SuppressWarnings("unchecked")
|
||||
ValueOperations<String, String> valueOperations = Mockito.mock(ValueOperations.class);
|
||||
Mockito.when(redisTemplate.opsForValue()).thenReturn(valueOperations);
|
||||
AgentRuntimeRouteRegistry registry = registry(redisTemplate, properties("node-a"));
|
||||
|
||||
registry.removeRun("request-1");
|
||||
|
||||
@@ -5,17 +5,24 @@ import com.mybatisflex.core.update.UpdateChain;
|
||||
import com.mybatisflex.core.util.LambdaGetter;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.agent.distributed.AgentRuntimeCommandProducer;
|
||||
import tech.easyflow.agent.distributed.AgentRuntimeRouteRegistry;
|
||||
import tech.easyflow.agent.entity.Agent;
|
||||
import tech.easyflow.agent.runtime.AgentRunRegistry;
|
||||
import tech.easyflow.agent.runtime.hitl.AgentHitlPendingService;
|
||||
import tech.easyflow.agent.service.AgentKnowledgeBindingService;
|
||||
import tech.easyflow.agent.service.AgentService;
|
||||
import tech.easyflow.agent.service.AgentToolBindingService;
|
||||
import tech.easyflow.agent.support.AgentBindingLockExecutor;
|
||||
import tech.easyflow.ai.enums.PublishStatus;
|
||||
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.math.BigInteger;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
@@ -82,27 +89,50 @@ public class AgentApprovalSubjectHandlerTest {
|
||||
}
|
||||
|
||||
/**
|
||||
* 审批删除 Agent 前必须同步清理工具绑定和知识库绑定，避免留下孤儿数据。
|
||||
* 审批删除 Agent 时必须在同一配置锁内取消运行并清理关联数据。
|
||||
*/
|
||||
@Test
|
||||
public void beforeRemoveShouldCleanAgentBindings() {
|
||||
public void removeResourceShouldCancelRunsAndCleanBindings() {
|
||||
AtomicInteger toolRemoveCalls = new AtomicInteger();
|
||||
AtomicInteger knowledgeRemoveCalls = new AtomicInteger();
|
||||
AgentToolBindingService toolBindingService = proxy(AgentToolBindingService.class, toolRemoveCalls);
|
||||
AgentKnowledgeBindingService knowledgeBindingService = proxy(AgentKnowledgeBindingService.class, knowledgeRemoveCalls);
|
||||
AgentService agentService = mock(AgentService.class);
|
||||
AgentRunRegistry runRegistry = mock(AgentRunRegistry.class);
|
||||
AgentHitlPendingService pendingService = mock(AgentHitlPendingService.class);
|
||||
AgentRuntimeRouteRegistry routeRegistry = mock(AgentRuntimeRouteRegistry.class);
|
||||
AgentRuntimeCommandProducer commandProducer = mock(AgentRuntimeCommandProducer.class);
|
||||
when(routeRegistry.findOwnerNodesByAgent("1001")).thenReturn(Set.of("node-a", "node-b"));
|
||||
when(routeRegistry.currentNodeId()).thenReturn("node-a");
|
||||
AgentApprovalSubjectHandler handler = new AgentApprovalSubjectHandler(
|
||||
null,
|
||||
new ObjectMapper(),
|
||||
null,
|
||||
agentService,
|
||||
toolBindingService,
|
||||
knowledgeBindingService,
|
||||
null
|
||||
null,
|
||||
immediateLockExecutor(),
|
||||
runRegistry,
|
||||
pendingService,
|
||||
routeRegistry,
|
||||
commandProducer
|
||||
);
|
||||
|
||||
handler.beforeRemove(BigInteger.valueOf(1001));
|
||||
handler.removeResource(BigInteger.valueOf(1001));
|
||||
|
||||
Assert.assertEquals(1, toolRemoveCalls.get());
|
||||
Assert.assertEquals(1, knowledgeRemoveCalls.get());
|
||||
verify(pendingService).cancelByAgentId(
|
||||
BigInteger.valueOf(1001),
|
||||
"Agent 已删除，待审批运行已取消"
|
||||
);
|
||||
verify(runRegistry).cancelAgent("1001");
|
||||
verify(commandProducer).sendCancelAgent(
|
||||
"node-b",
|
||||
"1001",
|
||||
"Agent 已删除，待审批运行已取消"
|
||||
);
|
||||
verify(agentService).removeById(BigInteger.valueOf(1001));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -112,16 +142,36 @@ public class AgentApprovalSubjectHandlerTest {
|
||||
* @return Agent 审批资源处理器
|
||||
*/
|
||||
private static AgentApprovalSubjectHandler handler(AgentService agentService) {
|
||||
AgentRuntimeRouteRegistry routeRegistry = mock(AgentRuntimeRouteRegistry.class);
|
||||
when(routeRegistry.findOwnerNodesByAgent("1001")).thenReturn(Set.of());
|
||||
return new AgentApprovalSubjectHandler(
|
||||
null,
|
||||
new ObjectMapper(),
|
||||
agentService,
|
||||
null,
|
||||
null,
|
||||
null
|
||||
null,
|
||||
immediateLockExecutor(),
|
||||
mock(AgentRunRegistry.class),
|
||||
mock(AgentHitlPendingService.class),
|
||||
routeRegistry,
|
||||
mock(AgentRuntimeCommandProducer.class)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建同步执行任务的 Agent 配置锁测试桩。
|
||||
*
|
||||
* @return Agent 配置锁执行器
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
private static AgentBindingLockExecutor immediateLockExecutor() {
|
||||
AgentBindingLockExecutor executor = mock(AgentBindingLockExecutor.class);
|
||||
when(executor.execute(any(BigInteger.class), any(Supplier.class)))
|
||||
.thenAnswer(invocation -> ((Supplier<?>) invocation.getArgument(1)).get());
|
||||
return executor;
|
||||
}
|
||||
|
||||
/**
|
||||
* 准备字段更新对象。
|
||||
*
|
||||
|
||||
@@ -1246,6 +1246,11 @@ public class AgentRunServiceDraftAndHitlTest {
|
||||
cancelByRequestIdCount++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancelByAgentId(BigInteger agentId, String reason) {
|
||||
// 测试桩无需处理。
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteByChatSessionId(BigInteger chatSessionId) {
|
||||
// 测试桩无需处理。
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
package tech.easyflow.agent.runtime;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
import tech.easyflow.agent.entity.Agent;
|
||||
import tech.easyflow.agent.entity.AgentToolBinding;
|
||||
import tech.easyflow.agent.service.AgentService;
|
||||
import tech.easyflow.ai.enums.PublishStatus;
|
||||
import tech.easyflow.chatlog.service.ChatSessionQueryService;
|
||||
import tech.easyflow.common.entity.LoginAccount;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
|
||||
import java.lang.reflect.Field;
|
||||
import java.math.BigInteger;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* {@link AgentRunService} 公共 API 租户边界测试。
|
||||
*/
|
||||
public class AgentRunServicePublicTest {
|
||||
|
||||
/**
|
||||
* 验证 API Key 不能运行其他租户的 Agent。
|
||||
*
|
||||
* @throws Exception 注入测试依赖失败
|
||||
*/
|
||||
@Test
|
||||
public void chatPublicShouldRejectCrossTenantAgent() throws Exception {
|
||||
BigInteger agentId = BigInteger.valueOf(1001);
|
||||
Agent agent = new Agent();
|
||||
agent.setId(agentId);
|
||||
agent.setTenantId(BigInteger.valueOf(2001));
|
||||
AgentService agentService = Mockito.mock(AgentService.class);
|
||||
Mockito.when(agentService.getById(agentId)).thenReturn(agent);
|
||||
AgentRunService service = new AgentRunService();
|
||||
setField(service, "agentService", agentService);
|
||||
AgentChatRequest request = new AgentChatRequest();
|
||||
request.setAgentId(agentId);
|
||||
request.setPrompt("hello");
|
||||
LoginAccount account = new LoginAccount();
|
||||
account.setId(BigInteger.valueOf(3001));
|
||||
account.setTenantId(BigInteger.valueOf(2002));
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> service.chatPublic(request, account)
|
||||
);
|
||||
|
||||
Assert.assertEquals("Agent 不存在或不可用", exception.getMessage());
|
||||
Mockito.verify(agentService, Mockito.never()).getPublishedView(agentId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 公共 API 没有恢复入口时必须拒绝包含 HITL 工具的 Agent。
|
||||
*
|
||||
* @throws Exception 注入测试依赖失败
|
||||
*/
|
||||
@Test
|
||||
public void chatPublicShouldRejectHitlTool() throws Exception {
|
||||
BigInteger agentId = BigInteger.valueOf(1001);
|
||||
BigInteger tenantId = BigInteger.valueOf(2001);
|
||||
Agent liveAgent = new Agent();
|
||||
liveAgent.setId(agentId);
|
||||
liveAgent.setTenantId(tenantId);
|
||||
liveAgent.setStatus(1);
|
||||
liveAgent.setPublishStatus(PublishStatus.PUBLISHED.getCode());
|
||||
AgentToolBinding binding = new AgentToolBinding();
|
||||
binding.setEnabled(true);
|
||||
binding.setHitlEnabled(true);
|
||||
Agent publishedAgent = new Agent();
|
||||
publishedAgent.setId(agentId);
|
||||
publishedAgent.setToolBindings(List.of(binding));
|
||||
AgentService agentService = Mockito.mock(AgentService.class);
|
||||
Mockito.when(agentService.getById(agentId)).thenReturn(liveAgent);
|
||||
Mockito.when(agentService.getPublishedView(agentId)).thenReturn(publishedAgent);
|
||||
AgentRunService service = new AgentRunService();
|
||||
setField(service, "agentService", agentService);
|
||||
setField(service, "chatSessionQueryService", Mockito.mock(ChatSessionQueryService.class));
|
||||
AgentChatRequest request = new AgentChatRequest();
|
||||
request.setAgentId(agentId);
|
||||
request.setPrompt("hello");
|
||||
LoginAccount account = new LoginAccount();
|
||||
account.setId(BigInteger.valueOf(3001));
|
||||
account.setTenantId(tenantId);
|
||||
|
||||
BusinessException exception = Assert.assertThrows(
|
||||
BusinessException.class,
|
||||
() -> service.chatPublic(request, account)
|
||||
);
|
||||
|
||||
Assert.assertEquals("公共 Agent API 暂不支持需要执行确认的工具", exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 写入被测对象私有字段。
|
||||
*
|
||||
* @param target 被测对象
|
||||
* @param fieldName 字段名
|
||||
* @param value 字段值
|
||||
* @throws Exception 字段不存在或不可访问
|
||||
*/
|
||||
private void setField(Object target, String fieldName, Object value) throws Exception {
|
||||
Field field = target.getClass().getDeclaredField(fieldName);
|
||||
field.setAccessible(true);
|
||||
field.set(target, value);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package tech.easyflow.agent.runtime;
|
||||
|
||||
import com.mybatisflex.core.query.QueryWrapper;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.agent.entity.Agent;
|
||||
import tech.easyflow.agent.service.AgentService;
|
||||
import tech.easyflow.ai.enums.PublishStatus;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
|
||||
import java.math.BigInteger;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* {@link AgentRunStartGuard} 单元测试。
|
||||
*/
|
||||
public class AgentRunStartGuardTest {
|
||||
|
||||
/**
|
||||
* 已启用且已发布的 Agent 允许启动正式运行。
|
||||
*/
|
||||
@Test
|
||||
public void publishedAgentShouldBeRunnable() {
|
||||
AgentService agentService = mock(AgentService.class);
|
||||
Agent agent = agent(1, PublishStatus.PUBLISHED);
|
||||
when(agentService.getOne(any(QueryWrapper.class))).thenReturn(agent);
|
||||
|
||||
new AgentRunStartGuard(agentService).assertRunnable(BigInteger.valueOf(1001));
|
||||
}
|
||||
|
||||
/**
|
||||
* 已下线 Agent 必须拒绝新运行。
|
||||
*/
|
||||
@Test(expected = BusinessException.class)
|
||||
public void offlineAgentShouldBeRejected() {
|
||||
AgentService agentService = mock(AgentService.class);
|
||||
when(agentService.getOne(any(QueryWrapper.class)))
|
||||
.thenReturn(agent(1, PublishStatus.OFFLINE));
|
||||
|
||||
new AgentRunStartGuard(agentService).assertRunnable(BigInteger.valueOf(1001));
|
||||
}
|
||||
|
||||
/**
|
||||
* 已禁用 Agent 必须拒绝新运行。
|
||||
*/
|
||||
@Test(expected = BusinessException.class)
|
||||
public void disabledAgentShouldBeRejected() {
|
||||
AgentService agentService = mock(AgentService.class);
|
||||
when(agentService.getOne(any(QueryWrapper.class)))
|
||||
.thenReturn(agent(0, PublishStatus.PUBLISHED));
|
||||
|
||||
new AgentRunStartGuard(agentService).assertRunnable(BigInteger.valueOf(1001));
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建测试 Agent。
|
||||
*
|
||||
* @param status 启用状态
|
||||
* @param publishStatus 发布状态
|
||||
* @return 测试 Agent
|
||||
*/
|
||||
private static Agent agent(Integer status, PublishStatus publishStatus) {
|
||||
Agent agent = new Agent();
|
||||
agent.setId(BigInteger.valueOf(1001));
|
||||
agent.setStatus(status);
|
||||
agent.setPublishStatus(publishStatus.getCode());
|
||||
return agent;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
package tech.easyflow.agent.service.impl;
|
||||
|
||||
import com.mybatisflex.core.query.QueryWrapper;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mockito;
|
||||
import tech.easyflow.agent.entity.Agent;
|
||||
import tech.easyflow.agent.service.AgentCategoryService;
|
||||
import tech.easyflow.agent.service.AgentDependencyAccessService;
|
||||
import tech.easyflow.ai.entity.DocumentCollection;
|
||||
import tech.easyflow.ai.entity.Mcp;
|
||||
import tech.easyflow.ai.entity.Plugin;
|
||||
import tech.easyflow.ai.entity.PluginItem;
|
||||
import tech.easyflow.ai.entity.Workflow;
|
||||
import tech.easyflow.ai.enums.PublishStatus;
|
||||
import tech.easyflow.ai.mapper.PluginMapper;
|
||||
import tech.easyflow.ai.service.DocumentCollectionService;
|
||||
import tech.easyflow.ai.service.McpService;
|
||||
import tech.easyflow.ai.service.ModelService;
|
||||
import tech.easyflow.ai.service.PluginItemService;
|
||||
import tech.easyflow.ai.service.PluginVisibilityService;
|
||||
import tech.easyflow.ai.service.WorkflowService;
|
||||
import tech.easyflow.system.service.CategoryPermissionService;
|
||||
import tech.easyflow.system.service.ResourceAccessService;
|
||||
|
||||
import java.math.BigInteger;
|
||||
import java.util.Locale;
|
||||
|
||||
/**
|
||||
* Agent 绑定资源状态锁测试。
|
||||
*/
|
||||
public class AgentBindingValidationLockTest {
|
||||
|
||||
/**
|
||||
* 验证工作流绑定使用锁定读校验最新发布状态。
|
||||
*
|
||||
* @throws Exception 反射调用失败
|
||||
*/
|
||||
@Test
|
||||
public void workflowBindingShouldValidateWithForUpdate() {
|
||||
Workflow workflow = new Workflow();
|
||||
workflow.setPublishStatus(PublishStatus.PUBLISHED.getCode());
|
||||
workflow.setTenantId(BigInteger.ONE);
|
||||
WorkflowService workflowService = Mockito.mock(WorkflowService.class);
|
||||
Mockito.when(workflowService.getOne(Mockito.any(QueryWrapper.class))).thenReturn(workflow);
|
||||
AgentDependencyAccessService service = createService(
|
||||
workflowService,
|
||||
Mockito.mock(PluginItemService.class),
|
||||
Mockito.mock(PluginMapper.class),
|
||||
Mockito.mock(PluginVisibilityService.class),
|
||||
Mockito.mock(McpService.class),
|
||||
Mockito.mock(DocumentCollectionService.class),
|
||||
Mockito.mock(ResourceAccessService.class)
|
||||
);
|
||||
|
||||
service.requireWorkflow(agent(), BigInteger.valueOf(1001));
|
||||
|
||||
ArgumentCaptor<QueryWrapper> queryCaptor = ArgumentCaptor.forClass(QueryWrapper.class);
|
||||
Mockito.verify(workflowService).getOne(queryCaptor.capture());
|
||||
Assert.assertTrue(queryCaptor.getValue().toSQL().toUpperCase(Locale.ROOT).contains("FOR UPDATE"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证知识库绑定使用锁定读校验最新发布状态。
|
||||
*
|
||||
* @throws Exception 反射调用失败
|
||||
*/
|
||||
@Test
|
||||
public void knowledgeBindingShouldValidateWithForUpdate() {
|
||||
DocumentCollection knowledge = new DocumentCollection();
|
||||
knowledge.setPublishStatus(PublishStatus.PUBLISHED.getCode());
|
||||
knowledge.setTenantId(BigInteger.ONE);
|
||||
DocumentCollectionService knowledgeService = Mockito.mock(DocumentCollectionService.class);
|
||||
Mockito.when(knowledgeService.getOne(Mockito.any(QueryWrapper.class))).thenReturn(knowledge);
|
||||
AgentDependencyAccessService service = createService(
|
||||
Mockito.mock(WorkflowService.class),
|
||||
Mockito.mock(PluginItemService.class),
|
||||
Mockito.mock(PluginMapper.class),
|
||||
Mockito.mock(PluginVisibilityService.class),
|
||||
Mockito.mock(McpService.class),
|
||||
knowledgeService,
|
||||
Mockito.mock(ResourceAccessService.class)
|
||||
);
|
||||
|
||||
service.requireKnowledge(agent(), BigInteger.valueOf(2001));
|
||||
|
||||
ArgumentCaptor<QueryWrapper> queryCaptor = ArgumentCaptor.forClass(QueryWrapper.class);
|
||||
Mockito.verify(knowledgeService).getOne(queryCaptor.capture());
|
||||
Assert.assertTrue(queryCaptor.getValue().toSQL().toUpperCase(Locale.ROOT).contains("FOR UPDATE"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证插件绑定使用锁定读校验最新启用状态。
|
||||
*
|
||||
* @throws Exception 反射调用失败
|
||||
*/
|
||||
@Test
|
||||
public void pluginBindingShouldValidateWithForUpdate() {
|
||||
BigInteger pluginId = BigInteger.valueOf(30);
|
||||
BigInteger pluginItemId = BigInteger.valueOf(3001);
|
||||
PluginItem pluginItem = pluginItem(pluginId);
|
||||
PluginItemService pluginItemService = Mockito.mock(PluginItemService.class);
|
||||
Mockito.when(pluginItemService.getById(pluginItemId)).thenReturn(pluginItem);
|
||||
Mockito.when(pluginItemService.getOne(Mockito.any(QueryWrapper.class))).thenReturn(pluginItem);
|
||||
Plugin plugin = new Plugin();
|
||||
plugin.setId(pluginId);
|
||||
plugin.setTenantId(1L);
|
||||
PluginMapper pluginMapper = Mockito.mock(PluginMapper.class);
|
||||
Mockito.when(pluginMapper.selectOneByQuery(Mockito.any(QueryWrapper.class))).thenReturn(plugin);
|
||||
AgentDependencyAccessService service = createService(
|
||||
Mockito.mock(WorkflowService.class),
|
||||
pluginItemService,
|
||||
pluginMapper,
|
||||
Mockito.mock(PluginVisibilityService.class),
|
||||
Mockito.mock(McpService.class),
|
||||
Mockito.mock(DocumentCollectionService.class),
|
||||
Mockito.mock(ResourceAccessService.class)
|
||||
);
|
||||
|
||||
service.requirePluginItem(agent(), pluginItemId);
|
||||
|
||||
ArgumentCaptor<QueryWrapper> pluginQueryCaptor = ArgumentCaptor.forClass(QueryWrapper.class);
|
||||
Mockito.verify(pluginMapper).selectOneByQuery(pluginQueryCaptor.capture());
|
||||
Assert.assertTrue(pluginQueryCaptor.getValue().toSQL().toUpperCase(Locale.ROOT).contains("FOR UPDATE"));
|
||||
ArgumentCaptor<QueryWrapper> itemQueryCaptor = ArgumentCaptor.forClass(QueryWrapper.class);
|
||||
Mockito.verify(pluginItemService).getOne(itemQueryCaptor.capture());
|
||||
Assert.assertTrue(itemQueryCaptor.getValue().toSQL().toUpperCase(Locale.ROOT).contains("FOR UPDATE"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 MCP 绑定使用锁定读校验最新启用状态。
|
||||
*
|
||||
* @throws Exception 反射调用失败
|
||||
*/
|
||||
@Test
|
||||
public void mcpBindingShouldValidateWithForUpdate() {
|
||||
Mcp mcp = new Mcp();
|
||||
mcp.setStatus(true);
|
||||
mcp.setTenantId(BigInteger.ONE);
|
||||
McpService mcpService = Mockito.mock(McpService.class);
|
||||
Mockito.when(mcpService.getOne(Mockito.any(QueryWrapper.class))).thenReturn(mcp);
|
||||
AgentDependencyAccessService service = createService(
|
||||
Mockito.mock(WorkflowService.class),
|
||||
Mockito.mock(PluginItemService.class),
|
||||
Mockito.mock(PluginMapper.class),
|
||||
Mockito.mock(PluginVisibilityService.class),
|
||||
mcpService,
|
||||
Mockito.mock(DocumentCollectionService.class),
|
||||
Mockito.mock(ResourceAccessService.class)
|
||||
);
|
||||
|
||||
service.requireMcp(agent(), BigInteger.valueOf(4001));
|
||||
|
||||
ArgumentCaptor<QueryWrapper> queryCaptor = ArgumentCaptor.forClass(QueryWrapper.class);
|
||||
Mockito.verify(mcpService).getOne(queryCaptor.capture());
|
||||
Assert.assertTrue(queryCaptor.getValue().toSQL().toUpperCase(Locale.ROOT).contains("FOR UPDATE"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建依赖资源校验服务。
|
||||
*
|
||||
* @param workflowService 工作流服务
|
||||
* @param pluginItemService 插件工具服务
|
||||
* @param pluginMapper 插件 Mapper
|
||||
* @param pluginVisibilityService 插件可见性服务
|
||||
* @param mcpService MCP 服务
|
||||
* @param documentCollectionService 知识库服务
|
||||
* @param resourceAccessService 资源权限服务
|
||||
* @return 依赖资源校验服务
|
||||
*/
|
||||
private AgentDependencyAccessService createService(
|
||||
WorkflowService workflowService,
|
||||
PluginItemService pluginItemService,
|
||||
PluginMapper pluginMapper,
|
||||
PluginVisibilityService pluginVisibilityService,
|
||||
McpService mcpService,
|
||||
DocumentCollectionService documentCollectionService,
|
||||
ResourceAccessService resourceAccessService) {
|
||||
return new AgentDependencyAccessService(
|
||||
Mockito.mock(ModelService.class),
|
||||
workflowService,
|
||||
pluginItemService,
|
||||
pluginMapper,
|
||||
pluginVisibilityService,
|
||||
mcpService,
|
||||
documentCollectionService,
|
||||
Mockito.mock(AgentCategoryService.class),
|
||||
Mockito.mock(CategoryPermissionService.class),
|
||||
resourceAccessService
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建同租户 Agent。
|
||||
*
|
||||
* @return Agent
|
||||
*/
|
||||
private Agent agent() {
|
||||
Agent agent = new Agent();
|
||||
agent.setTenantId(BigInteger.ONE);
|
||||
return agent;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建启用的插件工具。
|
||||
*
|
||||
* @param pluginId 插件 ID
|
||||
* @return 插件工具
|
||||
*/
|
||||
private PluginItem pluginItem(BigInteger pluginId) {
|
||||
PluginItem pluginItem = new PluginItem();
|
||||
pluginItem.setPluginId(pluginId);
|
||||
pluginItem.setStatus(1);
|
||||
return pluginItem;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
package tech.easyflow.agent.service.impl;
|
||||
|
||||
import com.mybatisflex.core.query.QueryWrapper;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
import tech.easyflow.agent.entity.Agent;
|
||||
import tech.easyflow.agent.entity.AgentToolBinding;
|
||||
import tech.easyflow.agent.enums.AgentToolType;
|
||||
import tech.easyflow.agent.service.AgentKnowledgeBindingService;
|
||||
import tech.easyflow.agent.service.AgentService;
|
||||
import tech.easyflow.agent.service.AgentToolBindingService;
|
||||
import tech.easyflow.agent.support.AgentBindingLockExecutor;
|
||||
|
||||
import java.math.BigInteger;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Map;
|
||||
import java.util.List;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* {@link AgentResourceBindingProviderImpl} 批量解绑锁顺序测试。
|
||||
*/
|
||||
public class AgentResourceBindingProviderImplTest {
|
||||
|
||||
/**
|
||||
* 验证工作流批量解绑按 Agent ID 升序获取绑定锁。
|
||||
*/
|
||||
@Test
|
||||
public void unbindWorkflowShouldAcquireAgentLocksInAscendingOrder() {
|
||||
AgentService agentService = Mockito.mock(AgentService.class);
|
||||
AgentToolBindingService toolBindingService = Mockito.mock(AgentToolBindingService.class);
|
||||
AgentKnowledgeBindingService knowledgeBindingService =
|
||||
Mockito.mock(AgentKnowledgeBindingService.class);
|
||||
AgentBindingLockExecutor lockExecutor = Mockito.mock(AgentBindingLockExecutor.class);
|
||||
Mockito.when(toolBindingService.list(Mockito.any(QueryWrapper.class)))
|
||||
.thenReturn(List.of(
|
||||
workflowBinding(3),
|
||||
workflowBinding(1),
|
||||
workflowBinding(2),
|
||||
workflowBinding(1)
|
||||
));
|
||||
List<BigInteger> lockOrder = new ArrayList<>();
|
||||
Mockito.doAnswer(invocation -> {
|
||||
lockOrder.add(invocation.getArgument(0));
|
||||
Supplier<?> task = invocation.getArgument(1);
|
||||
return task.get();
|
||||
}).when(lockExecutor).execute(Mockito.any(BigInteger.class), Mockito.any());
|
||||
AgentResourceBindingProviderImpl provider = new AgentResourceBindingProviderImpl(
|
||||
agentService,
|
||||
toolBindingService,
|
||||
knowledgeBindingService,
|
||||
lockExecutor
|
||||
);
|
||||
|
||||
provider.unbindWorkflow(BigInteger.TEN);
|
||||
|
||||
Assert.assertEquals(
|
||||
List.of(BigInteger.ONE, BigInteger.TWO, BigInteger.valueOf(3)),
|
||||
lockOrder
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 已发布快照中的引用必须参与资源删除影响检查。
|
||||
*/
|
||||
@Test
|
||||
public void listAgentsByWorkflowIdShouldIncludeSnapshotOnlyReference() {
|
||||
AgentService agentService = Mockito.mock(AgentService.class);
|
||||
AgentToolBindingService toolBindingService = Mockito.mock(AgentToolBindingService.class);
|
||||
AgentKnowledgeBindingService knowledgeBindingService =
|
||||
Mockito.mock(AgentKnowledgeBindingService.class);
|
||||
AgentBindingLockExecutor lockExecutor = Mockito.mock(AgentBindingLockExecutor.class);
|
||||
BigInteger agentId = BigInteger.valueOf(7);
|
||||
Agent agent = new Agent();
|
||||
agent.setId(agentId);
|
||||
agent.setName("已发布智能体");
|
||||
agent.setPublishedSnapshotJson(Map.of(
|
||||
"toolBindings",
|
||||
List.of(Map.of(
|
||||
"toolType", AgentToolType.WORKFLOW.name(),
|
||||
"targetId", BigInteger.TEN
|
||||
))
|
||||
));
|
||||
Mockito.when(toolBindingService.list(Mockito.any(QueryWrapper.class)))
|
||||
.thenReturn(List.of());
|
||||
Mockito.when(agentService.list(Mockito.any(QueryWrapper.class)))
|
||||
.thenReturn(List.of(agent));
|
||||
Mockito.when(agentService.listByIds(Mockito.anyCollection()))
|
||||
.thenReturn(List.of(agent));
|
||||
AgentResourceBindingProviderImpl provider = new AgentResourceBindingProviderImpl(
|
||||
agentService,
|
||||
toolBindingService,
|
||||
knowledgeBindingService,
|
||||
lockExecutor
|
||||
);
|
||||
|
||||
var result = provider.listAgentsByWorkflowId(BigInteger.TEN);
|
||||
|
||||
Assert.assertEquals(1, result.size());
|
||||
Assert.assertEquals(agentId, result.get(0).getId());
|
||||
Assert.assertEquals("已发布智能体", result.get(0).getTitle());
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造工作流工具绑定。
|
||||
*
|
||||
* @param agentId Agent ID
|
||||
* @return 工具绑定
|
||||
*/
|
||||
private static AgentToolBinding workflowBinding(long agentId) {
|
||||
AgentToolBinding binding = new AgentToolBinding();
|
||||
binding.setAgentId(BigInteger.valueOf(agentId));
|
||||
binding.setToolType(AgentToolType.WORKFLOW.name());
|
||||
binding.setTargetId(BigInteger.TEN);
|
||||
return binding;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
package tech.easyflow.agent.support;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.transaction.support.TransactionSynchronization;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import tech.easyflow.common.cache.RedisLockExecutor;
|
||||
|
||||
import java.math.BigInteger;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* {@link AgentBindingLockExecutor} 事务锁生命周期测试。
|
||||
*/
|
||||
public class AgentBindingLockExecutorTest {
|
||||
|
||||
/**
|
||||
* 验证活动事务内的绑定锁延迟到事务完成后释放。
|
||||
*/
|
||||
@Test
|
||||
public void executeShouldReleaseLockAfterTransactionCompletion() {
|
||||
RedisLockExecutor redisLockExecutor = Mockito.mock(RedisLockExecutor.class);
|
||||
RedisLockExecutor.LockHandle lockHandle = Mockito.mock(RedisLockExecutor.LockHandle.class);
|
||||
ScheduledExecutorService renewExecutor = Mockito.mock(ScheduledExecutorService.class);
|
||||
ScheduledFuture<?> renewTask = Mockito.mock(ScheduledFuture.class);
|
||||
Mockito.when(redisLockExecutor.acquire(
|
||||
Mockito.anyString(),
|
||||
Mockito.any(Duration.class),
|
||||
Mockito.any(Duration.class)
|
||||
)).thenReturn(lockHandle);
|
||||
Mockito.doReturn(renewTask).when(renewExecutor).scheduleWithFixedDelay(
|
||||
Mockito.any(Runnable.class),
|
||||
Mockito.anyLong(),
|
||||
Mockito.anyLong(),
|
||||
Mockito.eq(TimeUnit.MILLISECONDS)
|
||||
);
|
||||
Mockito.when(lockHandle.renew()).thenReturn(true);
|
||||
AgentBindingLockExecutor executor = new AgentBindingLockExecutor(redisLockExecutor, renewExecutor);
|
||||
TransactionSynchronizationManager.initSynchronization();
|
||||
TransactionSynchronizationManager.setActualTransactionActive(true);
|
||||
try {
|
||||
String result = executor.execute(BigInteger.ONE, () -> "ok");
|
||||
|
||||
Assert.assertEquals("ok", result);
|
||||
Mockito.verify(lockHandle, Mockito.never()).release();
|
||||
Mockito.verify(renewTask, Mockito.never()).cancel(false);
|
||||
ArgumentCaptor<Runnable> renewCaptor = ArgumentCaptor.forClass(Runnable.class);
|
||||
Mockito.verify(renewExecutor).scheduleWithFixedDelay(
|
||||
renewCaptor.capture(),
|
||||
Mockito.anyLong(),
|
||||
Mockito.anyLong(),
|
||||
Mockito.eq(TimeUnit.MILLISECONDS)
|
||||
);
|
||||
renewCaptor.getValue().run();
|
||||
Mockito.verify(lockHandle).renew();
|
||||
List<TransactionSynchronization> synchronizations =
|
||||
TransactionSynchronizationManager.getSynchronizations();
|
||||
Assert.assertEquals(1, synchronizations.size());
|
||||
synchronizations.get(0).afterCompletion(TransactionSynchronization.STATUS_COMMITTED);
|
||||
Mockito.verify(renewTask).cancel(false);
|
||||
Mockito.verify(lockHandle).release();
|
||||
} finally {
|
||||
TransactionSynchronizationManager.setActualTransactionActive(false);
|
||||
TransactionSynchronizationManager.clearSynchronization();
|
||||
executor.destroy();
|
||||
}
|
||||
Mockito.verify(renewExecutor).shutdownNow();
|
||||
}
|
||||
|
||||
/**
|
||||
* 同一事务内重复进入相同 Agent 锁时只能获取一次 Redis 锁。
|
||||
*/
|
||||
@Test
|
||||
public void executeShouldReuseSameAgentLockWithinTransaction() {
|
||||
RedisLockExecutor redisLockExecutor = Mockito.mock(RedisLockExecutor.class);
|
||||
RedisLockExecutor.LockHandle lockHandle = Mockito.mock(RedisLockExecutor.LockHandle.class);
|
||||
ScheduledExecutorService renewExecutor = Mockito.mock(ScheduledExecutorService.class);
|
||||
ScheduledFuture<?> renewTask = Mockito.mock(ScheduledFuture.class);
|
||||
Mockito.when(redisLockExecutor.acquire(
|
||||
Mockito.anyString(),
|
||||
Mockito.any(Duration.class),
|
||||
Mockito.any(Duration.class)
|
||||
)).thenReturn(lockHandle);
|
||||
Mockito.doReturn(renewTask).when(renewExecutor).scheduleWithFixedDelay(
|
||||
Mockito.any(Runnable.class),
|
||||
Mockito.anyLong(),
|
||||
Mockito.anyLong(),
|
||||
Mockito.eq(TimeUnit.MILLISECONDS)
|
||||
);
|
||||
AgentBindingLockExecutor executor = new AgentBindingLockExecutor(redisLockExecutor, renewExecutor);
|
||||
TransactionSynchronizationManager.initSynchronization();
|
||||
TransactionSynchronizationManager.setActualTransactionActive(true);
|
||||
try {
|
||||
String result = executor.execute(
|
||||
BigInteger.ONE,
|
||||
() -> executor.execute(BigInteger.ONE, () -> "nested")
|
||||
);
|
||||
|
||||
Assert.assertEquals("nested", result);
|
||||
Mockito.verify(redisLockExecutor, Mockito.times(1)).acquire(
|
||||
Mockito.anyString(),
|
||||
Mockito.any(Duration.class),
|
||||
Mockito.any(Duration.class)
|
||||
);
|
||||
List<TransactionSynchronization> synchronizations =
|
||||
TransactionSynchronizationManager.getSynchronizations();
|
||||
Assert.assertEquals(1, synchronizations.size());
|
||||
synchronizations.get(0).afterCompletion(TransactionSynchronization.STATUS_COMMITTED);
|
||||
} finally {
|
||||
TransactionSynchronizationManager.unbindResourceIfPossible(
|
||||
"easyflow:lock:agent:binding:" + BigInteger.ONE
|
||||
);
|
||||
TransactionSynchronizationManager.setActualTransactionActive(false);
|
||||
TransactionSynchronizationManager.clearSynchronization();
|
||||
executor.destroy();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 锁续期失败后事务提交必须被阻止。
|
||||
*/
|
||||
@Test
|
||||
public void renewalFailureShouldPreventTransactionCommit() {
|
||||
RedisLockExecutor redisLockExecutor = Mockito.mock(RedisLockExecutor.class);
|
||||
RedisLockExecutor.LockHandle lockHandle = Mockito.mock(RedisLockExecutor.LockHandle.class);
|
||||
ScheduledExecutorService renewExecutor = Mockito.mock(ScheduledExecutorService.class);
|
||||
ScheduledFuture<?> renewTask = Mockito.mock(ScheduledFuture.class);
|
||||
Mockito.when(redisLockExecutor.acquire(
|
||||
Mockito.anyString(),
|
||||
Mockito.any(Duration.class),
|
||||
Mockito.any(Duration.class)
|
||||
)).thenReturn(lockHandle);
|
||||
ArgumentCaptor<Runnable> renewCaptor = ArgumentCaptor.forClass(Runnable.class);
|
||||
Mockito.doReturn(renewTask).when(renewExecutor).scheduleWithFixedDelay(
|
||||
renewCaptor.capture(),
|
||||
Mockito.anyLong(),
|
||||
Mockito.anyLong(),
|
||||
Mockito.eq(TimeUnit.MILLISECONDS)
|
||||
);
|
||||
Mockito.when(lockHandle.renew()).thenReturn(false);
|
||||
AgentBindingLockExecutor executor = new AgentBindingLockExecutor(redisLockExecutor, renewExecutor);
|
||||
TransactionSynchronizationManager.initSynchronization();
|
||||
TransactionSynchronizationManager.setActualTransactionActive(true);
|
||||
try {
|
||||
executor.execute(BigInteger.ONE, () -> "ok");
|
||||
renewCaptor.getValue().run();
|
||||
TransactionSynchronization synchronization =
|
||||
TransactionSynchronizationManager.getSynchronizations().get(0);
|
||||
|
||||
IllegalStateException exception = Assert.assertThrows(
|
||||
IllegalStateException.class,
|
||||
() -> synchronization.beforeCommit(false)
|
||||
);
|
||||
|
||||
Assert.assertEquals("Agent 绑定锁已失效，事务禁止提交", exception.getMessage());
|
||||
synchronization.afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
|
||||
} finally {
|
||||
TransactionSynchronizationManager.unbindResourceIfPossible(
|
||||
"easyflow:lock:agent:binding:" + BigInteger.ONE
|
||||
);
|
||||
TransactionSynchronizationManager.setActualTransactionActive(false);
|
||||
TransactionSynchronizationManager.clearSynchronization();
|
||||
executor.destroy();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user