From ba17e34b84c04ad91c75334ad97fcedfbf983b3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sun, 30 Aug 2026 22:23:33 +0800 Subject: [PATCH] =?UTF-8?q?task-79:=20=E4=B8=BA=E5=AF=B9=E8=B1=A1=E5=AD=98?= =?UTF-8?q?=E5=82=A8=E3=80=81=E6=95=B0=E6=8D=AE=E5=BA=93=E5=92=8C=E9=98=9F?= =?UTF-8?q?=E5=88=97=E5=A2=9E=E5=8A=A0=E6=95=85=E9=9A=9C=E6=B3=A8=E5=85=A5?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/object/FaultInjectionTest.java | 263 ++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/FaultInjectionTest.java diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/FaultInjectionTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/FaultInjectionTest.java new file mode 100644 index 00000000..ac085cf3 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/FaultInjectionTest.java @@ -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.emptyProvider(), + FaultInjectionTest.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.emptyProvider(), + FaultInjectionTest.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.emptyProvider(), + FaultInjectionTest.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.emptyProvider(), + FaultInjectionTest.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.emptyProvider(), + FaultInjectionTest.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.emptyProvider(), + FaultInjectionTest.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 ObjectProvider objectProvider(T value) { + ObjectProvider provider = mock(ObjectProvider.class); + when(provider.getIfAvailable()).thenReturn(value); + return provider; + } + + @SuppressWarnings("unchecked") + private static ObjectProvider emptyProvider() { + ObjectProvider provider = mock(ObjectProvider.class); + when(provider.getIfAvailable()).thenReturn(null); + return provider; + } +}