#[cfg(all(test, feature = "hotpath"))]
pub mod tests {
use std::process::Command;
use hotpath::json::{JsonReport, JsonStreamsList};
fn parse_streams(stdout: &str) -> JsonStreamsList {
let json_start = stdout.find('{').expect("No JSON report in output");
let report: JsonReport = serde_json::Deserializer::from_str(&stdout[json_start..])
.into_iter::<JsonReport>()
.next()
.expect("No JSON value in output")
.expect("Failed to parse JSON report");
report.streams.expect("No streams section in report")
}
#[test]
fn test_default_mode_aggregates_per_callsite() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"agg_streams",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}\nStderr:\n{}",
output.status,
String::from_utf8_lossy(&output.stderr)
);
let stdout = String::from_utf8_lossy(&output.stdout);
let streams = parse_streams(&stdout);
let agg = streams
.data
.iter()
.find(|s| !s.has_custom_label)
.expect("aggregated entry not found");
assert_eq!(agg.instances, 4, "4 streams created at the call site");
assert_eq!(agg.closed_instances, 4, "all streams completed");
assert_eq!(agg.state, None, "aggregated entries report no state");
assert_eq!(agg.items_yielded, 20, "summed across instances");
assert_eq!(agg.iter, 0, "aggregated entries carry no iter suffix");
for label in ["itered", "itered-2", "itered-3"] {
let entry = streams
.data
.iter()
.find(|s| s.label == label)
.unwrap_or_else(|| panic!("per-instance entry {label} not found"));
assert_eq!(entry.instances, 1);
assert_eq!(entry.items_yielded, 2);
}
assert_eq!(streams.data.len(), 4, "one aggregated + three per-instance");
}
#[test]
fn test_iter_param_compiles_without_feature() {
let output = Command::new("cargo")
.args(["build", "-p", "test-streams", "--example", "agg_streams"])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"feature-off build of `stream!(..., iter = true)` failed:\n{}",
String::from_utf8_lossy(&output.stderr)
);
}
#[test]
fn test_basic_streams_output() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"basic_streams",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}",
output.status
);
let stdout = String::from_utf8_lossy(&output.stdout);
let all_expected = [
"number-stream",
"text-stream",
"repeat-stream",
"Stream example completed!",
"Stream yield statistics",
"5", "4", "3", "Yielded",
];
for expected in all_expected {
assert!(
stdout.contains(expected),
"Expected:\n{expected}\n\nGot:\n{stdout}",
);
}
}
#[test]
fn test_streams_closed_state() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"basic_streams",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}",
output.status
);
let stdout = String::from_utf8_lossy(&output.stdout);
let closed_count = stdout.matches("| closed").count();
assert!(
closed_count >= 3,
"Expected at least 3 'closed' states for streams, found {}.\nOutput:\n{}",
closed_count,
stdout
);
}
#[test]
fn test_data_endpoints() {
use hotpath::json::JsonStreamsList;
use std::{thread::sleep, time::Duration};
let mut child = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"basic_streams",
"--features",
"hotpath",
])
.env("HOTPATH_METRICS_PORT", "6774")
.env("TEST_SLEEP_SECONDS", "10")
.spawn()
.expect("Failed to spawn command");
let mut json_text = String::new();
let mut last_error = None;
let all_expected = ["basic_streams.rs", "number-stream", "text-stream"];
for _attempt in 0..12 {
sleep(Duration::from_millis(750));
match ureq::get("http://localhost:6774/streams").call() {
Ok(mut response) => {
json_text = response
.body_mut()
.read_to_string()
.expect("Failed to read response body");
last_error = None;
if all_expected.iter().all(|e| json_text.contains(e)) {
break;
}
}
Err(e) => {
last_error = Some(format!("Request error: {}", e));
}
}
}
if let Some(error) = last_error {
let _ = child.kill();
panic!("Failed after 12 retries: {}", error);
}
for expected in all_expected {
assert!(
json_text.contains(expected),
"Expected:\n{expected}\n\nGot:\n{json_text}",
);
}
let streams: JsonStreamsList =
serde_json::from_str(&json_text).expect("Failed to parse streams JSON");
if let Some(stream) = streams.data.first() {
let logs_url = format!("http://localhost:6774/streams/{}/logs", stream.id);
let response = ureq::get(&logs_url)
.call()
.expect("Failed to call /streams/:id/logs endpoint");
assert_eq!(
response.status(),
200,
"Expected status 200 for /streams/:id/logs endpoint"
);
}
let _ = child.kill();
let _ = child.wait();
}
#[test]
fn test_guard_timeout_output() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"guard_timeout_streams",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Process did not exit successfully.\n\nstderr:\n{}",
String::from_utf8_lossy(&output.stderr)
);
let stdout = String::from_utf8_lossy(&output.stdout);
let expected_content = ["[hotpath]", "| streams", "timeout-stream"];
for expected in expected_content {
assert!(
stdout.contains(expected),
"Expected:\n{expected}\n\nGot:\n{stdout}",
);
}
}
#[test]
fn test_format_none_suppresses_output() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"basic_streams",
"--features",
"hotpath",
])
.env("HOTPATH_OUTPUT_FORMAT", "none")
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Process did not exit successfully.\n\nstderr:\n{}",
String::from_utf8_lossy(&output.stderr)
);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
stdout.contains("Stream example completed!"),
"Application output should still be present.\nGot:\n{stdout}"
);
let not_expected = [
"[hotpath]",
"number-stream",
"text-stream",
"Stream yield statistics",
];
for not_exp in not_expected {
assert!(
!stdout.contains(not_exp),
"Stream output should be suppressed with HOTPATH_OUTPUT_FORMAT=none.\nFound: {not_exp}\nGot:\n{stdout}"
);
}
}
#[test]
fn test_streams_file_output() {
use std::fs;
use std::path::Path;
let output_path = "tmp/streams_output_test.json";
fs::create_dir_all("tmp").ok();
if Path::new(output_path).exists() {
fs::remove_file(output_path).ok();
}
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"streams_file_output",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Process did not exit successfully.\n\nstderr:\n{}",
String::from_utf8_lossy(&output.stderr)
);
assert!(
Path::new(output_path).exists(),
"Output file was not created at {}",
output_path
);
let file_content = fs::read_to_string(output_path).expect("Failed to read output file");
let expected_content = ["number-stream", "\"items_yielded\""];
for expected in expected_content {
assert!(
file_content.contains(expected),
"Expected:\n{expected}\n\nGot:\n{file_content}",
);
}
fs::remove_file(output_path).ok();
}
#[test]
fn test_same_line_call_sites_stay_distinct() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-streams",
"--example",
"same_line_streams",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}\nStderr:\n{}",
output.status,
String::from_utf8_lossy(&output.stderr)
);
let stdout = String::from_utf8_lossy(&output.stdout);
let streams = parse_streams(&stdout);
let a = streams
.data
.iter()
.find(|s| s.label == "same-line-a")
.expect("same-line-a stream not found");
let b = streams
.data
.iter()
.find(|s| s.label == "same-line-b")
.expect("same-line-b stream not found");
assert_ne!(a.id, b.id, "same-line call sites must not share an entry");
assert_eq!(
a.items_yielded, 3,
"counts must not merge across call sites"
);
assert_eq!(
b.items_yielded, 7,
"counts must not merge across call sites"
);
assert_eq!(a.instances, 1);
assert_eq!(b.instances, 1);
assert_eq!(a.closed_instances, 1);
assert_eq!(b.closed_instances, 1);
assert_eq!(a.source, b.source, "one physical line renders one source");
assert_eq!(
a.source.matches(':').count(),
1,
"source must stay file:line"
);
}
}