Add base64 encoding support to tedge flows test

Signed-off-by: Didier Wenzek <[email protected]>

Author: didier-wenzek

crates/core/tedge/src/cli/flows/cli.rs

@@ -1,3 +1,4 @@
+use crate::cli::flows::decode_message;
 use crate::cli::flows::list::ListCommand;
 use crate::cli::flows::test::TestCommand;
 use crate::command::BuildCommand;
@@ -8,7 +9,6 @@ use anyhow::Context;
 use anyhow::Error;
 use std::path::PathBuf;
 use tedge_config::TEdgeConfig;
-use tedge_flows::flow::Message;
 use tedge_flows::MessageProcessor;
 
 #[derive(clap::Subcommand, Debug)]
@@ -46,6 +46,14 @@ pub enum TEdgeFlowsCli {
         #[clap(long = "final-on-interval")]
         final_on_interval: bool,
 
+        /// The input payloads are base64 encoded and have to be decoded first
+        #[clap(long = "base64-input")]
+        base64_input: bool,
+
+        /// The output payloads have to be base64 encoded before being displayed
+        #[clap(long = "base64-output")]
+        base64_output: bool,
+
         /// Topic of the message sample
         ///
         /// If none is provided, messages are read from stdin expecting a line per message:
@@ -71,16 +79,16 @@ impl BuildCommand for TEdgeFlowsCli {
                 flows_dir,
                 flow,
                 final_on_interval,
+                base64_input,
+                base64_output,
                 topic,
                 payload,
             } => {
                 let flows_dir = flows_dir.unwrap_or_else(|| Self::default_flows_dir(config));
                 let message = match (topic, payload) {
-                    (Some(topic), Some(payload)) => Some(Message {
-                        topic,
-                        payload: payload.into_bytes(),
-                        timestamp: None,
-                    }),
+                    (Some(topic), Some(payload)) => {
+                        Some(decode_message(topic, payload, base64_input)?)
+                    }
                     (Some(_), None) => Err(anyhow!("Missing sample payload"))?,
                     (None, Some(_)) => Err(anyhow!("Missing sample topic"))?,
                     (None, None) => None,
@@ -90,6 +98,8 @@ impl BuildCommand for TEdgeFlowsCli {
                     flow,
                     message,
                     final_on_interval,
+                    base64_input,
+                    base64_output,
                 }
                 .into_boxed())
             }

crates/core/tedge/src/cli/flows/mod.rs

@@ -2,4 +2,27 @@ mod cli;
 mod list;
 mod test;
 
+use base64::prelude::*;
 pub use cli::TEdgeFlowsCli;
+use tedge_flows::flow::Message;
+
+fn decode_message(
+    topic: String,
+    payload: String,
+    base64_payload: bool,
+) -> Result<Message, anyhow::Error> {
+    let payload = if base64_payload {
+        BASE64_STANDARD.decode(payload.as_bytes())?
+    } else {
+        payload.into_bytes()
+    };
+    Ok(Message::new(topic, payload))
+}
+
+fn encode_message(mut message: Message, base64_payload: bool) -> Message {
+    if base64_payload {
+        message.payload = BASE64_STANDARD.encode(message.payload).into_bytes();
+    };
+
+    message
+}

crates/core/tedge/src/cli/flows/test.rs

@@ -1,3 +1,5 @@
+use crate::cli::flows::decode_message;
+use crate::cli::flows::encode_message;
 use crate::cli::flows::TEdgeFlowsCli;
 use crate::command::Command;
 use crate::log::MaybeFancy;
@@ -15,6 +17,8 @@ pub struct TestCommand {
     pub flow: Option<PathBuf>,
     pub message: Option<Message>,
     pub final_on_interval: bool,
+    pub base64_input: bool,
+    pub base64_output: bool,
 }
 
 #[async_trait::async_trait]
@@ -36,7 +40,7 @@ impl Command for TestCommand {
             self.process(&mut processor, message, &timestamp).await;
         } else {
             let mut stdin = BufReader::new(tokio::io::stdin());
-            while let Some(message) = next_message(&mut stdin).await {
+            while let Some(message) = next_message(self.base64_input, &mut stdin).await {
                 let timestamp = DateTime::now();
                 self.process(&mut processor, &message, &timestamp).await;
             }
@@ -61,7 +65,7 @@ impl TestCommand {
             .await
             .into_iter()
             .map(|(_, v)| v)
-            .for_each(print)
+            .for_each(|msg| print(self.base64_output, msg))
     }
 
     async fn tick(&self, processor: &mut MessageProcessor, timestamp: &DateTime) {
@@ -70,15 +74,15 @@ impl TestCommand {
             .await
             .into_iter()
             .map(|(_, v)| v)
-            .for_each(print)
+            .for_each(|msg| print(self.base64_output, msg))
     }
 }
 
-fn print(messages: Result<Vec<Message>, FlowError>) {
+fn print(base64_output: bool, messages: Result<Vec<Message>, FlowError>) {
     match messages {
         Ok(messages) => {
             for message in messages {
-                println!("{message}");
+                println!("{}", encode_message(message, base64_output));
             }
         }
         Err(err) => {
@@ -87,7 +91,7 @@ fn print(messages: Result<Vec<Message>, FlowError>) {
     }
 }
 
-fn parse(line: String) -> Result<Option<Message>, Error> {
+fn parse(base64_input: bool, line: String) -> Result<Option<Message>, Error> {
     let line = line.trim();
     if line.is_empty() {
         return Ok(None);
@@ -101,12 +105,9 @@ fn parse(line: String) -> Result<Option<Message>, Error> {
 
     let topic = line[1..closing_bracket].to_string();
     let payload = line[closing_bracket + 1..].to_string();
+    let message = decode_message(topic, payload, base64_input)?;
 
-    Ok(Some(Message {
-        topic,
-        payload: payload.into_bytes(),
-        timestamp: None,
-    }))
+    Ok(Some(message))
 }
 
 async fn next_line(input: &mut BufReader<Stdin>) -> Option<String> {
@@ -127,9 +128,9 @@ async fn next_line(input: &mut BufReader<Stdin>) -> Option<String> {
         }
     }
 }
-async fn next_message(input: &mut BufReader<Stdin>) -> Option<Message> {
+async fn next_message(base64_input: bool, input: &mut BufReader<Stdin>) -> Option<Message> {
     let line = next_line(input).await?;
-    match parse(line) {
+    match parse(base64_input, line) {
         Ok(message) => message,
         Err(err) => {
             eprintln!("Fail to parse input message {}", err);

crates/extensions/tedge_flows/src/flow.rs

@@ -218,21 +218,26 @@ impl TryFrom<OffsetDateTime> for DateTime {
 }
 
 impl Message {
-    #[cfg(test)]
-    pub(crate) fn new(topic: &str, payload: &str) -> Self {
+    pub fn new(topic: impl ToString, payload: impl Into<Vec<u8>>) -> Self {
         Message {
             topic: topic.to_string(),
-            payload: payload.to_string().into_bytes(),
-            timestamp: Some(DateTime::now()),
+            payload: payload.into(),
+            timestamp: None,
         }
     }
 
+    #[cfg(test)]
+    pub fn sent_now(mut self) -> Self {
+        self.timestamp = Some(DateTime::now());
+        self
+    }
+
     #[cfg(test)]
     pub(crate) fn new_binary(topic: &str, payload: impl Into<Vec<u8>>) -> Self {
         Message {
             topic: topic.to_string(),
             payload: payload.into(),
-            timestamp: Some(DateTime::now()),
+            timestamp: None,
         }
     }
 

crates/extensions/tedge_flows/src/js_script.rs

@@ -155,7 +155,7 @@ mod tests {
         let js = "export function onMessage(msg) { return [msg]; };";
         let (runtime, script) = runtime_with(js).await;
 
-        let input = Message::new("te/main/device///m/", "hello world");
+        let input = Message::new("te/main/device///m/", "hello world").sent_now();
         let output = input.clone();
         assert_eq!(
             script
@@ -171,7 +171,7 @@ mod tests {
         let js = "export function onMessage(msg) { return msg; };";
         let (runtime, script) = runtime_with(js).await;
 
-        let input = Message::new("te/main/device///m/", "hello world");
+        let input = Message::new("te/main/device///m/", "hello world").sent_now();
         let output = input.clone();
         assert_eq!(
             script
@@ -255,11 +255,10 @@ export function onMessage(message, config) {
             "collectd/h/memory/percent-used",
             "1748440192.104:19.9289468288182",
         );
-        let mut output = Message::new(
+        let output = Message::new(
             "te/device/main///m/collectd",
             r#"{"time": 1748440192.104, "memory": {"percent-used": 19.9289468288182}}"#,
         );
-        output.timestamp = None;
         assert_eq!(
             script
                 .on_message(&runtime, &DateTime::now(), &input)
@@ -279,8 +278,7 @@ export async function onMessage(message, config) {
         let (runtime, script) = runtime_with(js).await;
 
         let input = Message::new("dummy", "content");
-        let mut output = Message::new("foo/bar", r#"{foo:"bar"}"#);
-        output.timestamp = None;
+        let output = Message::new("foo/bar", r#"{foo:"bar"}"#);
         assert_eq!(
             script
                 .on_message(&runtime, &DateTime::now(), &input)
@@ -364,8 +362,7 @@ export async function onMessage(message, config) {
         let (runtime, script) = runtime_with(js).await;
 
         let input = Message::new_binary("encoded", [240, 159, 146, 150]);
-        let mut output = Message::new("decoded", "💖");
-        output.timestamp = None;
+        let output = Message::new("decoded", "💖");
         assert_eq!(
             script
                 .on_message(&runtime, &DateTime::now(), &input)
@@ -389,8 +386,7 @@ export async function onMessage(message, config) {
         let (runtime, script) = runtime_with(js).await;
 
         let input = Message::new("decoded", "💖");
-        let mut output = Message::new_binary("encoded", [240, 159, 146, 150]);
-        output.timestamp = None;
+        let output = Message::new_binary("encoded", [240, 159, 146, 150]);
         assert_eq!(
             script
                 .on_message(&runtime, &DateTime::now(), &input)

tests/RobotFramework/tests/tedge_flows/tedge_flows.robot

@@ -38,6 +38,26 @@ Translate complex tedge json to c8y json
     ...    ${transformed_msg}
     ...    [c8y/measurement/measurements/create] {"type":"environment","time":"2025-06-27T08:11:05.301804125Z","temperature":{"temperature":258},"location":{"latitude":32.54,"longitude":-117.67,"altitude":98.6},"pressure":{"pressure":98}}
 
+Using base64 to encode tedge flows input
+    ${encoded_input}    Execute Command
+    ...    echo -n '{"time":"2025-06-27T08:11:05.301804125Z", "temperature": 258}' | base64 --wrap\=0
+    ${transformed_msg}    Execute Command
+    ...    tedge flows test --base64-input te/device/main///m/env ${encoded_input}
+    ...    strip=True
+    Should Be Equal
+    ...    ${transformed_msg}
+    ...    [c8y/measurement/measurements/create] {"type":"env","time":"2025-06-27T08:11:05.301804125Z","temperature":{"temperature":258}}
+
+Using base64 to encode tedge flows output
+    ${encoded_output}    Execute Command
+    ...    echo -n '{"type":"env","time":"2025-06-27T08:11:05.301804125Z","temperature":{"temperature":258}}' | base64 --wrap\=0
+    ${transformed_msg}    Execute Command
+    ...    tedge flows test --base64-output te/device/main///m/env '{"time":"2025-06-27T08:11:05.301804125Z", "temperature": 258}'
+    ...    strip=True
+    Should Be Equal
+    ...    ${transformed_msg}
+    ...    [c8y/measurement/measurements/create] ${encoded_output}
+
 Units are configured using topic metadata
     ${transformed_msg}    Execute Command
     ...    cat /etc/tedge/flows/measurements.samples | awk '{ print $2 }' FS\='INPUT:' | tedge flows test