task-79: 为对象存储、数据库和队列增加故障注入测试
This commit is contained in:
+263
@@ -0,0 +1,263 @@
|
||||
package com.nanri.aiimage.modules.file.service.object;
|
||||
|
||||
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||
import com.nanri.aiimage.config.CozeTaskQueueGate;
|
||||
import com.nanri.aiimage.config.TransientStorageProperties;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskFileJobMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskFileJobDispatchEvent;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
|
||||
import io.minio.GetObjectResponse;
|
||||
import io.minio.MinioClient;
|
||||
import io.minio.PutObjectArgs;
|
||||
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.core.task.TaskRejectedException;
|
||||
|
||||
import java.lang.reflect.Field;
|
||||
import java.util.concurrent.Semaphore;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* Task 79：为对象存储（RustFS/MinIO）、数据库（任务文件作业表）和队列
|
||||
* （Coze 执行队列）增加故障注入测试。全部通过 mock 依赖注入故障，验证
|
||||
* 错误可恢复、重试有界、信号量/等待槽释放、非法输入不创建无效资源。
|
||||
*/
|
||||
class FaultInjectionTest {
|
||||
|
||||
@BeforeAll
|
||||
static void initializeTableInfo() {
|
||||
TableInfoHelper.initTableInfo(
|
||||
new MapperBuilderAssistant(new MybatisConfiguration(), ""),
|
||||
TaskFileJobEntity.class);
|
||||
}
|
||||
|
||||
// ---------- 1. 对象存储默认路径：一次成功写入占用并释放上传与总预算信号量 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_normal_default_path() throws Exception {
|
||||
TransientStorageProperties properties = configuredProperties();
|
||||
properties.setMaxConcurrentUploads(2);
|
||||
properties.setMaxTotalConcurrentOperations(2);
|
||||
MinioClient client = mock(MinioClient.class);
|
||||
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||
properties, FaultInjectionTest.<MeterRegistry>emptyProvider(),
|
||||
FaultInjectionTest.<RustfsDeleteRetryService>emptyProvider(), () -> client);
|
||||
|
||||
String key = service.uploadText("task/a.json", "{\"k\":1}", false);
|
||||
|
||||
assertEquals("task/a.json", key);
|
||||
verify(client, times(1)).putObject(any(PutObjectArgs.class));
|
||||
assertEquals(2, semaphorePermits(service, "uploadSemaphore"), "成功路径释放上传信号量");
|
||||
assertEquals(2, semaphorePermits(service, "totalSemaphore"), "成功路径释放总预算信号量");
|
||||
}
|
||||
|
||||
// ---------- 2. 对象存储批量：多对象全部成功，信号量全部释放，顺序稳定 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_normal_multiple_items() throws Exception {
|
||||
TransientStorageProperties properties = configuredProperties();
|
||||
properties.setMaxConcurrentUploads(3);
|
||||
MinioClient client = mock(MinioClient.class);
|
||||
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||
properties, FaultInjectionTest.<MeterRegistry>emptyProvider(),
|
||||
FaultInjectionTest.<RustfsDeleteRetryService>emptyProvider(), () -> client);
|
||||
|
||||
service.uploadText("task/1.json", "1", false);
|
||||
service.uploadText("task/2.json", "2", false);
|
||||
service.uploadText("task/3.json", "3", false);
|
||||
|
||||
verify(client, times(3)).putObject(any(PutObjectArgs.class));
|
||||
assertEquals(3, semaphorePermits(service, "uploadSemaphore"), "批量全部释放上传信号量");
|
||||
}
|
||||
|
||||
// ---------- 3. 对象存储幂等：故障后重试一次成功，不产生重复对象，信号量精确释放 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_normal_repeated_operation_is_idempotent() throws Exception {
|
||||
TransientStorageProperties properties = configuredProperties();
|
||||
properties.setMaxConcurrentUploads(2);
|
||||
properties.setUploadMaxRetries(2);
|
||||
properties.setBaseRetryDelayMillis(0);
|
||||
properties.setRetryJitterMillis(0);
|
||||
MinioClient client = mock(MinioClient.class);
|
||||
doThrow(new IllegalStateException("first put failed"))
|
||||
.doReturn(null)
|
||||
.when(client).putObject(any(PutObjectArgs.class));
|
||||
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||
properties, FaultInjectionTest.<MeterRegistry>emptyProvider(),
|
||||
FaultInjectionTest.<RustfsDeleteRetryService>emptyProvider(), () -> client);
|
||||
|
||||
String key = service.uploadText("task/retry.json", "{}", false);
|
||||
|
||||
assertEquals("task/retry.json", key);
|
||||
verify(client, times(2)).putObject(any(PutObjectArgs.class));
|
||||
assertEquals(2, semaphorePermits(service, "uploadSemaphore"), "重试路径信号量仍精确释放");
|
||||
}
|
||||
|
||||
// ---------- 4. 对象存储空输入：空对象安全跳过，不产生请求 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_boundary_empty_input() throws Exception {
|
||||
TransientStorageProperties properties = configuredProperties();
|
||||
MinioClient client = mock(MinioClient.class);
|
||||
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||
properties, FaultInjectionTest.<MeterRegistry>emptyProvider(),
|
||||
FaultInjectionTest.<RustfsDeleteRetryService>emptyProvider(), () -> client);
|
||||
|
||||
service.deleteObject(null);
|
||||
service.deleteObject(" ");
|
||||
|
||||
verify(client, never()).removeObject(any());
|
||||
}
|
||||
|
||||
// ---------- 5. 对象存储单元素：单对象读取不依赖批量路径，信号量正确 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_boundary_single_item() throws Exception {
|
||||
TransientStorageProperties properties = configuredProperties();
|
||||
properties.setMaxConcurrentReads(1);
|
||||
properties.setReadMaxRetries(1);
|
||||
MinioClient client = mock(MinioClient.class);
|
||||
when(client.getObject(any(io.minio.GetObjectArgs.class)))
|
||||
.thenAnswer(invocation -> {
|
||||
GetObjectResponse response = mock(GetObjectResponse.class);
|
||||
when(response.readAllBytes()).thenReturn("{}".getBytes());
|
||||
return response;
|
||||
});
|
||||
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||
properties, FaultInjectionTest.<MeterRegistry>emptyProvider(),
|
||||
FaultInjectionTest.<RustfsDeleteRetryService>emptyProvider(), () -> client);
|
||||
|
||||
String content = service.readObjectAsString("task/single.json");
|
||||
|
||||
assertEquals("{}", content);
|
||||
assertEquals(1, semaphorePermits(service, "readSemaphore"), "单元素读取释放信号量");
|
||||
}
|
||||
|
||||
// ---------- 6. 对象存储上限/超限：总预算耗尽立即拒绝，不进入重试 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_boundary_limit_and_overflow() throws Exception {
|
||||
TransientStorageProperties properties = configuredProperties();
|
||||
properties.setMaxTotalConcurrentOperations(1);
|
||||
properties.setMaxConcurrentUploads(1);
|
||||
properties.setUploadMaxRetries(3);
|
||||
properties.setAcquirePermitTimeoutMillis(0);
|
||||
MinioClient client = mock(MinioClient.class);
|
||||
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||
properties, FaultInjectionTest.<MeterRegistry>emptyProvider(),
|
||||
FaultInjectionTest.<RustfsDeleteRetryService>emptyProvider(), () -> client);
|
||||
|
||||
service.uploadText("task/one.json", "{}", false);
|
||||
|
||||
// 手动占用总预算许可，模拟并发场景下预算耗尽（串行调用间许可已释放）
|
||||
semaphore(service, "totalSemaphore").tryAcquire();
|
||||
IllegalStateException rejected = assertThrows(IllegalStateException.class,
|
||||
() -> service.uploadText("task/two.json", "{}", false));
|
||||
|
||||
assertTrue(rejected.getMessage().contains("concurrency limit"), "超限必须明确拒绝");
|
||||
verify(client, times(1)).putObject(any(PutObjectArgs.class));
|
||||
}
|
||||
|
||||
// ---------- 7. 非法参数：DB 入队非法参数安全拒绝，队列 null 任务抛可识别异常 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_invalid_input_rejected() {
|
||||
TaskFileJobMapper mapper = mock(TaskFileJobMapper.class);
|
||||
ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class);
|
||||
com.nanri.aiimage.modules.task.service.TaskFileJobService service =
|
||||
new com.nanri.aiimage.modules.task.service.TaskFileJobService(mapper, publisher);
|
||||
|
||||
TaskFileJobEntity rejected = service.enqueueAssembleResult(null, "SIMILAR_ASIN", 23110L, "t-1");
|
||||
TaskFileJobEntity rejectedBlank = service.enqueueAssembleResult(20553L, " ", 23110L, "t-1");
|
||||
|
||||
assertNull(rejected, "非法参数安全拒绝");
|
||||
assertNull(rejectedBlank, "非法参数安全拒绝");
|
||||
verify(mapper, never()).insert(any(TaskFileJobEntity.class));
|
||||
verify(mapper, never()).selectOne(any());
|
||||
verify(publisher, never()).publishEvent(any(TaskFileJobDispatchEvent.class));
|
||||
|
||||
CozeTaskQueueGate gate = new CozeTaskQueueGate(mock(TaskExecutor.class), 1, emptyProvider());
|
||||
IllegalArgumentException ex = assertThrows(IllegalArgumentException.class, () -> gate.execute(null));
|
||||
assertTrue(ex.getMessage().contains("不能为 null"), "null 任务必须抛出可识别错误消息");
|
||||
}
|
||||
|
||||
// ---------- 8. 队列故障注入：执行器拒绝释放等待槽并记录指标，恢复后正常 ----------
|
||||
|
||||
@Test
|
||||
void test_task_079_object_storage_dependency_failure_releases_resources() throws Exception {
|
||||
TaskExecutor delegate = mock(TaskExecutor.class);
|
||||
doThrow(new TaskRejectedException("executor full"))
|
||||
.doAnswer(invocation -> {
|
||||
((Runnable) invocation.getArgument(0)).run();
|
||||
return null;
|
||||
})
|
||||
.when(delegate).execute(any(Runnable.class));
|
||||
SimpleMeterRegistry registry = new SimpleMeterRegistry();
|
||||
CozeTaskQueueGate gate = new CozeTaskQueueGate(delegate, 2, objectProvider(registry));
|
||||
Runnable task = () -> {
|
||||
};
|
||||
|
||||
// 第一次提交被执行器拒绝：等待槽释放、拒绝指标记录
|
||||
assertThrows(TaskRejectedException.class, () -> gate.execute(task));
|
||||
assertEquals(0, gate.waiting(), "执行器拒绝后等待槽必须释放");
|
||||
assertTrue(registry.counter("aiimage.coze-task.submit.rejected.total", "reason", "delegate-rejected").count() > 0);
|
||||
|
||||
// 第二次提交恢复成功：等待槽正常占用并释放
|
||||
gate.execute(task);
|
||||
assertEquals(0, gate.waiting(), "成功执行后等待槽释放");
|
||||
}
|
||||
|
||||
private static TransientStorageProperties configuredProperties() {
|
||||
TransientStorageProperties properties = new TransientStorageProperties();
|
||||
properties.setEndpoint("http://127.0.0.1:9000");
|
||||
properties.setBucket("bucket");
|
||||
properties.setAccessKeyId("ak");
|
||||
properties.setAccessKeySecret("sk");
|
||||
properties.setBaseRetryDelayMillis(0);
|
||||
properties.setRetryJitterMillis(0);
|
||||
return properties;
|
||||
}
|
||||
|
||||
private static int semaphorePermits(RustfsObjectStorageService service, String fieldName) throws Exception {
|
||||
return semaphore(service, fieldName).availablePermits();
|
||||
}
|
||||
|
||||
private static Semaphore semaphore(RustfsObjectStorageService service, String fieldName) throws Exception {
|
||||
Field field = RustfsObjectStorageService.class.getDeclaredField(fieldName);
|
||||
field.setAccessible(true);
|
||||
return (Semaphore) field.get(service);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static <T> ObjectProvider<T> objectProvider(T value) {
|
||||
ObjectProvider<T> provider = mock(ObjectProvider.class);
|
||||
when(provider.getIfAvailable()).thenReturn(value);
|
||||
return provider;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static <T> ObjectProvider<T> emptyProvider() {
|
||||
ObjectProvider<T> provider = mock(ObjectProvider.class);
|
||||
when(provider.getIfAvailable()).thenReturn(null);
|
||||
return provider;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user