Skip to content

Commit 3553bce

Browse files
reswqareswqa
committed
[FLINK-31223][test] Abstract SqlClientTestBase.
1 parent d3123fd commit 3553bce

File tree

3 files changed

+133
-87
lines changed

3 files changed

+133
-87
lines changed

‎flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java

Lines changed: 1 addition & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -19,49 +19,30 @@
1919
package org.apache.flink.table.client;
2020

2121
import org.apache.flink.configuration.Configuration;
22-
import org.apache.flink.core.testutils.CommonTestUtils;
23-
import org.apache.flink.table.client.cli.TerminalUtils;
2422
import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointExtension;
2523
import org.apache.flink.table.gateway.service.utils.SqlGatewayServiceExtension;
2624
import org.apache.flink.util.FileUtils;
2725
import org.apache.flink.util.Preconditions;
2826

29-
import org.jline.terminal.Size;
30-
import org.jline.terminal.Terminal;
31-
import org.junit.jupiter.api.AfterEach;
32-
import org.junit.jupiter.api.BeforeEach;
3327
import org.junit.jupiter.api.Order;
3428
import org.junit.jupiter.api.Test;
3529
import org.junit.jupiter.api.extension.RegisterExtension;
36-
import org.junit.jupiter.api.io.TempDir;
3730

38-
import java.io.ByteArrayInputStream;
39-
import java.io.ByteArrayOutputStream;
4031
import java.io.File;
41-
import java.io.IOException;
42-
import java.io.OutputStream;
4332
import java.net.URL;
44-
import java.nio.charset.StandardCharsets;
4533
import java.nio.file.Files;
46-
import java.nio.file.Path;
4734
import java.nio.file.Paths;
48-
import java.nio.file.StandardOpenOption;
4935
import java.util.Arrays;
5036
import java.util.Collections;
51-
import java.util.HashMap;
5237
import java.util.List;
53-
import java.util.Map;
5438

55-
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
5639
import static org.apache.flink.configuration.DeploymentOptions.TARGET;
5740
import static org.apache.flink.core.testutils.CommonTestUtils.assertThrows;
5841
import static org.assertj.core.api.Assertions.assertThat;
5942
import static org.assertj.core.api.Assertions.assertThatThrownBy;
6043

