moved mqtt tests from Python to Rust #1011 (#1128)

Signed-off-by: initard <solo@softwareag.com>
Co-authored-by: Lukasz Woznicki <lukasz.woznicki@softwareag.com>

Co-authored-by: initard <solo@softwareag.com>
Co-authored-by: Lukasz Woznicki <lukasz.woznicki@softwareag.com>

15 files changed, 62 insertions, 491 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 4cf63670..27e25416 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2878,6 +2878,7 @@ dependencies = [
  "futures",
  "hyper",
  "mockito",
+ "mqtt_tests",
  "pem",
  "predicates 2.1.1",
  "reqwest",
@@ -2893,6 +2894,7 @@ dependencies = [
  "tempfile",
  "test-case",
  "thiserror",
+ "tokio",
  "toml",
  "url",
  "which",
diff --git a/crates/core/tedge/Cargo.toml b/crates/core/tedge/Cargo.toml
index b2778458..e0e5ae36 100644
--- a/crates/core/tedge/Cargo.toml
+++ b/crates/core/tedge/Cargo.toml
@@ -38,10 +38,12 @@ which = "4.2"
 assert_cmd = "2.0"
 assert_matches = "1.5"
 mockito = "0.31"
+mqtt_tests = { path = "../../tests/mqtt_tests" }
 pem = "1.0"
 predicates = "2.1"
 tempfile = "3.2"
 test-case = "2.0"
+tokio = { version = "1.12" }
 
 [features]
 integration-test = []
diff --git a/crates/core/tedge/tests/mqtt.rs b/crates/core/tedge/tests/mqtt.rs
index 46eca827..b0e438ad 100644
--- a/crates/core/tedge/tests/mqtt.rs
+++ b/crates/core/tedge/tests/mqtt.rs
@@ -1,64 +1,84 @@
 #[cfg(test)]
-#[cfg(feature = "mosquitto-available")]
 mod tests {
 
     // These test cases need mosquitto on localhost on GH hosted machine.
 
+    use std::{io::Write, time::Duration};
+
     use assert_cmd::assert::OutputAssertExt;
     use assert_cmd::Command;
     use predicates::prelude::predicate;
+    use tedge_config::TEdgeConfigLocation;
+    use test_case::test_case;
 
-    #[test]
-    fn test_cli_pub_basic() -> Result<(), Box<dyn std::error::Error>> {
-        let mut cmd = Command::cargo_bin("tedge")?;
-        let assert = cmd
-            .args(&["mqtt", "pub", "topic", "message"])
-            .unwrap()
-            .assert();
+    const TEST_TIMEOUT_MS: Duration = Duration::from_millis(5000);
 
-        assert.success().code(predicate::eq(0));
-        Ok(())
+    fn make_config(port: u16) -> Result<tempfile::TempDir, anyhow::Error> {
+        let dir = tempfile::TempDir::new().unwrap();
+        let toml_conf = &format!("[mqtt]\nport = {port}");
+
+        let config_location = TEdgeConfigLocation::from_custom_root(dir.path());
+        let mut file = std::fs::File::create(config_location.tedge_config_file_path())?;
+        file.write_all(toml_conf.as_bytes())?;
+        Ok(dir)
     }
 
-    #[test]
-    fn test_cli_pub_qos() -> Result<(), Box<dyn std::error::Error>> {
+    #[test_case(Some("0"))]
+    #[test_case(Some("1"))]
+    #[test_case(Some("2"))]
+    #[test_case(None)]
+    #[tokio::test]
+    async fn test_cli_pub_basic(qos: Option<&str>) -> Result<(), anyhow::Error> {
+        let broker = mqtt_tests::test_mqtt_broker();
+        let tmpfile = make_config(broker.port)?;
+
+        let mut messages = broker.messages_published_on("topic").await;
+
         let mut cmd = Command::cargo_bin("tedge")?;
-        let assert = cmd
-            .args(&["mqtt", "pub", "topic", "message"])
-            .args(&["--qos", "1"])
-            .unwrap()
-            .assert();
+        cmd.args(&["--config-dir", tmpfile.path().to_str().unwrap()])
+            .args(&["mqtt", "pub", "topic", "message"]);
+
+        if let Some(qos) = qos {
+            cmd.args(&["--qos", qos]);
+        }
+        let assert = cmd.unwrap().assert();
+
+        mqtt_tests::assert_received_all_expected(&mut messages, TEST_TIMEOUT_MS, &["message"])
+            .await;
 
         assert.success().code(predicate::eq(0));
         Ok(())
     }
 
-    #[test]
-    fn test_cli_sub_basic() -> Result<(), Box<dyn std::error::Error>> {
-        let mut cmd = Command::cargo_bin("tedge")?;
-        let err = cmd
-            .args(&["mqtt", "sub", "topic"])
-            .timeout(std::time::Duration::from_secs(1))
-            .unwrap_err();
+    #[test_case(Some("0"))]
+    #[test_case(Some("1"))]
+    #[test_case(Some("2"))]
+    #[test_case(None)]
+    fn mqtt_pub_no_broker_running(qos: Option<&str>) {
+        let mut cmd = Command::cargo_bin("tedge").unwrap();
+        cmd.args(&["mqtt", "pub", "topic", "message"])
+            .timeout(std::time::Duration::from_secs(1));
 
-        let output = err.as_output().unwrap();
-        assert_eq!(None, output.status.code());
+        if let Some(qos) = qos {
+            cmd.args(&["--qos", qos]);
+        }
 
-        Ok(())
+        let _output = cmd.assert().code(predicate::eq(1));
     }
 
-    #[test]
-    fn test_cli_sub_qos() -> Result<(), Box<dyn std::error::Error>> {
-        let mut cmd = Command::cargo_bin("tedge")?;
-        let err = cmd
-            .args(&["mqtt", "sub", "topic"])
-            .args(&["--qos", "1"])
-            .timeout(std::time::Duration::from_secs(1))
-            .unwrap_err();
+    #[test_case(Some("0"))]
+    #[test_case(Some("1"))]
+    #[test_case(Some("2"))]
+    #[test_case(None)]
+    fn mqtt_sub_no_broker_running(qos: Option<&str>) {
+        let mut cmd = Command::cargo_bin("tedge").unwrap();
+        cmd.args(&["mqtt", "sub", "topic"])
+            .timeout(std::time::Duration::from_secs(1));
 
-        let output = err.as_output().unwrap();
-        assert_eq!(None, output.status.code());
+        if let Some(qos) = qos {
+            cmd.args(&["--qos", qos]);
+        }
 
-        Ok(())
+        let _output = cmd.assert().code(predicate::eq(1));
     }
 }
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_noqos/pysystest.xml b/tests/PySys/local_mqtt/local_mqtt_pub_fail_noqos/pysystest.xml
deleted file mode 100644
index 16ffff7d..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_noqos/pysystest.xml
+++ /dev/null
@@ -1,25 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<pysystest type="auto">
-
-  <description>
-	<title>Check error code when mosquitto is not running with no QoS set</title>
-	<purpose><![CDATA[
-]]>
-	</purpose>
-  </description>
-  <classification>
-	<groups inherit="true">
-	  <group></group>
-	</groups>
-	<modes inherit="true">
-	</modes>
-  </classification>
-  <data>
-	<class name="PySysTest" module="run"/>
-  </data>
-  <traceability>
-	<requirements>
-	  <requirement id=""/>
-	</requirements>
-  </traceability>
-</pysystest>
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_noqos/run.py b/tests/PySys/local_mqtt/local_mqtt_pub_fail_noqos/run.py
deleted file mode 100644
index 5bd76d54..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_noqos/run.py
+++ /dev/null
@@ -1,48 +0,0 @@
-from pysys.basetest import BaseTest
-import os
-
-"""
-Validate local publishing while no mosquitto is running
-
-Given a configured system
-When we stop mosquitto
-When we publish something without specifing qos level
-Then we expect an error code
-Then we restart mosquitto
-"""
-
-
-class PySysTest(BaseTest):
-    def setup(self):
-        self.tedge = "/usr/bin/tedge"
-        self.sudo = "/usr/bin/sudo"
-        self.systemctl = "/usr/bin/systemctl"
-        self.environ = {"HOME": os.environ.get("HOME")}
-
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "stop", "mosquitto"],
-            stdouterr="stop",
-        )
-
-        self.addCleanupFunction(self.mycleanup)
-
-    def execute(self):
-
-        pub = self.startProcess(
-            command=self.sudo,
-            arguments=[self.tedge, "mqtt", "pub", "atopic", "amessage"],
-            stdouterr="tedge_pub_fail",
-            expectedExitStatus="==1",
-            environs=self.environ,
-        )
-
-        # validate exit status with the expected status from calling startProcess
-        self.assertThat("value" + pub.expectedExitStatus, value=pub.exitStatus)
-
-    def mycleanup(self):
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "start", "mosquitto"],
-            stdouterr="start",
-        )
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos0/pysystest.xml b/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos0/pysystest.xml
deleted file mode 100644
index 900deb47..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos0/pysystest.xml
+++ /dev/null
@@ -1,25 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<pysystest type="auto">
-
-  <description>
-	<title>Check error code when mosquitto is not running with QoS set to 0</title>
-	<purpose><![CDATA[
-]]>
-	</purpose>
-  </description>
-  <classification>
-	<groups inherit="true">
-	  <group></group>
-	</groups>
-	<modes inherit="true">
-	</modes>
-  </classification>
-  <data>
-	<class name="PySysTest" module="run"/>
-  </data>
-  <traceability>
-	<requirements>
-	  <requirement id=""/>
-	</requirements>
-  </traceability>
-</pysystest>
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos0/run.py b/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos0/run.py
deleted file mode 100644
index 4af53c0a..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos0/run.py
+++ /dev/null
@@ -1,48 +0,0 @@
-from pysys.basetest import BaseTest
-import os
-
-"""
-Validate local publishing while no mosquitto is running
-
-Given a configured system
-When we stop mosquitto
-When we publish something with qos 0
-Then we expect an error code
-Then we restart mosquitto
-"""
-
-
-class PySysTest(BaseTest):
-    def setup(self):
-        self.tedge = "/usr/bin/tedge"
-        self.sudo = "/usr/bin/sudo"
-        self.systemctl = "/usr/bin/systemctl"
-        self.environ = {"HOME": os.environ.get("HOME")}
-
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "stop", "mosquitto"],
-            stdouterr="stop",
-        )
-
-        self.addCleanupFunction(self.mycleanup)
-
-    def execute(self):
-
-        pub = self.startProcess(
-            command=self.tedge,
-            arguments=["mqtt", "pub", "--qos", "0", "atopic", "amessage"],
-            stdouterr="tedge_pub_fail",
-            expectedExitStatus="==1",
-            environs=self.environ,
-        )
-
-        # validate exit status with the expected status from calling startProcess
-        self.assertThat("value" + pub.expectedExitStatus, value=pub.exitStatus)
-
-    def mycleanup(self):
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "start", "mosquitto"],
-            stdouterr="start",
-        )
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos1/pysystest.xml b/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos1/pysystest.xml
deleted file mode 100644
index 9a46a9c2..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos1/pysystest.xml
+++ /dev/null
@@ -1,25 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<pysystest type="auto">
-
-  <description>
-	<title>Check error code when mosquitto is not running with QoS set to 1</title>
-	<purpose><![CDATA[
-]]>
-	</purpose>
-  </description>
-  <classification>
-	<groups inherit="true">
-	  <group></group>
-	</groups>
-	<modes inherit="true">
-	</modes>
-  </classification>
-  <data>
-	<class name="PySysTest" module="run"/>
-  </data>
-  <traceability>
-	<requirements>
-	  <requirement id=""/>
-	</requirements>
-  </traceability>
-</pysystest>
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos1/run.py b/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos1/run.py
deleted file mode 100644
index 48d095b4..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos1/run.py
+++ /dev/null
@@ -1,47 +0,0 @@
-from pysys.basetest import BaseTest
-import os
-
-"""
-Validate local publishing while no mosquitto is running
-
-Given a configured system
-When we stop mosquitto
-When we publish something with qos 1
-Then we expect an error code
-Then we restart mosquitto
-"""
-
-
-class PySysTest(BaseTest):
-    def setup(self):
-        self.tedge = "/usr/bin/tedge"
-        self.sudo = "/usr/bin/sudo"
-        self.systemctl = "/usr/bin/systemctl"
-        self.environ = {"HOME": os.environ.get("HOME")}
-
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "stop", "mosquitto"],
-            stdouterr="stop",
-        )
-
-        self.addCleanupFunction(self.mycleanup)
-
-    def execute(self):
-
-        pub = self.startProcess(
-            command=self.tedge,
-            arguments=["mqtt", "pub", "--qos", "1", "atopic", "amessage"],
-            stdouterr="tedge_pub_fail",
-            expectedExitStatus="==1",
-        )
-
-        # validate exit status with the expected status from calling startProcess
-        self.assertThat("value" + pub.expectedExitStatus, value=pub.exitStatus)
-
-    def mycleanup(self):
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "start", "mosquitto"],
-            stdouterr="start",
-        )
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos2/pysystest.xml b/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos2/pysystest.xml
deleted file mode 100644
index a60a8ed4..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos2/pysystest.xml
+++ /dev/null
@@ -1,25 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<pysystest type="auto">
-
-  <description>
-	<title>Check error code when mosquitto is not running with QoS set to 2</title>
-	<purpose><![CDATA[
-]]>
-	</purpose>
-  </description>
-  <classification>
-	<groups inherit="true">
-	  <group></group>
-	</groups>
-	<modes inherit="true">
-	</modes>
-  </classification>
-  <data>
-	<class name="PySysTest" module="run"/>
-  </data>
-  <traceability>
-	<requirements>
-	  <requirement id=""/>
-	</requirements>
-  </traceability>
-</pysystest>
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos2/run.py b/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos2/run.py
deleted file mode 100644
index b249eaaa..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_fail_qos2/run.py
+++ /dev/null
@@ -1,48 +0,0 @@
-from pysys.basetest import BaseTest
-import os
-
-"""
-Validate local publishing while no mosquitto is running
-
-Given a configured system
-When we stop mosquitto
-When we publish something with qos 2
-Then we expect an error code
-Then we restart mosquitto
-"""
-
-
-class PySysTest(BaseTest):
-    def setup(self):
-        self.tedge = "/usr/bin/tedge"
-        self.sudo = "/usr/bin/sudo"
-        self.systemctl = "/usr/bin/systemctl"
-        self.environ = {"HOME": os.environ.get("HOME")}
-
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "stop", "mosquitto"],
-            stdouterr="stop",
-        )
-
-        self.addCleanupFunction(self.mycleanup)
-
-    def execute(self):
-
-        pub = self.startProcess(
-            command=self.tedge,
-            arguments=["mqtt", "pub", "--qos", "2", "atopic", "amessage"],
-            stdouterr="tedge_pub_fail",
-            expectedExitStatus="==1",
-            environs=self.environ,
-        )
-
-        # validate exit status with the expected status from calling startProcess
-        self.assertThat("value" + pub.expectedExitStatus, value=pub.exitStatus)
-
-    def mycleanup(self):
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "start", "mosquitto"],
-            stdouterr="start",
-        )
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_sub_with_sudo/pysystest.xml b/tests/PySys/local_mqtt/local_mqtt_pub_sub_with_sudo/pysystest.xml
deleted file mode 100644
index 045d3015..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_sub_with_sudo/pysystest.xml
+++ /dev/null
@@ -1,25 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<pysystest type="auto">
-
-  <description>
-	<title>Validate local publishing and subscribing</title>
-	<purpose><![CDATA[
-]]>
-	</purpose>
-  </description>
-  <classification>
-	<groups inherit="true">
-	  <group></group>
-	</groups>
-	<modes inherit="true">
-	</modes>
-  </classification>
-  <data>
-	<class name="PySysTest" module="run"/>
-  </data>
-  <traceability>
-	<requirements>
-	  <requirement id=""/>
-	</requirements>
-  </traceability>
-</pysystest>
diff --git a/tests/PySys/local_mqtt/local_mqtt_pub_sub_with_sudo/run.py b/tests/PySys/local_mqtt/local_mqtt_pub_sub_with_sudo/run.py
deleted file mode 100644
index c78e934e..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_pub_sub_with_sudo/run.py
+++ /dev/null
@@ -1,62 +0,0 @@
-import pysys
-from pysys.basetest import BaseTest
-
-import time
-import os
-
-"""
-Validate local publishing and subscribing:
-
-Given a configured system
-When we start the bridge and the mapper
-When we start sudo tedge sub in the background
-When we start sudo tedge pub to publish a message
-When we start sudo tedge pub to publish another message
-Then we find the messages in the output of tedge sub
-"""
-
-
-class PySysTest(BaseTest):
-    def execute(self):
-        tedge = "/usr/bin/tedge"
-        sudo = "/usr/bin/sudo"
-
-        sub = self.startProcess(
-            command=sudo,
-            arguments=[tedge, "mqtt", "sub", "atopic"],
-            stdouterr="tedge_sub",
-            background=True,
-        )
-
-        # Wait for a small amount of time to give tedge sub time
-        # to initialize. This is a heuristic measure.
-        # Without an additional wait we observe failures in 1% of the test
-        # runs.
-        time.sleep(0.1)
-
-        pub = self.startProcess(
-            command=sudo,
-            arguments=[tedge, "mqtt", "pub", "atopic", "amessage"],
-            stdouterr="tedge_pub2",
-        )
-
-        pub = self.startProcess(
-            command=sudo,
-            arguments=[tedge, "mqtt", "pub", "atopic", "the message"],
-            stdouterr="tedge_pub3",
-        )
-
-        # wait for a while before killing the subscribers
-        time.sleep(1)
-
-        # Kill the subscriber process explicitly with sudo as PySys does
-        # not have the rights to do it
-        kill = self.startProcess(
-            command=sudo,
-            arguments=["killall", "tedge"],
-            stdouterr="kill_out",
-        )
-
-    def validate(self):
-        self.assertGrep("tedge_sub.out", "amessage", contains=True)
-        self.assertGrep("tedge_sub.out", "the message", contains=True)
diff --git a/tests/PySys/local_mqtt/local_mqtt_sub_fail/pysystest.xml b/tests/PySys/local_mqtt/local_mqtt_sub_fail/pysystest.xml
deleted file mode 100644
index a5a96522..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_sub_fail/pysystest.xml
+++ /dev/null
@@ -1,25 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<pysystest type="auto">
-
-  <description>
-	<title>Validate local publishing when mosquitto is stopped</title>
-	<purpose><![CDATA[
-]]>
-	</purpose>
-  </description>
-  <classification>
-	<groups inherit="true">
-	  <group></group>
-	</groups>
-	<modes inherit="true">
-	</modes>
-  </classification>
-  <data>
-	<class name="PySysTest" module="run"/>
-  </data>
-  <traceability>
-	<requirements>
-	  <requirement id=""/>
-	</requirements>
-  </traceability>
-</pysystest>
diff --git a/tests/PySys/local_mqtt/local_mqtt_sub_fail/run.py b/tests/PySys/local_mqtt/local_mqtt_sub_fail/run.py
deleted file mode 100644
index 0dc74356..00000000
--- a/tests/PySys/local_mqtt/local_mqtt_sub_fail/run.py
+++ /dev/null
@@ -1,50 +0,0 @@
-from pysys.basetest import BaseTest
-
-import time
-import os
-
-"""
-Validate local subscribing while no mosquitto is running
-
-Given a configured system
-When we stop mosquitto
-When we subscribe to something
-Then we expect an error code
-Then we restart mosquitto
-"""
-
-
-class PySysTest(BaseTest):
-    def setup(self):
-        self.tedge = "/usr/bin/tedge"
-        self.sudo = "/usr/bin/sudo"
-        self.systemctl = "/usr/bin/systemctl"
-        self.environ = {"HOME": os.environ.get("HOME")}
-
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "stop", "mosquitto"],
-            stdouterr="stop",
-        )
-
-        self.addCleanupFunction(self.mycleanup)
-
-    def execute(self):
-
-        pub = self.startProcess(
-            command=self.tedge,
-            arguments=["mqtt", "sub", "atopic"],
-            stdouterr="tedge_sub_fail",
-            expectedExitStatus="==1",
-            environs=self.environ,
-        )
-
-        # validate exit status with the expected status from calling startProcess
-        self.assertThat("value" + pub.expectedExitStatus, value=pub.exitStatus)
-
-    def mycleanup(self):
-        self.startProcess(
-            command=self.sudo,
-            arguments=[self.systemctl, "start", "mosquitto"],
-            stdouterr="start",
-        )