refactor test cases

risingwavelabs · Jun 5, 2023 · 833a753 · 833a753
1 parent d9f9d49
commit 833a753
Show file tree

Hide file tree

Showing 10 changed files with 88 additions and 58 deletions.
diff --git a/.../connector-node/connector-api/src/main/java/com/risingwave/connector/api/TableSchema.java b/.../connector-node/connector-api/src/main/java/com/risingwave/connector/api/TableSchema.java
@@ -14,7 +14,6 @@
 
 package com.risingwave.connector.api;
 
-import com.google.common.collect.Lists;
 import com.risingwave.connector.api.sink.SinkRow;
 import com.risingwave.proto.ConnectorServiceProto;
 import com.risingwave.proto.Data;
@@ -75,37 +74,6 @@ public List<ColumnDesc> getColumnDescs() {
         return columnDescs;
     }
 
-    public static TableSchema getMockTableSchema() {
-        return new TableSchema(
-                Lists.newArrayList("id", "name"),
-                Lists.newArrayList(
-                        Data.DataType.newBuilder().setTypeName(TypeName.INT32).build(),
-                        Data.DataType.newBuilder().setTypeName(TypeName.VARCHAR).build()),
-                Lists.newArrayList("id"));
-    }
-
-    public static ConnectorServiceProto.TableSchema getMockTableProto() {
-        return ConnectorServiceProto.TableSchema.newBuilder()
-                .addColumns(
-                        ConnectorServiceProto.TableSchema.Column.newBuilder()
-                                .setName("id")
-                                .setDataType(
-                                        Data.DataType.newBuilder()
-                                                .setTypeName(TypeName.INT32)
-                                                .build())
-                                .build())
-                .addColumns(
-                        ConnectorServiceProto.TableSchema.Column.newBuilder()
-                                .setName("name")
-                                .setDataType(
-                                        Data.DataType.newBuilder()
-                                                .setTypeName(TypeName.VARCHAR)
-                                                .build())
-                                .build())
-                .addAllPkIndices(List.of(1))
-                .build();
-    }
-
     public Object getFromRow(String columnName, SinkRow row) {
         return row.get(columnIndices.get(columnName));
     }

diff --git a/...ctor-node/risingwave-connector-test/src/test/java/com/risingwave/connector/TestUtils.java b/...ctor-node/risingwave-connector-test/src/test/java/com/risingwave/connector/TestUtils.java
@@ -0,0 +1,59 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed 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 com.risingwave.connector;
+
+import com.google.common.collect.Lists;
+import com.risingwave.connector.api.TableSchema;
+import com.risingwave.proto.ConnectorServiceProto;
+import com.risingwave.proto.Data;
+import java.util.List;
+
+public class TestUtils {
+
+    public static TableSchema getMockTableSchema() {
+        return new TableSchema(
+                Lists.newArrayList("id", "name"),
+                Lists.newArrayList(
+                        Data.DataType.newBuilder()
+                                .setTypeName(Data.DataType.TypeName.INT32)
+                                .build(),
+                        Data.DataType.newBuilder()
+                                .setTypeName(Data.DataType.TypeName.VARCHAR)
+                                .build()),
+                Lists.newArrayList("id"));
+    }
+
+    public static ConnectorServiceProto.TableSchema getMockTableProto() {
+        return ConnectorServiceProto.TableSchema.newBuilder()
+                .addColumns(
+                        ConnectorServiceProto.TableSchema.Column.newBuilder()
+                                .setName("id")
+                                .setDataType(
+                                        Data.DataType.newBuilder()
+                                                .setTypeName(Data.DataType.TypeName.INT32)
+                                                .build())
+                                .build())
+                .addColumns(
+                        ConnectorServiceProto.TableSchema.Column.newBuilder()
+                                .setName("name")
+                                .setDataType(
+                                        Data.DataType.newBuilder()
+                                                .setTypeName(Data.DataType.TypeName.VARCHAR)
+                                                .build())
+                                .build())
+                .addAllPkIndices(List.of(1))
+                .build();
+    }
+}
diff --git a/...isingwave/connector/DeserializerTest.java → ...wave/connector/sink/DeserializerTest.java b/...isingwave/connector/DeserializerTest.java → ...wave/connector/sink/DeserializerTest.java
@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.risingwave.connector;
+package com.risingwave.connector.sink;
 
-import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.JsonDeserializer;
+import com.risingwave.connector.TestUtils;
 import com.risingwave.connector.api.sink.SinkRow;
 import com.risingwave.proto.ConnectorServiceProto;
 import com.risingwave.proto.ConnectorServiceProto.SinkStreamRequest.WriteBatch.JsonPayload;
