You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2026/10/01 18:03:49 UTC
(shardingsphere) branch master updated: Reuse log capture extension for expected exception tests (#40029)
This is an automated email from the ASF dual-hosted git repository.
terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 6a0e7d6289b Reuse log capture extension for expected exception tests (#40029)
6a0e7d6289b is described below
commit 6a0e7d6289b0de650008e30fdd0844245375d6e3
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Fri Oct 2 02:03:42 2026 +0800
Reuse log capture extension for expected exception tests (#40029)
---
.../calcite/processor/SQLFederationProcessor.java | 3 +-
mcp/bootstrap/pom.xml | 6 ++
.../transport/MCPTransportErrorFactoryTest.java | 14 +++-
.../MCPResourceSpecificationFactoryTest.java | 14 +++-
.../tool/MCPToolSpecificationFactoryTest.java | 14 +++-
.../server/http/StreamableHttpMCPServletTest.java | 25 ++++++--
.../netty/CDCChannelInboundHandlerTest.java | 74 ++++++++++++++++------
.../extension/log/LogCaptureAssertion.java | 16 +++++
.../extension/log/LogCaptureExtension.java | 18 +++++-
.../extension/log/LogCaptureSettings.java | 46 ++++++++++++++
10 files changed, 195 insertions(+), 35 deletions(-)
diff --git a/kernel/sql-federation/provider/calcite/src/main/java/org/apache/shardingsphere/sqlfederation/provider/calcite/processor/SQLFederationProcessor.java b/kernel/sql-federation/provider/calcite/src/main/java/org/apache/shardingsphere/sqlfederation/provider/calcite/processor/SQLFederationProcessor.java
index 0bda7b52fe8..07b31a87932 100644
--- a/kernel/sql-federation/provider/calcite/src/main/java/org/apache/shardingsphere/sqlfederation/provider/calcite/processor/SQLFederationProcessor.java
+++ b/kernel/sql-federation/provider/calcite/src/main/java/org/apache/shardingsphere/sqlfederation/provider/calcite/processor/SQLFederationProcessor.java
@@ -146,8 +146,7 @@ public final class SQLFederationProcessor {
public void release(final String currentDatabaseName, final String currentSchemaName, final QueryContext queryContext, final SchemaPlus schemaPlus) {
Collection<SimpleTableSegment> simpleTables = queryContext.getSqlStatementContext().getTablesContext().getSimpleTables();
for (SimpleTableSegment each : simpleTables) {
- Table table = getTable(currentDatabaseName, currentSchemaName, schemaPlus,
- each, queryContext.getSqlStatementContext().getSqlStatement().getDatabaseType(), queryContext.getSql());
+ Table table = getTable(currentDatabaseName, currentSchemaName, schemaPlus, each, queryContext.getSqlStatementContext().getSqlStatement().getDatabaseType(), queryContext.getSql());
if (table instanceof SQLFederationTable) {
((SQLFederationTable) table).clearScanImplementor();
}
diff --git a/mcp/bootstrap/pom.xml b/mcp/bootstrap/pom.xml
index 0660306e4cb..8613cc0d24a 100644
--- a/mcp/bootstrap/pom.xml
+++ b/mcp/bootstrap/pom.xml
@@ -50,6 +50,12 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.shardingsphere</groupId>
+ <artifactId>shardingsphere-test-infra-framework</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>io.modelcontextprotocol.sdk</groupId>
diff --git a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/MCPTransportErrorFactoryTest.java b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/MCPTransportErrorFactoryTest.java
index b97d764f29e..722f79e38cf 100644
--- a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/MCPTransportErrorFactoryTest.java
+++ b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/MCPTransportErrorFactoryTest.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.mcp.bootstrap.transport;
+import ch.qos.logback.classic.spi.ThrowableProxy;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import org.apache.shardingsphere.mcp.api.exception.MCPInvalidRequestException;
@@ -24,14 +25,21 @@ import org.apache.shardingsphere.mcp.bootstrap.transport.server.http.validator.M
import org.apache.shardingsphere.mcp.core.protocol.exception.UnsupportedResourceUriException;
import org.apache.shardingsphere.mcp.core.protocol.exception.UnsupportedToolException;
import org.apache.shardingsphere.mcp.support.database.exception.DatabaseCapabilityNotFoundException;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureAssertion;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureExtension;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureSettings;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.util.Map;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertFalse;
+@ExtendWith(LogCaptureExtension.class)
+@LogCaptureSettings(suppressOutput = true)
class MCPTransportErrorFactoryTest {
@Test
@@ -77,11 +85,13 @@ class MCPTransportErrorFactoryTest {
}
@Test
- void assertCreateErrorWithUnexpectedError() {
- McpError actual = MCPTransportErrorFactory.createError(new IllegalStateException("foo_message"));
+ void assertCreateErrorWithUnexpectedError(final LogCaptureAssertion logCaptureAssertion) {
+ IllegalStateException expectedException = new IllegalStateException("foo_message");
+ McpError actual = MCPTransportErrorFactory.createError(expectedException);
assertThat(actual.getJsonRpcError().code(), is(McpSchema.ErrorCodes.INTERNAL_ERROR));
assertThat(actual.getJsonRpcError().message(), is("Service is temporarily unavailable."));
assertFalse(String.valueOf(actual.getJsonRpcError().data()).contains("foo_message"));
+ logCaptureAssertion.assertErrorLog(actualException -> assertThat(((ThrowableProxy) actualException).getThrowable(), sameInstance(expectedException)));
}
@Test
diff --git a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/resource/MCPResourceSpecificationFactoryTest.java b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/resource/MCPResourceSpecificationFactoryTest.java
index 7d6915a7fa0..b725c83b73b 100644
--- a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/resource/MCPResourceSpecificationFactoryTest.java
+++ b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/resource/MCPResourceSpecificationFactoryTest.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.mcp.bootstrap.transport.capability.resource;
+import ch.qos.logback.classic.spi.ThrowableProxy;
import io.modelcontextprotocol.server.McpServerFeatures.SyncResourceSpecification;
import io.modelcontextprotocol.server.McpServerFeatures.SyncResourceTemplateSpecification;
import io.modelcontextprotocol.server.McpSyncServerExchange;
@@ -37,7 +38,11 @@ import org.apache.shardingsphere.mcp.core.session.MCPSessionManager;
import org.apache.shardingsphere.mcp.support.database.capability.MCPDatabaseCapabilityProvider;
import org.apache.shardingsphere.mcp.support.descriptor.MCPShardingSphereMetadataKeys;
import org.apache.shardingsphere.mcp.support.protocol.payload.MCPMapPayload;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureAssertion;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureExtension;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureSettings;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
@@ -50,6 +55,7 @@ import java.util.Optional;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.isA;
+import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -61,6 +67,8 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.when;
+@ExtendWith(LogCaptureExtension.class)
+@LogCaptureSettings(value = "org.apache.shardingsphere.mcp.bootstrap.transport.MCPTransportErrorFactory", suppressOutput = true)
class MCPResourceSpecificationFactoryTest {
@Test
@@ -155,12 +163,13 @@ class MCPResourceSpecificationFactoryTest {
}
@Test
- void assertReadResourceSanitizesRuntimeFailure() {
+ void assertReadResourceSanitizesRuntimeFailure(final LogCaptureAssertion logCaptureAssertion) {
MCPResourceDescriptor descriptor = new MCPResourceDescriptor("shardingsphere://runtime-error", "runtime-error", "Runtime Error", "Read runtime error.", "application/json",
MCPResourceAnnotations.EMPTY, Map.of());
try (MockedStatic<ResourceDefinitionRegistry> mocked = mockStatic(ResourceDefinitionRegistry.class)) {
mocked.when(ResourceDefinitionRegistry::getSupportedResourceDescriptors).thenReturn(List.of(descriptor));
- mocked.when(() -> ResourceDefinitionRegistry.dispatch(any(MCPFeatureRuntimeRequestContext.class), eq("shardingsphere://runtime-error"))).thenThrow(new RuntimeException("runtime failure"));
+ RuntimeException expectedException = new RuntimeException("runtime failure");
+ mocked.when(() -> ResourceDefinitionRegistry.dispatch(any(MCPFeatureRuntimeRequestContext.class), eq("shardingsphere://runtime-error"))).thenThrow(expectedException);
SyncResourceSpecification actualSpecification = findResourceSpecification(
new MCPResourceSpecificationFactory(createRuntimeContext()).createResourceSpecifications(), "shardingsphere://runtime-error");
McpError actual = assertThrows(McpError.class,
@@ -168,6 +177,7 @@ class MCPResourceSpecificationFactoryTest {
assertThat(actual.getJsonRpcError().code(), is(McpSchema.ErrorCodes.INTERNAL_ERROR));
assertThat(actual.getJsonRpcError().message(), is("Service is temporarily unavailable."));
assertFalse(String.valueOf(actual.getJsonRpcError().data()).contains("runtime failure"));
+ logCaptureAssertion.assertErrorLog(actualException -> assertThat(((ThrowableProxy) actualException).getThrowable(), sameInstance(expectedException)));
}
}
diff --git a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/tool/MCPToolSpecificationFactoryTest.java b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/tool/MCPToolSpecificationFactoryTest.java
index 0eba0daff1e..14bef5ae1c5 100644
--- a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/tool/MCPToolSpecificationFactoryTest.java
+++ b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/capability/tool/MCPToolSpecificationFactoryTest.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.mcp.bootstrap.transport.capability.tool;
+import ch.qos.logback.classic.spi.ThrowableProxy;
import io.modelcontextprotocol.server.McpServerFeatures.SyncToolSpecification;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
@@ -31,7 +32,11 @@ import org.apache.shardingsphere.mcp.core.tool.handler.ToolDefinitionRegistry;
import org.apache.shardingsphere.mcp.support.database.exception.DatabaseCapabilityNotFoundException;
import org.apache.shardingsphere.mcp.support.descriptor.MCPShardingSphereMetadataKeys;
import org.apache.shardingsphere.mcp.support.protocol.payload.MCPMapPayload;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureAssertion;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureExtension;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureSettings;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.MockedStatic;
import java.util.List;
@@ -39,6 +44,7 @@ import java.util.Map;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -50,6 +56,8 @@ import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
+@ExtendWith(LogCaptureExtension.class)
+@LogCaptureSettings(value = "org.apache.shardingsphere.mcp.bootstrap.transport.MCPTransportErrorFactory", suppressOutput = true)
class MCPToolSpecificationFactoryTest extends AbstractMCPToolSpecificationFactoryTest {
@Test
@@ -131,15 +139,17 @@ class MCPToolSpecificationFactoryTest extends AbstractMCPToolSpecificationFactor
}
@Test
- void assertCreateToolSpecificationsSanitizeUnexpectedError() {
+ void assertCreateToolSpecificationsSanitizeUnexpectedError(final LogCaptureAssertion logCaptureAssertion) {
try (MockedStatic<ToolDefinitionRegistry> mockedToolDefinitionRegistry = mockStatic(ToolDefinitionRegistry.class)) {
MCPToolDefinition toolDefinition = mockSupportedTool(mockedToolDefinitionRegistry, createToolDescriptorWithoutOutputSchema("fixture_ping"));
+ IllegalStateException expectedException = new IllegalStateException("sensitive detail");
mockedToolDefinitionRegistry.when(() -> ToolDefinitionRegistry.dispatch(any(MCPFeatureRuntimeRequestContext.class), eq(toolDefinition), eq(Map.of())))
- .thenThrow(new IllegalStateException("sensitive detail"));
+ .thenThrow(expectedException);
McpError actual = assertThrows(McpError.class, () -> callTool(createToolSpecification(MCPTransportType.STDIO), createExchange(), "fixture_ping", Map.of()));
assertThat(actual.getJsonRpcError().code(), is(ErrorCodes.INTERNAL_ERROR));
assertThat(actual.getJsonRpcError().message(), is("Service is temporarily unavailable."));
assertFalse(String.valueOf(actual.getJsonRpcError().data()).contains("sensitive detail"));
+ logCaptureAssertion.assertErrorLog(actualException -> assertThat(((ThrowableProxy) actualException).getThrowable(), sameInstance(expectedException)));
}
}
diff --git a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/server/http/StreamableHttpMCPServletTest.java b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/server/http/StreamableHttpMCPServletTest.java
index f428cdde09b..86b4515cc2e 100644
--- a/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/server/http/StreamableHttpMCPServletTest.java
+++ b/mcp/bootstrap/src/test/java/org/apache/shardingsphere/mcp/bootstrap/transport/server/http/StreamableHttpMCPServletTest.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.mcp.bootstrap.transport.server.http;
+import ch.qos.logback.classic.spi.ThrowableProxy;
import io.modelcontextprotocol.server.transport.HttpServletStreamableServerTransportProvider;
import io.modelcontextprotocol.spec.HttpHeaders;
import io.modelcontextprotocol.spec.McpSchema;
@@ -32,9 +33,13 @@ import org.apache.shardingsphere.mcp.bootstrap.config.SessionAttributionSourceCo
import org.apache.shardingsphere.mcp.bootstrap.transport.MCPTransportConstants;
import org.apache.shardingsphere.mcp.bootstrap.transport.MCPTransportJsonMapperFactory;
import org.apache.shardingsphere.mcp.core.session.MCPSessionManager;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureAssertion;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureExtension;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureSettings;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -73,6 +78,8 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+@ExtendWith(LogCaptureExtension.class)
+@LogCaptureSettings(suppressOutput = true)
class StreamableHttpMCPServletTest {
private static final String ACCEPT = "application/json, text/event-stream";
@@ -202,17 +209,19 @@ class StreamableHttpMCPServletTest {
@ParameterizedTest(name = "{0}")
@MethodSource("requestMethods")
- void assertServiceHandleUncommittedException(final String name, final String requestMethod) throws IOException {
+ void assertServiceHandleUncommittedException(final String name, final String requestMethod, final LogCaptureAssertion logCaptureAssertion) throws IOException {
HttpServletRequest request = mock(HttpServletRequest.class);
when(request.getMethod()).thenReturn(requestMethod);
when(request.getHeaderNames()).thenReturn(Collections.enumeration(List.of("Origin")));
when(request.getHeaders("Origin")).thenReturn(Collections.enumeration(List.of("https://example.com")));
HttpServletResponse response = mock(HttpServletResponse.class);
- doThrow(new IOException("foo_failure")).when(response).getWriter();
+ IOException expectedException = new IOException("foo_failure");
+ doThrow(expectedException).when(response).getWriter();
StreamableHttpMCPServlet actual = createServlet(mock(MCPSessionManager.class));
assertDoesNotThrow(() -> actual.service(request, response));
verify(response).reset();
verify(response).setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+ logCaptureAssertion.assertErrorLog(actualException -> assertThat(((ThrowableProxy) actualException).getThrowable(), sameInstance(expectedException)));
}
@ParameterizedTest(name = "{0}")
@@ -367,19 +376,21 @@ class StreamableHttpMCPServletTest {
}
@Test
- void assertServicePostRollbackSessionWhenDelegateFails() throws ServletException, IOException {
+ void assertServicePostRollbackSessionWhenDelegateFails(final LogCaptureAssertion logCaptureAssertion) throws ServletException, IOException {
MCPSessionManager sessionManager = new MCPSessionManager(Map.of());
HttpServletRequest request = mock(HttpServletRequest.class);
when(request.getMethod()).thenReturn("POST");
when(request.getHeaderNames()).thenReturn(Collections.emptyEnumeration());
HttpServletResponse response = mock(HttpServletResponse.class);
StreamableHttpMCPServlet actual = createServlet(sessionManager);
+ ServletException expectedException = new ServletException("foo_failure");
doAnswer(invocation -> {
((HttpServletResponse) invocation.getArgument(1)).setHeader(HttpHeaders.MCP_SESSION_ID, "session-id");
- throw new ServletException("foo_failure");
+ throw expectedException;
}).when(getDelegate()).service(any(HttpServletRequest.class), any(HttpServletResponse.class));
assertDoesNotThrow(() -> actual.service(request, response));
assertFalse(sessionManager.hasSession("session-id"));
+ logCaptureAssertion.assertErrorLog(actualException -> assertThat(((ThrowableProxy) actualException).getThrowable(), sameInstance(expectedException)));
}
@Test
@@ -514,7 +525,7 @@ class StreamableHttpMCPServletTest {
}
@Test
- void assertServiceDeletePreserveSessionWhenDelegateFails() throws ServletException, IOException {
+ void assertServiceDeletePreserveSessionWhenDelegateFails(final LogCaptureAssertion logCaptureAssertion) throws ServletException, IOException {
MCPSessionManager sessionManager = new MCPSessionManager(Collections.emptyMap());
sessionManager.createSession(new MCPSessionIdentity("session-id", "", "", Map.of()));
HttpServletRequest request = mock(HttpServletRequest.class);
@@ -523,9 +534,11 @@ class StreamableHttpMCPServletTest {
when(request.getHeader(HttpHeaders.MCP_SESSION_ID)).thenReturn("session-id");
HttpServletResponse response = mock(HttpServletResponse.class);
StreamableHttpMCPServlet actual = createServlet(sessionManager);
- doThrow(new ServletException("foo_failure")).when(getDelegate()).service(any(HttpServletRequest.class), any(HttpServletResponse.class));
+ ServletException expectedException = new ServletException("foo_failure");
+ doThrow(expectedException).when(getDelegate()).service(any(HttpServletRequest.class), any(HttpServletResponse.class));
assertDoesNotThrow(() -> actual.service(request, response));
assertTrue(sessionManager.hasSession("session-id"));
+ logCaptureAssertion.assertErrorLog(actualException -> assertThat(((ThrowableProxy) actualException).getThrowable(), sameInstance(expectedException)));
}
@Test
diff --git a/proxy/frontend/core/src/test/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandlerTest.java b/proxy/frontend/core/src/test/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandlerTest.java
index 998c5c92794..16f82defd33 100644
--- a/proxy/frontend/core/src/test/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandlerTest.java
+++ b/proxy/frontend/core/src/test/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandlerTest.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.proxy.frontend.netty;
+import ch.qos.logback.classic.spi.IThrowableProxy;
import com.google.common.hash.Hashing;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@@ -31,6 +32,8 @@ import org.apache.shardingsphere.authority.model.ShardingSpherePrivileges;
import org.apache.shardingsphere.authority.rule.AuthorityRule;
import org.apache.shardingsphere.data.pipeline.cdc.context.CDCConnectionContext;
import org.apache.shardingsphere.data.pipeline.cdc.exception.CDCExceptionWrapper;
+import org.apache.shardingsphere.data.pipeline.cdc.exception.CDCLoginFailedException;
+import org.apache.shardingsphere.data.pipeline.cdc.exception.EmptyCDCLoginRequestBodyException;
import org.apache.shardingsphere.data.pipeline.cdc.generator.CDCResponseUtils;
import org.apache.shardingsphere.data.pipeline.cdc.handler.CDCBackendHandler;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.AckStreamingRequestBody;
@@ -62,6 +65,9 @@ import org.apache.shardingsphere.infra.metadata.user.ShardingSphereUser;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
import org.apache.shardingsphere.proxy.frontend.protocol.FrontDatabaseProtocolTypeFactory;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureAssertion;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureExtension;
+import org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureSettings;
import org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
import org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.BeforeEach;
@@ -96,7 +102,8 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-@ExtendWith(AutoMockExtension.class)
+@ExtendWith({AutoMockExtension.class, LogCaptureExtension.class})
+@LogCaptureSettings(suppressOutput = true)
@StaticMockSettings({ProxyContext.class, FrontDatabaseProtocolTypeFactory.class})
@MockitoSettings(strictness = Strictness.LENIENT)
class CDCChannelInboundHandlerTest {
@@ -163,27 +170,37 @@ class CDCChannelInboundHandlerTest {
}
@Test
- void assertExceptionCaughtWithWrapperClosesChannel() {
+ void assertExceptionCaughtWithWrapperClosesChannel(final LogCaptureAssertion logCaptureAssertion) {
ChannelHandlerContext context = mock(ChannelHandlerContext.class, RETURNS_DEEP_STUBS);
ChannelFuture channelFuture = mock(ChannelFuture.class);
when(context.writeAndFlush(any())).thenReturn(channelFuture);
handler.exceptionCaught(context, new CDCExceptionWrapper("request-id", new PipelineInvalidParameterException("invalid")));
verify(context).writeAndFlush(argThat(argument -> "request-id".equals(((CDCResponse) argument).getRequestId())));
verify(channelFuture).addListener(ChannelFutureListener.CLOSE);
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("invalid"));
+ }
+
+ private void assertExpectedError(final LogCaptureAssertion logCaptureAssertion, final Throwable expectedException) {
+ logCaptureAssertion.assertErrorLog(actualException -> {
+ IThrowableProxy actualCause = CDCExceptionWrapper.class.getName().equals(actualException.getClassName()) ? actualException.getCause() : actualException;
+ assertThat(actualCause.getClassName(), is(expectedException.getClass().getName()));
+ assertThat(actualCause.getMessage(), is(expectedException.getMessage()));
+ });
}
@Test
- void assertExceptionCaughtWithNonWrapperException() {
+ void assertExceptionCaughtWithNonWrapperException(final LogCaptureAssertion logCaptureAssertion) {
ChannelHandlerContext context = mock(ChannelHandlerContext.class, RETURNS_DEEP_STUBS);
ChannelFuture channelFuture = mock(ChannelFuture.class);
when(context.channel().attr(CONNECTION_CONTEXT_KEY).get()).thenReturn(new CDCConnectionContext(user));
when(context.writeAndFlush(any())).thenReturn(channelFuture);
handler.exceptionCaught(context, new RuntimeException("error"));
verify(channelFuture, never()).addListener(ChannelFutureListener.CLOSE);
+ assertExpectedError(logCaptureAssertion, new RuntimeException("error"));
}
@Test
- void assertLoginRequestFailed() {
+ void assertLoginRequestFailed(final LogCaptureAssertion logCaptureAssertion) {
CDCRequest request = CDCRequest.newBuilder().setType(Type.LOGIN).setLoginRequestBody(LoginRequestBody.newBuilder()
.setBasicBody(BasicBody.newBuilder().setUsername("root2").build()).build()).build();
channel.writeInbound(request);
@@ -193,10 +210,11 @@ class CDCChannelInboundHandlerTest {
assertThat(loginResult.getStatus(), is(Status.FAILED));
assertThat(loginResult.getErrorCode(), is(XOpenSQLState.DATA_SOURCE_REJECTED_CONNECTION_ATTEMPT.getValue()));
assertFalse(channel.isOpen());
+ assertExpectedError(logCaptureAssertion, new CDCLoginFailedException());
}
@Test
- void assertIllegalLoginRequest() {
+ void assertIllegalLoginRequest(final LogCaptureAssertion logCaptureAssertion) {
CDCRequest request = CDCRequest.newBuilder().setType(Type.LOGIN).setVersion(1).setRequestId("test").build();
channel.writeInbound(request);
CDCResponse greeting = channel.readOutbound();
@@ -205,10 +223,11 @@ class CDCChannelInboundHandlerTest {
assertThat(loginResult.getStatus(), is(Status.FAILED));
assertThat(loginResult.getErrorCode(), is(XOpenSQLState.NOT_FOUND.getValue()));
assertFalse(channel.isOpen());
+ assertExpectedError(logCaptureAssertion, new EmptyCDCLoginRequestBodyException());
}
@Test
- void assertLoginRequestBodyWithoutBasicBody() {
+ void assertLoginRequestBodyWithoutBasicBody(final LogCaptureAssertion logCaptureAssertion) {
channel.writeInbound(CDCRequest.newBuilder().setType(Type.LOGIN).setRequestId("login-without-basic-body")
.setLoginRequestBody(LoginRequestBody.newBuilder().setType(LoginRequestBody.LoginType.BASIC).build()).build());
CDCResponse greeting = channel.readOutbound();
@@ -217,6 +236,7 @@ class CDCChannelInboundHandlerTest {
assertThat(loginResult.getStatus(), is(Status.FAILED));
assertThat(loginResult.getErrorCode(), is(XOpenSQLState.NOT_FOUND.getValue()));
assertFalse(channel.isOpen());
+ assertExpectedError(logCaptureAssertion, new EmptyCDCLoginRequestBodyException());
}
@Test
@@ -289,41 +309,45 @@ class CDCChannelInboundHandlerTest {
}
@Test
- void assertStreamDataRequestBodyMissing() {
+ void assertStreamDataRequestBodyMissing(final LogCaptureAssertion logCaptureAssertion) {
CDCRequest request = CDCRequest.newBuilder().setType(Type.STREAM_DATA).setRequestId("stream-request").build();
CDCResponse response = writeRequestWithContext(request);
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("Stream data request body is empty"));
}
@Test
- void assertStreamDataRequestDatabaseMissing() {
+ void assertStreamDataRequestDatabaseMissing(final LogCaptureAssertion logCaptureAssertion) {
CDCRequest request = createStreamDataRequest("");
CDCResponse response = writeRequestWithContext(request);
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("Database is empty"));
}
@Test
- void assertStreamDataRequestSourceSchemaTableMissing() {
+ void assertStreamDataRequestSourceSchemaTableMissing(final LogCaptureAssertion logCaptureAssertion) {
StreamDataRequestBody body = StreamDataRequestBody.newBuilder().setDatabase("logic_db").build();
CDCRequest request = CDCRequest.newBuilder().setType(Type.STREAM_DATA).setRequestId("stream-request").setStreamDataRequestBody(body).build();
CDCResponse response = writeRequestWithContext(request);
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("Source schema table is empty"));
}
@Test
- void assertStreamDataRequestMissingAuthorityRule() {
+ void assertStreamDataRequestMissingAuthorityRule(final LogCaptureAssertion logCaptureAssertion) {
mockProxyContext(new RuleMetaData(Collections.emptyList()));
CDCRequest request = createStreamDataRequest("logic_db");
CDCResponse response = writeRequestWithContext(request);
SQLException expectedException = SQLExceptionTransformEngine.toSQLException(new MissingRequiredRuleException("authority"), FrontDatabaseProtocolTypeFactory.getDatabaseType());
assertThat(response.getErrorCode(), is(expectedException.getSQLState()));
+ assertExpectedError(logCaptureAssertion, new MissingRequiredRuleException("authority"));
}
@Test
- void assertStreamDataRequestPrivilegesNotFound() {
+ void assertStreamDataRequestPrivilegesNotFound(final LogCaptureAssertion logCaptureAssertion) {
when(authorityRule.findPrivileges(any())).thenReturn(Optional.empty());
CDCRequest request = createStreamDataRequest("logic_db");
CDCResponse response = writeRequestWithContext(request);
@@ -331,15 +355,17 @@ class CDCChannelInboundHandlerTest {
SQLException expectedException = SQLExceptionTransformEngine.toSQLException(new AccessDeniedException(grantee.getUsername(), grantee.getHostname(), false),
FrontDatabaseProtocolTypeFactory.getDatabaseType());
assertThat(response.getErrorCode(), is(expectedException.getSQLState()));
+ assertExpectedError(logCaptureAssertion, new AccessDeniedException(user.getGrantee().getUsername(), user.getGrantee().getHostname(), false));
}
@Test
- void assertStreamDataRequestPrivilegesWithoutDatabasePermission() {
+ void assertStreamDataRequestPrivilegesWithoutDatabasePermission(final LogCaptureAssertion logCaptureAssertion) {
when(privileges.hasPrivileges("logic_db")).thenReturn(false);
CDCRequest request = createStreamDataRequest("logic_db");
CDCResponse response = writeRequestWithContext(request);
SQLException expectedException = SQLExceptionTransformEngine.toSQLException(new UnknownDatabaseException("logic_db"), FrontDatabaseProtocolTypeFactory.getDatabaseType());
assertThat(response.getErrorCode(), is(expectedException.getSQLState()));
+ assertExpectedError(logCaptureAssertion, new UnknownDatabaseException("logic_db"));
}
private CDCRequest createStreamDataRequest(final String database) {
@@ -349,7 +375,7 @@ class CDCChannelInboundHandlerTest {
}
@Test
- void assertStreamDataRequestWithSchemaNotFoundException() {
+ void assertStreamDataRequestWithSchemaNotFoundException(final LogCaptureAssertion logCaptureAssertion) {
channel.attr(CONNECTION_CONTEXT_KEY).set(new CDCConnectionContext(user));
CDCRequest request = createStreamDataRequest("logic_db");
when(backendHandler.streamData(any(), any(), any(), any())).thenThrow(new SchemaNotFoundException("foo_schema"));
@@ -358,10 +384,11 @@ class CDCChannelInboundHandlerTest {
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getRequestId(), is("stream-request"));
assertThat(response.getErrorCode(), is(XOpenSQLState.NOT_FOUND.getValue()));
+ assertExpectedError(logCaptureAssertion, new SchemaNotFoundException("foo_schema"));
}
@Test
- void assertStreamDataRequestWithInvalidParameterException() {
+ void assertStreamDataRequestWithInvalidParameterException(final LogCaptureAssertion logCaptureAssertion) {
channel.attr(CONNECTION_CONTEXT_KEY).set(new CDCConnectionContext(user));
CDCRequest request = createStreamDataRequest("logic_db");
when(backendHandler.streamData(any(), any(), any(), any())).thenThrow(new PipelineInvalidParameterException("invalid job configuration"));
@@ -370,6 +397,7 @@ class CDCChannelInboundHandlerTest {
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getRequestId(), is("stream-request"));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("invalid job configuration"));
}
@Test
@@ -387,20 +415,22 @@ class CDCChannelInboundHandlerTest {
}
@Test
- void assertAckStreamingRequestBodyMissing() {
+ void assertAckStreamingRequestBodyMissing(final LogCaptureAssertion logCaptureAssertion) {
CDCRequest request = CDCRequest.newBuilder().setType(Type.ACK_STREAMING).setRequestId("ack-request").build();
CDCResponse response = writeRequestWithContext(request);
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("Ack request body is empty"));
}
@Test
- void assertAckStreamingRequestAckIdMissing() {
+ void assertAckStreamingRequestAckIdMissing(final LogCaptureAssertion logCaptureAssertion) {
AckStreamingRequestBody.Builder bodyBuilder = AckStreamingRequestBody.newBuilder();
bodyBuilder.setAckId("");
CDCResponse response = writeRequestWithContext(CDCRequest.newBuilder().setType(Type.ACK_STREAMING).setRequestId("ack-request").setAckStreamingRequestBody(bodyBuilder.build()).build());
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("Ack request is empty"));
}
@Test
@@ -415,20 +445,22 @@ class CDCChannelInboundHandlerTest {
}
@Test
- void assertStartStreamingRequestBodyMissing() {
+ void assertStartStreamingRequestBodyMissing(final LogCaptureAssertion logCaptureAssertion) {
CDCRequest request = CDCRequest.newBuilder().setType(Type.START_STREAMING).setRequestId("start-request").build();
CDCResponse response = writeRequestWithContext(request);
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("Start streaming request body is empty"));
}
@Test
- void assertStartStreamingRequestIdMissing() {
+ void assertStartStreamingRequestIdMissing(final LogCaptureAssertion logCaptureAssertion) {
StartStreamingRequestBody.Builder bodyBuilder = StartStreamingRequestBody.newBuilder();
bodyBuilder.setStreamingId("");
CDCResponse response = writeRequestWithContext(CDCRequest.newBuilder().setType(Type.START_STREAMING).setRequestId("start-request").setStartStreamingRequestBody(bodyBuilder.build()).build());
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("Streaming id is empty"));
}
@Test
@@ -445,7 +477,7 @@ class CDCChannelInboundHandlerTest {
}
@Test
- void assertStartStreamingRequestWithInvalidParameterException() {
+ void assertStartStreamingRequestWithInvalidParameterException(final LogCaptureAssertion logCaptureAssertion) {
channel.attr(CONNECTION_CONTEXT_KEY).set(new CDCConnectionContext(user));
when(backendHandler.getDatabaseNameByJobId("job-1")).thenReturn("logic_db");
doThrow(new PipelineInvalidParameterException("invalid job configuration")).when(backendHandler).startStreaming(any(), any(), any());
@@ -456,6 +488,7 @@ class CDCChannelInboundHandlerTest {
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getRequestId(), is("start-request"));
assertThat(response.getErrorCode(), is(XOpenSQLState.INVALID_PARAMETER_VALUE.getValue()));
+ assertExpectedError(logCaptureAssertion, new PipelineInvalidParameterException("invalid job configuration"));
}
@Test
@@ -517,7 +550,7 @@ class CDCChannelInboundHandlerTest {
}
@Test
- void assertDropStreamingRequestFailed() {
+ void assertDropStreamingRequestFailed(final LogCaptureAssertion logCaptureAssertion) {
CDCConnectionContext connectionContext = new CDCConnectionContext(user);
connectionContext.setJobId("job-1");
channel.attr(CONNECTION_CONTEXT_KEY).set(connectionContext);
@@ -529,6 +562,7 @@ class CDCChannelInboundHandlerTest {
assertThat(response.getStatus(), is(Status.FAILED));
assertThat(response.getRequestId(), is("drop-request"));
assertThat(connectionContext.getJobId(), is("job-1"));
+ assertExpectedError(logCaptureAssertion, new PipelineInternalException("failed"));
}
@Test
diff --git a/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureAssertion.java b/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureAssertion.java
index e5e7fab7a29..1c974e3402f 100644
--- a/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureAssertion.java
+++ b/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureAssertion.java
@@ -19,9 +19,12 @@ package org.apache.shardingsphere.test.infra.framework.extension.log;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.IThrowableProxy;
import lombok.RequiredArgsConstructor;
import java.util.List;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
@@ -59,4 +62,17 @@ public final class LogCaptureAssertion {
assertThat(loggingEvents.get(actualLogEventIndex).getMessage(), is(expectedMessage));
}
}
+
+ /**
+ * Assert that exactly one ERROR log exists and consume it only after the throwable assertion succeeds.
+ * A failed assertion leaves the log available for output during test cleanup.
+ *
+ * @param throwableAssertion assertion identifying the expected exception
+ */
+ public void assertErrorLog(final Consumer<IThrowableProxy> throwableAssertion) {
+ List<ILoggingEvent> actualErrors = loggingEvents.stream().filter(each -> Level.ERROR == each.getLevel()).collect(Collectors.toList());
+ assertThat(actualErrors.size(), is(1));
+ throwableAssertion.accept(actualErrors.get(0).getThrowableProxy());
+ loggingEvents.remove(actualErrors.get(0));
+ }
}
diff --git a/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureExtension.java b/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureExtension.java
index 50fa6ed3066..7702feabb34 100644
--- a/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureExtension.java
+++ b/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureExtension.java
@@ -36,19 +36,35 @@ public final class LogCaptureExtension implements BeforeEachCallback, AfterEachC
private Logger logger;
+ private boolean suppressOutput;
+
+ private boolean originalAdditivity;
+
@Override
public void beforeEach(final ExtensionContext context) {
- String logTopic = context.getRequiredTestClass().getName().substring(0, context.getRequiredTestClass().getName().lastIndexOf("Test"));
+ LogCaptureSettings logCaptureSettings = context.getRequiredTestClass().getAnnotation(LogCaptureSettings.class);
+ String logTopic = null == logCaptureSettings || logCaptureSettings.value().isEmpty()
+ ? context.getRequiredTestClass().getName().substring(0, context.getRequiredTestClass().getName().lastIndexOf("Test"))
+ : logCaptureSettings.value();
logger = (Logger) LoggerFactory.getLogger(logTopic);
listAppender = new ListAppender<>();
listAppender.start();
logger.addAppender(listAppender);
+ suppressOutput = null != logCaptureSettings && logCaptureSettings.suppressOutput();
+ if (suppressOutput) {
+ originalAdditivity = logger.isAdditive();
+ logger.setAdditive(false);
+ }
}
@Override
public void afterEach(final ExtensionContext context) {
if (null != logger && null != listAppender) {
logger.detachAppender(listAppender);
+ if (suppressOutput) {
+ logger.setAdditive(originalAdditivity);
+ listAppender.list.forEach(logger::callAppenders);
+ }
}
}
diff --git a/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureSettings.java b/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureSettings.java
new file mode 100644
index 00000000000..8e19f3f11f4
--- /dev/null
+++ b/test/infra/framework/src/main/java/org/apache/shardingsphere/test/infra/framework/extension/log/LogCaptureSettings.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.test.infra.framework.extension.log;
+
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+/**
+ * Log capture settings.
+ */
+@Inherited
+@Retention(RetentionPolicy.RUNTIME)
+public @interface LogCaptureSettings {
+
+ /**
+ * Log topic.
+ *
+ * @return log topic
+ */
+ String value() default "";
+
+ /**
+ * Whether to defer output to ancestor appenders until test cleanup.
+ * Logs consumed by successful assertions are omitted; all remaining logs are replayed.
+ * Tests using the same logger must run sequentially.
+ *
+ * @return whether to suppress captured output
+ */
+ boolean suppressOutput() default false;
+}