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 2024/10/27 13:48:45 UTC
(shardingsphere) branch master updated: Add test cases on OpenGaussIncrementalPositionManager (#33431)
This is an automated email from the ASF dual-hosted git repository.
zhangliang 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 cf790187bdc Add test cases on OpenGaussIncrementalPositionManager (#33431)
cf790187bdc is described below
commit cf790187bdc538f97b0ef1cb48ac0200d5e2d2b3
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Sun Oct 27 21:48:39 2024 +0800
Add test cases on OpenGaussIncrementalPositionManager (#33431)
* Add test cases on OpenGaussIncrementalPositionManager
* Add test cases on OpenGaussIncrementalPositionManager
---
.../OpenGaussIncrementalPositionManagerTest.java | 100 +++++++++++++++++++++
.../PostgreSQLIncrementalPositionManagerTest.java | 97 +++++++++-----------
2 files changed, 144 insertions(+), 53 deletions(-)
diff --git a/kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/incremental/wal/position/OpenGaussIncrementalPositionManagerTest.java b/kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/incremental/wal/position/OpenGaussIncrementalPositionManagerTest.java
new file mode 100644
index 00000000000..cae661e50b8
--- /dev/null
+++ b/kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/incremental/wal/position/OpenGaussIncrementalPositionManagerTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.data.pipeline.opengauss.ingest.incremental.wal.position;
+
+import org.apache.shardingsphere.data.pipeline.core.ingest.position.DialectIncrementalPositionManager;
+import org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.WALPosition;
+import org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.position.slot.PostgreSQLSlotManager;
+import org.apache.shardingsphere.infra.database.core.spi.DatabaseTypedSPILoader;
+import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.test.fixture.jdbc.MockedDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Answers;
+import org.mockito.Mock;
+import org.mockito.internal.configuration.plugins.Plugins;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.opengauss.replication.LogSequenceNumber;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class OpenGaussIncrementalPositionManagerTest {
+
+ private static final String LSN = "0/14EFDB8";
+
+ private final DatabaseType databaseType = TypedSPILoader.getService(DatabaseType.class, "openGauss");
+
+ private final DialectIncrementalPositionManager incrementalPositionManager = DatabaseTypedSPILoader.getService(DialectIncrementalPositionManager.class, databaseType);
+
+ private DataSource dataSource;
+
+ @Mock(answer = Answers.RETURNS_DEEP_STUBS)
+ private Connection connection;
+
+ @Mock
+ private PostgreSQLSlotManager slotManager;
+
+ @BeforeEach
+ void setUp() throws ReflectiveOperationException {
+ dataSource = new MockedDataSource(connection);
+ Plugins.getMemberAccessor().set(OpenGaussIncrementalPositionManager.class.getDeclaredField("slotManager"), incrementalPositionManager, slotManager);
+ }
+
+ @Test
+ void assertInitWithData() {
+ WALPosition actual = (WALPosition) incrementalPositionManager.init(LSN);
+ assertThat(actual.getLogSequenceNumber().toString(), is("OpenGaussLogSequenceNumber(logSequenceNumber=LSN{0/14EFDB8})"));
+ }
+
+ @Test
+ void assertInitWithDataSource() throws SQLException {
+ PreparedStatement preparedStatement = mockPreparedStatement();
+ when(connection.prepareStatement("SELECT PG_CURRENT_XLOG_LOCATION()")).thenReturn(preparedStatement);
+ WALPosition actual = (WALPosition) incrementalPositionManager.init(dataSource, "");
+ assertThat(actual.getLogSequenceNumber().get(), is(LogSequenceNumber.valueOf(LSN)));
+ verify(slotManager).create(connection, "");
+ }
+
+ private PreparedStatement mockPreparedStatement() throws SQLException {
+ PreparedStatement result = mock(PreparedStatement.class);
+ ResultSet resultSet = mock(ResultSet.class);
+ when(result.executeQuery()).thenReturn(resultSet);
+ when(resultSet.next()).thenReturn(true, false);
+ when(resultSet.getString(1)).thenReturn(LSN);
+ return result;
+ }
+
+ @Test
+ void assertDestroy() throws SQLException {
+ incrementalPositionManager.destroy(dataSource, "");
+ verify(slotManager).dropIfExisted(connection, "");
+ }
+}
diff --git a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/position/PostgreSQLIncrementalPositionManagerTest.java b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/position/PostgreSQLIncrementalPositionManagerTest.java
index 8cf7d63b279..e1698a3822d 100644
--- a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/position/PostgreSQLIncrementalPositionManagerTest.java
+++ b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/position/PostgreSQLIncrementalPositionManagerTest.java
@@ -17,15 +17,21 @@
package org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.position;
-import lombok.SneakyThrows;
+import org.apache.shardingsphere.data.pipeline.core.exception.PipelineInternalException;
+import org.apache.shardingsphere.data.pipeline.core.ingest.position.DialectIncrementalPositionManager;
import org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.WALPosition;
+import org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.position.slot.PostgreSQLSlotManager;
+import org.apache.shardingsphere.infra.database.core.spi.DatabaseTypedSPILoader;
+import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.test.fixture.jdbc.MockedDataSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Answers;
import org.mockito.Mock;
+import org.mockito.internal.configuration.plugins.Plugins;
import org.mockito.junit.jupiter.MockitoExtension;
-import org.mockito.junit.jupiter.MockitoSettings;
-import org.mockito.quality.Strictness;
import org.postgresql.replication.LogSequenceNumber;
import javax.sql.DataSource;
@@ -43,94 +49,79 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
-@MockitoSettings(strictness = Strictness.LENIENT)
class PostgreSQLIncrementalPositionManagerTest {
private static final String POSTGRESQL_96_LSN = "0/14EFDB8";
private static final String POSTGRESQL_10_LSN = "0/1634520";
- @Mock(extraInterfaces = AutoCloseable.class)
+ private final DatabaseType databaseType = TypedSPILoader.getService(DatabaseType.class, "PostgreSQL");
+
+ private final DialectIncrementalPositionManager incrementalPositionManager = DatabaseTypedSPILoader.getService(DialectIncrementalPositionManager.class, databaseType);
+
private DataSource dataSource;
- @Mock
+ @Mock(answer = Answers.RETURNS_DEEP_STUBS)
private Connection connection;
@Mock
private DatabaseMetaData databaseMetaData;
+ @Mock
+ private PostgreSQLSlotManager slotManager;
+
@BeforeEach
- void setUp() throws SQLException {
- when(dataSource.getConnection()).thenReturn(connection);
- when(connection.getCatalog()).thenReturn("sharding_db");
+ void setUp() throws SQLException, ReflectiveOperationException {
+ dataSource = new MockedDataSource(connection);
when(connection.getMetaData()).thenReturn(databaseMetaData);
- PreparedStatement lsn96PreparedStatement = mockPostgreSQL96LSN();
- when(connection.prepareStatement("SELECT * FROM pg_create_logical_replication_slot(?, ?)")).thenReturn(mock(PreparedStatement.class));
- when(connection.prepareStatement("SELECT PG_CURRENT_XLOG_LOCATION()")).thenReturn(lsn96PreparedStatement);
- PreparedStatement lsn10PreparedStatement = mockPostgreSQL10LSN();
- when(connection.prepareStatement("SELECT PG_CURRENT_WAL_LSN()")).thenReturn(lsn10PreparedStatement);
+ Plugins.getMemberAccessor().set(PostgreSQLIncrementalPositionManager.class.getDeclaredField("slotManager"), incrementalPositionManager, slotManager);
}
@Test
- void assertGetCurrentPositionOnPostgreSQL96() throws SQLException {
- mockSlotExistsOrNot(false);
+ void assertInitWithData() {
+ WALPosition actual = (WALPosition) incrementalPositionManager.init(POSTGRESQL_96_LSN);
+ assertThat(actual.getLogSequenceNumber().toString(), is("PostgreSQLLogSequenceNumber(logSequenceNumber=LSN{0/14EFDB8})"));
+ }
+
+ @Test
+ void assertInitWithPostgreSQL96() throws SQLException {
when(databaseMetaData.getDatabaseMajorVersion()).thenReturn(9);
when(databaseMetaData.getDatabaseMinorVersion()).thenReturn(6);
- WALPosition actual = new PostgreSQLIncrementalPositionManager().init(dataSource, "");
+ PreparedStatement preparedStatement = mockPreparedStatement(POSTGRESQL_96_LSN);
+ when(connection.prepareStatement("SELECT PG_CURRENT_XLOG_LOCATION()")).thenReturn(preparedStatement);
+ WALPosition actual = (WALPosition) incrementalPositionManager.init(dataSource, "");
assertThat(actual.getLogSequenceNumber().get(), is(LogSequenceNumber.valueOf(POSTGRESQL_96_LSN)));
+ verify(slotManager).create(connection, "");
}
@Test
- void assertGetCurrentPositionOnPostgreSQL10() throws SQLException {
- mockSlotExistsOrNot(false);
+ void assertInitWithPostgreSQL10() throws SQLException {
when(databaseMetaData.getDatabaseMajorVersion()).thenReturn(10);
- WALPosition actual = new PostgreSQLIncrementalPositionManager().init(dataSource, "");
+ PreparedStatement preparedStatement = mockPreparedStatement(POSTGRESQL_10_LSN);
+ when(connection.prepareStatement("SELECT PG_CURRENT_WAL_LSN()")).thenReturn(preparedStatement);
+ WALPosition actual = (WALPosition) incrementalPositionManager.init(dataSource, "");
assertThat(actual.getLogSequenceNumber().get(), is(LogSequenceNumber.valueOf(POSTGRESQL_10_LSN)));
+ verify(slotManager).create(connection, "");
}
@Test
- void assertGetCurrentPositionThrowException() throws SQLException {
- mockSlotExistsOrNot(false);
- when(databaseMetaData.getDatabaseMajorVersion()).thenReturn(9);
- when(databaseMetaData.getDatabaseMinorVersion()).thenReturn(4);
- assertThrows(RuntimeException.class, () -> new PostgreSQLIncrementalPositionManager().init(dataSource, ""));
+ void assertInitFailed() throws SQLException {
+ when(databaseMetaData.getDatabaseMajorVersion()).thenReturn(1);
+ assertThrows(PipelineInternalException.class, () -> incrementalPositionManager.init(dataSource, ""));
}
- @SneakyThrows(SQLException.class)
- private PreparedStatement mockPostgreSQL96LSN() {
+ private PreparedStatement mockPreparedStatement(final String lsn) throws SQLException {
PreparedStatement result = mock(PreparedStatement.class);
ResultSet resultSet = mock(ResultSet.class);
when(result.executeQuery()).thenReturn(resultSet);
when(resultSet.next()).thenReturn(true, false);
- when(resultSet.getString(1)).thenReturn(POSTGRESQL_96_LSN);
+ when(resultSet.getString(1)).thenReturn(lsn);
return result;
}
- @SneakyThrows(SQLException.class)
- private PreparedStatement mockPostgreSQL10LSN() {
- PreparedStatement result = mock(PreparedStatement.class);
- ResultSet resultSet = mock(ResultSet.class);
- when(result.executeQuery()).thenReturn(resultSet);
- when(resultSet.next()).thenReturn(true, false);
- when(resultSet.getString(1)).thenReturn(POSTGRESQL_10_LSN);
- return result;
- }
-
- @SneakyThrows(SQLException.class)
- private void mockSlotExistsOrNot(final boolean exists) {
- PreparedStatement preparedStatement = mock(PreparedStatement.class);
- when(connection.prepareStatement("SELECT slot_name, database FROM pg_replication_slots WHERE slot_name=? AND plugin=?")).thenReturn(preparedStatement);
- ResultSet resultSet = mock(ResultSet.class);
- when(preparedStatement.executeQuery()).thenReturn(resultSet);
- when(resultSet.next()).thenReturn(exists);
- }
-
@Test
- void assertDestroyWhenSlotExists() throws SQLException {
- mockSlotExistsOrNot(true);
- PreparedStatement preparedStatement = mock(PreparedStatement.class);
- when(connection.prepareStatement("SELECT pg_drop_replication_slot(?)")).thenReturn(preparedStatement);
- new PostgreSQLIncrementalPositionManager().destroy(dataSource, "");
- verify(preparedStatement).execute();
+ void assertDestroy() throws SQLException {
+ incrementalPositionManager.destroy(dataSource, "");
+ verify(slotManager).dropIfExisted(connection, "");
}
}