Merge pull request #189 from code0-tech/185-add-body-validation

Add Test Execution Body Validation

Author: raphael-goetz

Cargo.lock

@@ -11,7 +11,7 @@ env_logger = "0.11.8"
 prost = "0.14.1"
 tonic = "0.14.1"
 tucana = { version = "0.0.33", features = ["all"] }
-code0-flow = { version = "0.0.14" }
+code0-flow = { version = "0.0.16" }
 serde_json = "1.0.140"
 async-nats = "0.42.0"
 tonic-health = "0.14.1"

Cargo.toml

@@ -7,12 +7,14 @@
 use futures::StreamExt;
 use prost::Message;
 use std::sync::Arc;
+use std::time::SystemTime;
+use code0_flow::flow_validator::verify_flow;
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::Request;
 use tonic::transport::Channel;
 use tucana::sagittarius::execution_logon_request::Data;
 use tucana::sagittarius::execution_service_client::ExecutionServiceClient;
-use tucana::sagittarius::{ExecutionLogonRequest, Logon, TestExecutionResponse};
+use tucana::sagittarius::{ExecutionLogonRequest, Log, Logon, TestExecutionResponse};
 use tucana::shared::{ExecutionFlow, ValidationFlow, Value};
 
 pub struct SagittariusTestExecutionServiceClient {
@@ -90,11 +92,38 @@ impl SagittariusTestExecutionServiceClient {
                 let validation_flow = match self.load_validation_flow(request.flow_id).await {
                     Some(flow) => flow,
                     None => {
-                        return;
+                        continue;
                     }
                 };
 
                 let uuid = uuid::Uuid::new_v4().to_string();
+
+                if let Some(body) = &request.body {
+                   if let Err(rule_violations) = verify_flow(validation_flow.clone(), body.clone()) {
+
+                       let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis().to_string();
+                       let log = Log {
+                         level: "error".to_string(),
+                          timestamp: now,
+                          message: rule_violations.to_string(),
+                      };
+
+                       let execution_result = ExecutionLogonRequest {
+                           data: Some(Data::Response(TestExecutionResponse {
+                               flow_id: request.flow_id,
+                               execution_uuid: uuid,
+                               result: None,
+                               logs: vec![log],
+                           })),
+                       };
+
+                       if let Err(err) = tx.send(execution_result).await {
+                           log::error!("Failed to send ExecutionLogonResponse: {:?}", err);
+                       }
+                       continue
+                    }
+                }
+
                 let execution_flow = ExecutionFlow {
                     flow_id: request.flow_id,
                     starting_node: validation_flow.starting_node,