tests: initial integration test for DefaultMessageBuilder

Signed-off-by: Dale Lane <dale.lane@uk.ibm.com>

Author: dalelane

src/integration/java/com/ibm/eventstreams/connect/mqsink/AbstractJMSContextIT.java

@@ -0,0 +1,72 @@
+package com.ibm.eventstreams.connect.mqsink;
+
+import java.util.concurrent.TimeoutException;
+
+import javax.jms.JMSContext;
+
+import org.junit.ClassRule;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.WaitingConsumer;
+
+import com.ibm.mq.jms.MQConnectionFactory;
+import com.ibm.msg.client.wmq.WMQConstants;
+
+
+/**
+ * Helper class for integration tests that have a dependency on JMSContext.
+ * 
+ *  It starts a queue manager in a test container, and uses it to create
+ *  a JMSContext instance, that can be used in tests.
+ */
+public abstract class AbstractJMSContextIT {
+
+    private static final String QMGR_NAME = "MYQMGR";
+    
+    @ClassRule
+    public static GenericContainer<?> MQ_CONTAINER = new GenericContainer<>("icr.io/ibm-messaging/mq:latest")
+        .withEnv("LICENSE", "accept")
+        .withEnv("MQ_QMGR_NAME", QMGR_NAME)
+        .withEnv("MQ_ENABLE_EMBEDDED_WEB_SERVER", "false")
+        .withExposedPorts(1414);
+    
+    private JMSContext jmsContext;
+    
+    
+    /**
+     * Returns a JMS context pointing at a developer queue manager running in a 
+     * test container. 
+     */
+    public JMSContext getJmsContext() throws Exception {
+        if (jmsContext == null) {
+            waitForQueueManagerStartup();
+            
+            MQConnectionFactory mqcf = new MQConnectionFactory();
+            mqcf.setTransportType(WMQConstants.WMQ_CM_CLIENT);
+            mqcf.setChannel("DEV.APP.SVRCONN");
+            mqcf.setQueueManager(QMGR_NAME);
+            mqcf.setConnectionNameList("localhost(" + getMQPort().toString() + ")");
+         
+            jmsContext = mqcf.createContext();
+        }
+        
+        return jmsContext;
+    }
+    
+    
+    /**
+     * Gets the host port that has been mapped to the default MQ 1414 port in the test container.
+     */
+    private Integer getMQPort() {
+        return MQ_CONTAINER.getMappedPort(1414);                
+    }
+
+    /** 
+     * Waits until we see a log line in the queue manager test container that indicates
+     *  the queue manager is ready.
+     */
+    private void waitForQueueManagerStartup() throws TimeoutException {
+        WaitingConsumer logConsumer = new WaitingConsumer();
+        MQ_CONTAINER.followOutput(logConsumer);
+        logConsumer.waitUntil(logline -> logline.getUtf8String().contains("AMQ5975I"));
+    }    
+}

src/integration/java/com/ibm/eventstreams/connect/mqsink/DefaultMessageBuilderIT.java

