Files
baiyue-firstleg/event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventInboxServiceTest.java
T
2026-08-05 16:26:10 +08:00

174 lines
7.7 KiB
Java

package com.baiyue.eventbridge.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
import com.baiyue.common.event.RegionCode;
import com.baiyue.eventbridge.domain.BridgeInbox;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.AfterEach;
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.dao.DuplicateKeyException;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.test.util.ReflectionTestUtils;
import java.time.Instant;
import java.time.LocalDateTime;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@DisplayName("跨区域事件收件箱")
@SpringJUnitConfig(classes = {EventInboxService.class, EventInboxServiceTest.TestConfig.class})
@TestPropertySource(properties = "baiyue.region=OVERSEAS")
class EventInboxServiceTest {
@Resource
private ObjectMapper objectMapper;
@MockBean
private InboxStore store;
@Resource
private EventInboxService service;
@Configuration(proxyBeanMethods = false)
static class TestConfig {
@Bean
ObjectMapper objectMapper() {
return new ObjectMapper();
}
}
@AfterEach
void resetRegion() {
ReflectionTestUtils.setField(service, "localRegion", "OVERSEAS");
}
@Test
void firstEventPersistsAndReturnsReceived() {
EventEnvelope envelope = envelope("sha256:placeholder");
envelope = withHash(envelope);
assertThat(service.receive(envelope).result()).isEqualTo("已接收");
verify(store).insert(any(BridgeInbox.class));
}
@Test
void identicalEventAndDigestDoesNotPersistAgain() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
BridgeInbox existing = existing(envelope, envelope.payloadHash());
when(store.findByEventId(envelope.eventId())).thenReturn(existing);
assertThat(service.receive(envelope).result()).isEqualTo("重复");
}
@Test
void sameEventWithDifferentDigestRejectsOverwrite() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
when(store.findByEventId(envelope.eventId()))
.thenReturn(existing(envelope, "sha256:" + "0".repeat(64)));
assertThatThrownBy(() -> service.receive(envelope))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("拒绝覆盖");
}
@Test
void sameEventAndPayloadWithDifferentTenantRejectsOverwrite() {
EventEnvelope original = withHash(envelope("sha256:placeholder"));
EventEnvelope otherTenant = withTenant(original, "20002");
when(store.findByEventId(otherTenant.eventId()))
.thenReturn(existing(original, original.payloadHash()));
assertThatThrownBy(() -> service.receive(otherTenant))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("信封身份不一致");
}
@Test
void sameEventAndPayloadWithDifferentAggregateVersionRejectsOverwrite() {
EventEnvelope original = withHash(envelope("sha256:placeholder"));
EventEnvelope otherEnvelope = withAggregateVersion(original, 2L);
when(store.findByEventId(otherEnvelope.eventId()))
.thenReturn(existing(original, original.payloadHash()));
assertThatThrownBy(() -> service.receive(otherEnvelope))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("信封身份不一致");
}
@Test
void duplicateInsertAfterRaceIsTreatedAsDuplicate() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
when(store.findByEventId(envelope.eventId()))
.thenReturn(null, existing(envelope, envelope.payloadHash()));
doThrow(new DuplicateKeyException("并发重复")).when(store).insert(any());
assertThat(service.receive(envelope).result()).isEqualTo("重复");
}
@Test
void targetRegionDifferentFromCurrentDeploymentIsRejected() {
ReflectionTestUtils.setField(service, "localRegion", "CN");
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
assertThatThrownBy(() -> service.receive(envelope))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("目标区域");
}
private EventEnvelope envelope(String hash) {
return new EventEnvelope("550e8400-e29b-41d4-a716-446655440000", "OrderApproved", 1,
Instant.parse("2026-07-29T08:30:00Z"), RegionCode.CN, RegionCode.OVERSEAS,
"20001", "ORDER", "90001", 1, null, "trace-1", "flow-1", null,
objectMapper.createObjectNode().put("orderId", "90001"), hash);
}
private EventEnvelope withHash(EventEnvelope source) {
String hash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(source.payload()));
return new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(), source.occurredAt(),
source.sourceRegion(), source.targetRegion(), source.tenantId(), source.aggregateType(),
source.aggregateId(), source.aggregateVersion(), source.handoverVersion(), source.traceId(),
source.correlationId(), source.causationId(), source.payload(), hash);
}
private EventEnvelope withTenant(EventEnvelope source, String tenantId) {
return new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(), source.occurredAt(),
source.sourceRegion(), source.targetRegion(), tenantId, source.aggregateType(),
source.aggregateId(), source.aggregateVersion(), source.handoverVersion(), source.traceId(),
source.correlationId(), source.causationId(), source.payload(), source.payloadHash());
}
private EventEnvelope withAggregateVersion(EventEnvelope source, Long aggregateVersion) {
return new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(), source.occurredAt(),
source.sourceRegion(), source.targetRegion(), source.tenantId(), source.aggregateType(),
source.aggregateId(), aggregateVersion, source.handoverVersion(), source.traceId(),
source.correlationId(), source.causationId(), source.payload(), source.payloadHash());
}
private BridgeInbox existing(EventEnvelope envelope, String hash) {
BridgeInbox inbox = new BridgeInbox();
inbox.setEventId(envelope.eventId());
inbox.setTenantId(envelope.tenantId());
inbox.setEventType(envelope.eventType());
inbox.setEventVersion(envelope.eventVersion());
inbox.setSourceRegion(envelope.sourceRegion().name());
inbox.setTargetRegion(envelope.targetRegion().name());
inbox.setAggregateType(envelope.aggregateType());
inbox.setAggregateId(envelope.aggregateId());
inbox.setAggregateVersion(envelope.aggregateVersion());
inbox.setHandoverVersion(envelope.handoverVersion());
inbox.setPayloadHash(hash);
inbox.setReceivedAt(LocalDateTime.of(2026, 7, 29, 18, 0));
return inbox;
}
}