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>

2 files changed, 60 insertions, 38 deletions

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));
     }
 }