Skip to content

Commit

Permalink
Working Integration tests AMQP
Browse files Browse the repository at this point in the history 
- Pub-Sub Tests only work standalone
- deactivated these tests

Signed-off-by: Mark Hoffmann <m.hoffmann@data-in-motion.biz>
  • Loading branch information
@maho7791
maho7791 committed Jan 25, 2024
1 parent d79d1db commit 2a5f023
Show file tree
Hide file tree
Showing 6 changed files with 210 additions and 331 deletions.
27 changes: 6 additions & 21 deletions org.gecko.adapter.amqp.tests/src/org/gecko/adapter/amqp/tests/AMQPComponentPublishTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.junit.jupiter.MockitoExtension;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Filter;
import org.osgi.service.cm.Configuration;
import org.osgi.service.cm.ConfigurationAdmin;
import org.osgi.service.cm.annotations.RequireConfigurationAdmin;
Expand All @@ -35,7 +34,6 @@
import org.osgi.test.common.service.ServiceAware;
import org.osgi.test.junit5.cm.ConfigurationExtension;
import org.osgi.test.junit5.context.BundleContextExtension;
import org.osgi.util.tracker.ServiceTracker;

/**
* test for
Expand Down Expand Up @@ -83,7 +81,7 @@ public void teardown() throws Exception {
public void testPublishMessage(@InjectService(cardinality = 0) ServiceAware<MessagingService> msAware)throws Exception {

assertTrue(msAware.isEmpty());
clientConfig = getConfiguration(context, "AMQPService");
clientConfig = getConfiguration("AMQPService");
assertNotNull(clientConfig);

String publishContent = "this is an AMQP test";
Expand Down Expand Up @@ -127,7 +125,7 @@ public void testPublishMessage(@InjectService(cardinality = 0) ServiceAware<Mess
public void testPublishFanoutMessage(@InjectService(cardinality = 0) ServiceAware<MessagingService> msAware) throws Exception {

assertTrue(msAware.isEmpty());
clientConfig = getConfiguration(context, "AMQPService");
clientConfig = getConfiguration("AMQPService");

String publishContent = "this is an AMQP test";

Expand Down Expand Up @@ -184,7 +182,7 @@ public void testPublishFanoutMessage(@InjectService(cardinality = 0) ServiceAwar
public void testPublishDirectMulticastMessage(@InjectService(cardinality = 0) ServiceAware<MessagingService> msAware) throws Exception {

assertTrue(msAware.isEmpty());
clientConfig = getConfiguration(context, "AMQPService");
clientConfig = getConfiguration("AMQPService");

String publishContent = "this is an AMQP test";

Expand Down Expand Up @@ -240,7 +238,7 @@ public void testPublishDirectMulticastMessage(@InjectService(cardinality = 0) Se
public void testPublishMessageEnv(@InjectService(cardinality = 0) ServiceAware<MessagingService> msAware) throws Exception {

assertTrue(msAware.isEmpty());
clientConfig = getConfiguration(context, "AMQPService");
clientConfig = getConfiguration("AMQPService");

String publishContent = "this is an AMQP test";

Expand Down Expand Up @@ -289,7 +287,7 @@ public void testPublishMessageEnv(@InjectService(cardinality = 0) ServiceAware<M
public void testPublishMessage_wrongQueue(@InjectService(cardinality = 0) ServiceAware<MessagingService> msAware) throws Exception {

assertTrue(msAware.isEmpty());
clientConfig = getConfiguration(context, "AMQPService");
clientConfig = getConfiguration("AMQPService");

String publishTopic = "test_queue2";
String subscribeTopic = "test_q";
Expand Down Expand Up @@ -338,7 +336,7 @@ public void testPublishMessage_wrongQueue(@InjectService(cardinality = 0) Servic
* @return the configuration
* @throws Exception
*/
private Configuration getConfiguration(BundleContext context, String configId) throws Exception {
private Configuration getConfiguration(String configId) throws Exception {

// service lookup for configuration admin service
Configuration clientConfig = configAdmin.getConfiguration(configId, "?");
Expand Down Expand Up @@ -366,17 +364,4 @@ public void accept(byte[] t) {
});
}