@@ -0,0 +1,124 @@
+package com.ibm.eventstreams.connect.mqsink;
+
+import static org.junit.Assert.assertEquals;
+
+import java.nio.ByteBuffer;
+
+import javax.jms.BytesMessage;
+import javax.jms.Message;
+import javax.jms.TextMessage;
+
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.ibm.eventstreams.connect.mqsink.builders.DefaultMessageBuilder;
+import com.ibm.eventstreams.connect.mqsink.builders.MessageBuilder;
+
+public class DefaultMessageBuilderIT extends AbstractJMSContextIT {
+
+    private MessageBuilder builder;
+    
+    @Before
+    public void prepareMessageBuilder() {
+        builder = new DefaultMessageBuilder();
+    }
+    
+    private SinkRecord generateSinkRecord(Schema valueSchema, Object value) {
+        final String TOPIC = "TOPIC.NAME";
+        final int PARTITION = 0;
+        final long OFFSET = 0;
+        final Schema KEY_SCHEMA = Schema.STRING_SCHEMA;
+        final String KEY = "mykey";
+        
+        return new SinkRecord(TOPIC, PARTITION, 
+                              KEY_SCHEMA, KEY,
+                              valueSchema, value, 
+                              OFFSET);
+    }
+    
+    
+    @Test
+    public void buildEmptyMessageWithoutSchema() throws Exception {
+        createAndVerifyEmptyMessage(null);
+    }
+    @Test
+    public void buildEmptyMessageWithSchema() throws Exception {
+        createAndVerifyEmptyMessage(Schema.STRING_SCHEMA);
+    }    
+    @Test
+    public void buildTextMessageWithoutSchema() throws Exception {
+        createAndVerifyStringMessage(null, "Hello World");
+    }
+    @Test
+    public void buildTextMessageWithSchema() throws Exception {
+        createAndVerifyStringMessage(Schema.STRING_SCHEMA, "Hello World with a schema");
+    }
+    @Test
+    public void buildIntMessageWithoutSchema() throws Exception {
+        createAndVerifyIntegerMessage(null, 1234);
+    }
+    @Test
+    public void buildIntMessageWithSchema() throws Exception {
+        createAndVerifyIntegerMessage(Schema.INT32_SCHEMA, 1234);
+    }
+    @Test
+    public void buildByteArrayMessageWithoutSchema() throws Exception {
+        String TEST_MESSAGE = "This is a test";
+        createAndVerifyByteMessage(null, TEST_MESSAGE.getBytes(), TEST_MESSAGE);
+    }
+    @Test
+    public void buildByteArrayMessageWithSchema() throws Exception {
+        String TEST_MESSAGE = "This is another test";
+        createAndVerifyByteMessage(Schema.BYTES_SCHEMA, TEST_MESSAGE.getBytes(), TEST_MESSAGE);
+    }
+    @Test
+    public void buildByteBufferMessageWithoutSchema() throws Exception {
+        String TEST_MESSAGE = "This is also a test!";
+        byte[] payload = TEST_MESSAGE.getBytes();
+        ByteBuffer value = ByteBuffer.allocate(payload.length);
+        value.put(payload);
+        createAndVerifyByteMessage(null, value, TEST_MESSAGE);
+    }
+    @Test
+    public void buildByteBufferMessageWithSchema() throws Exception {
+        String TEST_MESSAGE = "This is a bytebuffer test";
+        byte[] payload = TEST_MESSAGE.getBytes();
+        ByteBuffer value = ByteBuffer.allocate(payload.length);
+        value.put(payload);
+        createAndVerifyByteMessage(Schema.BYTES_SCHEMA, value, TEST_MESSAGE);
+    }
+    
+
+    private void createAndVerifyEmptyMessage(Schema valueSchema) throws Exception {
+        Message message = builder.fromSinkRecord(getJmsContext(), generateSinkRecord(valueSchema, null));
+        assertEquals(null, message.getBody(String.class));
+    }
+    
+    private void createAndVerifyStringMessage(Schema valueSchema, String value) throws Exception {
+        Message message = builder.fromSinkRecord(getJmsContext(), generateSinkRecord(valueSchema, value));
+        assertEquals(value, message.getBody(String.class));
+        
+        TextMessage textmessage = (TextMessage) message;
+        assertEquals(value, textmessage.getText());
+    }
+    
+    private void createAndVerifyIntegerMessage(Schema valueSchema, Integer value) throws Exception {
+        Message message = builder.fromSinkRecord(getJmsContext(), generateSinkRecord(valueSchema, value));
+        assertEquals(value, new Integer(message.getBody(String.class)));
+    }
+    
+    private void createAndVerifyByteMessage(Schema valueSchema, Object value, String valueAsString) throws Exception {
+        Message message = builder.fromSinkRecord(getJmsContext(), generateSinkRecord(valueSchema, value));
+        
+        BytesMessage byteMessage = (BytesMessage) message;
+        byteMessage.reset();
+
+        byte[] byteData = null;
+        byteData = new byte[(int) byteMessage.getBodyLength()];
+        byteMessage.readBytes(byteData);
+        String stringMessage =  new String(byteData);
+        assertEquals(valueAsString, stringMessage);
+    }
+}