use crate::door::assistant_test_support::{records, *};
use serde_json::json;
use std::sync::Arc;
fn next_item(stream: &mut Stream) -> Result<String> {
loop {
let event = stream.event()?;
if event.contains("event: item\n") {
return Ok(event);
}
anyhow::ensure!(event.contains("event: replayed\n"), "{event}");
}
}
fn pair() -> (String, String) {
(PERSON.into(), "wards".into())
}
#[test]
fn r3_01_buffered_message_cannot_be_skipped_by_a_later_live_cursor() -> Result {
let f = Fixture::new(json!({}))?;
let hub = f.door.assistant()?;
let initial_store = records::start(&f, records::rows(Vec::new()), |_| Ok(()))?;
let mut first = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=9",
&json!({}),
))?;
first.event()?;
initial_store.finish()?;
let mut message = item(10);
message["durable"] = json!(false);
hub.item(&pair(), &message)?;
let buffered = next_item(&mut first)?;
assert!(
!buffered.contains("id:"),
"non-durable item advanced cursor: {buffered}"
);
hub.item(&pair(), &item(11))?;
let later = next_item(&mut first)?;
assert!(
!later.contains("id: conversation:11"),
"gap advanced cursor: {later}"
);
first.close()?;
let store = records::start(&f, records::rows(vec![item(10), item(11)]), |_| Ok(()))?;
let mut request = f.request(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
);
request
.headers
.push(("last-event-id".into(), "conversation:9".into()));
let mut resumed = Stream::open(crate::door::route::answer(
&f.door,
&f.site,
&request,
crate::serve::pass::now(),
))?;
resumed.event()?;
let ten = next_item(&mut resumed)?;
assert!(
ten.contains("id: conversation:10\n") && ten.contains("item 10"),
"{ten}"
);
assert!(next_item(&mut resumed)?.contains("id: conversation:11\n"));
resumed.close()?;
store.finish()?;
Ok(())
}
#[test]
fn r3_02_last_event_id_overrides_the_first_open_after_parameter() -> Result {
let f = Fixture::new(json!({}))?;
let store = records::start(&f, records::rows((6..=9).map(item).collect()), |_| Ok(()))?;
let mut request = f.request(
"GET",
"/events?conversation=conversation&after=5",
&json!({}),
);
request
.headers
.push(("last-event-id".into(), "conversation:8".into()));
let mut stream = Stream::open(crate::door::route::answer(
&f.door,
&f.site,
&request,
crate::serve::pass::now(),
))?;
stream.event()?;
let event = next_item(&mut stream)?;
assert!(
event.contains("id: conversation:9\n"),
"resent an acknowledged item: {event}"
);
stream.close()?;
store.finish()?;
Ok(())
}
#[test]
fn r3_06_five_hundred_records_use_at_most_nine_page_reads() -> Result {
let f = Fixture::new(json!({}))?;
let store = records::start(&f, records::rows((1..=500).map(item).collect()), |_| Ok(()))?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
))?;
stream.event()?;
for seq in 1..=500 {
assert!(next_item(&mut stream)?.contains(&format!("id: conversation:{seq}\n")));
}
stream.close()?;
let calls = store.finish()?;
let reads = calls
.iter()
.filter(|r| r["operation"]["method"] == "documents.list")
.count();
assert!(
reads <= 500_usize.div_ceil(64) + 1,
"500-item replay made {reads} page reads"
);
assert_eq!(
calls
.iter()
.filter(|r| r["operation"]["method"] == "session.pass")
.count(),
1
);
Ok(())
}
#[test]
fn r3_06_conversation_list_reads_only_meta_content_in_nine_round_trips() -> Result {
let f = Fixture::new(json!({}))?;
let mut items = Vec::new();
for conv in ["a", "b", "c"] {
items.push(records::meta(conv));
for seq in 1..=500 {
let mut v = item(seq);
v["conversation"] = json!(conv);
items.push(v);
}
}
let store = records::start(&f, records::rows(items), |_| Ok(()))?;
let text = written(f.answer("GET", "/conversations", &json!({})))?;
assert!(text.starts_with("HTTP/1.1 200 "), "{text}");
let calls = store.finish()?;
assert!(
calls.len() <= 9,
"three conversations required {} round trips",
calls.len()
);
assert!(
calls
.iter()
.filter(|r| r["operation"]["method"] == "documents.list")
.all(|r| r["operation"]["params"]["content"] == "none"),
"item bodies were read"
);
assert_eq!(
calls
.iter()
.filter(|r| r["operation"]["method"] == "documents.get")
.count(),
3
);
Ok(())
}
#[test]
fn r3_07_durable_live_burst_during_replay_never_lag_detaches() -> Result {
let f = Fixture::new(json!({}))?;
let hub = f.door.assistant()?;
let rows = records::rows((1..=200).map(item).collect());
let store = records::start(&f, Arc::clone(&rows), move |rows| {
for seq in 201..=300 {
let v = item(seq);
rows.lock()
.map_err(|_| anyhow::anyhow!("rows poisoned"))?
.insert(format!("{PERSON}/conversation/{seq:020}"), v.clone());
hub.item(&pair(), &v)?;
}
Ok(())
})?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
))?;
stream.event()?;
for seq in 1..=300 {
let e = next_item(&mut stream)?;
assert!(e.contains(&format!("id: conversation:{seq}\n")), "{e}");
}
let calls = Arc::clone(&store.calls);
let hub = f.door.assistant()?;
rows.lock()
.map_err(|_| anyhow::anyhow!("rows"))?
.insert(format!("{PERSON}/conversation/{:020}", 301), item(301));
hub.item(&pair(), &item(301))?;
let mut event = stream.event()?;
if event.contains("event: replayed") {
event = stream.event()?;
}
assert!(
!event.contains("lagged") && event.contains("item 301"),
"{event}"
);
stream.close()?;
store.finish()?;
assert!(
!calls
.lock()
.map_err(|_| anyhow::anyhow!("calls"))?
.is_empty()
);
Ok(())
}
#[test]
fn r3_08_meta_requires_created_at_ms() -> Result {
let f = Fixture::new(json!({}))?;
let mut meta = records::meta("conversation");
meta.as_object_mut()
.ok_or_else(|| anyhow::anyhow!("meta"))?
.remove("created_at_ms");
let store = records::start(&f, records::rows(vec![meta]), |_| Ok(()))?;
let response = written(f.answer("GET", "/conversations", &json!({})))?;
store.finish()?;
assert!(
response.contains("assistant_record_invalid"),
"meta without created_at_ms admitted: {response}"
);
Ok(())
}
#[test]
fn r3_11_another_conversations_session_transition_reaches_the_tab() -> Result {
let f = Fixture::new(json!({}))?;
let store = records::start(&f, records::rows(Vec::new()), |_| Ok(()))?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
))?;
stream.event()?;
store.finish()?;
let hub = f.door.assistant()?;
for (seq, kind) in [(1, "session_opening"), (2, "session_opened")] {
let mut v = item(seq);
v["conversation"] = json!("next");
v["kind"] = json!(kind);
hub.item(&pair(), &v)?;
}
hub.item(&pair(), &item(1))?;
let first = next_item(&mut stream)?;
assert!(
first.contains("session_opening"),
"transition filtered: {first}"
);
assert!(next_item(&mut stream)?.contains("session_opened"));
stream.close()
}
#[test]
fn r3_12_replayed_marker_has_the_studio_shape() -> Result {
let f = Fixture::new(json!({}))?;
let store = records::start(&f, records::rows(vec![item(1)]), |_| Ok(()))?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
))?;
stream.event()?;
next_item(&mut stream)?;
store.finish()?;
f.door.assistant()?.item(&pair(), &item(2))?;
let marker = stream.event()?;
assert!(
marker.contains("event: replayed\ndata: {\"conversation\":\"conversation\",\"through\":1}"),
"{marker}"
);
stream.close()
}
#[test]
fn r3_14_a_live_frame_sized_item_is_replayable_despite_json_escaping() -> Result {
let f = Fixture::new(json!({"frame_bytes":4096,"events_after_bytes":4096}))?;
let mut value = item(1);
value["text"] = json!("\"".repeat(1950));
let frame = json!({"item":value}).to_string();
assert!(frame.len() < 4096 && frame.len() > 3900);
let store = records::start(&f, records::rows(vec![value]), |_| Ok(()))?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
))?;
stream.event()?;
let event = stream.event()?;
assert!(
event.contains("id: conversation:1"),
"live-admissible record refused: {event}"
);
stream.close()?;
store.finish()?;
Ok(())
}
#[test]
fn r3_gap_no_person_or_agent_pass_is_returned_to_the_browser() -> Result {
let f = Fixture::new(json!({}))?;
let mut seat = f.seat("agent-pass", "wards")?;
let mut stream = Stream::open(f.answer("GET", "/events", &json!({})))?;
let mut text = stream.event()?;
seat.item(1)?;
text.push_str(&stream.event()?);
text.push_str(&written(f.answer("POST", "/start", &json!({})))?);
for secret in ["person-pass", "refresh", &agent_token("agent-pass")?] {
assert!(!text.contains(secret), "credential reached browser");
}
stream.close()
}
#[test]
fn r3_06_pages_halve_before_any_oversized_page_is_delivered() -> Result {
let f = Fixture::new(json!({"events_after_bytes":512}))?;
let mut items = Vec::new();
for seq in 1..=12 {
let mut value = item(seq);
value["text"] = json!("x".repeat(80));
items.push(value);
}
let store = records::start(&f, records::rows(items), |_| Ok(()))?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
))?;
stream.event()?;
for seq in 1..=12 {
assert!(next_item(&mut stream)?.contains(&format!("id: conversation:{seq}\n")));
}
stream.close()?;
let calls = store.finish()?;
let limits: Vec<_> = calls
.iter()
.filter(|r| r["operation"]["method"] == "documents.list")
.filter_map(|r| r["operation"]["params"]["limit"].as_u64())
.collect();
assert!(
limits.starts_with(&[64, 32, 16, 8, 4, 2]),
"page limits were {limits:?}"
);
Ok(())
}
#[test]
fn r3_06_one_record_over_the_decoded_byte_bound_is_named() -> Result {
let f = Fixture::new(json!({"events_after_bytes":512}))?;
let mut value = item(1);
value["text"] = json!("x".repeat(1000));
let store = records::start(&f, records::rows(vec![value]), |_| Ok(()))?;
let mut stream = Stream::open(f.answer(
"GET",
"/events?conversation=conversation&after=0",
&json!({}),
))?;
stream.event()?;
let refused = stream.event()?;
assert!(
refused.contains("assistant_record_item_too_large"),
"{refused}"
);
stream.close()?;
store.finish()?;
Ok(())
}