6144
/** Tests for {@link SqlClient}. */
62-
class SqlClientTest {
63-
64-
@TempDir private Path tempFolder;
45+
class SqlClientTest extends SqlClientTestBase {
6546

6647
@RegisterExtension
6748
@Order(1)
@@ -78,34 +59,6 @@ class SqlClientTest {
7859
private static final SqlGatewayRestEndpointExtension SQL_GATEWAY_REST_ENDPOINT_EXTENSION =
7960
new SqlGatewayRestEndpointExtension(SQL_GATEWAY_SERVICE_EXTENSION::getService);
8061

81-
private Map<String, String> originalEnv;
82-
83-
private String historyPath;
84-
85-
@BeforeEach
86-
void before() throws IOException {
87-
originalEnv = System.getenv();
88-
89-
// prepare conf dir
90-
File confFolder = Files.createTempDirectory(tempFolder, "conf").toFile();
91-
File confYaml = new File(confFolder, "config.yaml");
92-
if (!confYaml.createNewFile()) {
93-
throw new IOException("Can't create testing config.yaml file.");
94-
}
95-
96-
// adjust the test environment for the purposes of this test
97-
Map<String, String> map = new HashMap<>(System.getenv());
98-
map.put(ENV_FLINK_CONF_DIR, confFolder.getAbsolutePath());
99-
CommonTestUtils.setEnv(map);
100-
101-
historyPath = Files.createTempFile(tempFolder, "history", "").toFile().getPath();
102-
}
103-
104-
@AfterEach
105-
void after() {
106-
CommonTestUtils.setEnv(originalEnv);
107-
}
108-
10962
@Test
11063
void testEmbeddedWithOptions() throws Exception {
11164
String[] args = new String[] {"embedded", "-hist", historyPath};
@@ -332,42 +285,4 @@ private void runTestCliHelp(String[] args, String expected) throws Exception {
332285
.toURI())));
333286
assertThat(runSqlClient(args)).isEqualTo(actual);
334287
}
335-
336-
private String runSqlClient(String[] args) throws Exception {
337-
return runSqlClient(args, "QUIT;\n", false);
338-
}
339-
340-
private String runSqlClient(String[] args, String statements, boolean printInput)
341-
throws Exception {
342-
try (OutputStream out = new ByteArrayOutputStream();
343-
Terminal terminal =
344-
TerminalUtils.createDumbTerminal(
345-
new ByteArrayInputStream(
346-
statements.getBytes(StandardCharsets.UTF_8)),
347-
out)) {
348-
if (printInput) {
349-
// The default terminal has an empty size. Here increase the terminal to allow
350-
// the line reader print the input string.
351-
terminal.setSize(new Size(160, 80));
352-
}
353-
SqlClient.startClient(args, () -> terminal);
354-
return out.toString().replace("\r\n", System.lineSeparator());
355-
}
356-
}
357-
358-
private String createSqlFile(List<String> statements, String name) throws IOException {
359-
// create sql file
360-
File sqlFileFolder = Files.createTempDirectory(tempFolder, "sql-file").toFile();
361-
File sqlFile = new File(sqlFileFolder, name);
362-
if (!sqlFile.createNewFile()) {
363-
throw new IOException(String.format("Can't create testing %s.", name));
364-
}
365-
String sqlFilePath = sqlFile.getPath();
366-
Files.write(
367-
Paths.get(sqlFilePath),
368-
statements,
369-
StandardCharsets.UTF_8,
370-
StandardOpenOption.APPEND);
371-
return sqlFilePath;
372-
}
373288
}

