feat: added execution test

Author: raphael-goetz

crates/taurus/src/app/worker.rs

@@ -342,3 +342,196 @@ fn now_unix_ms() -> i64 {
         .unwrap_or(0)
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Mutex;
+    use std::time::Duration;
+
+    use prost::Message;
+    use serde::Deserialize;
+    use taurus_core::runtime::engine::ExecutionEngine;
+    use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter;
+    use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime;
+    use tucana::shared::{
+        ValidationFlow, execution_result,
+        helper::value::{from_json_value, to_json_value},
+    };
+
+    #[derive(Deserialize)]
+    struct FixtureInput {
+        input: Option<serde_json::Value>,
+        expected_result: serde_json::Value,
+    }
+
+    #[derive(Deserialize)]
+    struct FlowFixture {
+        inputs: Vec<FixtureInput>,
+        flow: ValidationFlow,
+    }
+
+    static NATS_TEST_LOCK: Mutex<()> = Mutex::new(());
+
+    #[test]
+    #[ignore = "requires a running NATS server at NATS_URL or nats://127.0.0.1:4222"]
+    fn test_execution_request_returns_execution_result_over_nats() {
+        let _lock = NATS_TEST_LOCK
+            .lock()
+            .expect("NATS test lock should not be poisoned");
+        runtime().block_on(async {
+            let client = connect_test_nats().await;
+            let worker = spawn_test_worker(client.clone());
+            wait_for_worker_subscription().await;
+
+            let execution_id = ExecutionId::new_v4();
+            let fixture = load_fixture("flows/01_return_object.json");
+            let expected_result = fixture.inputs[0].expected_result.clone();
+            let flow = execution_flow_from_fixture(fixture);
+            let response = request_execution_result(&client, execution_id, flow.encode_to_vec())
+                .await
+                .expect("test execution request should receive an ExecutionResult response");
+
+            worker.abort();
+
+            assert_eq!(response.execution_identifier, execution_id.to_string());
+            assert_eq!(response.flow_id, flow.flow_id);
+            assert!(response.started_at > 0);
+            assert!(response.finished_at >= response.started_at);
+            assert_eq!(response.input, None);
+            assert!(response.node_execution_results.is_empty());
+
+            match response.result {
+                Some(execution_result::Result::Success(value)) => {
+                    assert_eq!(to_json_value(value), expected_result);
+                }
+                other => panic!("expected successful test execution result, got {:?}", other),
+            }
+        });
+    }
+
+    #[test]
+    #[ignore = "requires a running NATS server at NATS_URL or nats://127.0.0.1:4222"]
+    fn test_execution_request_returns_decode_error_over_nats() {
+        let _lock = NATS_TEST_LOCK
+            .lock()
+            .expect("NATS test lock should not be poisoned");
+        runtime().block_on(async {
+            let client = connect_test_nats().await;
+            let worker = spawn_test_worker(client.clone());
+            wait_for_worker_subscription().await;
+
+            let execution_id = ExecutionId::new_v4();
+            let response =
+                request_execution_result(&client, execution_id, b"not protobuf".to_vec())
+                    .await
+                    .expect("malformed test execution request should receive an error response");
+
+            worker.abort();
+
+            assert_eq!(response.execution_identifier, execution_id.to_string());
+            assert_eq!(response.flow_id, 0);
+            assert!(response.started_at > 0);
+            assert!(response.finished_at >= response.started_at);
+            assert_eq!(response.input, None);
+            assert!(response.node_execution_results.is_empty());
+
+            match response.result {
+                Some(execution_result::Result::Error(error)) => {
+                    assert_eq!(error.code, "T-TAURUS-000001");
+                    assert_eq!(error.category, "ExecutionFlowDecodeError");
+                    assert_eq!(
+                        error.message,
+                        "Failed to decode test execution flow payload"
+                    );
+                }
+                other => panic!("expected decode error result, got {:?}", other),
+            }
+        });
+    }
+
+    fn runtime() -> tokio::runtime::Runtime {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should build")
+    }
+
+    async fn connect_test_nats() -> async_nats::Client {
+        let nats_url =
+            std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string());
+        async_nats::connect(&nats_url)
+            .await
+            .unwrap_or_else(|err| panic!("failed to connect to NATS at {nats_url}: {err}"))
+    }
+
+    fn spawn_test_worker(client: async_nats::Client) -> tokio::task::JoinHandle<()> {
+        let engine = ExecutionEngine::new();
+        let nats_remote = NATSRemoteRuntime::new(client.clone());
+        let runtime_emitter = NATSRespondEmitter::new(client.clone());
+        spawn_worker(client, engine, nats_remote, runtime_emitter, None)
+    }
+
+    async fn wait_for_worker_subscription() {
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+
+    async fn request_execution_result(
+        client: &async_nats::Client,
+        execution_id: ExecutionId,
+        payload: Vec<u8>,
+    ) -> Result<ExecutionResult, String> {
+        let subject = format!("test_executions.{execution_id}");
+
+        for attempt in 1..=10 {
+            match tokio::time::timeout(
+                Duration::from_secs(2),
+                client.request(subject.clone(), payload.clone().into()),
+            )
+            .await
+            {
+                Ok(Ok(message)) => {
+                    return ExecutionResult::decode(&*message.payload)
+                        .map_err(|err| format!("failed to decode ExecutionResult: {err}"));
+                }
+                Ok(Err(err)) if attempt < 10 => {
+                    log::debug!(
+                        "test execution request attempt {} failed before subscription was ready: {:?}",
+                        attempt,
+                        err
+                    );
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                }
+                Ok(Err(err)) => return Err(format!("NATS request failed: {err}")),
+                Err(_) if attempt < 10 => {
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                }
+                Err(_) => return Err("timed out waiting for test execution response".to_string()),
+            }
+        }
+
+        Err("test execution request did not complete".to_string())
+    }
+
+    fn load_fixture(path: &str) -> FlowFixture {
+        let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("../..")
+            .join(path);
+        let content = std::fs::read_to_string(&path)
+            .unwrap_or_else(|err| panic!("failed to read fixture {}: {err}", path.display()));
+        serde_json::from_str(&content)
+            .unwrap_or_else(|err| panic!("failed to parse fixture {}: {err}", path.display()))
+    }
+
+    fn execution_flow_from_fixture(fixture: FlowFixture) -> ExecutionFlow {
+        ExecutionFlow {
+            flow_id: fixture.flow.flow_id,
+            project_id: fixture.flow.project_id,
+            starting_node_id: fixture.flow.starting_node_id,
+            node_functions: fixture.flow.node_functions,
+            input_value: fixture
+                .inputs
+                .first()
+                .and_then(|input| input.input.clone().map(from_json_value)),
+        }
+    }
+}