Retained messages tests, plus fix

Wiebe Cazemier
1 parent 121e04de
Showing 9 changed files with 292 additions and 8 deletions
FlashMQTests/FlashMQTests.pro
FlashMQTests/mainappthread.cpp
FlashMQTests/mainappthread.h
FlashMQTests/tst_maintests.cpp
FlashMQTests/twoclienttestcontext.cpp
FlashMQTests/twoclienttestcontext.h
mainapp.cpp
mainapp.h
subscriptionstore.cpp
 QT += testlib
 QT -= gui
-Qt += network
+QT += network
+QT += qmqtt
  
 DEFINES += TESTING
  
@@ -12,7 +13,34 @@ CONFIG -= app_bundle
 TEMPLATE = app
  
 SOURCES +=  tst_maintests.cpp \
-    ../cirbuf.cpp
+    ../MqttPacket.cpp \
+    ../cirbuf.cpp \
+    ../client.cpp \
+    ../exceptions.cpp \
+    ../mainapp.cpp \
+    ../mqttpacket.cpp \
+    ../retainedmessage.cpp \
+    ../rwlockguard.cpp \
+    ../subscriptionstore.cpp \
+    ../threaddata.cpp \
+    ../types.cpp \
+    ../utils.cpp \
+    mainappthread.cpp \
+    twoclienttestcontext.cpp
+
  
 HEADERS += \
-    ../cirbuf.h
+    ../cirbuf.h \
+    ../client.h \
+    ../exceptions.h \
+    ../forward_declarations.h \
+    ../mainapp.h \
+    ../mqttpacket.h \
+    ../retainedmessage.h \
+    ../rwlockguard.h \
+    ../subscriptionstore.h \
+    ../threaddata.h \
+    ../types.h \
+    ../utils.h \
+    mainappthread.h \
+    twoclienttestcontext.h
+#include "mainappthread.h"
+
+MainAppThread::MainAppThread(QObject *parent) : QThread(parent)
+{
+    appInstance = MainApp::getMainApp();
+}
+
+void MainAppThread::run()
+{
+    appInstance->start();
+}
+
+void MainAppThread::stopApp()
+{
+    appInstance->quit();
+}
+
+void MainAppThread::waitForStarted()
+{
+    int n = 0;
+    while(!appInstance->getStarted())
+    {
+        QThread::msleep(10);
+
+        if (n++ > 500)
+            throw new std::runtime_error("Waiting for app to start failed.");
+    }
+}
+#ifndef MAINAPPTHREAD_H
+#define MAINAPPTHREAD_H
+
+#include <QObject>
+#include <QThread>
+#include <mainapp.h>
+
+class MainAppThread : public QThread
+{
+    Q_OBJECT
+    MainApp *appInstance = nullptr;
+public:
+    explicit MainAppThread(QObject *parent = nullptr);
+
+public slots:
+    void run() override;
+    void stopApp();
+    void waitForStarted();
+
+signals:
+
+};
+
+#endif // MAINAPPTHREAD_H
 #include <QtTest>
  
+
+#include <QtQmqtt/qmqtt.h>
+#include <QScopedPointer>
+#include <QHostInfo>
+
 #include "cirbuf.h"
+#include "mainapp.h"
+#include "mainappthread.h"
+#include "twoclienttestcontext.h"
  
 class MainTests : public QObject
 {
     Q_OBJECT
  
+    MainAppThread mainApp;
+
 public:
     MainTests();
     ~MainTests();
  
 private slots:
-    void test_case1();
+    void cleanupTestCase();
+
     void test_circbuf();
     void test_circbuf_unwrapped_doubling();
     void test_circbuf_wrapped_doubling();
     void test_circbuf_full_wrapped_buffer_doubling();
  
+    void test_retained();
+    void test_retained_changed();
+    void test_retained_removed();
+
 };
  
 MainTests::MainTests()
 {
-
+    mainApp.start();
+    mainApp.waitForStarted();
 }
  
 MainTests::~MainTests()
