Skip to content

Commit 5b03c3c

Browse files
dalelanedalelane
committed
test: new test to cover the change in JMSReader stop behaviour
Signed-off-by: Dale Lane <dale.lane@uk.ibm.com>
1 parent 23d0a43 commit 5b03c3c

File tree

5 files changed

+207
-8
lines changed

5 files changed

+207
-8
lines changed

‎src/integration/java/com/ibm/eventstreams/connect/mqsource/MQSourceTaskAuthIT.java

Lines changed: 49 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import static org.junit.Assert.assertArrayEquals;
1919
import static org.junit.Assert.assertEquals;
2020
import static org.junit.Assert.assertNull;
21+
import static org.junit.Assert.assertTrue;
2122

2223
import java.util.Arrays;
2324
import java.util.HashMap;
@@ -33,6 +34,8 @@
3334
import org.testcontainers.containers.GenericContainer;
3435
import org.testcontainers.containers.output.WaitingConsumer;
3536

37+
import com.ibm.eventstreams.connect.mqsource.utils.MQQueueManagerAttrs;
38+
import com.ibm.eventstreams.connect.mqsource.utils.SourceTaskStopper;
3639
import com.ibm.mq.MQException;
3740
import com.ibm.mq.MQMessage;
3841
import com.ibm.mq.MQQueue;
@@ -45,21 +48,19 @@ public class MQSourceTaskAuthIT {
4548
private static final String QUEUE_NAME = "DEV.QUEUE.2";
4649
private static final String CHANNEL_NAME = "DEV.APP.SVRCONN";
4750
private static final String APP_PASSWORD = "MySuperSecretPassword";
51+
private static final String ADMIN_PASSWORD = "MyAdminPassword";
4852

4953

5054
@ClassRule
5155
public static GenericContainer<?> MQ_CONTAINER = new GenericContainer<>("icr.io/ibm-messaging/mq:latest")
5256
.withEnv("LICENSE", "accept")
5357
.withEnv("MQ_QMGR_NAME", QMGR_NAME)
54-
.withEnv("MQ_ENABLE_EMBEDDED_WEB_SERVER", "false")
5558
.withEnv("MQ_APP_PASSWORD", APP_PASSWORD)
56-
.withExposedPorts(1414);
59+
.withEnv("MQ_ADMIN_PASSWORD", ADMIN_PASSWORD)
60+
.withExposedPorts(1414, 9443);
5761

5862

59-
@Test
60-
public void testAuthenticatedQueueManager() throws Exception {
61-
waitForQueueManagerStartup();
62-
63+
private Map<String, String> getConnectorProps() {
6364
Map<String, String> connectorProps = new HashMap<>();
6465
connectorProps.put("mq.queue.manager", QMGR_NAME);
6566
connectorProps.put("mq.connection.mode", "client");
@@ -71,9 +72,15 @@ public void testAuthenticatedQueueManager() throws Exception {
7172
connectorProps.put("mq.password", APP_PASSWORD);
7273
connectorProps.put("mq.message.body.jms", "false");
7374
connectorProps.put("mq.record.builder", "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder");
75+
return connectorProps;
76+
}
77+
78+
@Test
79+
public void testAuthenticatedQueueManager() throws Exception {
80+
waitForQueueManagerStartup();
7481

7582
MQSourceTask newConnectTask = new MQSourceTask();
76-
newConnectTask.start(connectorProps);
83+
newConnectTask.start(getConnectorProps());
7784

7885
MQMessage message1 = new MQMessage();
7986
message1.writeString("hello");
@@ -98,6 +105,41 @@ public void testAuthenticatedQueueManager() throws Exception {
98105
}
99106

100107

108+
109+
@Test
110+
public void verifyJmsConnClosed() throws Exception {
111+
112+
int restApiPortNumber = MQ_CONTAINER.getMappedPort(9443);
113+
114+
// count number of connections to the qmgr at the start
115+
int numQmgrConnectionsBefore = MQQueueManagerAttrs.getNumConnections(QMGR_NAME, restApiPortNumber, ADMIN_PASSWORD);
116+
117+
// start the source connector so that it connects to the qmgr
118+
MQSourceTask connectTask = new MQSourceTask();
119+
connectTask.start(getConnectorProps());
120+
121+
// count number of connections to the qmgr now - it should have increased
122+
int numQmgrConnectionsDuring = MQQueueManagerAttrs.getNumConnections(QMGR_NAME, restApiPortNumber, ADMIN_PASSWORD);
123+
124+
// stop the source connector so it disconnects from the qmgr
125+
connectTask.stop();
126+
127+
// count number of connections to the qmgr now - it should have decreased
128+
int numQmgrConnectionsAfter = MQQueueManagerAttrs.getNumConnections(QMGR_NAME, restApiPortNumber, ADMIN_PASSWORD);
129+
130+
// verify number of connections changed as expected
131+
assertTrue("connections should have increased after starting the source task",
132+
numQmgrConnectionsDuring > numQmgrConnectionsBefore);
133+
assertTrue("connections should have decreased after calling stop()",
134+
numQmgrConnectionsAfter < numQmgrConnectionsDuring);
135+
136+
// cleanup
137+
SourceTaskStopper stopper = new SourceTaskStopper(connectTask);
138+
stopper.run();
139+
}
140+
141+
142+
101143
private void waitForQueueManagerStartup() throws TimeoutException {
102144
WaitingConsumer logConsumer = new WaitingConsumer();
103145
MQ_CONTAINER.followOutput(logConsumer);

‎src/integration/java/com/ibm/eventstreams/connect/mqsource/MQSourceTaskIT.java

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,9 @@
3434
import org.junit.After;
3535
import org.junit.Test;
3636

37+
import com.ibm.eventstreams.connect.mqsource.utils.SourceTaskStopper;
38+
39+
3740
public class MQSourceTaskIT extends AbstractJMSContextIT {
3841

3942
private MQSourceTask connectTask = null;

‎src/integration/java/com/ibm/eventstreams/connect/mqsource/utils/JsonRestApi.java

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
/**
2+
* Copyright 2022 IBM Corporation
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.ibm.eventstreams.connect.mqsource.utils;
17+
18+
import java.io.BufferedReader;
19+
import java.io.IOException;
20+
import java.io.InputStream;
21+
import java.io.InputStreamReader;
22+
import java.io.OutputStream;
23+
import java.io.Reader;
24+
import java.net.URL;
25+
import java.nio.charset.Charset;
26+
import java.security.KeyManagementException;
27+
import java.security.NoSuchAlgorithmException;
28+
import java.security.cert.X509Certificate;
29+
import java.util.Base64;
30+
31+
import javax.net.ssl.HostnameVerifier;
32+
import javax.net.ssl.HttpsURLConnection;
33+
import javax.net.ssl.SSLContext;
34+
import javax.net.ssl.SSLSession;
35+
import javax.net.ssl.SSLSocketFactory;
36+
import javax.net.ssl.TrustManager;
37+
import javax.net.ssl.X509TrustManager;
38+
39+
import org.json.JSONException;
40+
import org.json.JSONObject;
41+
42+
public class JsonRestApi {
43+
44+
public static JSONObject jsonPost(String url, String username, String password, String payload) throws IOException, KeyManagementException, NoSuchAlgorithmException, JSONException {
45+
URL urlObj = new URL(url);
46+
HttpsURLConnection urlConnection = (HttpsURLConnection) urlObj.openConnection();
47+
urlConnection.setHostnameVerifier(new IgnoreCertVerifier());
48+
urlConnection.setSSLSocketFactory(getTrustAllCertsFactory());
49+
urlConnection.setRequestProperty("Authorization", getAuthHeader(username, password));
50+
urlConnection.setRequestProperty("Content-Type", "application/json");
51+
urlConnection.setRequestProperty("ibm-mq-rest-csrf-token", "junit");
52+
urlConnection.setDoOutput(true);
53+
54+
try(OutputStream os = urlConnection.getOutputStream()) {
55+
byte[] input = payload.getBytes("utf-8");
56+
os.write(input, 0, input.length);
57+
}
58+
59+
try (InputStream input = urlConnection.getInputStream()){
60+
BufferedReader re = new BufferedReader(new InputStreamReader(input, Charset.forName("utf-8")));
61+
return new JSONObject(read(re));
62+
}
63+
}
64+
65+
66+
private static String read(Reader re) throws IOException {
67+
StringBuilder str = new StringBuilder();
68+
int ch;
69+
do {
70+
ch = re.read();
71+
str.append((char) ch);
72+
} while (ch != -1);
73+
return str.toString();
74+
}
75+
76+
77+
private static String getAuthHeader(String username, String password) {
78+
String userpass = username + ":" + password;
79+
String basicAuth = "Basic " + new String(Base64.getEncoder().encode(userpass.getBytes()));
80+
return basicAuth;
81+
}
82+
83+
private static class IgnoreCertVerifier implements HostnameVerifier {
84+
@Override
85+
public boolean verify(String host, SSLSession session) {
86+
return true;
87+
}
88+
}
89+
90+
private static SSLSocketFactory getTrustAllCertsFactory() throws NoSuchAlgorithmException, KeyManagementException {
91+
TrustManager[] trustAllCerts = new TrustManager[] {
92+
new X509TrustManager() {
93+
public java.security.cert.X509Certificate[] getAcceptedIssuers() { return null; }
94+
public void checkClientTrusted(X509Certificate[] certs, String authType) { }
95+
public void checkServerTrusted(X509Certificate[] certs, String authType) { }
96+
}
97+
};
98+
SSLContext sc = SSLContext.getInstance("SSL");
99+
sc.init(null, trustAllCerts, new java.security.SecureRandom());
100+
return sc.getSocketFactory();
101+
}
102+
}

‎src/integration/java/com/ibm/eventstreams/connect/mqsource/utils/MQQueueManagerAttrs.java

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/**
2+
* Copyright 2022 IBM Corporation
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.ibm.eventstreams.connect.mqsource.utils;
17+
18+
import java.io.IOException;
19+
import java.security.KeyManagementException;
20+
import java.security.NoSuchAlgorithmException;
21+
22+
import org.json.JSONException;
23+
import org.json.JSONObject;
24+
25+
public class MQQueueManagerAttrs {
26+
27+
private static final String REQ_GET_SVRCONNS = "{"
28+
+ " \"type\": \"runCommand\","
29+
+ " \"parameters\": {"
30+
+ " \"command\": \"display conn(*) where (channel EQ 'DEV.APP.SVRCONN')\""
31+
+ " }"
32+
+ "}";
33+
34+
35+
public static int getNumConnections(String qmgrname, int portnum, String password) throws KeyManagementException, NoSuchAlgorithmException, IOException, JSONException {
36+
String url = "https://localhost:" + portnum + "/ibmmq/rest/v2/admin/action/qmgr/" + qmgrname + "/mqsc";
37+
JSONObject connectionInfo = JsonRestApi.jsonPost(url, "admin", password, REQ_GET_SVRCONNS);
38+
39+
int completionCode = connectionInfo.getInt("overallCompletionCode");
40+
int reasonCode = connectionInfo.getInt("overallReasonCode");
41+
42+
if (completionCode == 2 && reasonCode == 3008) {
43+
return 0;
44+
}
45+
else if (completionCode == 0 && reasonCode == 0) {
46+
return connectionInfo.getJSONArray("commandResponse").length();
47+
}
48+
else {
49+
return -1;
50+
}
51+
}
52+
}

‎src/integration/java/com/ibm/eventstreams/connect/mqsource/SourceTaskStopper.java renamed to ‎src/integration/java/com/ibm/eventstreams/connect/mqsource/utils/SourceTaskStopper.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
* See the License for the specific language governing permissions and
1414
* limitations under the License.
1515
*/
16-
package com.ibm.eventstreams.connect.mqsource;
16+
package com.ibm.eventstreams.connect.mqsource.utils;
1717

1818
import java.util.concurrent.ExecutorService;
1919
import java.util.concurrent.Executors;

0 commit comments

Comments
 (0)