use std::collections::BTreeMap;
use std::error::Error as StdError;
use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
use std::path::Path;
use haematite_query::handle::{Error, StartupStage};
use haematite_query::{Refusal, page::BINDING_NAMES};
use serde_json::value::RawValue;
use super::{caller_failure, open, request};
use crate::serve::query::Settings;
type TestResult = Result<(), Box<dyn StdError>>;
const FRAME: NonZeroU32 = NonZeroU32::MIN.saturating_add(8191);
fn settings() -> Settings {
Settings {
time_ms: NonZeroU64::MIN.saturating_add(59_999),
memory_bytes: NonZeroU64::MIN.saturating_add(999_999),
outputs_per_record: NonZeroU64::MIN.saturating_add(6),
scan_records: NonZeroU64::MIN.saturating_add(99),
max_running: NonZeroUsize::MIN.saturating_add(1),
worker: Path::new("/usr/bin/true").to_owned(),
}
}
#[test]
fn the_common_start_preserves_every_worker_limit_filter_parameter_and_binding() -> TestResult {
let limits = settings();
let parameters = BTreeMap::from([("now".to_owned(), RawValue::from_string("123".to_owned())?)]);
let start = request(&limits, FRAME, "$now".to_owned(), None, parameters);
assert_eq!(start.protocol, haematite_query::PROTOCOL);
assert_eq!(start.filter, "$now");
assert_eq!(start.limits.time_ms, limits.time_ms);
assert_eq!(start.limits.memory_bytes, limits.memory_bytes.get());
assert_eq!(
start.limits.outputs_per_input,
limits.outputs_per_record.get()
);
assert_eq!(start.limits.frame_bytes, FRAME);
assert_eq!(
start.parameters.get("now").ok_or("parameter lost")?.get(),
"123"
);
assert_eq!(start.binding_names, BINDING_NAMES);
Ok(())
}
#[test]
fn a_start_failed_frame_keeps_its_worker_code_even_if_the_ending_has_another_code() {
for code in [
Refusal::QueryCompile,
Refusal::QueryFunctionRefused,
Refusal::QueryProtocol,
] {
let error = Error::new(Refusal::QueryStopped, "worker diagnostic")
.at_start(StartupStage::Failed(code), Some(123));
let failure = caller_failure(&error);
assert_eq!(
failure.code,
match code {
Refusal::QueryCompile => "query_compile",
Refusal::QueryFunctionRefused => "query_function_refused",
_ => "query_protocol",
}
);
assert!(failure.message.contains("worker diagnostic"));
}
}
#[test]
fn time_and_memory_endings_keep_the_document_names_before_other_start_causes() {
for (code, expected) in [
(Refusal::QueryTimeLimit, "query_time_limit"),
(Refusal::QueryMemoryLimit, "query_memory_limit"),
] {
let error = Error::new(code, "allowance exhausted")
.at_start(StartupStage::Failed(Refusal::QueryCompile), Some(123));
assert_eq!(caller_failure(&error).code, expected);
}
}
#[test]
fn an_identity_mismatch_is_a_caller_engine_mismatch() {
let error = Error::new(Refusal::QueryEngineMismatch, "versions differ")
.at_start(StartupStage::Identity, Some(123));
assert_eq!(caller_failure(&error).code, "query_engine_mismatch");
}
#[test]
fn every_other_start_failure_is_unavailable_with_the_operator_name() {
for (stage, detail) in [
(StartupStage::Spawn, "query_worker_not_executable"),
(StartupStage::Owner, "query_worker_owner_failed"),
(StartupStage::UnexpectedFrame, "query_worker_not_ready"),
(StartupStage::Stopping, "query_worker_stop_failed"),
(StartupStage::SendStart, "query_worker_owner_failed"),
(StartupStage::ReadReady, "query_worker_owner_failed"),
] {
let error = Error::new(Refusal::QueryStopped, "same diagnostic").at_start(stage, Some(123));
let failure = caller_failure(&error);
assert_eq!(failure.code, "query_engine_unavailable");
assert!(failure.message.contains(detail));
}
}
#[test]
fn every_carried_worker_code_matches_its_existing_wire_name() -> TestResult {
for code in [
Refusal::QueryCompile,
Refusal::QueryRuntime,
Refusal::QueryOutputLimit,
Refusal::QueryValueTooLarge,
Refusal::QueryFunctionRefused,
Refusal::QueryProtocol,
Refusal::QueryEngineMismatch,
Refusal::QueryEngineUnavailable,
Refusal::QueryTimeLimit,
Refusal::QueryMemoryLimit,
Refusal::QueryStopped,
Refusal::QueryScanLimit,
] {
let error = Error::new(Refusal::QueryStopped, "worker diagnostic")
.at_start(StartupStage::Failed(code), Some(123));
assert_eq!(
serde_json::to_string(&caller_failure(&error).code)?,
serde_json::to_string(&code)?
);
}
Ok(())
}
#[test]
fn unavailable_keeps_missing_provenance_status_and_stderr_reader_errors() {
let mut error = Error::new(Refusal::QueryStopped, "diagnostic");
error.status = Some("exit status: 2".to_owned());
error.stderr_error = Some("reader failed".to_owned());
let refusal = caller_failure(&error);
assert_eq!(refusal.code, "query_engine_unavailable");
assert!(refusal.message.contains("query_worker_provenance_missing"));
assert!(refusal.message.contains("exit status: 2"));
assert!(refusal.message.contains("reader failed"));
}
#[test]
fn a_fresh_start_returns_typed_error_and_a_waited_status() -> TestResult {
let error = open(&settings(), FRAME, ".".to_owned(), None, BTreeMap::new())
.err()
.ok_or("true answered Ready")?;
assert!(error.status.is_some());
assert!(error.startup.ok_or("startup evidence lost")?.pid.is_some());
let failure = caller_failure(&error);
assert_eq!(failure.code, "query_engine_unavailable");
assert!(failure.message.contains("query_worker_exited_before_ready"));
Ok(())
}
#[test]
fn boundary_failures_keep_their_detail_in_an_unclassified_spawn_error() -> TestResult {
let mut limits = settings();
limits.worker = Path::new("relative-worker").to_owned();
let error = open(&limits, FRAME, ".".to_owned(), None, BTreeMap::new())
.err()
.ok_or("relative worker accepted")?;
assert_eq!(error.refusal.code, Refusal::QueryEngineUnavailable);
let startup = error.startup.ok_or("startup evidence lost")?;
assert_eq!(startup.stage, StartupStage::Spawn);
assert!(startup.pid.is_none());
assert!(
caller_failure(&error)
.message
.contains("query_worker_path_invalid")
);
Ok(())
}