flowrs-core 1.0.2

Core components of the flowrs framework for directed graph workflows
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166

#[cfg(test)]
mod tests {
    use crate::action::*;
    use crate::error::*;
    use crate::node::*;
    use async_trait::async_trait;

    // Simple test context
    #[derive(Debug, Clone)]
    struct TestContext {
        value: i32,
    }

    // Test node that adds a value and returns a complete outcome
    struct AddNode {
        id: String,
    }

    impl AddNode {
        fn new(id: impl Into<String>) -> Self {
            Self { id: id.into() }
        }
    }

    #[async_trait]
    impl Node<TestContext, DefaultAction> for AddNode {
        type Output = i32;

        fn id(&self) -> String {
            self.id.clone()
        }

        async fn process(
            &self,
            ctx: &mut TestContext,
        ) -> Result<NodeOutcome<Self::Output, DefaultAction>, FlowrsError> {
            ctx.value += 1;
            Ok(NodeOutcome::Success(ctx.value))
        }
    }

    // Test node that checks if value is positive and returns different actions based on that
    struct CheckPositiveNode {
        id: String,
    }

    impl CheckPositiveNode {
        fn new(id: impl Into<String>) -> Self {
            Self { id: id.into() }
        }
    }

    #[async_trait]
    impl Node<TestContext, DefaultAction> for CheckPositiveNode {
        type Output = bool;

        fn id(&self) -> String {
            self.id.clone()
        }

        async fn process(
            &self,
            ctx: &mut TestContext,
        ) -> Result<NodeOutcome<Self::Output, DefaultAction>, FlowrsError> {
            if ctx.value > 0 {
                Ok(NodeOutcome::RouteToAction(DefaultAction::Next))
            } else {
                Ok(NodeOutcome::RouteToAction(DefaultAction::Error))
            }
        }
    }

    // Test node that fails during processing
    struct FailingNode {
        id: String,
    }

    impl FailingNode {
        fn new(id: impl Into<String>) -> Self {
            Self { id: id.into() }
        }
    }

    #[async_trait]
    impl Node<TestContext, DefaultAction> for FailingNode {
        type Output = bool;

        fn id(&self) -> String {
            self.id.clone()
        }

        async fn process(
            &self,
            _ctx: &mut TestContext,
        ) -> Result<NodeOutcome<Self::Output, DefaultAction>, FlowrsError> {
            Err(FlowrsError::node_execution(
                self.id(),
                "This node always fails",
            ))
        }
    }

    #[tokio::test]
    async fn test_add_node() {
        let node = AddNode::new("add");
        let mut ctx = TestContext { value: 5 };

        let result = node.process(&mut ctx).await;
        assert!(result.is_ok());

        match result.unwrap() {
            NodeOutcome::Success(value) => {
                assert_eq!(value, 6);
                assert_eq!(ctx.value, 6);
            }
            _ => panic!("Expected Success outcome"),
        }
    }

    #[tokio::test]
    async fn test_check_positive_node() {
        let node = CheckPositiveNode::new("check_positive");

        // Test with positive value
        let mut ctx_positive = TestContext { value: 5 };
        let result_positive = node.process(&mut ctx_positive).await;
        assert!(result_positive.is_ok());

        match result_positive.unwrap() {
            NodeOutcome::RouteToAction(action) => {
                assert_eq!(action, DefaultAction::Next);
            }
            _ => panic!("Expected RouteToAction outcome"),
        }

        // Test with negative value
        let mut ctx_negative = TestContext { value: -5 };
        let result_negative = node.process(&mut ctx_negative).await;
        assert!(result_negative.is_ok());

        match result_negative.unwrap() {
            NodeOutcome::RouteToAction(action) => {
                assert_eq!(action, DefaultAction::Error);
            }
            _ => panic!("Expected RouteToAction outcome"),
        }
    }

    #[tokio::test]
    async fn test_failing_node() {
        let node = FailingNode::new("failing");
        let mut ctx = TestContext { value: 0 };

        let result = node.process(&mut ctx).await;
        assert!(result.is_err());

        let error = result.unwrap_err();
        match error {
            FlowrsError::NodeExecution(msg) => {
                assert!(msg.contains("failing"));
                assert!(msg.contains("This node always fails"));
            }
            _ => panic!("Expected NodeExecution error"),
        }
    }
}