This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
commit 5cd404b56c3c0eb09665072134bc6c30a7c0fccc Author: Martijn Visser <[email protected]> AuthorDate: Mon Sep 21 13:54:25 2026 +0200 [FLINK-40729][tests] Replace Mockito with reusable test implementations The Flink code style asks for reusable test implementations rather than Mockito, which had three users here. JdbcTestBase.getRuntimeContext had no callers at all, so both overloads go. BaseJdbcWriterTest mocked WriterInitContext and JobInfo. JdbcWriter takes an InitContext, which is three methods, so TestingInitContext implements that instead. Its serializer now comes from the existing getSerializer helper rather than from a mock returning null. The comment about changes between 1.18 and 1.19 no longer applies on a Flink 2.x branch. JdbcDialectConverterTest needed a ResultSet that hands a timestamp column a LocalDateTime, which no driver would return. TestingResultSet answers getObject(int) from a fixed row and throws on anything else, so a test that starts to depend on another method fails rather than reading a default. Generated-by: Claude Code (Claude Opus 5) --- .../apache/flink/connector/jdbc/JdbcTestBase.java | 25 ---------- .../database/dialect/JdbcDialectConverterTest.java | 8 ++-- .../datastream/sink/writer/BaseJdbcWriterTest.java | 19 ++++---- .../jdbc/testutils/TestingInitContext.java | 54 ++++++++++++++++++++++ .../connector/jdbc/testutils/TestingResultSet.java | 52 +++++++++++++++++++++ 5 files changed, 117 insertions(+), 41 deletions(-) diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/JdbcTestBase.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/JdbcTestBase.java index 5bb49aef..93d12d1e 100644 --- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/JdbcTestBase.java +++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/JdbcTestBase.java @@ -18,21 +18,12 @@ package org.apache.flink.connector.jdbc; import org.apache.flink.api.common.ExecutionConfig; -import org.apache.flink.api.common.JobID; -import org.apache.flink.api.common.JobInfo; -import org.apache.flink.api.common.JobInfoImpl; -import org.apache.flink.api.common.TaskInfo; -import org.apache.flink.api.common.TaskInfoImpl; -import org.apache.flink.api.common.functions.RuntimeContext; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.connector.jdbc.derby.DerbyTestBase; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; -import org.mockito.Mockito; - -import static org.mockito.Mockito.doReturn; /** * Base class for JDBC test using DDL from {@link JdbcTestFixture}. It uses create tables before @@ -50,22 +41,6 @@ public abstract class JdbcTestBase implements DerbyTestBase { JdbcTestFixture.cleanUpDatabasesStatic(getMetadata()); } - public static RuntimeContext getRuntimeContext(Boolean reused) { - ExecutionConfig config = getExecutionConfig(reused); - RuntimeContext context = Mockito.mock(RuntimeContext.class); - doReturn(config.isObjectReuseEnabled()).when(context).isObjectReuseEnabled(); - return context; - } - - public static RuntimeContext getRuntimeContext(JobID jobId) { - RuntimeContext context = getRuntimeContext(false); - JobInfo jobInfo = new JobInfoImpl(jobId, "test_jobName"); - TaskInfo taskInfo = new TaskInfoImpl("test_taskName", 4, 1, 4, 0); - doReturn(jobInfo).when(context).getJobInfo(); - doReturn(taskInfo).when(context).getTaskInfo(); - return context; - } - public static ExecutionConfig getExecutionConfig(boolean reused) { ExecutionConfig config = new ExecutionConfig(); if (reused) { diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/database/dialect/JdbcDialectConverterTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/database/dialect/JdbcDialectConverterTest.java index f4261afd..4266d115 100644 --- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/database/dialect/JdbcDialectConverterTest.java +++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/database/dialect/JdbcDialectConverterTest.java @@ -18,13 +18,13 @@ package org.apache.flink.connector.jdbc.core.database.dialect; +import org.apache.flink.connector.jdbc.testutils.TestingResultSet; import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.TimestampType; import org.junit.jupiter.api.Test; -import org.mockito.Mockito; import java.sql.ResultSet; import java.time.LocalDateTime; @@ -48,10 +48,8 @@ class JdbcDialectConverterTest { } }; - ResultSet resultSet = Mockito.mock(ResultSet.class); - Mockito.when(resultSet.getObject(1)).thenReturn(123); - Mockito.when(resultSet.getObject(2)) - .thenReturn(LocalDateTime.parse("2021-04-07T00:00:05.999")); + ResultSet resultSet = + TestingResultSet.of(123, LocalDateTime.parse("2021-04-07T00:00:05.999")); RowData res = rowConverter.toInternal(resultSet); assertThat(res.getInt(0)).isEqualTo(123); diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/BaseJdbcWriterTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/BaseJdbcWriterTest.java index 7a7f04d1..205f4cf9 100644 --- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/BaseJdbcWriterTest.java +++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/BaseJdbcWriterTest.java @@ -18,13 +18,14 @@ package org.apache.flink.connector.jdbc.core.datastream.sink.writer; import org.apache.flink.api.common.JobID; -import org.apache.flink.api.common.JobInfo; import org.apache.flink.api.common.TaskInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.connector.sink2.InitContext; import org.apache.flink.api.connector.sink2.SinkWriter; -import org.apache.flink.api.connector.sink2.WriterInitContext; import org.apache.flink.connector.base.DeliveryGuarantee; import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; +import org.apache.flink.connector.jdbc.JdbcTestBase; import org.apache.flink.connector.jdbc.core.datastream.sink.committer.JdbcCommitable; import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider; import org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement; @@ -32,13 +33,13 @@ import org.apache.flink.connector.jdbc.datasource.statements.SimpleJdbcQueryStat import org.apache.flink.connector.jdbc.derby.DerbyTestBase; import org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer; import org.apache.flink.connector.jdbc.testutils.TableManaged; +import org.apache.flink.connector.jdbc.testutils.TestingInitContext; import org.apache.flink.connector.jdbc.testutils.tables.templates.BooksTable; import org.apache.flink.connector.testutils.source.TestingTaskInfo; import org.apache.flink.util.StringUtils; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; -import org.mockito.Mockito; import java.util.Arrays; import java.util.Collections; @@ -47,7 +48,6 @@ import java.util.stream.Collectors; import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.doReturn; /** * Base smoke tests for the {@link @@ -87,17 +87,14 @@ abstract class BaseJdbcWriterTest implements DerbyTestBase { @BeforeEach void init() throws Exception { - // We have to mock this because we have changes between 1.18 and 1.19 - WriterInitContext sinkContext = Mockito.mock(WriterInitContext.class); - JobInfo jobInfo = Mockito.mock(JobInfo.class); - doReturn(jobInfo).when(sinkContext).getJobInfo(); - doReturn(JobID.fromHexString(JOBID)).when(jobInfo).getJobId(); TaskInfo taskInfo = new TestingTaskInfo("test_task", 4, 1, 4, 0, "test_subTask", "id"); - doReturn(taskInfo).when(sinkContext).getTaskInfo(); + InitContext sinkContext = + new TestingInitContext(JobID.fromHexString(JOBID), "test_jobName", taskInfo); JdbcOutputSerializer<BooksTable.BookEntry> outputSerializer = JdbcOutputSerializer.of( - sinkContext.createInputSerializer(), sinkContext.isObjectReuseEnabled()); + JdbcTestBase.getSerializer( + TypeInformation.of(BooksTable.BookEntry.class), false)); JdbcQueryStatement<BooksTable.BookEntry> queryStatement = new SimpleJdbcQueryStatement<>( diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/testutils/TestingInitContext.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/testutils/TestingInitContext.java new file mode 100644 index 00000000..412b4881 --- /dev/null +++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/testutils/TestingInitContext.java @@ -0,0 +1,54 @@ +/* + * 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.flink.connector.jdbc.testutils; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobInfo; +import org.apache.flink.api.common.JobInfoImpl; +import org.apache.flink.api.common.TaskInfo; +import org.apache.flink.api.connector.sink2.InitContext; + +import java.util.OptionalLong; + +/** An {@link InitContext} carrying a fixed job and task, for sink writers under test. */ +public class TestingInitContext implements InitContext { + + private final JobInfo jobInfo; + private final TaskInfo taskInfo; + + public TestingInitContext(JobID jobId, String jobName, TaskInfo taskInfo) { + this.jobInfo = new JobInfoImpl(jobId, jobName); + this.taskInfo = taskInfo; + } + + @Override + public OptionalLong getRestoredCheckpointId() { + return OptionalLong.empty(); + } + + @Override + public JobInfo getJobInfo() { + return jobInfo; + } + + @Override + public TaskInfo getTaskInfo() { + return taskInfo; + } +} diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/testutils/TestingResultSet.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/testutils/TestingResultSet.java new file mode 100644 index 00000000..43050c85 --- /dev/null +++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/testutils/TestingResultSet.java @@ -0,0 +1,52 @@ +/* + * 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.flink.connector.jdbc.testutils; + +import java.lang.reflect.Proxy; +import java.sql.ResultSet; + +/** + * A {@link ResultSet} over a single fixed row, for converters that read their columns through + * {@link ResultSet#getObject(int)}. It lets a test hand a column a value no driver would return for + * it, such as a {@link java.time.LocalDateTime} for a timestamp. Any other method throws, so a test + * that starts to depend on one fails instead of silently reading a default. + */ +public final class TestingResultSet { + + private TestingResultSet() {} + + /** Returns a {@link ResultSet} whose {@code getObject(i)} answers {@code columns[i - 1]}. */ + public static ResultSet of(Object... columns) { + return (ResultSet) + Proxy.newProxyInstance( + TestingResultSet.class.getClassLoader(), + new Class<?>[] {ResultSet.class}, + (proxy, method, args) -> { + if (method.getName().equals("getObject") + && args != null + && args.length == 1 + && args[0] instanceof Integer) { + return columns[(Integer) args[0] - 1]; + } + throw new UnsupportedOperationException( + "TestingResultSet only answers getObject(int), not " + + method.getName()); + }); + } +}
