fix(mq): B-04 显式 mq-enabled 默认关+最终凭据链+注册应用核验+动态 binding——平台 host 注入不再自动启用 MQ 消费
This commit is contained in:
chenxing
@@ -0,0 +1,94 @@
|
||||
package com.yunhai.svchall.service.impl;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.rabbitmq.client.Channel;
|
||||
import com.yunhai.svchall.entity.FcAnswerInbox;
|
||||
import com.yunhai.svchall.model.FcApplyModels.SnapshotEnvelope;
|
||||
import com.yunhai.svchall.repository.FcAnswerInboxRepository;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.InOrder;
|
||||
import org.springframework.amqp.core.Message;
|
||||
import org.springframework.amqp.core.MessageProperties;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Optional;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.mockito.ArgumentMatchers.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/** Local JSON handoff and manual ACK/NACK ordering; not real broker or transaction commit evidence. */
|
||||
class FcAnswerConsumerMqTest {
|
||||
private final ObjectMapper mapper = new ObjectMapper();
|
||||
private AnswerProjectionStore store;
|
||||
private FcSubmissionClient client;
|
||||
private FcAnswerInboxRepository rows;
|
||||
private FcAnswerConsumer consumer;
|
||||
private Channel channel;
|
||||
private final String json = "{\"protocolVersion\":\"fc.answer.snapshot.v2\",\"eventId\":\"evt\","
|
||||
+ "\"eventType\":\"updated\",\"sourceAppKey\":\"100025\",\"consumerAppKey\":\"100018\","
|
||||
+ "\"consumerSchoolId\":27,\"grantVersion\":4,\"formUuid\":\"form\",\"formVersion\":1,"
|
||||
+ "\"answerUuid\":\"answer\",\"answerRevision\":8,\"isDeleted\":false,\"contentHash\":\"hash\","
|
||||
+ "\"occurredAt\":\"2026-10-23T10:00:00+08:00\",\"snapshot\":{\"schId\":27,\"appCode\":\"owner\","
|
||||
+ "\"code\":\"leave\",\"submitterId\":41,\"subjectId\":42,\"status\":\"processed\","
|
||||
+ "\"values\":{\"reason\":\"leave\"},\"answerNo\":\"LEAVE-8\",\"createTime\":\"2026-10-23T10:00:00\","
|
||||
+ "\"updateTime\":\"2026-10-23T10:01:00\",\"intent\":{\"intentId\":\"intent\"}}}";
|
||||
|
||||
@BeforeEach void setUp() {
|
||||
store = mock(AnswerProjectionStore.class); client = mock(FcSubmissionClient.class);
|
||||
rows = mock(FcAnswerInboxRepository.class); channel = mock(Channel.class);
|
||||
consumer = new FcAnswerConsumer(store, client, rows, mapper);
|
||||
ReflectionTestUtils.setField(consumer, "consumerAppKey", "100018");
|
||||
}
|
||||
|
||||
@Test void validJsonPreservesEveryAnswerCoordinateBeforeAck() throws Exception {
|
||||
when(store.applySnapshot(any())).thenReturn(new AnswerProjectionStore.Disposition(
|
||||
AnswerProjectionStore.Outcome.REJECTED, "evt", FcAnswerInbox.STATE_REJECTED, 8L, "hash", null, null));
|
||||
consumer.onMessage(message(json), channel, 5L);
|
||||
ArgumentCaptor<SnapshotEnvelope> envelope = ArgumentCaptor.forClass(SnapshotEnvelope.class);
|
||||
InOrder order = inOrder(store, channel); order.verify(store).applySnapshot(envelope.capture());
|
||||
order.verify(channel).basicAck(5L, false);
|
||||
// Compare the complete JSON wire representation; DTO LongNode vs parsed IntNode is not data loss.
|
||||
com.fasterxml.jackson.databind.JsonNode decoded = mapper.readTree(mapper.writeValueAsBytes(envelope.getValue()));
|
||||
assertThat(decoded).isEqualTo(mapper.readTree(json));
|
||||
verifyNoInteractions(client);
|
||||
}
|
||||
|
||||
@Test void invalidJsonIsAckedAsPoisonWithoutPretendingItWasApplied() throws Exception {
|
||||
consumer.onMessage(message("{broken"), channel, 6L);
|
||||
verify(channel).basicAck(6L, false); verify(channel, never()).basicNack(anyLong(), anyBoolean(), anyBoolean());
|
||||
verifyNoInteractions(store, client);
|
||||
}
|
||||
|
||||
@Test void failedProjectionIsNackedWithoutAckOrFalseAppliedReport() throws Exception {
|
||||
when(store.applySnapshot(any())).thenThrow(new IllegalStateException("database unavailable"));
|
||||
consumer.onMessage(message(json), channel, 7L);
|
||||
verify(channel).basicNack(7L, false, false); verify(channel, never()).basicAck(anyLong(), anyBoolean());
|
||||
verifyNoInteractions(client);
|
||||
}
|
||||
|
||||
@Test void reportFailureKeepsTheCommittedOutcomeAndSchedulesOnlyReportRetry() throws Exception {
|
||||
when(store.applySnapshot(any())).thenReturn(new AnswerProjectionStore.Disposition(
|
||||
AnswerProjectionStore.Outcome.APPLIED, "evt", FcAnswerInbox.STATE_APPLIED, 8L, "hash", 42, null));
|
||||
FcAnswerInbox row = new FcAnswerInbox(); row.setState(FcAnswerInbox.STATE_APPLIED);
|
||||
when(rows.findByEventIdAndConsumerAppKey("evt", "100018")).thenReturn(Optional.of(row));
|
||||
consumer.onMessage(message(json), channel, 8L);
|
||||
InOrder order = inOrder(store, channel, client);
|
||||
order.verify(store).applySnapshot(any()); order.verify(channel).basicAck(8L, false);
|
||||
order.verify(client).reportApplied(any());
|
||||
assertThat(row.getState()).isEqualTo(FcAnswerInbox.STATE_APPLIED);
|
||||
assertThat(row.getReportState()).isEqualTo(FcAnswerInbox.REPORT_FAILED);
|
||||
assertThat(row.getReportAttempts()).isEqualTo(1);
|
||||
assertThat(row.getNextReportAt()).isAfter(LocalDateTime.now());
|
||||
verify(channel, never()).basicNack(anyLong(), anyBoolean(), anyBoolean());
|
||||
}
|
||||
|
||||
private Message message(String body) {
|
||||
MessageProperties props = new MessageProperties(); props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
|
||||
return new Message(body.getBytes(StandardCharsets.UTF_8), props);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package com.yunhai.svchall.service.impl;
|
||||
|
||||
import com.yunhai.svchall.config.FcAnswerMqConfiguration;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.expression.spel.standard.SpelExpressionParser;
|
||||
import org.springframework.mock.env.MockEnvironment;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/** B-04 默认关闭的 Spring 条件装配回归，不启动真实 broker。 */
|
||||
class ShMqExplicitEnableTest {
|
||||
private final ApplicationContextRunner context = new ApplicationContextRunner()
|
||||
.withUserConfiguration(FcAnswerMqConfiguration.class, FcAnswerConsumer.class);
|
||||
|
||||
@Test
|
||||
void platformHostInjectionAloneDoesNotCreateTopologyOrConsumer() {
|
||||
context.withPropertyValues("spring.rabbitmq.host=platform-broker").run(ctx -> {
|
||||
assertThat(ctx).hasNotFailed();
|
||||
assertThat(ctx).doesNotHaveBean(FcAnswerConsumer.class);
|
||||
assertThat(ctx).doesNotHaveBean(FcAnswerMqConfiguration.class);
|
||||
assertThat(ctx).doesNotHaveBean("fcAnswerContainerFactory");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void explicitFalseStillWinsWithPlatformHost() {
|
||||
context.withPropertyValues("spring.rabbitmq.host=platform-broker", "svchall.fc.mq-enabled=false")
|
||||
.run(ctx -> assertThat(ctx).hasNotFailed().doesNotHaveBean(FcAnswerConsumer.class)
|
||||
.doesNotHaveBean(FcAnswerMqConfiguration.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void enablingWithoutBrokerDoesNotCreateConsumerOrFactory() {
|
||||
context.withPropertyValues("svchall.fc.mq-enabled=true", "spring.rabbitmq.host=")
|
||||
.run(ctx -> assertThat(ctx).hasNotFailed().doesNotHaveBean(FcAnswerConsumer.class)
|
||||
.doesNotHaveBean("fcAnswerContainerFactory"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void bothProductionConditionsRequireTheSameExplicitTrueAndHost() {
|
||||
String topology = FcAnswerMqConfiguration.class.getAnnotation(ConditionalOnExpression.class).value();
|
||||
String consumer = FcAnswerConsumer.class.getAnnotation(ConditionalOnExpression.class).value();
|
||||
assertThat(consumer).isEqualTo(topology).contains("svchall.fc.mq-enabled:false");
|
||||
MockEnvironment env = new MockEnvironment().withProperty("svchall.fc.mq-enabled", "true")
|
||||
.withProperty("spring.rabbitmq.host", "broker");
|
||||
assertThat(new SpelExpressionParser().parseExpression(env.resolvePlaceholders(topology))
|
||||
.getValue(Boolean.class)).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
void bindingRemainsSnapshotPrefixForOwnConsumerIdentity() {
|
||||
FcAnswerMqConfiguration config = new FcAnswerMqConfiguration();
|
||||
ReflectionTestUtils.setField(config, "consumerAppKey", "100018");
|
||||
assertThat(config.fcSnapshotBinding().getRoutingKey()).isEqualTo("fc.snapshot.v2.100018.*");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
package com.yunhai.svchall.service.impl;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import com.yunhai.platform.openapi.base.OpenApiAppClient;
|
||||
import com.yunhai.svchall.config.FcAnswerMqConfiguration;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.env.YamlPropertySourceLoader;
|
||||
import org.springframework.core.env.MapPropertySource;
|
||||
import org.springframework.core.env.StandardEnvironment;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import java.util.Collections;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/** Binding object and registry/config contract checks; actual broker routing is NOT RUN. */
|
||||
class ShMqRoutingIdentityTest {
|
||||
private final ObjectMapper mapper = new ObjectMapper();
|
||||
|
||||
@Test void yamlUsesTheFinalSigningPropertyInsteadOfTheStaleEnvironmentFallback() throws Exception {
|
||||
StandardEnvironment env = new StandardEnvironment();
|
||||
new YamlPropertySourceLoader().load("application", new ClassPathResource("application.yml"))
|
||||
.forEach(env.getPropertySources()::addLast);
|
||||
env.getPropertySources().addFirst(new MapPropertySource("final-platform-key",
|
||||
Collections.singletonMap("yh.app-auth.app-key", "100008")));
|
||||
assertThat(env.getProperty("svchall.fc.consumer-app-key")).isEqualTo("100008");
|
||||
assertThat(env.getProperty("svchall.fc.mq-enabled")).isEqualTo("false");
|
||||
}
|
||||
|
||||
@Test void matchingCredentialBuildsTheCanonical100018BindingWithoutUsingThePrimaryId() {
|
||||
FcAnswerMqConfiguration config = config("100018", "100018", registered(700018, 100018, "svc-hall"));
|
||||
assertThatCode(config::validateIdentity).doesNotThrowAnyException();
|
||||
assertThat(config.fcSnapshotBinding().getRoutingKey()).isEqualTo("fc.snapshot.v2.100018.*");
|
||||
assertThat(config.fcSnapshotBinding().getExchange()).isEqualTo("fc.answer.delivery.v2");
|
||||
}
|
||||
|
||||
@Test void stale100018CannotSurviveA100008FinalSigningConfiguration() {
|
||||
FcAnswerMqConfiguration config = config("100018", "100008", registered(700008, 100008, "svc-hall"));
|
||||
assertThatThrownBy(config::validateIdentity).hasMessageContaining("CONSUMER_CREDENTIAL_MISMATCH");
|
||||
}
|
||||
|
||||
@Test void configuredCredentialDeterminesBindingWithoutHardcoding100008() {
|
||||
FcAnswerMqConfiguration config = config("100008", "100008", registered(700008, 100008, "svc-hall"));
|
||||
config.validateIdentity();
|
||||
assertThat(config.fcSnapshotBinding().getRoutingKey()).isEqualTo("fc.snapshot.v2.100008.*")
|
||||
.isNotEqualTo("fc.snapshot.v2.100018.*");
|
||||
}
|
||||
|
||||
@Test void aCredentialRegisteredToAnotherBusinessApplicationIsRejected() {
|
||||
assertThatThrownBy(config("100018", "100018", registered(700025, 100025, "form-center"))::validateIdentity)
|
||||
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
|
||||
}
|
||||
|
||||
@Test void businessPrimaryIdIsNeverComparedEqualToOrSubstitutedForTheCredential() {
|
||||
assertThatThrownBy(config("700018", "700018", registered(700018, 100018, "svc-hall"))::validateIdentity)
|
||||
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
|
||||
}
|
||||
|
||||
@Test void emptyRegistryOrMissingPrimaryIdFailsClosedOnlyWhenMqIsExplicitlyEnabled() {
|
||||
assertThatThrownBy(config("100018", "100018", null)::validateIdentity)
|
||||
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
|
||||
ObjectNode incomplete = registered(700018, 100018, "svc-hall"); incomplete.remove("id");
|
||||
assertThatThrownBy(config("100018", "100018", incomplete)::validateIdentity)
|
||||
.hasMessageContaining("CONSUMER_APPLICATION_MISMATCH");
|
||||
}
|
||||
|
||||
@Test void routeWildcardsOrExtraWordsCannotExpandTheConsumersBinding() {
|
||||
for (String key : new String[]{"", "100018.*", "100018.#", "100018.27"}) {
|
||||
FcAnswerMqConfiguration config = config(key, key, null);
|
||||
assertThatThrownBy(config::fcSnapshotBinding).hasMessageContaining("INVALID_CONSUMER_ROUTE_KEY");
|
||||
}
|
||||
}
|
||||
|
||||
private ObjectNode registered(int id, int credential, String code) {
|
||||
return mapper.createObjectNode().put("id", id).put("appKey", credential).put("appCode", code);
|
||||
}
|
||||
|
||||
private FcAnswerMqConfiguration config(String consumer, String signing, ObjectNode application) {
|
||||
FcAnswerMqConfiguration config = new FcAnswerMqConfiguration();
|
||||
ReflectionTestUtils.setField(config, "consumerAppKey", consumer);
|
||||
ReflectionTestUtils.setField(config, "signingAppKey", signing);
|
||||
ReflectionTestUtils.setField(config, "applicationCode", "svc-hall");
|
||||
OpenApiAppClient registry = mock(OpenApiAppClient.class);
|
||||
when(registry.getAppByCode("svc-hall")).thenReturn(application);
|
||||
ReflectionTestUtils.setField(config, "appRegistryClient", registry);
|
||||
return config;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user