Skip to content

Commit

Permalink
Tests for sub + wildcards and unsub
Browse files Browse the repository at this point in the history
  • Loading branch information
@NehaSelvan1512
NehaSelvan1512 committed Sep 29, 2023
1 parent 60ee769 commit e834f56
Showing 1 changed file with 101 additions and 12 deletions.
113 changes: 101 additions & 12 deletions plugins/mqtt-stream/src/test/java/org/polypheny/db/mqtt/MqttClientBrokerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,13 @@

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicLong;
import org.junit.Before;
import org.junit.BeforeClass;
Expand Down Expand Up @@ -50,8 +53,8 @@ public static void init() {
initialSettings.clear();
initialSettings.put( "brokerAddress", "1883" );
initialSettings.put( "brokerPort", "1883" );
initialSettings.put( "commonCollectionName", "testCollection" );
initialSettings.put( "commonCollection", "true" );
initialSettings.put( "catchAllEntityName", "testCollection" );
initialSettings.put( "catchAllEntity", "true" );
initialSettings.put( "namespace", "testNamespace" );
initialSettings.put( "namespaceType", "DOCUMENT" );
initialSettings.put( "topics", "button" );
Expand All @@ -69,14 +72,14 @@ public static void init() {

}

/*
@Before
public void resetSettings() {

initialSettings.clear();
initialSettings.put( "brokerAddress", "1883" );
initialSettings.put( "brokerPort", "1883" );
initialSettings.put( "commonCollectionName", "testCollection" );
initialSettings.put( "commonCollection", "true" );
initialSettings.put( "catchAllEntityName", "testCollection" );
initialSettings.put( "catchAllEntity", "true" );
initialSettings.put( "namespace", "testNamespace" );
initialSettings.put( "namespaceType", "DOCUMENT" );
initialSettings.put( "topics", "" );
Expand All @@ -93,22 +96,108 @@ public void resetSettings() {
initialSettings );

changedSettings.clear();
changedSettings.put( "commonCollectionName", "testCollection" );
changedSettings.put( "commonCollection", "true" );
changedSettings.put( "catchAllEntityName", "testCollection" );
changedSettings.put( "catchAllEntity", "true" );
changedSettings.put( "namespace", "testNamespace" );
changedSettings.put( "topics", "" );
changedSettings.put( "filterQuery", "" );
}
*/


private void simulateIoTDevices() {
client.publish( "device1/online", "true" );
client.publish( "device1/sensor/measurements", "[28,30,35 ]" );
client.publish( "device1/sensor/measurements/unit", "C" );
client.publish( "device1/sensor/battery", "86" );

client.publish( "device2/online", "true" );
client.publish( "device2/location/info", "Basel" );
client.publish( "device2/sensor/info", "{\"wifi\":\"networkName\", \"mqtt\":{\"brokerIp\":\"127.0.0.1\", \"port\":1883}, \"deviceName\":\"device2\"}" );
}


@Test
public void simpleSubscribeUnsubscribeTest() {
changedSettings.replace( "topics", "device1/sensor/battery" );
//All subscribed topics so far are unsubscribed
client.updateSettings( changedSettings );
assertEquals( 1, client.getTopicsMap().size() );
assertEquals( 0, client.getTopicsMap().get( "device1/sensor/battery" ).intValue() );
simulateIoTDevices();
assertEquals( 1, client.getTopicsMap().get( "device1/sensor/battery" ).intValue() );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/sensor/battery", "86" } ) );
}


@Test
public void subscribeWithWildcardHashtagTest() {
changedSettings.replace( "topics", "#" );
client.updateSettings( changedSettings );
simulateIoTDevices();
assertEquals( 7, client.getTopicsMap().get( "#" ).intValue() );
}


@Test
public void subscribeWithWildcardHashtagAtEndTest() {
changedSettings.replace( "topics", "device1/#" );
client.updateSettings( changedSettings );
simulateIoTDevices();
assertEquals( 4, client.getTopicsMap().get( "device1/#" ).intValue() );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/sensor/battery", "86" } ) );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/online", "true" } ) );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/sensor/measurements", "[28,76,55 ]" } ) );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/sensor/info", "{\"wifi\":\"networkName\", \"mqtt\":{\"brokerIp\":\"127.0.0.1\", \"port\":1883}, \"deviceName\":\"device1\"}" } ) );
}


@Test
public void subscribeWithWildcardPlusAtEndTest() {
changedSettings.replace( "topics", "device1/sensor/+" );
client.updateSettings( changedSettings );
simulateIoTDevices();
assertEquals( 3, client.getTopicsMap().get( "device1/sensor/+" ).intValue() );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/sensor/battery", "86" } ) );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/sensor/measurements", "[28,76,55 ]" } ) );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/sensor/info", "{\"wifi\":\"networkName\", \"mqtt\":{\"brokerIp\":\"127.0.0.1\", \"port\":1883}, \"deviceName\":\"device1\"}" } ) );
}


@Test
public void subscribeWithWildcardPlusInMiddleTest() {
changedSettings.replace( "topics", "device2/+/info" );
client.updateSettings( changedSettings );
simulateIoTDevices();
assertEquals( 2, client.getTopicsMap().get( "device2/+/info" ).intValue() );
assertTrue( client.getMessageQueue().contains( new String[]{ "device2/location/info", "Basel" } ) );
assertTrue( client.getMessageQueue().contains( new String[]{ "device2/sensor/info", "{\"wifi\":\"networkName\", \"mqtt\":{\"brokerIp\":\"127.0.0.1\", \"port\":1883}, \"deviceName\":\"device2\"}" } ) );

}


@Test
public void connectionTest() {
Map<String, AtomicLong> topicsMap = client.getTopicsMap();
assertEquals( 1, topicsMap.size() );
assertEquals( 0, topicsMap.get( "button" ) );
public void subscribeWithWildcardPlusAtBeginningTest() {
changedSettings.replace( "topics", "+/online" );
client.updateSettings( changedSettings );
simulateIoTDevices();
assertEquals( 2, client.getTopicsMap().get( "+/online" ).intValue() );
assertTrue( client.getMessageQueue().contains( new String[]{ "device1/online", "true" } ) );
assertTrue( client.getMessageQueue().contains( new String[]{ "device2/online", "true" } ) );
}


//
// subscribe to topic
// + wildcards

// richtige query für wildcard topic holen
// collections werden zu beginn richtig erstellt
// catchAllEntity
// collectionPerTopic
// 1 msg handling test -> verify dass methoden aufgerufen werden





}

0 comments on commit e834f56

Please sign in to comment.