@@ -29,9 +45,9 @@ MainTests::~MainTests()
  
 }
  
-void MainTests::test_case1()
+void MainTests::cleanupTestCase()
 {
-
+    mainApp.stopApp();
 }
  
 void MainTests::test_circbuf()
@@ -243,6 +259,85 @@ void MainTests::test_circbuf_full_wrapped_buffer_doubling()
     QVERIFY(true);
 }
  
-QTEST_APPLESS_MAIN(MainTests)
+void MainTests::test_retained()
+{
+    TwoClientTestContext testContext;
+
+    QByteArray payload = "We are testing";
+    QString topic = "retaintopic";
+
+    testContext.connectSender();
+    testContext.publishRetained(topic, payload);
+    testContext.publishRetained("dummy2", "Nobody sees this");
+
+    testContext.connectReceiver();
+    testContext.subscribeReceiver("dummy");
+    testContext.subscribeReceiver(topic);
+    testContext.waitReceiverReceived();
+
+    QVERIFY2(testContext.receivedMessages.count() == 1, "There must be one message in the received list");
+
+    QMQTT::Message msg = testContext.receivedMessages.first();
+    QCOMPARE(msg.payload(), payload);
+    QVERIFY(msg.retain());
+
+    testContext.receivedMessages.clear();
+
+    testContext.publishRetained(topic, payload);
+    testContext.waitReceiverReceived();
+
+    QVERIFY2(testContext.receivedMessages.count() == 1, "There must be one message in the received list");
+    QMQTT::Message msg2 = testContext.receivedMessages.first();
+    QCOMPARE(msg2.payload(), payload);
+    QVERIFY2(!msg2.retain(), "Getting a retained message while already being subscribed must be marked as normal, not retain.");
+}
+
+void MainTests::test_retained_changed()
+{
+    TwoClientTestContext testContext;
+
+    QByteArray payload = "We are testing";
+    QString topic = "retaintopic";
+
+    testContext.connectSender();
+    testContext.publishRetained(topic, payload);
+
+    payload = "Changed payload";
+
+    testContext.publishRetained(topic, payload);
+
+    testContext.connectReceiver();
+    testContext.subscribeReceiver(topic);
+    testContext.waitReceiverReceived();
+
+    QVERIFY2(testContext.receivedMessages.count() == 1, "There must be one message in the received list");
+
+    QMQTT::Message msg = testContext.receivedMessages.first();
+    QCOMPARE(msg.payload(), payload);
+    QVERIFY(msg.retain());
+}
+
+void MainTests::test_retained_removed()
+{
+    TwoClientTestContext testContext;
+
+    QByteArray payload = "We are testing";
+    QString topic = "retaintopic";
+
+    testContext.connectSender();
+    testContext.publishRetained(topic, payload);
+
+    payload = "";
+
+    testContext.publishRetained(topic, payload);
+
+    testContext.connectReceiver();
+    testContext.subscribeReceiver(topic);
+    testContext.waitReceiverReceived();
+
+    QVERIFY2(testContext.receivedMessages.empty(), "We erased the retained message. We shouldn't have received any.");
+}
+
+QTEST_GUILESS_MAIN(MainTests)
  
 #include "tst_maintests.moc"