@@ -23,7 +24,7 @@
 
 public class DeserializerTest extends TestCase {
     public void testJsonDeserializer() {
-        JsonDeserializer deserializer = new JsonDeserializer(TableSchema.getMockTableSchema());
+        JsonDeserializer deserializer = new JsonDeserializer(TestUtils.getMockTableSchema());
         JsonPayload jsonPayload =
                 JsonPayload.newBuilder()
                         .addRowOps(

diff --git a/...om/risingwave/connector/FileSinkTest.java → ...singwave/connector/sink/FileSinkTest.java b/...om/risingwave/connector/FileSinkTest.java → ...singwave/connector/sink/FileSinkTest.java
@@ -12,13 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.risingwave.connector;
+package com.risingwave.connector.sink;
 
 import static com.risingwave.proto.Data.*;
 import static org.junit.Assert.*;
 
 import com.google.common.collect.Iterators;
-import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.FileSink;
+import com.risingwave.connector.FileSinkConfig;
+import com.risingwave.connector.TestUtils;
 import com.risingwave.connector.api.sink.ArraySinkRow;
 import java.io.IOException;
 import java.nio.file.Files;
@@ -36,7 +38,7 @@ public void testSync() throws IOException {
         }
 
         FileSinkConfig config = new FileSinkConfig(path);
-        FileSink sink = new FileSink(config, TableSchema.getMockTableSchema());
+        FileSink sink = new FileSink(config, TestUtils.getMockTableSchema());
         String filePath = sink.getSinkPath();
 
         Path file = Paths.get(filePath);
@@ -78,7 +80,7 @@ public void testWrite() throws IOException {
             Files.createDirectories(Paths.get(path));
         }
         FileSinkConfig config = new FileSinkConfig(path);
-        FileSink sink = new FileSink(config, TableSchema.getMockTableSchema());
+        FileSink sink = new FileSink(config, TestUtils.getMockTableSchema());
 
         String filePath = sink.getSinkPath();
         try {
@@ -107,7 +109,7 @@ public void testDrop() throws IOException {
             Files.createDirectories(Paths.get(path));
         }
         FileSinkConfig config = new FileSinkConfig(path);
-        FileSink sink = new FileSink(config, TableSchema.getMockTableSchema());
+        FileSink sink = new FileSink(config, TestUtils.getMockTableSchema());
 
         sink.drop();
 

diff --git a/...ave/connector/SinkStreamObserverTest.java → ...onnector/sink/SinkStreamObserverTest.java b/...ave/connector/SinkStreamObserverTest.java → ...onnector/sink/SinkStreamObserverTest.java
@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.risingwave.connector;
+package com.risingwave.connector.sink;
 
-import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.SinkStreamObserver;
+import com.risingwave.connector.TestUtils;
 import com.risingwave.proto.ConnectorServiceProto;
 import com.risingwave.proto.ConnectorServiceProto.SinkConfig;
 import com.risingwave.proto.Data.Op;
@@ -27,7 +28,7 @@ public class SinkStreamObserverTest {
 
     public SinkConfig fileSinkConfig =
             SinkConfig.newBuilder()
-                    .setTableSchema(TableSchema.getMockTableProto())
+                    .setTableSchema(TestUtils.getMockTableProto())
                     .setConnectorType("file")
                     .putAllProperties(Map.of("output.path", "/tmp/rw-connector"))
                     .build();

diff --git a/...or-test/src/test/java/com/risingwave/connector/sink/deltalake/DeltaLakeLocalSinkTest.java b/...or-test/src/test/java/com/risingwave/connector/sink/deltalake/DeltaLakeLocalSinkTest.java
@@ -19,7 +19,7 @@
 
 import com.google.common.collect.Iterators;
 import com.risingwave.connector.DeltaLakeSink;
-import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.TestUtils;
 import com.risingwave.connector.api.sink.ArraySinkRow;
 import io.delta.standalone.DeltaLog;
 import java.io.IOException;
@@ -43,7 +43,7 @@ private static DeltaLakeSink createMockSink(String location) {
         DeltaLakeSinkFactoryTest.createMockTable(location);
         Configuration conf = new Configuration();
         DeltaLog log = DeltaLog.forTable(conf, location);
-        return new DeltaLakeSink(TableSchema.getMockTableSchema(), conf, log);
+        return new DeltaLakeSink(TestUtils.getMockTableSchema(), conf, log);
     }
 
     private void validateTableWithSpark(String location, List<Row> rows, StructType schema) {

diff --git a/...-test/src/test/java/com/risingwave/connector/sink/deltalake/DeltaLakeSinkFactoryTest.java b/...-test/src/test/java/com/risingwave/connector/sink/deltalake/DeltaLakeSinkFactoryTest.java
@@ -15,7 +15,7 @@
 package com.risingwave.connector.sink.deltalake;
 
 import com.risingwave.connector.DeltaLakeSinkFactory;
-import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.TestUtils;
 import io.delta.standalone.DeltaLog;
 import io.delta.standalone.Operation;
 import io.delta.standalone.OptimisticTransaction;
@@ -49,7 +49,7 @@ public static void createMockTable(String location) {
         Configuration conf = new Configuration();
         DeltaLog log = DeltaLog.forTable(conf, location);
 
-        // should be synchronized with `TableSchema.getMockTableSchema()`;
+        // should be synchronized with `TestUtils.getMockTableSchema()`;
         StructType schema =
                 new StructType(
                         new StructField[] {
@@ -73,7 +73,7 @@ public void testCreate() throws IOException {
         createMockTable(location);
         DeltaLakeSinkFactory sinkFactory = new DeltaLakeSinkFactory();
         sinkFactory.create(
-                TableSchema.getMockTableSchema(),
+                TestUtils.getMockTableSchema(),
                 new HashMap<>() {
                     {
                         put("location", String.format("file://%s", location));

diff --git a/...ctor-test/src/test/java/com/risingwave/connector/sink/iceberg/IcebergSinkFactoryTest.java b/...ctor-test/src/test/java/com/risingwave/connector/sink/iceberg/IcebergSinkFactoryTest.java
@@ -18,7 +18,7 @@
 
 import com.risingwave.connector.IcebergSink;
 import com.risingwave.connector.IcebergSinkFactory;
-import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.TestUtils;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Paths;
@@ -64,7 +64,7 @@ public void testCreate() throws IOException {
         IcebergSink sink =
                 (IcebergSink)
                         sinkFactory.create(
-                                TableSchema.getMockTableSchema(),
+                                TestUtils.getMockTableSchema(),
                                 Map.of(
                                         "type",
                                         sinkMode,

diff --git a/...nector-test/src/test/java/com/risingwave/connector/sink/iceberg/IcebergSinkLocalTest.java b/...nector-test/src/test/java/com/risingwave/connector/sink/iceberg/IcebergSinkLocalTest.java
@@ -21,7 +21,7 @@
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Sets;
 import com.risingwave.connector.IcebergSink;
-import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.TestUtils;
 import com.risingwave.connector.api.sink.ArraySinkRow;
 import java.io.IOException;
 import java.nio.file.Files;
@@ -110,7 +110,7 @@ public void testSync() throws IOException {
         TableIdentifier tableIdentifier = TableIdentifier.of(databaseName, tableName);
         IcebergSink sink =
                 new IcebergSink(
-                        TableSchema.getMockTableSchema(),
+                        TestUtils.getMockTableSchema(),
                         hadoopCatalog,
                         hadoopCatalog.loadTable(tableIdentifier),
                         FileFormat.PARQUET);
@@ -153,7 +153,7 @@ public void testWrite() throws IOException {
         TableIdentifier tableIdentifier = TableIdentifier.of(databaseName, tableName);
         IcebergSink sink =
                 new IcebergSink(
-                        TableSchema.getMockTableSchema(),
+                        TestUtils.getMockTableSchema(),
                         hadoopCatalog,
                         hadoopCatalog.loadTable(tableIdentifier),
                         FileFormat.PARQUET);
@@ -189,7 +189,7 @@ public void testDrop() throws IOException {
         TableIdentifier tableIdentifier = TableIdentifier.of(databaseName, tableName);
         IcebergSink sink =
                 new IcebergSink(
-                        TableSchema.getMockTableSchema(),
+                        TestUtils.getMockTableSchema(),
                         hadoopCatalog,
                         hadoopCatalog.loadTable(tableIdentifier),
                         FileFormat.PARQUET);

diff --git a/...-test/src/test/java/com/risingwave/connector/sink/iceberg/UpsertIcebergSinkLocalTest.java b/...-test/src/test/java/com/risingwave/connector/sink/iceberg/UpsertIcebergSinkLocalTest.java
@@ -20,7 +20,6 @@
 
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Sets;
-import com.risingwave.connector.api.TableSchema;
 import com.risingwave.connector.api.sink.ArraySinkRow;
 import java.io.IOException;
 import java.nio.file.Files;
@@ -109,7 +108,7 @@ public void testSync() throws IOException {
         TableIdentifier tableIdentifier = TableIdentifier.of(databaseName, tableName);
         UpsertIcebergSink sink =
                 new UpsertIcebergSink(
-                        TableSchema.getMockTableSchema(),
+                        TestUtils.getMockTableSchema(),
                         hadoopCatalog,
                         hadoopCatalog.loadTable(tableIdentifier),
                         FileFormat.PARQUET);
@@ -152,7 +151,7 @@ public void testWrite() throws IOException {
         TableIdentifier tableIdentifier = TableIdentifier.of(databaseName, tableName);
         UpsertIcebergSink sink =
                 new UpsertIcebergSink(
-                        TableSchema.getMockTableSchema(),
+                        TestUtils.getMockTableSchema(),
                         hadoopCatalog,
                         hadoopCatalog.loadTable(tableIdentifier),
                         FileFormat.PARQUET);
@@ -198,7 +197,7 @@ public void testDrop() throws IOException {
         TableIdentifier tableIdentifier = TableIdentifier.of(databaseName, tableName);
         UpsertIcebergSink sink =
                 new UpsertIcebergSink(
-                        TableSchema.getMockTableSchema(),
+                        TestUtils.getMockTableSchema(),
                         hadoopCatalog,
                         hadoopCatalog.loadTable(tableIdentifier),
                         FileFormat.PARQUET);