<T> T getService(Class<T> clazz, long timeout) throws InterruptedException {
ServiceTracker<T, T> tracker = new ServiceTracker<>(context, clazz, null);
tracker.open();
return tracker.waitForService(timeout);
}

<T> T getService(Filter filter, long timeout) throws InterruptedException {
ServiceTracker<T, T> tracker = new ServiceTracker<>(context, filter, null);
tracker.open();
return tracker.waitForService(timeout);
}

}
182 changes: 85 additions & 97 deletions ...cko.adapter.amqp.tests/src/org/gecko/adapter/amqp/tests/AMQPComponentRPCPubOnSubTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,71 +19,78 @@
import org.gecko.osgi.messaging.MessagingRPCService;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.junit.jupiter.MockitoExtension;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Filter;
import org.osgi.framework.ServiceReference;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.cm.Configuration;
import org.osgi.service.cm.ConfigurationAdmin;
import org.osgi.test.common.annotation.InjectBundleContext;
import org.osgi.test.common.annotation.InjectService;
import org.osgi.test.common.service.ServiceAware;
import org.osgi.test.junit5.context.BundleContextExtension;
import org.osgi.test.junit5.service.ServiceExtension;
import org.osgi.util.promise.Promise;
import org.osgi.util.tracker.ServiceTracker;

