package com.baiyue.audit.mq; import com.baiyue.audit.repository.AuditDeadLetterRepository; import com.baiyue.audit.service.AuditDeadLetterCompensator; import com.baiyue.audit.service.AuditIngestService; import com.fasterxml.jackson.databind.ObjectMapper; import jakarta.annotation.Resource; import org.junit.jupiter.api.Test; import org.springframework.boot.test.mock.mockito.MockBean; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @SpringJUnitConfig(classes = { AuditMqReliabilityTest.Config.class, AuditRocketMqConsumer.class, AuditDeadLetterConsumer.class, AuditDeadLetterCompensator.class }) @TestPropertySource(properties = { "baiyue.audit.rocketmq.consumer-enabled=true", "baiyue.audit.rocketmq.dead-letter-consumer-enabled=true", "baiyue.audit.rocketmq.dead-letter-compensation-enabled=true", "baiyue.audit.rocketmq.topic=test-audit", "baiyue.audit.rocketmq.consumer-group=test-audit-consumer", "baiyue.audit.rocketmq.dead-letter-topic=%DLQ%test-audit-consumer", "baiyue.audit.rocketmq.dead-letter-group=test-audit-dlq", "spring.application.name=audit-service-test" }) class AuditMqReliabilityTest { private static final String PAYLOAD = """ {"eventId":"event-1","eventType":"ROLE_UPDATE","resourceType":"ROLE", "durationMs":10,"success":true,"occurredAt":"2026-08-05T00:00:00Z"} """; @MockBean private AuditIngestService service; @MockBean private AuditDeadLetterRepository repository; @Resource private AuditRocketMqConsumer consumer; @Resource private AuditDeadLetterConsumer deadLetterConsumer; @Resource private AuditDeadLetterCompensator compensator; @Test void consumesAuditEvent() { consumer.onMessage(PAYLOAD); verify(service).ingest(any()); } @Test void rejectsInvalidPayload() { assertThatThrownBy(() -> consumer.onMessage("not-json")) .isInstanceOf(IllegalArgumentException.class); } @Test void storesDeadLetter() { deadLetterConsumer.onMessage(PAYLOAD); verify(repository).save("event-1", PAYLOAD); } @Test void compensatesDeadLetter() { when(repository.pendingIds(50)).thenReturn(java.util.List.of(7L)); when(repository.claim(eq(7L), anyString(), eq(60))).thenReturn(true); when(repository.find(7L)).thenReturn( new AuditDeadLetterRepository.DeadRow(7L, "event-1", PAYLOAD, 1)); compensator.compensate(); verify(service).ingest(any()); verify(repository).processed(eq(7L), anyString()); } @Test void retriesFailedCompensation() { when(repository.pendingIds(50)).thenReturn(java.util.List.of(8L)); when(repository.claim(eq(8L), anyString(), eq(60))).thenReturn(true); when(repository.find(8L)).thenReturn( new AuditDeadLetterRepository.DeadRow(8L, "event-2", "not-json", 1)); compensator.compensate(); verify(repository).failed(eq(8L), anyString(), eq(5), anyString()); } @Configuration static class Config { @Bean ObjectMapper objectMapper() { return new ObjectMapper().findAndRegisterModules(); } } }