use std::num::NonZeroU64;
use serde_json::{Value, json};
use super::support::{Fixture, TestResult, request};
use super::{documents, rows, values};
use crate::serve::query::Continuation;
const SIX: NonZeroU64 = NonZeroU64::MIN.saturating_add(5);
const ONE: NonZeroU64 = NonZeroU64::MIN;
const QUERY: &str = "{ patients { name } wards { code } }";
fn fixture() -> Result<Fixture, Box<dyn std::error::Error>> {
Fixture::with_wards(
&documents(),
&[
("w0", json!({"code":"4B", "private":"hidden"})),
("w1", json!({"code":"3A", "private":"hidden"})),
],
)
}
fn expected() -> Vec<Value> {
vec![
json!({"name":"Ada"}),
json!({"name":"Bea"}),
json!({"name":"Cy"}),
json!({"code":"4B"}),
json!({"code":"3A"}),
]
}
#[test]
fn several_roots_stream_cut_records_in_root_order_with_their_own_origins() -> TestResult {
let fixture = fixture()?;
let answer = fixture.run(&request(QUERY, SIX), &[])?;
assert_eq!(values(&answer)?, expected());
assert!(answer.get("next").ok_or("missing next")?.is_null());
assert_eq!(
answer.get("records_read").ok_or("missing reads")?,
&json!(5)
);
assert_eq!(answer.get("stages").ok_or("missing stages")?, &json!(["{"]));
let actual = rows(&answer)?;
assert_eq!(actual.len(), 5);
for (row, (collection, id)) in actual.iter().zip([
("patients", "p0"),
("patients", "p1"),
("patients", "p2"),
("wards", "w0"),
("wards", "w1"),
]) {
let origin = row.get("origin").ok_or("missing origin")?;
assert_eq!(
origin.get("collection").ok_or("missing collection")?,
&json!(collection),
);
assert_eq!(origin.get("id").ok_or("missing id")?, &json!(id));
let written = fixture
.written
.get(&format!("{collection}/{id}"))
.ok_or("origin was never written")?;
assert_eq!(
origin.get("version").ok_or("missing version")?,
&serde_json::to_value(written)?,
);
}
let reverse = fixture.run(&request("{ wards { code } patients { name } }", SIX), &[])?;
assert_eq!(
values(&reverse)?,
vec![
json!({"code":"4B"}),
json!({"code":"3A"}),
json!({"name":"Ada"}),
json!({"name":"Bea"}),
json!({"name":"Cy"}),
]
);
Ok(())
}
#[test]
fn second_root_worker_input_is_cut_and_retains_its_bindings() -> TestResult {
let fixture = fixture()?;
let answer = fixture.run(&request(&format!("{QUERY} | '.'"), SIX), &[])?;
assert_eq!(values(&answer)?, expected());
assert_eq!(fixture.inputs()?, expected());
let frames = fixture.frames()?;
assert_eq!(frames.len(), 5);
for (frame, id) in frames[3..].iter().zip(["w0", "w1"]) {
let bindings = frame.get("bindings").ok_or("missing Input bindings")?;
assert_eq!(
bindings.get("collection").ok_or("missing collection")?,
&json!("wards"),
);
assert_eq!(bindings.get("id").ok_or("missing id")?, &json!(id));
let written = fixture
.written
.get(&format!("wards/{id}"))
.ok_or("Input was never written")?;
assert_eq!(
bindings.get("version").ok_or("missing version")?,
&serde_json::to_value(written)?,
);
assert_eq!(bindings.get("score").ok_or("missing score")?, &Value::Null);
assert!(
frame
.get("value")
.ok_or("missing Input value")?
.get("private")
.is_none()
);
}
Ok(())
}
#[test]
fn missing_second_root_is_named_before_any_worker_starts() -> TestResult {
let fixture = fixture()?;
let error = fixture
.run(
&request("{ patients { name } absent { code } } | '.'", SIX),
&[],
)
.err()
.ok_or("missing second collection accepted")?;
assert_eq!(error.code, "collection_missing");
assert_eq!(
error.message,
"collection absent does not exist at the captured version",
);
assert!(!fixture.worker_started());
Ok(())
}
#[test]
fn missing_single_root_is_named_before_any_worker_starts() -> TestResult {
let fixture = fixture()?;
let error = fixture
.run(&request("{ absent { code } } | '.'", SIX), &[])
.err()
.ok_or("missing collection accepted")?;
assert_eq!(error.code, "collection_missing");
assert_eq!(
error.message,
"collection absent does not exist at the captured version",
);
assert!(!fixture.worker_started());
Ok(())
}
#[test]
fn second_root_must_exist_at_the_captured_version_even_if_it_exists_now() -> TestResult {
let fixture = fixture()?;
let mut ask = request(&format!("{QUERY} | '.'"), SIX);
ask.selection.version = Some(fixture.patients_only);
let error = fixture
.run(&ask, &[])
.err()
.ok_or("future collection accepted")?;
assert_eq!(error.code, "collection_missing");
assert_eq!(
error.message,
"collection wards does not exist at the captured version",
);
assert!(!fixture.worker_started());
Ok(())
}
#[test]
fn roots_filter_before_projection_and_access_still_checks_each_record() -> TestResult {
let fixture = fixture()?;
let answer = fixture.run(
&request(
"{ patients(name: \"Ada\") { ward } wards(code: \"4B\") { code } } | '.'",
SIX,
),
&[],
)?;
assert_eq!(
values(&answer)?,
vec![
json!({"ward":{"code":"4B", "beds":12}}),
json!({"code":"4B"})
]
);
assert_eq!(
answer.get("records_read").ok_or("missing reads")?,
&json!(5)
);
let fixture = self::fixture()?;
let answer = fixture.run(
&request(&format!("{QUERY} | '.'"), SIX),
&[("patients", "p1"), ("wards", "w0")],
)?;
assert_eq!(
values(&answer)?,
vec![
json!({"name":"Ada"}),
json!({"name":"Cy"}),
json!({"code":"3A"})
]
);
assert_eq!(fixture.inputs()?, values(&answer)?);
assert_eq!(
answer.get("records_read").ok_or("missing reads")?,
&json!(5)
);
Ok(())
}
#[test]
fn roots_replay_across_the_collection_boundary_without_repeating_records() -> TestResult {
let fixture = fixture()?;
let mut ask = request(&format!("{QUERY} | '.'"), ONE);
let mut result = Vec::new();
for index in 0..5 {
let answer = fixture.run(&ask, &[])?;
let page = values(&answer)?;
assert_eq!(page.len(), 1);
result.extend(page);
let next = answer.get("next").ok_or("missing next")?;
if index == 4 {
assert!(next.is_null());
} else {
let next = next.as_str().ok_or("missing continuation")?;
let token = Continuation::decode(next, &fixture.seal)?;
let (collection, id) = if index < 2 {
("patients", format!("p{}", index + 1))
} else {
("wards", format!("w{}", index - 2))
};
assert_eq!(token.resume.pending.collection, collection);
assert_eq!(token.resume.pending.id, id);
ask.continuation = Some(next.to_owned());
}
}
assert_eq!(result, expected());
Ok(())
}
#[test]
fn every_root_shares_the_same_scan_allowance() -> TestResult {
for limit in [
NonZeroU64::MIN.saturating_add(2),
NonZeroU64::MIN.saturating_add(3),
] {
let mut fixture = fixture()?;
fixture.scan(limit);
let mut ask = request(QUERY, SIX);
let first = fixture.run(&ask, &[])?;
let split = usize::try_from(limit.get())?;
assert_eq!(values(&first)?, expected()[..split]);
assert_eq!(
first.get("records_read").ok_or("missing reads")?,
&json!(limit.get()),
);
let next = first
.get("next")
.and_then(Value::as_str)
.ok_or("missing continuation")?;
let token = Continuation::decode(next, &fixture.seal)?;
let (collection, id) = if limit.get() == 3 {
("patients", "p2")
} else {
("wards", "w0")
};
assert_eq!(token.resume.pending.collection, collection);
assert_eq!(token.resume.pending.id, id);
assert!(token.resume.passed);
ask.continuation = Some(next.to_owned());
let second = fixture.run(&ask, &[])?;
assert_eq!(values(&second)?, expected()[split..]);
assert_eq!(
second.get("records_read").ok_or("missing reads")?,
&json!(5)
);
assert!(second.get("next").ok_or("missing next")?.is_null());
}
Ok(())
}
#[test]
fn second_roots_arguments_are_checked_against_its_captured_schema() -> TestResult {
use crate::serve::query::stages_tests_support::Fixture as SchemaFixture;
let mut fixture = SchemaFixture::new(
Some(json!({"type":"object", "properties":{"name":{"type":"string"}}})),
&[],
)?;
fixture.no_worker();
let error = fixture
.run(
&request(
"{ wards { code } patients(ward: \"4B\") { name } } | '.'",
SIX,
),
&[],
)
.err()
.ok_or("unknown property accepted")?;
assert_eq!(error.code, "query_field_unknown");
assert_eq!(
error.message,
"collection patients has no top-level property ward"
);
Ok(())
}