+#include "twoclienttestcontext.h"
+
+#include <QEventLoop>
+#include <QTimer>
+
+TwoClientTestContext::TwoClientTestContext(QObject *parent) : QObject(parent)
+{
+    QHostInfo targetHostInfo = QHostInfo::fromName("localhost");
+    QHostAddress targetHost(targetHostInfo.addresses().first());
+    sender.reset(new QMQTT::Client(targetHost));
+    receiver.reset(new QMQTT::Client(targetHost));
+}
+
+void TwoClientTestContext::publishRetained(const QString &topic, const QByteArray &payload)
+{
+    QMQTT::Message msg;
+    msg.setTopic(topic);
+    msg.setRetain(true);
+    msg.setQos(0);
+    msg.setPayload(payload);
+    sender->publish(msg);
+}
+
+void TwoClientTestContext::connectSender()
+{
+    sender->connectToHost();
+    QEventLoop waiter;
+    connect(sender.data(), &QMQTT::Client::connected, &waiter, &QEventLoop::quit);
+    waiter.exec();
+}
+
+void TwoClientTestContext::connectReceiver()
+{
+    connect(receiver.data(), &QMQTT::Client::received, this, &TwoClientTestContext::onReceiverReceived);
+
+    receiver->connectToHost();
+    QEventLoop waiter;
+    connect(receiver.data(), &QMQTT::Client::connected, &waiter, &QEventLoop::quit);
+    waiter.exec();
+}
+
+void TwoClientTestContext::disconnectReceiver()
+{
+    receiver->disconnectFromHost();
+    QEventLoop waiter;
+    connect(sender.data(), &QMQTT::Client::disconnected, &waiter, &QEventLoop::quit);
+    waiter.exec();
+}
+
+void TwoClientTestContext::subscribeReceiver(const QString &topic)
+{
+    receiver->subscribe(topic);
+}
+
+void TwoClientTestContext::waitReceiverReceived()
+{
+    QEventLoop waiter;
+    QTimer timeout;
+    timeout.setSingleShot(true);
+    timeout.setInterval(1000);
+    connect(&timeout, &QTimer::timeout, &waiter, &QEventLoop::quit);
+    connect(receiver.data(), &QMQTT::Client::received, &waiter, &QEventLoop::quit);
+    timeout.start();
+    waiter.exec();
+}
+
+void TwoClientTestContext::onReceiverReceived(const QMQTT::Message &message)
+{
+    receivedMessages.append(message);
+}
+#ifndef RETAINTESTCONTEXT_H
+#define RETAINTESTCONTEXT_H
+
+#include <QObject>
+#include <QtQmqtt/qmqtt.h>
+#include <QHostInfo>
+
+class TwoClientTestContext : public QObject
+{
+    Q_OBJECT
+
+    QScopedPointer<QMQTT::Client> sender;
+    QScopedPointer<QMQTT::Client> receiver;
+
+private slots:
+    void onReceiverReceived(const QMQTT::Message& message);
+
+public:
+    explicit TwoClientTestContext(QObject *parent = nullptr);
+    void publishRetained(const QString &topic, const QByteArray &payload);
+    void connectSender();
+    void connectReceiver();
+    void disconnectReceiver();
+    void subscribeReceiver(const QString &topic);
+    void waitReceiverReceived();
+
+    QList<QMQTT::Message> receivedMessages;
+
+signals:
+
+};
+
+#endif // RETAINTESTCONTEXT_H
@@ -171,6 +171,7 @@ void MainApp::start()
  
     uint next_thread_index = 0;
  
+    started = true;
     while (running)
     {
         int num_fds = epoll_wait(epoll_fd_accept, events, MAX_EVENTS, 100);
@@ -21,6 +21,7 @@ class MainApp
 {
     static MainApp *instance;
  
+    bool started = false;
     bool running = true;
     std::vector<std::shared_ptr<ThreadData>> threads;
     std::shared_ptr<SubscriptionStore> subscriptionStore;
@@ -32,6 +33,7 @@ public:
     static MainApp *getMainApp();
     void start();
     void quit();
+    bool getStarted() const {return started;}
 };
  
 #endif // MAINAPP_H
@@ -148,6 +148,9 @@ void SubscriptionStore::setRetainedMessage(const std::string &amp;topic, const std::
         return;
     }
  
+    if (retained_found)
+        retainedMessages.erase(rm);
+
     retainedMessages.insert(std::move(rm));
 }