earl 0.5.2

AI-safe CLI for AI agents
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

//! Tests for the streaming output pipeline and backwards compatibility.

use std::path::Path;

use earl::output::stream::render_streaming_output;
use earl::template::parser::parse_template_hcl;
use earl::template::schema::{ResultDecode, ResultTemplate};
use earl_core::{Redactor, StreamChunk};
use serde_json::Map;
use tokio::sync::mpsc;

// ── render_streaming_output tests ───────────────────────────

#[tokio::test]
async fn json_chunks_are_processed_without_error() {
    let (tx, rx) = mpsc::channel::<StreamChunk>(16);

    tokio::spawn(async move {
        tx.send(StreamChunk {
            data: br#"{"msg":"hello"}"#.to_vec(),
            content_type: Some("application/json".to_string()),
        })
        .await
        .unwrap();
        tx.send(StreamChunk {
            data: br#"{"msg":"world"}"#.to_vec(),
            content_type: Some("application/json".to_string()),
        })
        .await
        .unwrap();
    });

    let result_template = ResultTemplate::default();
    let args = Map::new();
    let redactor = Redactor::new(vec![]);

    let result = render_streaming_output(rx, &result_template, &args, &redactor, true, None).await;
    assert!(result.is_ok(), "should process JSON chunks without error");
}

#[tokio::test]
async fn malformed_json_chunk_is_skipped_and_processing_continues() {
    let (tx, rx) = mpsc::channel::<StreamChunk>(16);

    tokio::spawn(async move {
        tx.send(StreamChunk {
            data: b"not valid json".to_vec(),
            content_type: Some("application/json".to_string()),
        })
        .await
        .unwrap();
        tx.send(StreamChunk {
            data: br#"{"ok":true}"#.to_vec(),
            content_type: Some("application/json".to_string()),
        })
        .await
        .unwrap();
    });

    let result_template = ResultTemplate {
        decode: ResultDecode::Json,
        extract: None,
        output: "{{ result }}".to_string(),
        result_alias: None,
    };
    let args = Map::new();
    let redactor = Redactor::new(vec![]);

    let result = render_streaming_output(rx, &result_template, &args, &redactor, true, None).await;
    assert!(result.is_ok(), "should skip bad chunks and continue");
}

#[tokio::test]
async fn empty_channel_returns_ok() {
    let (_tx, rx) = mpsc::channel::<StreamChunk>(1);
    drop(_tx);

    let result_template = ResultTemplate::default();
    let args = Map::new();
    let redactor = Redactor::new(vec![]);

    let result = render_streaming_output(rx, &result_template, &args, &redactor, false, None).await;
    assert!(result.is_ok(), "empty channel should return Ok");
}

// ── Backwards compatibility tests ───────────────────────────

#[test]
#[cfg(feature = "http")]
fn non_streaming_template_defaults_to_false() {
    let hcl_src = r#"
version = 1
provider = "demo"

command "fetch" {
  title = "Fetch"
  summary = "Fetch data"
  description = "Non-streaming."

  annotations {
    mode = "read"
    secrets = []
  }

  operation {
    protocol = "http"
    method = "GET"
    url = "https://api.example.com/data"
  }

  result {
    output = "{{ result }}"
  }
}
"#;
    let template_file = parse_template_hcl(hcl_src, Path::new(".")).unwrap();
    let cmd = template_file.commands.get("fetch").unwrap();
    assert!(
        !cmd.operation.is_streaming(),
        "should default to false when stream field is absent"
    );
}

#[test]
#[cfg(feature = "http")]
fn template_with_stream_false_returns_false() {
    let hcl_src = r#"
version = 1
provider = "demo"

command "fetch" {
  title = "Fetch"
  summary = "Fetch data"
  description = "Explicitly non-streaming."

  annotations {
    mode = "read"
    secrets = []
  }

  operation {
    protocol = "http"
    method = "GET"
    url = "https://api.example.com/data"
    stream = false
  }

  result {
    output = "{{ result }}"
  }
}
"#;
    let template_file = parse_template_hcl(hcl_src, Path::new(".")).unwrap();
    let cmd = template_file.commands.get("fetch").unwrap();
    assert!(
        !cmd.operation.is_streaming(),
        "should return false when stream = false is explicit"
    );
}