use std::collections::BTreeMap;
use std::error::Error;
use std::num::NonZeroU64;
use haematite_query::page::{Allowance, Origin, Resume};
use serde_json::value::RawValue;
use serde_json::{Value, json};
use super::{Binding, Continuation, Engine, Options, Rows, Seal};
use crate::serve::contracts::Caller;
use crate::serve::protocol::Version;
type TestResult = Result<(), Box<dyn Error>>;
fn raw(value: &str) -> Result<Box<RawValue>, serde_json::Error> {
RawValue::from_string(value.to_owned())
}
#[derive(Clone)]
struct Request {
query: String,
parameters: BTreeMap<String, Box<RawValue>>,
engine: Engine,
caller: Caller,
allowance: Allowance,
}
impl Request {
fn new() -> Result<Self, Box<dyn Error>> {
Ok(Self {
query: ".private_original_question".to_owned(),
parameters: BTreeMap::from([("threshold".to_owned(), raw(r#"{"a":1,"b":[2,3]}"#)?)]),
engine: Engine {
protocol: 1,
worker: "0.1.0".to_owned(),
engine: "jaq".to_owned(),
jaq_core: "3.1.0".to_owned(),
jaq_std: "3.0.1".to_owned(),
jaq_json: "2.0.2".to_owned(),
},
caller: Caller::new(
"local",
"private-original-person",
Some("responsible-person".to_owned()),
)?,
allowance: Allowance {
rows: NonZeroU64::new(10).ok_or("zero rows")?,
bytes: NonZeroU64::new(1000).ok_or("zero bytes")?,
scan_records: 100,
},
})
}
fn binding(&self) -> Result<Binding, crate::serve::storage::Failure> {
Binding::new(
&self.query,
&self.parameters,
&self.engine,
&self.caller,
&self.allowance,
Rows::Values,
)
}
}
fn continuation(request: &Request) -> Result<Continuation, Box<dyn Error>> {
Ok(Continuation::new(
"main".to_owned(),
Version {
generation: 7,
sequence: 11,
},
Resume {
records_read: 3,
pending: Origin {
collection: "charts".to_owned(),
id: "chart-3".to_owned(),
version: raw(r#"{"generation":"7","sequence":"11"}"#)?,
score: Some(0.75),
},
discard_outputs: 2,
passed: false,
passed_counts: Vec::new(),
},
raw("1700000000.125")?,
request.binding()?,
)?)
}
#[test]
fn each_bound_kind_changed_alone_is_named_without_private_values() -> TestResult {
let original = Request::new()?;
let token = continuation(&original)?;
for kind in ["query", "parameters", "engine", "caller", "allowances"] {
let mut changed = original.clone();
match kind {
"query" => changed.query = ".other_question".to_owned(),
"parameters" => {
changed.parameters.insert("threshold".to_owned(), raw("4")?);
}
"engine" => changed.engine.engine = "different-engine".to_owned(),
"caller" => changed.caller = Caller::new("local", "other-person", None)?,
"allowances" => changed.allowance.rows = NonZeroU64::new(11).ok_or("zero rows")?,
other => return Err(format!("unknown test kind {other}").into()),
}
let error = token
.check_binding(&changed.binding()?)
.err()
.ok_or("changed binding accepted")?;
assert_eq!(error.code, format!("query_continuation_{kind}"));
assert!(error.message.contains(kind));
assert!(!error.message.contains("private-original-person"));
assert!(!error.message.contains("private_original_question"));
}
Ok(())
}
#[test]
fn each_engine_version_caller_member_and_allowance_is_bound() -> TestResult {
let original = Request::new()?;
let token = continuation(&original)?;
for member in [
"protocol",
"worker",
"jaq_core",
"jaq_std",
"jaq_json",
"issuer",
"subject",
"responsible",
"bytes",
"scan",
] {
let mut changed = original.clone();
let kind = match member {
"protocol" => {
changed.engine.protocol += 1;
"engine"
}
"worker" => {
changed.engine.worker.push('x');
"engine"
}
"jaq_core" => {
changed.engine.jaq_core.push('x');
"engine"
}
"jaq_std" => {
changed.engine.jaq_std.push('x');
"engine"
}
"jaq_json" => {
changed.engine.jaq_json.push('x');
"engine"
}
"issuer" => {
changed.caller = Caller::new(
"other-issuer",
original.caller.subject(),
original.caller.responsible().map(str::to_owned),
)?;
"caller"
}
"subject" => {
changed.caller = Caller::new(
original.caller.issuer(),
"other-subject",
original.caller.responsible().map(str::to_owned),
)?;
"caller"
}
"responsible" => {
changed.caller =
Caller::new(original.caller.issuer(), original.caller.subject(), None)?;
"caller"
}
"bytes" => {
changed.allowance.bytes = NonZeroU64::new(1001).ok_or("zero bytes")?;
"allowances"
}
"scan" => {
changed.allowance.scan_records += 1;
"allowances"
}
other => return Err(format!("unknown test member {other}").into()),
};
let error = token
.check_binding(&changed.binding()?)
.err()
.ok_or("unbound member")?;
assert_eq!(error.code, format!("query_continuation_{kind}"));
}
Ok(())
}
#[test]
fn the_row_form_is_bound_with_the_allowances() -> TestResult {
let request = Request::new()?;
let origins = Binding::new(
&request.query,
&request.parameters,
&request.engine,
&request.caller,
&request.allowance,
Rows::Origins,
)?;
let error = continuation(&request)?
.check_binding(&origins)
.err()
.ok_or("a values token was taken for origins")?;
assert_eq!(error.code, "query_continuation_allowances");
Ok(())
}
#[test]
fn parameter_object_order_and_whitespace_do_not_change_the_binding() -> TestResult {
let original = Request::new()?;
let mut reordered = original.clone();
reordered
.parameters
.insert("threshold".to_owned(), raw(r#"{ "b": [2, 3], "a": 1 }"#)?);
continuation(&original)?.check_binding(&reordered.binding()?)?;
Ok(())
}
#[test]
fn continuation_decode_and_encode_are_byte_identical_and_keep_replay_state() -> TestResult {
let request = Request::new()?;
let seal = Seal::new()?;
let encoded = continuation(&request)?.encode(&seal)?;
let decoded = Continuation::decode(&encoded, &seal)?;
decoded.check_binding(&request.binding()?)?;
assert_eq!(decoded.plain()?, continuation(&request)?.plain()?);
assert_eq!(decoded.branch, "main");
assert_eq!(
decoded.version,
Version {
generation: 7,
sequence: 11
}
);
assert_eq!(decoded.resume.records_read, 3);
assert_eq!(decoded.resume.discard_outputs, 2);
assert_eq!(decoded.resume.pending.id, "chart-3");
assert_eq!(decoded.now.get(), "1700000000.125");
Ok(())
}
#[test]
fn a_later_request_checks_branch_and_only_an_explicit_version() -> TestResult {
let token = continuation(&Request::new()?)?;
token.check_moment("main", None)?;
token.check_moment(
"main",
Some(Version {
generation: 7,
sequence: 11,
}),
)?;
for (branch, version) in [
("other-branch", None),
(
"main",
Some(Version {
generation: 8,
sequence: 11,
}),
),
(
"main",
Some(Version {
generation: 7,
sequence: 12,
}),
),
] {
let error = token
.check_moment(branch, version)
.err()
.ok_or("different moment accepted")?;
assert_eq!(error.code, "query_continuation_moment");
assert!(!error.message.contains("other-branch"));
}
assert_eq!(token.now.get(), "1700000000.125");
Ok(())
}
#[test]
fn truncated_padded_and_noncanonical_tokens_are_refused() -> TestResult {
let seal = Seal::new()?;
let encoded = continuation(&Request::new()?)?.encode(&seal)?;
let bytes = seal.open(&encoded)?;
let mut padded = bytes.clone();
padded.push(b' ');
let mut trailing = bytes;
trailing.extend_from_slice(b"{}");
for invalid in [
encoded
.get(..encoded.len() - 2)
.ok_or("encoded token too short")?
.to_owned(),
format!(" {encoded}"),
encoded.to_uppercase(),
seal.seal(&padded)?,
seal.seal(&trailing)?,
] {
assert!(Continuation::decode(&invalid, &seal).is_err());
}
Ok(())
}
fn replace_token(
seal: &Seal,
encoded: &str,
from: &str,
to: &str,
) -> Result<String, Box<dyn Error>> {
let bytes = String::from_utf8(seal.open(encoded)?)?;
if bytes.matches(from).count() != 1 {
return Err("token replacement did not name exactly one field".into());
}
Ok(seal.seal(bytes.replacen(from, to, 1).as_bytes())?)
}
#[test]
fn unknown_format_zero_scan_count_and_overflowing_score_are_refused() -> TestResult {
let seal = Seal::new()?;
let encoded = continuation(&Request::new()?)?.encode(&seal)?;
let format = replace_token(&seal, &encoded, "\"format\":1", "\"format\":2")?;
let error = Continuation::decode(&format, &seal)
.err()
.ok_or("unknown format accepted")?;
assert_eq!(error.code, "query_continuation_format");
let zero = replace_token(&seal, &encoded, "\"records_read\":3", "\"records_read\":0")?;
assert!(Continuation::decode(&zero, &seal).is_err());
let overflow = replace_token(&seal, &encoded, "\"score\":0.75", "\"score\":1e400")?;
assert!(Continuation::decode(&overflow, &seal).is_err());
let mut invalid = continuation(&Request::new()?)?;
invalid.resume.pending.score = Some(f64::NAN);
assert!(invalid.encode(&seal).is_err());
Ok(())
}
fn pending_p2(request: &Request) -> Result<Continuation, Box<dyn Error>> {
let mut token = continuation(request)?;
token.resume.pending.collection = "patients".to_owned();
token.resume.pending.id = "p2".to_owned();
Ok(token)
}
#[test]
fn a_sealed_token_names_no_collection_id_or_field_of_its_record() -> TestResult {
let token = pending_p2(&Request::new()?)?;
let encoded = token.encode(&Seal::new()?)?;
let plain = token.plain()?;
for hidden in [
&b"patients"[..],
&b"\"p2\""[..],
&b"\"id\":\"p2\""[..],
&b"\"pending\""[..],
&plain[..],
] {
let written = super::codec::encode(hidden);
assert!(
!encoded.contains(&written),
"{written} is readable in {encoded}"
);
}
Ok(())
}
#[test]
fn one_changed_hex_digit_anywhere_is_refused_as_invalid() -> TestResult {
let seal = Seal::new()?;
let encoded = pending_p2(&Request::new()?)?.encode(&seal)?;
for (index, digit) in encoded.char_indices() {
let other = if digit == '0' { "1" } else { "0" };
let mut changed = encoded.clone();
changed.replace_range(index..=index, other);
let error = Continuation::decode(&changed, &seal)
.err()
.ok_or_else(|| format!("a change at hex digit {index} was accepted"))?;
assert_eq!(error.code, "query_continuation_invalid");
}
Ok(())
}
#[test]
fn a_token_sealed_by_another_key_is_refused_as_invalid() -> TestResult {
let encoded = pending_p2(&Request::new()?)?.encode(&Seal::new()?)?;
let error = Continuation::decode(&encoded, &Seal::new()?)
.err()
.ok_or("another key opened the token")?;
assert_eq!(error.code, "query_continuation_invalid");
assert!(error.message.contains("restarted"), "{}", error.message);
Ok(())
}
#[test]
fn two_tokens_for_one_position_differ_and_open_alike() -> TestResult {
let seal = Seal::new()?;
let token = pending_p2(&Request::new()?)?;
let first = token.encode(&seal)?;
let second = token.encode(&seal)?;
assert_ne!(first, second);
let first = Continuation::decode(&first, &seal)?.plain()?;
let second = Continuation::decode(&second, &seal)?.plain()?;
assert_eq!(first, second);
Ok(())
}
#[test]
fn large_parameter_numbers_and_array_order_remain_distinct() -> TestResult {
for (first, second) in [
("18446744073709551617", "18446744073709551618"),
("[2,3]", "[3,2]"),
] {
let mut original = Request::new()?;
original
.parameters
.insert("threshold".to_owned(), raw(first)?);
let mut changed = original.clone();
changed
.parameters
.insert("threshold".to_owned(), raw(second)?);
let error = continuation(&original)?
.check_binding(&changed.binding()?)
.err()
.ok_or("different parameter values collided")?;
assert_eq!(error.code, "query_continuation_parameters");
}
Ok(())
}
#[test]
fn a_mismatched_aggregate_digest_and_unknown_fields_are_refused() -> TestResult {
let seal = Seal::new()?;
let encoded = continuation(&Request::new()?)?.encode(&seal)?;
let value: Value = serde_json::from_slice(&seal.open(&encoded)?)?;
let aggregate = value["binding"]["aggregate"]
.as_str()
.ok_or("aggregate missing")?;
let replacement = if aggregate.starts_with('0') {
"1".repeat(64)
} else {
"0".repeat(64)
};
let invalid = replace_token(
&seal,
&encoded,
&format!("\"aggregate\":\"{aggregate}\""),
&format!("\"aggregate\":\"{replacement}\""),
)?;
assert!(Continuation::decode(&invalid, &seal).is_err());
let unknown = replace_token(
&seal,
&encoded,
"\"format\":1",
"\"format\":1,\"unexpected\":true",
)?;
assert!(Continuation::decode(&unknown, &seal).is_err());
Ok(())
}
fn settings_value() -> Value {
json!({
"query_time_ms": 1000,
"query_memory_bytes": 64_000_000,
"query_outputs_per_record": 32,
"query_scan_records": 100,
"query_max_running": 2,
"query_worker": "/worker/haematite-jq"
})
}
#[test]
fn query_settings_have_all_present_and_all_absent_states() -> TestResult {
let absent: Options = serde_json::from_value(json!({}))?;
assert!(absent.resolve()?.is_none());
let present: Options = serde_json::from_value(settings_value())?;
let settings = present.resolve()?.ok_or("settings lost")?;
assert_eq!(settings.time_ms.get(), 1000);
assert_eq!(settings.memory_bytes.get(), 64_000_000);
assert_eq!(settings.outputs_per_record.get(), 32);
assert_eq!(settings.scan_records.get(), 100);
assert_eq!(settings.max_running.get(), 2);
assert_eq!(settings.worker.as_os_str(), "/worker/haematite-jq");
Ok(())
}
#[test]
fn each_missing_query_setting_refuses_and_names_the_missing_one() -> TestResult {
for name in super::settings::NAMES {
let mut value = settings_value();
value
.as_object_mut()
.ok_or("settings not object")?
.remove(name);
let options: Options = serde_json::from_value(value)?;
let error = options.resolve().err().ok_or("mixed settings accepted")?;
assert!(error.contains(name), "missing {name}: {error}");
}
let options: Options = serde_json::from_value(json!({"query_time_ms": 1000}))?;
let error = options.resolve().err().ok_or("single setting accepted")?;
for name in &super::settings::NAMES[1..] {
assert!(error.contains(*name));
}
Ok(())
}
#[test]
fn a_records_read_limit_below_two_is_refused_by_name() -> TestResult {
let mut one = settings_value();
one["query_scan_records"] = json!(1);
let options: Options = serde_json::from_value(one)?;
let error = options.resolve().err().ok_or("a limit of one accepted")?;
assert!(error.contains("query_scan_records"), "{error}");
assert!(error.contains("at least 2"), "{error}");
let mut two = settings_value();
two["query_scan_records"] = json!(2);
let options: Options = serde_json::from_value(two)?;
let settings = options.resolve()?.ok_or("no settings")?;
assert_eq!(settings.scan_records.get(), 2);
Ok(())
}
#[test]
fn a_passed_place_with_no_records_read_seals_and_opens() -> TestResult {
let seal = Seal::new()?;
let mut token = continuation(&Request::new()?)?;
token.resume.records_read = 0;
token.resume.discard_outputs = 0;
token.resume.passed = true;
let opened = Continuation::decode(&token.encode(&seal)?, &seal)?;
assert!(opened.resume.passed);
assert_eq!(opened.resume.records_read, 0);
token.resume.passed = false;
assert!(
token.encode(&seal).is_err(),
"a pending record read by no one"
);
Ok(())
}
#[test]
fn zero_query_settings_relative_paths_and_explicit_null_are_refused() -> TestResult {
for name in &super::settings::NAMES[..5] {
let mut value = settings_value();
value[*name] = json!(0);
let options: Options = serde_json::from_value(value)?;
assert!(
options
.resolve()
.err()
.ok_or("zero accepted")?
.contains(*name)
);
}
let mut relative = settings_value();
relative["query_worker"] = json!("haematite-jq");
let options: Options = serde_json::from_value(relative)?;
assert!(
options
.resolve()
.err()
.ok_or("relative path accepted")?
.contains("query_worker")
);
let error = serde_json::from_value::<Options>(json!({"query_time_ms": null}))
.err()
.ok_or("explicit null accepted")?;
assert!(error.to_string().contains("query_time_ms"));
Ok(())
}
#[cfg(unix)]
#[test]
fn mixed_settings_refuse_service_start_before_other_start_checks() -> TestResult {
let directory = tempfile::tempdir()?;
let socket = directory.path().join("service.sock");
let path = directory.path().join("serve.json");
let value = json!({
"data_dir": "relative-database", "socket": socket,
"branch_cache_bytes_per_shard": 1024,
"max_frame": 8192, "max_clients": 2, "max_page": 10, "timeout_ms": 1000,
"query_time_ms": 1000,
"create": {"shards": 1, "cache_bytes_total": 1024, "executor_threads": 1}
});
std::fs::write(&path, serde_json::to_vec(&value)?)?;
let args = [
"--config".to_owned(),
path.to_string_lossy().into_owned(),
"--create".to_owned(),
];
let error = crate::serve::cli::run(args.into_iter())
.err()
.ok_or("mixed settings started service")?;
for name in &super::settings::NAMES[1..] {
assert!(error.contains(*name));
}
assert!(!socket.exists());
Ok(())
}