‎flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTestBase.java

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.flink.table.client;
20+
21+
import org.apache.flink.core.testutils.CommonTestUtils;
22+
import org.apache.flink.table.client.cli.TerminalUtils;
23+
24+
import org.jline.terminal.Size;
25+
import org.jline.terminal.Terminal;
26+
import org.junit.jupiter.api.AfterEach;
27+
import org.junit.jupiter.api.BeforeEach;
28+
import org.junit.jupiter.api.io.TempDir;
29+
30+
import java.io.ByteArrayInputStream;
31+
import java.io.ByteArrayOutputStream;
32+
import java.io.File;
33+
import java.io.IOException;
34+
import java.io.OutputStream;
35+
import java.nio.charset.StandardCharsets;
36+
import java.nio.file.Files;
37+
import java.nio.file.Path;
38+
import java.nio.file.Paths;
39+
import java.nio.file.StandardOpenOption;
40+
import java.util.HashMap;
41+
import java.util.List;
42+
import java.util.Map;
43+
44+
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
45+
46+
/** Base class for test {@link SqlClient}. */
47+
class SqlClientTestBase {
48+
@TempDir private Path tempFolder;
49+
50+
protected String historyPath;
51+
52+
protected Map<String, String> originalEnv;
53+
54+
@BeforeEach
55+
void before() throws IOException {
56+
originalEnv = System.getenv();
57+
58+
// prepare conf dir
59+
File confFolder = Files.createTempDirectory(tempFolder, "conf").toFile();
60+
File confYaml = new File(confFolder, "config.yaml");
61+
if (!confYaml.createNewFile()) {
62+
throw new IOException("Can't create testing config.yaml file.");
63+
}
64+
writeConfigOptionsToConfYaml(confYaml.toPath());
65+
// adjust the test environment for the purposes of this test
66+
Map<String, String> map = new HashMap<>(System.getenv());
67+
map.put(ENV_FLINK_CONF_DIR, confFolder.getAbsolutePath());
68+
CommonTestUtils.setEnv(map);
69+
70+
historyPath = Files.createTempFile(tempFolder, "history", "").toFile().getPath();
71+
}
72+
73+
@AfterEach
74+
void after() {
75+
CommonTestUtils.setEnv(originalEnv);
76+
}
77+
78+
protected String createSqlFile(List<String> statements, String name) throws IOException {
79+
// create sql file
80+
File sqlFileFolder = Files.createTempDirectory(tempFolder, "sql-file").toFile();
81+
File sqlFile = new File(sqlFileFolder, name);
82+
if (!sqlFile.createNewFile()) {
83+
throw new IOException(String.format("Can't create testing %s.", name));
84+
}
85+
String sqlFilePath = sqlFile.getPath();
86+
Files.write(
87+
Paths.get(sqlFilePath),
88+
statements,
89+
StandardCharsets.UTF_8,
90+
StandardOpenOption.APPEND);
91+
return sqlFilePath;
92+
}
93+
94+
public static String runSqlClient(String[] args) throws Exception {
95+
return runSqlClient(args, "QUIT;\n", false);
96+
}
97+
98+
public static String runSqlClient(String[] args, String statements, boolean printInput)
99+
throws Exception {
100+
try (OutputStream out = new ByteArrayOutputStream();
101+
Terminal terminal =
102+
TerminalUtils.createDumbTerminal(
103+
new ByteArrayInputStream(
104+
statements.getBytes(StandardCharsets.UTF_8)),
105+
out)) {
106+
if (printInput) {
107+
// The default terminal has an empty size. Here increase the terminal to allow
108+
// the line reader print the input string.
109+
terminal.setSize(new Size(160, 80));
110+
}
111+
SqlClient.startClient(args, () -> terminal);
112+
return out.toString().replace("\r\n", System.lineSeparator());
113+
}
114+
}
115+
116+
protected void writeConfigOptionsToConfYaml(Path confYamlPath) throws IOException {
117+
// no-op for default.
118+
}
119+
}

‎flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointExtension.java

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030

3131
import java.net.InetAddress;
3232
import java.net.InetSocketAddress;
33+
import java.util.function.Consumer;
3334
import java.util.function.Supplier;
3435

3536
import static org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointTestUtils.getBaseConfig;
@@ -41,6 +42,8 @@ public class SqlGatewayRestEndpointExtension implements BeforeAllCallback, After
4142

4243
private final Supplier<SqlGatewayService> serviceSupplier;
4344

45+
private final Consumer<Configuration> flinkConfConsumer;
46+
4447
private SqlGatewayRestEndpoint sqlGatewayRestEndpoint;
4548
private SqlGatewayService sqlGatewayService;
4649
private String targetAddress;
@@ -59,13 +62,22 @@ public SqlGatewayService getSqlGatewayService() {
5962
}
6063

6164
public SqlGatewayRestEndpointExtension(Supplier<SqlGatewayService> serviceSupplier) {
65+
this(serviceSupplier, (conf) -> {});
66+
}
67+
68+
public SqlGatewayRestEndpointExtension(
69+
Supplier<SqlGatewayService> serviceSupplier,
70+
Consumer<Configuration> flinkConfConsumer) {
6271
this.serviceSupplier = serviceSupplier;
72+
this.flinkConfConsumer = flinkConfConsumer;
6373
}
6474

6575
@Override
6676
public void beforeAll(ExtensionContext context) {
6777
String address = InetAddress.getLoopbackAddress().getHostAddress();
68-
Configuration config = getBaseConfig(getFlinkConfig(address, address, "0"));
78+
Configuration flinkConfig = getFlinkConfig(address, address, "0");
79+
flinkConfConsumer.accept(flinkConfig);
80+
Configuration config = getBaseConfig(flinkConfig);
6981

7082
try {
7183
sqlGatewayService = serviceSupplier.get();

0 commit comments

Comments
 (0)