diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index 7f386cb9..164a39e9 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -22,12 +22,14 @@ import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.*; +import io.moquette.spi.security.IAuthorizator; import io.netty.buffer.Unpooled; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.*; import java.util.*; +import static io.moquette.spi.impl.ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE; @@ -44,9 +46,7 @@ abstract class AbstractProtocolProcessorCommonUtils { static final String TEST_USER = "fakeuser"; static final String TEST_PWD = "fakepwd"; - static final List EMPTY_OBSERVERS = Collections.emptyList(); - static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); - public static final String HELLO_WORLD_MQTT = "Hello world MQTT!!"; + private static final String HELLO_WORLD_MQTT = "Hello world MQTT!!"; EmbeddedChannel m_channel; ProtocolProcessor m_processor; @@ -55,7 +55,7 @@ abstract class AbstractProtocolProcessorCommonUtils { ISessionsStore m_sessionStore; ISubscriptionsDirectory subscriptions; MockAuthenticator m_mockAuthenticator; - protected SessionsRepository sessionsRepository; + SessionsRepository sessionsRepository; protected void initializeProcessorAndSubsystems() { m_channel = new EmbeddedChannel(); @@ -81,10 +81,10 @@ abstract class AbstractProtocolProcessorCommonUtils { subscriptions.init(sessionsRepository); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, sessionsRepository); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, sessionsRepository); } - protected void verifyNoPublishIsReceived() { + void verifyNoPublishIsReceived() { final Object messageReceived = m_channel.readOutbound(); assertNull("Received an out message from processor while not expected", messageReceived); } @@ -107,6 +107,16 @@ abstract class AbstractProtocolProcessorCommonUtils { verifySubscriptionExists(m_sessionStore, expectedSubscription); } + protected void subscribeWithFail(EmbeddedChannel channel, String topic, MqttQoS desiredQos) { + MqttSubscribeMessage subscribe = MqttMessageBuilders.subscribe() + .addSubscription(desiredQos, topic) + .messageId(1) + .build(); + this.m_processor.processSubscribe(channel, subscribe); + MqttSubAckMessage subAck = channel.readOutbound(); + assertEquals(MqttQoS.FAILURE.value(), (int) subAck.payload().grantedQoSLevels().get(0)); + } + protected static void verifySubscriptionExists(ISessionsStore sessionsStore, Subscription expectedSubscription) { final Subscription subscription = sessionsStore.subscriptionStore().reload(expectedSubscription); assertEquals(expectedSubscription, subscription); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index e84ed3c7..0cd03738 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -59,16 +59,16 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils static final String EVIL_TEST_USER = "eviluser"; static final String EVIL_TEST_PWD = "unsecret"; - static final List EMPTY_OBSERVERS = Collections.emptyList(); - static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); + public static final List EMPTY_OBSERVERS = Collections.emptyList(); + public static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); @Before - public void setUp() throws InterruptedException { + public void setUp() { initializeProcessorAndSubsystems(); } @Test - public void testPublishToItself() throws InterruptedException { + public void testPublishToItself() { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true @@ -101,7 +101,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils } @Test - public void testPublishToMultipleSubscribers() throws InterruptedException { + public void testPublishToMultipleSubscribers() { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); final Subscription subscriptionClient2 = new Subscription( FAKE_CLIENT_ID2, diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java index cc53ff6b..531876ad 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java @@ -16,6 +16,9 @@ package io.moquette.spi.impl; import io.moquette.server.netty.NettyUtils; +import io.moquette.spi.impl.security.PermitAllAuthorizator; +import io.moquette.spi.impl.subscriptions.Topic; +import io.moquette.spi.security.IAuthorizator; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.MqttQoS; import org.junit.Before; @@ -27,7 +30,7 @@ import static org.junit.Assert.assertEquals; public class ProtocolProcessor_InternalPublish_Test extends AbstractProtocolProcessorCommonUtils { @Before - public void setUp() throws InterruptedException { + public void setUp() { initializeProcessorAndSubsystems(); connect(); } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java index 70f33cb3..4ff3965b 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java @@ -27,7 +27,7 @@ import static org.junit.Assert.assertFalse; public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProcessorCommonUtils { @Before - public void setUp() throws InterruptedException { + public void setUp() { initializeProcessorAndSubsystems(); }