Lines
98.82 %
Functions
100 %
Branches
use super::*;
use std::time::Duration;
use tokio::runtime::Handle;
use tokio::time::sleep;
/// Polls `drain` until it returns at least one item or the budget
/// expires, yielding to let the worker run between polls.
async fn drain_one(eval: &mut ConsoleEval) -> Vec<DrainItem> {
for _ in 0..200 {
let items = eval.drain();
if !items.is_empty() {
return items;
}
sleep(Duration::from_millis(20)).await;
eval.drain()
/// Extract the wire string from a `DrainItem`, panicking if it's a Notice.
fn wire_of(item: &DrainItem) -> &str {
match item {
DrainItem::Routed { wire, .. } => wire,
DrainItem::Unroutable(w) => w,
DrainItem::Notice(n) => panic!("expected wire, got Notice({n:?})"),
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn real_session_evaluates_bare_form_via_envelope_wrapping() {
let mut eval = ConsoleEval::spawn(&Handle::current(), Uuid::nil()).expect("spawn session");
eval.submit("(+ 1 2)".to_string());
let items = drain_one(&mut eval).await;
assert_eq!(items.len(), 1, "got {items:?}");
let wire = wire_of(&items[0]);
assert!(wire.contains(":value 3"), "{wire:?}");
assert!(wire.contains(":id 1"), "{wire:?}");
async fn submit_increments_the_envelope_id() {
let first = drain_one(&mut eval).await;
let w0 = wire_of(&first[0]);
assert!(w0.contains(":id 1"), "{w0:?}");
eval.submit("(+ 2 2)".to_string());
let second = drain_one(&mut eval).await;
let w1 = wire_of(&second[0]);
assert!(w1.contains(":id 2"), "{w1:?}");
assert!(w1.contains(":value 4"), "{w1:?}");
async fn echo_actor_round_trips_the_submitted_form() {
let mut eval = ConsoleEval::echo(&Handle::current());
eval.submit("(foo)".to_string());
assert_eq!(wire, "(:id 1 :form (foo))");
async fn interrupt_cancels_an_inflight_eval() {
let ctx = ScriptCtx::new(Uuid::nil()).with_limits(ScriptLimits {
fuel: u64::MAX,
..ScriptLimits::default()
});
let mut eval = ConsoleEval::spawn_with_ctx(&Handle::current(), ctx).expect("spawn session");
eval.submit("(do ((i 0 (+ i 1))) ((>= i 2000000000) i))".to_string());
// Deliberately path-agnostic: this asserts the USER-VISIBLE contract, that
// an interrupt cancels the eval, whichever mechanism gets there — the
// in-Wasm epoch trap if the worker already started, the pre-start latch if
// it has not. Which one wins depends on host speed, so pinning either here
// would be a flake. The epoch trap itself is pinned deterministically by
// rpc's epoch_bumper_cancels_inflight_long_eval, which makes the pre-start
// path structurally unreachable instead of racing it.
sleep(Duration::from_millis(40)).await;
eval.interrupt();
assert!(wire.contains(":code interrupted"), "{wire:?}");
async fn interrupt_without_inflight_eval_does_not_panic() {
let eval = ConsoleEval::echo(&Handle::current());
async fn submit_after_worker_stops_reports_failure() {
let handle = eval.worker_handle();
handle.abort();
if handle.is_finished() {
break;
sleep(Duration::from_millis(10)).await;
assert!(
eval.submit("(x)".to_string()).is_none(),
"submit must return None once the worker has stopped"
);
async fn drain_surfaces_worker_stopped_notice_exactly_once() {
let first = eval.drain();
assert_eq!(first.len(), 1, "expected one item, got {first:?}");
matches!(&first[0], DrainItem::Notice(n) if n == WORKER_STOPPED_NOTICE),
"expected Notice, got {first:?}"
eval.drain().is_empty(),
"the notice must not repeat on later drains"
async fn worker_breaks_when_receiver_is_dropped_mid_eval() {
let ConsoleEval {
forms_tx,
results_rx,
..
} = eval;
// Dropping the receiver before any frame exists makes the send-failure
// break the only way out; forms_tx lives to the end of the test, so recv
// never returns None.
drop(results_rx);
forms_tx.send("(a)".to_string()).expect("queue a frame");
handle.is_finished(),
"worker must break out of its loop once the receiver is gone"
async fn dropping_the_eval_ends_the_worker() {
drop(eval);
return;
assert!(handle.is_finished(), "worker did not exit after drop");
async fn submit_returns_monotonic_ids() {
let id0 = eval.submit("(a)".to_string()).expect("id0");
let id1 = eval.submit("(b)".to_string()).expect("id1");
let id2 = eval.submit("(c)".to_string()).expect("id2");
assert_eq!(id0, 1, "ids start at 1; 0 is reserved for parse errors");
assert_eq!(id1, 2);
assert_eq!(id2, 3);
async fn drain_classifies_echo_request_frames_as_unroutable() {
eval.submit("(x)".to_string());
assert_eq!(items.len(), 1);
// Echo sends back "(:id 1 :form (x))" — a request envelope, not a
// response; parse_response rejects it (no :value/:error) → Unroutable.
matches!(&items[0], DrainItem::Unroutable(_)),
"echo frame must be Unroutable, got {items:?}"
#[test]
fn classify_response_positive_id_is_routed() {
let wire = "(:id 5 :value 42)".to_string();
let item = classify_response(wire.clone());
matches!(item, DrainItem::Routed { id: 5, .. }),
"expected Routed{{id:5}}, got {item:?}"
fn classify_response_id_zero_is_unroutable() {
let wire = "(:id 0 :value 42)".to_string();
let item = classify_response(wire);
matches!(item, DrainItem::Unroutable(_)),
"id 0 must be Unroutable"
fn classify_response_parse_error_is_unroutable() {
let wire = "((((".to_string();
"parse error must be Unroutable"
fn classify_response_string_id_is_unroutable() {
let wire = r#"(:id "abc" :value 1)"#.to_string();
"string id must be Unroutable"