QPID-8083 [System Tests] [REST/HTTP] Add end to end message test Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/bd1f9098 Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/bd1f9098 Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/bd1f9098 Branch: refs/heads/master Commit: bd1f90985703a5dedc37935ad833bece1f650ccd Parents: 0bb035e Author: Keith Wall Authored: Mon Feb 5 07:09:00 2018 +0000 Committer: Keith Wall Committed: Mon Feb 5 09:14:53 2018 +0000 ---------------------------------------------------------------------- systests/qpid-systests-http-management/pom.xml | 56 +++++ .../apache/qpid/tests/http/HttpTestBase.java | 57 ++++- .../apache/qpid/tests/http/HttpTestHelper.java | 66 +++++- .../resources/config-http-management-tests.json | 174 ++++++++------ .../http/endtoend/message/MessageTest.java | 234 +++++++++++++++++++ 5 files changed, 504 insertions(+), 83 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/bd1f9098/systests/qpid-systests-http-management/pom.xml ---------------------------------------------------------------------- diff --git a/systests/qpid-systests-http-management/pom.xml b/systests/qpid-systests-http-management/pom.xml index 8ea2625..7135629 100644 --- a/systests/qpid-systests-http-management/pom.xml +++ b/systests/qpid-systests-http-management/pom.xml @@ -85,6 +85,11 @@ org.apache.qpid + qpid-systests-jms-core + + + + org.apache.qpid qpid-broker-plugins-derby-store true test @@ -117,7 +122,58 @@ + + + addQpidJmsClientIfNecessary + + + !enableAmqp0-x + + + + + org.apache.qpid + qpid-jms-client + + + + + + addJms11IfNecessary + + + enableAmqp0-x + + + + + org.apache.qpid + qpid-client + + + + + + + + ${basedir}/src/test/java + + **/*.java/ + + + + ${basedir}/src/test/resources + + + ${basedir}/../../test-profiles/test_resources/ssl + + *.jks + + + + + org.apache.maven.plugins http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/bd1f9098/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestBase.java ---------------------------------------------------------------------- diff --git a/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestBase.java b/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestBase.java index 32100fa..7230400 100644 --- a/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestBase.java +++ b/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestBase.java @@ -20,13 +20,24 @@ package org.apache.qpid.tests.http; +import java.net.InetSocketAddress; + +import javax.jms.Connection; +import javax.jms.JMSException; +import javax.naming.NamingException; + import org.junit.After; import org.junit.Before; import org.junit.Rule; -import org.junit.internal.runners.TestMethod; -import org.junit.rules.MethodRule; import org.junit.rules.TestName; +import org.apache.qpid.server.model.Protocol; +import org.apache.qpid.systests.AmqpManagementFacade; +import org.apache.qpid.systests.ConnectionBuilder; +import org.apache.qpid.systests.JmsProvider; +import org.apache.qpid.systests.QpidJmsClient0xProvider; +import org.apache.qpid.systests.QpidJmsClientProvider; +import org.apache.qpid.tests.utils.BrokerAdmin; import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase; public abstract class HttpTestBase extends BrokerAdminUsingTestBase @@ -36,6 +47,8 @@ public abstract class HttpTestBase extends BrokerAdminUsingTestBase private HttpTestHelper _helper; + private JmsProvider _jmsProvider; + @Before public void setUpTestBase() throws Exception { @@ -45,6 +58,18 @@ public abstract class HttpTestBase extends BrokerAdminUsingTestBase _helper = new HttpTestHelper(getBrokerAdmin(), config != null && config.useVirtualHostAsHost() ? getVirtualHost() : null); + + Protocol protocol = getProtocol(); + AmqpManagementFacade managementFacade = new AmqpManagementFacade(protocol); + if (protocol == Protocol.AMQP_1_0) + { + _jmsProvider = new QpidJmsClientProvider(managementFacade); + } + else + { + _jmsProvider = new QpidJmsClient0xProvider(); + } + } @After @@ -63,6 +88,21 @@ public abstract class HttpTestBase extends BrokerAdminUsingTestBase return _helper; } + protected Connection getConnection() throws JMSException, NamingException + { + return getConnectionBuilder().build(); + } + + protected ConnectionBuilder getConnectionBuilder() + { + InetSocketAddress brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.AMQP); + return _jmsProvider.getConnectionBuilder() + .setHost(brokerAddress.getHostName()) + .setPort(brokerAddress.getPort()) + .setUsername(getBrokerAdmin().getValidUsername()) + .setPassword(getBrokerAdmin().getValidPassword()); + } + private HttpRequestConfig getHttpRequestConfig() throws Exception { HttpRequestConfig config = getClass().getMethod(_testName.getMethodName(), new Class[]{}).getAnnotation(HttpRequestConfig.class); @@ -73,4 +113,17 @@ public abstract class HttpTestBase extends BrokerAdminUsingTestBase return config; } + + protected static long getReceiveTimeout() + { + return Long.getLong("qpid.test_receive_timeout", 1000L); + } + + protected static Protocol getProtocol() + { + return Protocol.valueOf("AMQP_" + System.getProperty("broker.version", "0-9-1") + .replace('-', '_') + .replace('.', '_')); + } + } http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/bd1f9098/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestHelper.java ---------------------------------------------------------------------- diff --git a/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestHelper.java b/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestHelper.java index 3fc6c78..5fcb62d 100644 --- a/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestHelper.java +++ b/systests/qpid-systests-http-management/src/main/java/org/apache/qpid/tests/http/HttpTestHelper.java @@ -32,11 +32,20 @@ import java.net.HttpURLConnection; import java.net.MalformedURLException; import java.net.URL; import java.net.URLEncoder; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.security.cert.X509Certificate; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import javax.net.ssl.HostnameVerifier; +import javax.net.ssl.HttpsURLConnection; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSession; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; import javax.xml.bind.DatatypeConverter; import com.fasterxml.jackson.core.type.TypeReference; @@ -69,7 +78,7 @@ public class HttpTestHelper private final int _connectTimeout = Integer.getInteger("qpid.resttest_connection_timeout", 30000); private String _acceptEncoding; - private boolean _useSsl = false; + private boolean _tls = false; public HttpTestHelper(final BrokerAdmin admin) { @@ -85,27 +94,27 @@ public class HttpTestHelper _requestHostName = requestHostName; } - public int getHttpPort() + public void setTls(final boolean tls) { - return _httpPort; + _tls = tls; } - private String getHostName() + private int getHttpPort() { - return "localhost"; + return _httpPort; } - private String getProtocol() + private String getHostName() { - return _useSsl ? "https" : "http"; + return "localhost"; } - public String getManagementURL() + private String getManagementURL() { - return getProtocol() + "://" + getHostName() + ":" + getHttpPort(); + return (_tls ? "https" : "http") + "://" + getHostName() + ":" + getHttpPort(); } - public URL getManagementURL(String path) throws MalformedURLException + private URL getManagementURL(String path) throws MalformedURLException { return new URL(getManagementURL() + path); } @@ -118,6 +127,42 @@ public class HttpTestHelper } URL url = getManagementURL(path); HttpURLConnection httpCon = (HttpURLConnection) url.openConnection(); + if (httpCon instanceof HttpsURLConnection) + { + HttpsURLConnection httpsCon = (HttpsURLConnection) httpCon; + try + { + SSLContext sslContext = SSLContext.getInstance("TLS"); + TrustManager[] trustAllCerts = new TrustManager[] { + new X509TrustManager() + { + public X509Certificate[] getAcceptedIssuers() + { + X509Certificate[] issuers = new X509Certificate[0]; + return issuers; + } + + @Override + public void checkClientTrusted(X509Certificate[] certs, String authType) + { + } + + @Override + public void checkServerTrusted(X509Certificate[] certs, String authType) + { + } + } + }; + + sslContext.init(null, trustAllCerts, null); + httpsCon.setSSLSocketFactory(sslContext.getSocketFactory()); + httpsCon.setHostnameVerifier((s, sslSession) -> true); + } + catch (KeyManagementException | NoSuchAlgorithmException e) + { + throw new RuntimeException(e); + } + } httpCon.setConnectTimeout(_connectTimeout); if (_requestHostName != null) { @@ -388,4 +433,5 @@ public class HttpTestHelper { _acceptEncoding = acceptEncoding; } + } http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/bd1f9098/systests/qpid-systests-http-management/src/main/resources/config-http-management-tests.json ---------------------------------------------------------------------- diff --git a/systests/qpid-systests-http-management/src/main/resources/config-http-management-tests.json b/systests/qpid-systests-http-management/src/main/resources/config-http-management-tests.json index 84310f9..356c4bb 100644 --- a/systests/qpid-systests-http-management/src/main/resources/config-http-management-tests.json +++ b/systests/qpid-systests-http-management/src/main/resources/config-http-management-tests.json @@ -19,75 +19,107 @@ * */ { - "name" : "${broker.name}", - "modelVersion" : "7.0", - "authenticationproviders" : [ { - "name" : "anon", - "type" : "Anonymous" - }, { - "name" : "plain", - "type" : "Plain", - "secureOnlyMechanisms" : [], - "users" : [ { - "name" : "admin", - "type" : "managed", - "password" : "admin" - }, { - "name" : "guest", - "type" : "managed", - "password" : "guest" - } ] - } ], - "ports" : [ { - "name" : "AMQP", - "type" : "AMQP", - "authenticationProvider" : "plain", - "port" : "0", - "virtualhostaliases" : [ { - "name" : "defaultAlias", - "type" : "defaultAlias" - }, { - "name" : "hostnameAlias", - "type" : "hostnameAlias" - }, { - "name" : "nameAlias", - "type" : "nameAlias" - } ] - }, { - "name" : "ANONYMOUS_AMQP", - "type" : "AMQP", - "authenticationProvider" : "anon", - "port" : "0", - "virtualhostaliases" : [ { - "name" : "defaultAlias", - "type" : "defaultAlias", - "durable" : true - }, { - "name" : "hostnameAlias", - "type" : "hostnameAlias", - "durable" : true - }, { - "name" : "nameAlias", - "type" : "nameAlias", - "durable" : true - } ] - }, { - "name": "HTTP", - "authenticationProvider": "plain", - "port": "0", - "protocols": [ - "HTTP" - ], - "virtualhostaliases" : [ { - "name" : "nameAlias", - "type" : "nameAlias" - } ] - - }], - "plugins" : [ { - "type" : "MANAGEMENT-HTTP", - "name" : "httpManagement", - "httpBasicAuthenticationEnabled" : true - } ], - "virtualhostnodes" : [] + "name": "${broker.name}", + "modelVersion": "7.0", + "keystores": [ + { + "name": "systestsKeyStore", + "storeUrl": "classpath:java_broker_keystore.jks", + "password": "password" + } + ], + "authenticationproviders": [ + { + "name": "anon", + "type": "Anonymous" + }, + { + "name": "plain", + "type": "Plain", + "secureOnlyMechanisms": [], + "users": [ + { + "name": "admin", + "type": "managed", + "password": "admin" + }, + { + "name": "guest", + "type": "managed", + "password": "guest" + } + ] + } + ], + "ports": [ + { + "name": "AMQP", + "type": "AMQP", + "authenticationProvider": "plain", + "port": "0", + "virtualhostaliases": [ + { + "name": "defaultAlias", + "type": "defaultAlias" + }, + { + "name": "hostnameAlias", + "type": "hostnameAlias" + }, + { + "name": "nameAlias", + "type": "nameAlias" + } + ] + }, + { + "name": "ANONYMOUS_AMQP", + "type": "AMQP", + "authenticationProvider": "anon", + "port": "0", + "virtualhostaliases": [ + { + "name": "defaultAlias", + "type": "defaultAlias", + "durable": true + }, + { + "name": "hostnameAlias", + "type": "hostnameAlias", + "durable": true + }, + { + "name": "nameAlias", + "type": "nameAlias", + "durable": true + } + ] + }, + { + "name": "HTTP", + "authenticationProvider": "plain", + "keyStore": "systestsKeyStore", + "port": "0", + "protocols": [ + "HTTP" + ], + "transports": [ + "TCP", "SSL" + ], + "virtualhostaliases": [ + { + "name": "nameAlias", + "type": "nameAlias" + } + ] + } + ], + "plugins": [ + { + "type": "MANAGEMENT-HTTP", + "name": "httpManagement", + "httpBasicAuthenticationEnabled": true + } + ], + "virtualhostnodes": [] } http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/bd1f9098/systests/qpid-systests-http-management/src/test/java/org/apache/qpid/tests/http/endtoend/message/MessageTest.java ---------------------------------------------------------------------- diff --git a/systests/qpid-systests-http-management/src/test/java/org/apache/qpid/tests/http/endtoend/message/MessageTest.java b/systests/qpid-systests-http-management/src/test/java/org/apache/qpid/tests/http/endtoend/message/MessageTest.java new file mode 100644 index 0000000..cb92341 --- /dev/null +++ b/systests/qpid-systests-http-management/src/test/java/org/apache/qpid/tests/http/endtoend/message/MessageTest.java @@ -0,0 +1,234 @@ +/* + * + * 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.qpid.tests.http.endtoend.message; + +import static javax.servlet.http.HttpServletResponse.SC_OK; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; +import static org.junit.Assert.assertThat; +import static org.junit.Assume.assumeThat; + +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.IntStream; + +import javax.jms.BytesMessage; +import javax.jms.Connection; +import javax.jms.MapMessage; +import javax.jms.Message; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.StreamMessage; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.google.common.io.ByteStreams; +import org.hamcrest.CoreMatchers; +import org.junit.Before; +import org.junit.Test; + +import org.apache.qpid.server.model.Protocol; +import org.apache.qpid.tests.http.HttpRequestConfig; +import org.apache.qpid.tests.http.HttpTestBase; + +@HttpRequestConfig +public class MessageTest extends HttpTestBase +{ + + private static final String QUEUE_NAME = "myqueue"; + private static final TypeReference>> LIST_MAP_TYPE_REF = + new TypeReference>>() {}; + private static final TypeReference> MAP_TYPE_REF = + new TypeReference>() {}; + private static final TypeReference> LIST_TYPE_REF = + new TypeReference>() {}; + + @Before + public void setUp() + { + getBrokerAdmin().createQueue(QUEUE_NAME); + } + + @Test + public void getJmsMessage() throws Exception + { + getHelper().setTls(true); + + final String messageProperty = "myProp"; + final String messagePropertyValue = "myValue"; + + Connection connection = getConnection(); + try + { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME)); + Message message = session.createMessage(); + message.setStringProperty(messageProperty, messagePropertyValue); + producer.send(message); + } + finally + { + connection.close(); + } + + List> messages = getHelper().postJson("queue/myqueue/getMessageInfo", + Collections.singletonMap("includeHeaders", Boolean.TRUE), + LIST_MAP_TYPE_REF, SC_OK); + assertThat(messages.size(), is(equalTo(1))); + + Map message = messages.get(0); + @SuppressWarnings("unchecked") + Map headers = (Map) message.get("headers"); + assertThat(headers.get(messageProperty), is(equalTo(messagePropertyValue))); + } + + @Test + public void getJmsMapMessage() throws Exception + { + getHelper().setTls(true); + final String mapKey = "key"; + final String mapKeyValue = "value"; + + Connection connection = getConnection(); + try + { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME)); + MapMessage message = session.createMapMessage(); + message.setString(mapKey, mapKeyValue); + producer.send(message); + } + finally + { + connection.close(); + } + + List> messages = getHelper().postJson("queue/myqueue/getMessageInfo", + Collections.singletonMap("includeHeaders", Boolean.TRUE), + LIST_MAP_TYPE_REF, SC_OK); + assertThat(messages.size(), is(equalTo(1))); + Map message = messages.get(0); + int messageId = (int) message.get("id"); + + Map contentParams = new HashMap<>(); + contentParams.put("messageId", messageId); + contentParams.put("returnJson", Boolean.TRUE); + + Map content = getHelper().postJson("queue/myqueue/getMessageContent", + contentParams, + MAP_TYPE_REF, SC_OK); + assertThat(content.size(), is(equalTo(1))); + assertThat(content.get(mapKey), is(equalTo(mapKeyValue))); + } + + @Test + public void getJmsStreamMessage() throws Exception + { + getHelper().setTls(true); + + Connection connection = getConnection(); + try + { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME)); + StreamMessage message = session.createStreamMessage(); + message.writeLong(Long.MAX_VALUE); + message.writeBoolean(true); + message.writeString("Hello World"); + producer.send(message); + } + finally + { + connection.close(); + } + + List> messages = getHelper().postJson("queue/myqueue/getMessageInfo", + Collections.singletonMap("includeHeaders", Boolean.TRUE), + LIST_MAP_TYPE_REF, SC_OK); + assertThat(messages.size(), is(equalTo(1))); + Map message = messages.get(0); + int messageId = (int) message.get("id"); + + Map contentParams = new HashMap<>(); + contentParams.put("messageId", messageId); + contentParams.put("returnJson", Boolean.TRUE); + + List content = getHelper().postJson("queue/myqueue/getMessageContent", + contentParams, + LIST_TYPE_REF, SC_OK); + assertThat(content.size(), is(equalTo(3))); + assertThat(content.get(0), is(equalTo(Long.MAX_VALUE))); + assertThat(content.get(1), is(equalTo(Boolean.TRUE))); + assertThat(content.get(2), is(equalTo("Hello World"))); + } + + @Test + public void getJmsBytesMessage() throws Exception + { + getHelper().setTls(true); + + final byte[] content = new byte[512]; + IntStream.range(0, content.length).forEachOrdered(i -> content[i] = (byte) (i % 256)); + + Connection connection = getConnection(); + try + { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME)); + BytesMessage message = session.createBytesMessage(); + message.writeBytes(content); + producer.send(message); + } + finally + { + connection.close(); + } + + List> messages = getHelper().postJson("queue/myqueue/getMessageInfo", + Collections.singletonMap("includeHeaders", Boolean.TRUE), + LIST_MAP_TYPE_REF, SC_OK); + assertThat(messages.size(), is(equalTo(1))); + Map message = messages.get(0); + int messageId = (int) message.get("id"); + + HttpURLConnection httpCon = getHelper().openManagementConnection(String.format( + "queue/myqueue/getMessageContent?messageId=%d", messageId), "GET"); + httpCon.connect(); + + byte[] receivedContent; + try(InputStream is = httpCon.getInputStream()) + { + receivedContent = ByteStreams.toByteArray(is); + } + + assumeThat("AMQP1.0 messages return the AMQP type", + getProtocol(), is(not(equalTo(Protocol.AMQP_1_0)))); + + assertThat(receivedContent, is(equalTo(content))); + } + + +} --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org For additional commands, e-mail: commits-help@qpid.apache.org