Add a --processing-time option to tedge flows test

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

Author: didier-wenzek

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

@@ -9,6 +9,7 @@ use anyhow::Error;
 use camino::Utf8PathBuf;
 use tedge_config::TEdgeConfig;
 use tedge_flows::BaseFlowRegistry;
+use tedge_flows::DateTime;
 use tedge_flows::Message;
 use tedge_flows::MessageProcessor;
 
@@ -47,6 +48,11 @@ pub enum TEdgeFlowsCli {
         #[clap(long = "final-on-interval")]
         final_on_interval: bool,
 
+        /// Processing time to be used for the test
+        #[clap(long = "processing-time")]
+        #[arg(value_parser = parse_date)]
+        processing_time: Option<DateTime>,
+
         /// The input payloads are base64 encoded and have to be decoded first
         #[clap(long = "base64-input")]
         base64_input: bool,
@@ -80,6 +86,7 @@ impl BuildCommand for TEdgeFlowsCli {
                 flows_dir,
                 flow,
                 final_on_interval,
+                processing_time,
                 base64_input,
                 base64_output,
                 topic,
@@ -97,6 +104,7 @@ impl BuildCommand for TEdgeFlowsCli {
                     flow,
                     message,
                     final_on_interval,
+                    processing_time,
                     base64_input,
                     base64_output,
                 }
@@ -137,3 +145,9 @@ impl TEdgeFlowsCli {
         Ok(processor)
     }
 }
+
+fn parse_date(src: &str) -> Result<DateTime, String> {
+    let time = humantime::parse_rfc3339_weak(src)
+        .map_err(|e| format!("Unable to parse RFC3339 date: {e}"))?;
+    time.try_into()
+}

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

@@ -22,6 +22,7 @@ pub struct TestCommand {
     pub flow: Option<Utf8PathBuf>,
     pub message: Option<Message>,
     pub final_on_interval: bool,
+    pub processing_time: Option<DateTime>,
     pub base64_input: bool,
     pub base64_output: bool,
 }
@@ -41,13 +42,13 @@ impl Command for TestCommand {
             Some(flow) => TEdgeFlowsCli::load_file(&self.flows_dir, flow).await?,
         };
         if let Some(message) = &self.message {
-            let timestamp = DateTime::now();
+            let timestamp = self.processing_time.unwrap_or_else(|| DateTime::now());
             self.process(&mut processor, message.clone(), timestamp)
                 .await;
         } else {
             let mut stdin = BufReader::new(tokio::io::stdin());
             while let Some(message) = next_message(&mut stdin).await {
-                let timestamp = DateTime::now();
+                let timestamp = self.processing_time.unwrap_or_else(|| DateTime::now());
                 self.process(&mut processor, message, timestamp).await;
             }
         }

crates/extensions/tedge_flows/src/flow.rs

@@ -10,6 +10,7 @@ use serde_json::Value;
 use std::fmt::Display;
 use std::fmt::Formatter;
 use std::time::Duration;
+use std::time::SystemTime;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::Topic;
 use tedge_mqtt_ext::TopicFilter;
@@ -409,6 +410,20 @@ impl DateTime {
     }
 }
 
+impl TryFrom<SystemTime> for DateTime {
+    type Error = String;
+
+    fn try_from(time: SystemTime) -> Result<Self, Self::Error> {
+        match time.duration_since(SystemTime::UNIX_EPOCH) {
+            Ok(elapsed) => Ok(DateTime {
+                seconds: elapsed.as_secs(),
+                nanoseconds: 0,
+            }),
+            Err(_) => Err("SystemTime before UNIX EPOCH!".to_string()),
+        }
+    }
+}
+
 impl TryFrom<OffsetDateTime> for DateTime {
     type Error = FlowError;
 

tests/RobotFramework/tests/tedge_flows/flows/add_timestamp.js

@@ -2,7 +2,7 @@ export function onMessage (message) {
   let payload = JSON.parse(message.payload)
   if (!payload.time) {
     let timestamp = message.timestamp
-    payload.time = timestamp.seconds + (timestamp.nanoseconds / 1e9)
+    payload.time = timestamp.getTime() / 1000;
   }
 
   return [{

tests/RobotFramework/tests/tedge_flows/flows/drop_stragglers.js

@@ -12,7 +12,6 @@ export function onMessage (message, config) {
     }
 
     let timestamp = message.timestamp
-    let time = timestamp.seconds + (timestamp.nanoseconds / 1e9)
     let max = time + (config.max_advance || 1);
     let min = time - (config.max_delay || 10);
 

tests/RobotFramework/tests/tedge_flows/tedge_flows.robot

@@ -10,8 +10,8 @@ Test Tags           theme:tedge_flows
 
 *** Test Cases ***
 Add missing timestamps
-    ${transformed_msg}    Execute Command    tedge flows test te/device/main///m/ '{}'
-    Should Contain    ${transformed_msg}    item=time
+    ${transformed_msg}    Execute Command    tedge flows test --processing-time "2025-06-27 11:31:02" te/device/main///m/ '{}'
+    Should Contain    ${transformed_msg}    item="time":"2025-06-27T11:31:02.000Z"
 
 Convert timestamps to ISO
     ${transformed_msg}    Execute Command    tedge flows test te/device/main///m/ '{"time": 1751023862.000}'