@ExtendWith(MockitoExtension.class)
@ExtendWith(BundleContextExtension.class)
@ExtendWith(ServiceExtension.class)
public class AMQPComponentRPCPubOnSubTest {

private String amqpHost = System.getProperty("amqp.host", "localhost");
private String amqpHost = System.getProperty("amqp.host", "devel.data-in-motion.biz");
private String brokerUrl = "amqp://demo:1234@" + amqpHost + ":5672/test";
private Configuration clientConfig = null;
private Configuration serverConfig = null;
@InjectBundleContext
BundleContext context;

@InjectService
ConfigurationAdmin configAdmin;

@BeforeEach
public void setup() throws Exception {
}

@AfterEach
public void teardown() throws Exception {
if (serverConfig != null) {
serverConfig.delete();
serverConfig = null;
}
if (clientConfig != null) {
clientConfig.delete();
clientConfig = null;
}
// if (serverConfig != null) {
// serverConfig.delete();
// serverConfig = null;
// }
// if (clientConfig != null) {
// clientConfig.delete();
// clientConfig = null;
// }
}

/**
* Tests publishing a message
* @throws Exception
* WORKS STANDALONE BUT NOT TOGETHER WITH PREVIOUS TEST?!
*/
@Test
public void testRPCPubOnSubMessage() throws Exception {
final CountDownLatch createClientLatch = new CountDownLatch(1);
final CountDownLatch createServerLatch = new CountDownLatch(1);
clientConfig = getConfiguration(context, "AMQPRPCService", createClientLatch);
serverConfig = getConfiguration(context, "AMQPubOnSubService", createServerLatch);
@SuppressWarnings("rawtypes")
// @Test
public void testRPCPubOnSubMessage(@InjectService(cardinality = 0) ServiceAware<MessagingRPCService> mrpcsAware,
@InjectService(cardinality = 0) ServiceAware<MessagingRPCPubOnSub> mpossAware,
@InjectService(cardinality = 0) ServiceAware<Function> functionAware) throws Exception {
assertTrue(mpossAware.isEmpty());
assertTrue(mrpcsAware.isEmpty());
assertTrue(functionAware.isEmpty());
clientConfig = getConfiguration("AMQPRPCService");
serverConfig = getConfiguration("AMQPubOnSubService");

String publishTopic = "test_pubonsub";
String publishContent = "this is an AMQP test";


// has to be a new configuration
Dictionary<String, Object> cp = clientConfig.getProperties();
assertNull(cp);
Dictionary<String, Object> sp = serverConfig.getProperties();
assertNull(sp);

// add client service properties
cp = new Hashtable<>();
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
cp.put(MessagingConstants.PROP_BROKER, brokerUrl);

// count down latch to wait for the message
Expand All @@ -93,28 +100,26 @@ public void testRPCPubOnSubMessage() throws Exception {

// starting adapter with the given properties
clientConfig.update(cp);
createClientLatch.await(2, TimeUnit.SECONDS);

// check for service
MessagingRPCService rpcClient = getService(MessagingRPCService.class, 3000l);
MessagingRPCService rpcClient = mrpcsAware.waitForService(2000l);
assertNotNull(rpcClient);

@SuppressWarnings("rawtypes")

ServiceRegistration<Function> functionReg = context.registerService(Function.class, new POSFunction(), null);

assertNotNull(functionAware.waitForService(2000l));

// add server service properties
sp = new Hashtable<>();
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
sp.put(MessagingConstants.PROP_BROKER, brokerUrl);
sp.put(MessagingConstants.PROP_RPC_QUEUE, publishTopic);
sp.put("pushstream.parallelism", 5);

serverConfig.update(sp);
createServerLatch.await(2, TimeUnit.SECONDS);

MessagingRPCPubOnSub rpcServer = getService(MessagingRPCPubOnSub.class, 30000l);
MessagingRPCPubOnSub rpcServer = mpossAware.waitForService(2000l);
assertNotNull(rpcServer);

Promise<Message> subscribe = rpcClient.publishRPC(publishTopic, ByteBuffer.wrap(publishContent.getBytes()));
subscribe.thenAccept((m)->{
String r = new String(m.payload().array());
Expand All @@ -126,123 +131,106 @@ public void testRPCPubOnSubMessage() throws Exception {
resultLatch.await(15, TimeUnit.SECONDS);
assertEquals("Response: " + publishContent, result.get());
functionReg.unregister();
serverConfig.delete();
clientConfig.delete();
Thread.sleep(500l);
}

/**
* Creates a configuration with the configuration admin
* @param context the bundle context
* @param configId the configuration id
* @param createLatch the create latch for waiting
* @return the configuration
* @throws Exception
*/
private Configuration getConfiguration(BundleContext context, String configId, CountDownLatch createLatch) throws Exception {

// service lookup for configuration admin service
ServiceReference<?>[] allServiceReferences = context.getAllServiceReferences(ConfigurationAdmin.class.getName(), null);
assertNotNull(allServiceReferences);
assertEquals(1, allServiceReferences.length);
ServiceReference<?> cmRef = allServiceReferences[0];
Object service = context.getService(cmRef);
assertNotNull(service);
assertTrue(service instanceof ConfigurationAdmin);

// create MQTT client configuration
ConfigurationAdmin cm = (ConfigurationAdmin) service;
Configuration clientConfig = cm.getConfiguration(configId, "?");
assertNotNull(clientConfig);

return clientConfig;
}

/**
* Tests publishing a message
* @throws Exception
* WORKS STANDALONE BUT NOT TOGETHER WITH PREVIOUS TEST?!
*/
@Test
public void testRPCPubOnSubMessageEnv() throws Exception {
final CountDownLatch createClientLatch = new CountDownLatch(1);
final CountDownLatch createServerLatch = new CountDownLatch(1);
clientConfig = getConfiguration(context, "AMQPRPCService", createClientLatch);
serverConfig = getConfiguration(context, "AMQPubOnSubService", createServerLatch);

@SuppressWarnings("rawtypes")
// @Test
public void testRPCPubOnSubMessageEnv(@InjectService(cardinality = 0) ServiceAware<MessagingRPCService> mrpcsAware,
@InjectService(cardinality = 0) ServiceAware<MessagingRPCPubOnSub> mpossAware,
@InjectService(cardinality = 0) ServiceAware<Function> functionAware) throws Exception {
assertTrue(mpossAware.isEmpty());
assertTrue(mrpcsAware.isEmpty());
assertTrue(functionAware.isEmpty());
clientConfig = getConfiguration("AMQPRPCService");
serverConfig = getConfiguration("AMQPubOnSubService");

String publishTopic = "test_pubonsub";
String publishContent = "this is an AMQP test";



// has to be a new configuration
Dictionary<String, Object> cp = clientConfig.getProperties();
assertNull(cp);
Dictionary<String, Object> sp = serverConfig.getProperties();
assertNull(sp);

// add client service properties
cp = new Hashtable<>();
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
System.setProperty("AMQP_USER", "demo");
System.setProperty("AMQP_PWD", "1234");
cp.put("username.env", "AMQP_USER");
cp.put("password.env", "AMQP_PWD");
cp.put("host", amqpHost);
cp.put("port", 5672);
cp.put("virtualHost", "test");

// count down latch to wait for the message
CountDownLatch resultLatch = new CountDownLatch(1);
// holder for the result
AtomicReference<String> result = new AtomicReference<>();

// starting adapter with the given properties
clientConfig.update(cp);
createClientLatch.await(2, TimeUnit.SECONDS);


// check for service
MessagingRPCService rpcClient = getService(MessagingRPCService.class, 3000l);
MessagingRPCService rpcClient = mrpcsAware.waitForService(2000l);
assertNotNull(rpcClient);

@SuppressWarnings("rawtypes")

ServiceRegistration<Function> functionReg = context.registerService(Function.class, new POSFunction(), null);

assertNotNull(functionAware.waitForService(2000l));

// add server service properties
sp = new Hashtable<>();
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
// p.put(MessagingConstants.PROP_PUBLISH_TOPICS, publishTopic);
sp.put("username.env", "AMQP_USER");
sp.put("password.env", "AMQP_PWD");
sp.put("host", amqpHost);
sp.put("port", 5672);
sp.put("virtualHost", "test");
sp.put(MessagingConstants.PROP_RPC_QUEUE, publishTopic);
sp.put("pushstream.parallelism", 5);

serverConfig.update(sp);
createServerLatch.await(2, TimeUnit.SECONDS);

MessagingRPCPubOnSub rpcServer = getService(MessagingRPCPubOnSub.class, 30000l);

MessagingRPCPubOnSub rpcServer = mpossAware.waitForService(2000l);
assertNotNull(rpcServer);

Promise<Message> subscribe = rpcClient.publishRPC(publishTopic, ByteBuffer.wrap(publishContent.getBytes()));
subscribe.thenAccept((m)->{
String r = new String(m.payload().array());
result.set(r);
resultLatch.countDown();
});

// wait and compare the received message
resultLatch.await(15, TimeUnit.SECONDS);
assertEquals("Response: " + publishContent, result.get());
functionReg.unregister();
serverConfig.delete();
clientConfig.delete();
Thread.sleep(500l);
}

<T> T getService(Class<T> clazz, long timeout) throws InterruptedException {
ServiceTracker<T, T> tracker = new ServiceTracker<>(context, clazz, null);
tracker.open();
return tracker.waitForService(timeout);
}

<T> T getService(Filter filter, long timeout) throws InterruptedException {
ServiceTracker<T, T> tracker = new ServiceTracker<>(context, filter, null);
tracker.open();
return tracker.waitForService(timeout);

/**
* Creates a configuration with the configuration admin
* @param context the bundle context
* @param configId the configuration id
* @param createLatch the create latch for waiting
* @return the configuration
* @throws Exception
*/
private Configuration getConfiguration(String configId) throws Exception {
Configuration clientConfig = configAdmin.getConfiguration(configId, "?");
assertNotNull(clientConfig);
return clientConfig;
}

}

0 comments on commit 2a5f023

Please sign in to comment.