Brummel
3676eb51e5
The `pipe-buffered` function has been renamed to `pipe-series` to better reflect its functionality. This change involves updating the function name in the example file `examples/data_stream_pipe_buffered.myc` and across the Rust source code in `src/ast/rtl/streams.rs`. Additionally, test cases and documentation have been updated accordingly.
210 lines
7.6 KiB
Rust
210 lines
7.6 KiB
Rust
use myc::ast::environment::Environment;
|
|
use myc::ast::types::Value;
|
|
|
|
/// Verifies that bidirectional type inference correctly propagates stream inner types
|
|
/// into lambda parameters for `pipe` calls with a single stream in a vector `[src]`.
|
|
/// The lambda parameter should be inferred as the stream's Record type, not `Any`.
|
|
#[test]
|
|
fn test_pipe_infers_lambda_param_from_vector_input() {
|
|
let env = Environment::new();
|
|
let source = "(do
|
|
(def src (create-random-ohlc 42 5))
|
|
(pipe [src] (fn [tick] (.close tick)))
|
|
)";
|
|
let dump = env.dump_ast(source).expect("dump_ast failed");
|
|
|
|
// The lambda parameter `tick` should be inferred as the OHLC Record type,
|
|
// NOT as `Any`. This validates the Vector → Stream<T> → T hint resolution.
|
|
assert!(
|
|
!dump.contains("params: Tuple([Any])"),
|
|
"Lambda params should NOT be Any — bidirectional inference failed.\nDump:\n{dump}"
|
|
);
|
|
// The pipe call should return Stream(Float), not Stream(Any)
|
|
assert!(
|
|
dump.contains("Stream(Float)"),
|
|
"Pipe should produce Stream(Float), not Stream(Any).\nDump:\n{dump}"
|
|
);
|
|
|
|
// Also verify it runs correctly
|
|
let res = env.run_script(source);
|
|
assert!(res.is_ok(), "Script failed: {:?}", res.err());
|
|
}
|
|
|
|
/// Verifies bidirectional inference for pipe with multiple stream inputs in a tuple.
|
|
#[test]
|
|
fn test_pipe_infers_lambda_params_from_tuple_inputs() {
|
|
let env = Environment::new();
|
|
let source = "(do
|
|
(def src1 (create-random-ohlc 42 5))
|
|
(def src2 (create-random-ohlc 99 5))
|
|
(pipe [src1 src2] (fn [t1 t2] (+ (.close t1) (.close t2))))
|
|
)";
|
|
let res = env.run_script(source);
|
|
assert!(res.is_ok(), "Script failed: {:?}", res.err());
|
|
}
|
|
|
|
/// Verifies that PipeHook::finalize replaces the `pipe` identifier with a
|
|
/// pre-configured factory Constant — mirroring the SeriesHook pattern.
|
|
/// FAILS before PipeHook is implemented (callee is still `Identifier: pipe`).
|
|
#[test]
|
|
fn test_pipe_callee_is_factory_after_finalize() {
|
|
let env = Environment::new();
|
|
let dump = env.dump_ast("(do
|
|
(def ohlc (create-random-ohlc 42 5))
|
|
(pipe ohlc (fn [bar] (.close bar)))
|
|
)").expect("dump_ast failed");
|
|
|
|
assert!(
|
|
!dump.contains("Identifier: pipe"),
|
|
"PipeHook::finalize should replace `pipe` with a pre-configured factory Constant.\nDump:\n{dump}"
|
|
);
|
|
}
|
|
|
|
/// Field accessor on unknown field in pipe lambda must be a compile-time error.
|
|
/// The lambda param is inferred as the concrete Record type (not Any),
|
|
/// so the compiler can validate field names.
|
|
#[test]
|
|
fn test_pipe_field_typo_is_compile_error() {
|
|
let env = Environment::new();
|
|
let valid = "(do (def ohlc (create-random-ohlc 42 5)) (pipe ohlc (fn [bar] (.close bar))))";
|
|
assert!(env.run_script(valid).is_ok(), "Valid pipe should compile");
|
|
|
|
let typo = "(do (def ohlc (create-random-ohlc 42 5)) (pipe ohlc (fn [bar] (.nonexistent bar))))";
|
|
assert!(
|
|
env.run_script(typo).is_err(),
|
|
"Field typo in pipe lambda must be a compile error"
|
|
);
|
|
}
|
|
|
|
/// A chained pipe (output feeds another pipe) must preserve concrete Stream types.
|
|
#[test]
|
|
fn test_pipe_chain_preserves_concrete_types() {
|
|
let env = Environment::new();
|
|
let source = "(do
|
|
(def ohlc (create-random-ohlc 42 5))
|
|
(def closes (pipe ohlc (fn [bar] (.close bar))))
|
|
(pipe closes (fn [c] (* c 2.0)))
|
|
)";
|
|
let dump = env.dump_ast(source).expect("dump_ast failed");
|
|
assert!(
|
|
dump.contains("Stream(Float)"),
|
|
"Both pipe stages should yield Stream(Float).\nDump:\n{dump}"
|
|
);
|
|
assert!(env.run_script(source).is_ok(), "Chained pipe failed at runtime");
|
|
}
|
|
|
|
/// Pipe with 2 input streams but lambda expecting 1 parameter must be a compile error.
|
|
/// Regression: previously crashed at runtime with "No call frame" instead of a
|
|
/// clean compile-time diagnostic.
|
|
#[test]
|
|
fn test_pipe_param_count_mismatch_is_compile_error() {
|
|
let env = Environment::new();
|
|
let source = "(do
|
|
(def src (create-random-ohlc 42 5))
|
|
(pipe [src src] (fn [bar] (.close bar)))
|
|
)";
|
|
let res = env.run_script(source);
|
|
assert!(res.is_err(), "Pipe with 2 inputs but 1-param lambda must be a compile error");
|
|
|
|
let err = res.unwrap_err();
|
|
assert!(
|
|
err.contains("pipe") && err.contains("parameter"),
|
|
"Error should mention parameter mismatch, got: {err}"
|
|
);
|
|
}
|
|
|
|
/// Verifies that pipe with Optional return (filter pattern) still produces Stream(Float).
|
|
#[test]
|
|
fn test_pipeline_optional_type() {
|
|
let env = Environment::new();
|
|
// The lambda uses an `if` without an `else` returning a float constant.
|
|
// The TypeChecker should deduce `Optional(Float)` for the lambda body,
|
|
// and correctly unwrap it to `Series(Float)` for the pipeline output.
|
|
let source = "(do
|
|
(def src (create-random-ohlc 42 10))
|
|
(def filtered
|
|
(pipe [src]
|
|
(fn [tick]
|
|
(if true
|
|
42.0
|
|
)
|
|
)
|
|
)
|
|
)
|
|
filtered
|
|
)";
|
|
let res = env.run_script(source);
|
|
if let Err(e) = &res {
|
|
panic!("Script failed to compile/run: {:?}", e);
|
|
}
|
|
|
|
let val = res.unwrap();
|
|
if let Value::Stream(s) = val {
|
|
assert_eq!(s.stream_type_name(), "stream");
|
|
} else {
|
|
panic!("Expected a Stream(StreamNode)");
|
|
}
|
|
}
|
|
|
|
/// Verifies that `pipe-series` accumulates values into Series and only fires
|
|
/// the lambda once all series have reached the lookback depth (fill gate).
|
|
/// Lambda parameters must be Series objects, not individual values.
|
|
#[test]
|
|
fn test_pipe_series_fill_gate() {
|
|
let env = Environment::new();
|
|
// create-random-ohlc produces 10 ticks. With lookback 3, the lambda
|
|
// fires from tick 3 onwards (8 times). Each call returns the difference
|
|
// between the current and previous close: (.close (bars 0)) - (.close (bars 1))
|
|
let source = "(do
|
|
(def ohlc (create-random-ohlc 42 10))
|
|
(def result (pipe-series 3 [ohlc]
|
|
(fn [bars]
|
|
(- (.close (bars 0)) (.close (bars 1))))))
|
|
result
|
|
)";
|
|
let res = env.run_script(source);
|
|
assert!(res.is_ok(), "pipe-series script failed: {:?}", res.err());
|
|
|
|
let val = res.unwrap();
|
|
if let Value::Stream(s) = val {
|
|
assert_eq!(s.stream_type_name(), "stream");
|
|
} else {
|
|
panic!("Expected Stream from pipe-series, got: {:?}", val);
|
|
}
|
|
}
|
|
|
|
/// Verifies that pipe-series with multiple inputs waits for all series to fill.
|
|
#[test]
|
|
fn test_pipe_series_multi_input() {
|
|
let env = Environment::new();
|
|
let source = "(do
|
|
(def src1 (create-random-ohlc 42 10))
|
|
(def src2 (create-random-ohlc 99 10))
|
|
(def result (pipe-series 2 [src1 src2]
|
|
(fn [a b]
|
|
(+ (.close (a 0)) (.close (b 0))))))
|
|
result
|
|
)";
|
|
let res = env.run_script(source);
|
|
assert!(res.is_ok(), "pipe-series multi-input failed: {:?}", res.err());
|
|
}
|
|
|
|
/// Verifies that pipe-series lambda params are typed as Series, not raw values.
|
|
/// The arg hint resolver must wrap inner types in Series<T>.
|
|
#[test]
|
|
fn test_pipe_series_lambda_params_are_series() {
|
|
let env = Environment::new();
|
|
let source = "(do
|
|
(def ohlc (create-random-ohlc 42 5))
|
|
(pipe-series 3 [ohlc] (fn [bars] (.close (bars 0))))
|
|
)";
|
|
let dump = env.dump_ast(source).expect("dump_ast failed");
|
|
|
|
// Lambda param should be Series<Record>, not just Record or Any.
|
|
// The AST dump uses Debug format: "Series(Record(...))"
|
|
assert!(
|
|
dump.contains("Series(Record("),
|
|
"Lambda param should be typed as Series, not raw value.\nDump:\n{dump}"
|
|
);
|
|
}
|