tinyflows 0.2.0

A Rust-based workflow management solution.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

//! The `http_request` node: an outbound HTTP request.

use async_trait::async_trait;

use crate::data::Item;
use crate::error::Result;
use crate::nodes::{NodeContext, NodeExecutor, NodeOutput};

/// Performs an outbound HTTP request via [`crate::caps::HttpClient`].
#[derive(Debug, Default, Clone)]
pub struct HttpRequestNode;

#[async_trait]
impl NodeExecutor for HttpRequestNode {
    async fn execute(&self, ctx: NodeContext<'_>) -> Result<NodeOutput> {
        // The node's config is the request descriptor; the host's HttpClient interprets it.
        let conn = ctx
            .node
            .config
            .get("connection_ref")
            .and_then(serde_json::Value::as_str);
        let response = ctx.caps.http.request(ctx.node.config.clone(), conn).await?;
        Ok(NodeOutput::main(vec![Item::new(response)]))
    }
}

#[cfg(test)]
mod tests {
    use crate::caps::mock::mock_capabilities;
    use crate::compiler::compile;
    use crate::engine::run;
    use crate::model::{Edge, Node, NodeKind, WorkflowGraph};
    use serde_json::{Value, json};

    fn wf(kind: NodeKind, config: Value) -> WorkflowGraph {
        WorkflowGraph {
            nodes: vec![
                Node {
                    id: "t".into(),
                    kind: NodeKind::Trigger,
                    type_version: 1,
                    name: "t".into(),
                    config: Value::Null,
                    ports: vec![],
                    position: None,
                },
                Node {
                    id: "n".into(),
                    kind,
                    type_version: 1,
                    name: "n".into(),
                    config,
                    ports: vec![],
                    position: None,
                },
            ],
            edges: vec![Edge {
                from_node: "t".into(),
                from_port: "main".into(),
                to_node: "n".into(),
                to_port: "main".into(),
            }],
            ..Default::default()
        }
    }

    #[tokio::test]
    async fn http_request_returns_mock_response() {
        let graph = wf(
            NodeKind::HttpRequest,
            json!({ "method": "GET", "url": "https://example.com" }),
        );
        let compiled = compile(&graph).expect("compile");
        let out = run(&compiled, json!({ "seed": 1 }), &mock_capabilities())
            .await
            .expect("run");
        assert_eq!(out.output["nodes"]["n"]["items"][0]["json"]["status"], 200);
    }

    use super::HttpRequestNode;
    use crate::data::Item;
    use crate::nodes::{NodeContext, NodeExecutor};

    #[tokio::test]
    async fn echoes_method_url_and_threads_connection() {
        let node = Node {
            id: "n".into(),
            kind: NodeKind::HttpRequest,
            type_version: 1,
            name: "n".into(),
            config: json!({ "method": "POST", "url": "https://api.test/x", "connection_ref": "http:acct_2" }),
            ports: vec![],
            position: None,
        };
        let input = vec![Item::new(json!({ "seed": 1 }))];
        let caps = mock_capabilities();
        let run_meta = Value::Null;
        let ctx = NodeContext {
            node: &node,
            input: &input,
            run: &run_meta,
            caps: &caps,
        };
        let out = HttpRequestNode.execute(ctx).await.expect("execute");
        assert_eq!(out.items.len(), 1);
        // The mock HTTP client echoes the request descriptor and the conn ref.
        assert_eq!(out.items[0].json["status"], 200);
        assert_eq!(out.items[0].json["request"]["method"], "POST");
        assert_eq!(out.items[0].json["request"]["url"], "https://api.test/x");
        assert_eq!(out.items[0].json["connection"], "http:acct_2");
    }

    #[tokio::test]
    async fn missing_connection_ref_is_null() {
        let node = Node {
            id: "n".into(),
            kind: NodeKind::HttpRequest,
            type_version: 1,
            name: "n".into(),
            config: json!({ "method": "GET", "url": "u" }),
            ports: vec![],
            position: None,
        };
        let input = vec![];
        let caps = mock_capabilities();
        let run_meta = Value::Null;
        let ctx = NodeContext {
            node: &node,
            input: &input,
            run: &run_meta,
            caps: &caps,
        };
        let out = HttpRequestNode.execute(ctx).await.expect("execute");
        assert_eq!(out.items[0].json["connection"], Value::Null);
    }
}