use std::error::Error as StdError;
use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
use std::path::{Path, PathBuf};
use std::process::Command;
use haematite_query::handle::{Error, StartupStage};
use haematite_query::{Ready, Refusal};
use super::{Prepared, classify, prepare, run};
use crate::serve::query::Settings;
type TestResult = Result<(), Box<dyn StdError>>;
const FRAME: NonZeroU32 = NonZeroU32::MIN.saturating_add(8191);
fn settings(path: &Path) -> Settings {
Settings {
time_ms: NonZeroU64::MIN.saturating_add(59_999),
memory_bytes: NonZeroU64::MIN.saturating_add(64 * 1024 * 1024 - 1),
outputs_per_record: NonZeroU64::MIN.saturating_add(31),
scan_records: NonZeroU64::MIN.saturating_add(99),
max_running: NonZeroUsize::MIN.saturating_add(1),
worker: path.to_owned(),
}
}
fn reaped(error: &super::ProbeError) -> TestResult {
let ending = error.ending.as_ref().ok_or("missing worker ending")?;
assert!(
ending.status.is_some(),
"owner did not report a waited status"
);
let startup = ending.startup.ok_or("missing typed startup evidence")?;
let pid = startup.pid.ok_or("missing spawned worker PID")?;
let status = Command::new("/bin/kill")
.args(["-0", &pid.to_string()])
.status()?;
assert!(!status.success(), "worker {pid} remains alive");
Ok(())
}
#[test]
fn a_failed_probe_is_kept_by_name_with_its_settings() -> TestResult {
let directory = tempfile::tempdir()?;
let absent = directory.path().join("absent-worker");
let prepared = prepare(settings(&absent), FRAME);
let refusal = prepared.ready().err().ok_or("missing worker was ready")?;
assert_eq!(refusal.code, "query_worker_missing");
assert_eq!(prepared.settings().worker, absent);
Ok(())
}
#[test]
fn a_missing_worker_and_a_directory_have_distinct_start_refusals() -> TestResult {
let directory = tempfile::tempdir()?;
let missing = run(&settings(&directory.path().join("absent-worker")), FRAME)
.err()
.ok_or("missing worker accepted")?;
assert_eq!(missing.failure.code, "query_worker_missing");
assert!(missing.ending.is_none());
let nonfile = run(&settings(directory.path()), FRAME)
.err()
.ok_or("directory accepted")?;
assert_eq!(nonfile.failure.code, "query_worker_not_file");
assert!(nonfile.ending.is_none());
Ok(())
}
#[test]
fn a_nonexecutable_regular_file_is_named_without_spawning_a_worker() -> TestResult {
let file = tempfile::NamedTempFile::new()?;
std::fs::write(file.path(), b"ordinary data")?;
let error = run(&settings(file.path()), FRAME)
.err()
.ok_or("nonexecutable accepted")?;
assert_eq!(error.failure.code, "query_worker_not_executable");
let ending = error.ending.as_ref().ok_or("missing spawn refusal")?;
assert_eq!(
ending.startup.ok_or("missing startup evidence")?.stage,
StartupStage::Spawn
);
assert!(
ending
.startup
.ok_or("missing startup evidence")?
.pid
.is_none()
);
Ok(())
}
#[test]
fn a_worker_exiting_before_ready_is_named_and_reaped() -> TestResult {
let error = run(&settings(Path::new("/usr/bin/true")), FRAME)
.err()
.ok_or("true answered Ready")?;
assert_eq!(error.failure.code, "query_worker_exited_before_ready");
reaped(&error)
}
#[test]
fn an_echoed_start_frame_is_not_ready_and_the_worker_is_reaped() -> TestResult {
let error = run(&settings(Path::new("/bin/cat")), FRAME)
.err()
.ok_or("cat answered Ready")?;
assert_eq!(error.failure.code, "query_worker_not_ready");
assert_eq!(
error
.ending
.as_ref()
.ok_or("missing ending")?
.startup
.ok_or("missing startup")?
.stage,
StartupStage::UnexpectedFrame
);
reaped(&error)
}
#[test]
fn a_worker_that_never_answers_ready_hits_the_production_deadline_and_is_reaped() -> TestResult {
let mut limits = settings(Path::new("/usr/bin/wc"));
limits.time_ms = NonZeroU64::MIN.saturating_add(49);
let error = run(&limits, FRAME).err().ok_or("wc answered Ready")?;
assert_eq!(error.failure.code, "query_worker_start_timeout");
assert!(error.failure.message.contains("query_time_ms"));
if error.ending.is_some() {
reaped(&error)?;
}
Ok(())
}
#[test]
fn a_failed_frame_carries_its_typed_code_into_the_probe_refusal() {
let limits = settings(Path::new("/worker"));
let error = Error::new(Refusal::QueryCompile, "invalid probe program")
.at_start(StartupStage::Failed(Refusal::QueryCompile), Some(123));
let refusal = classify(error, &limits);
assert_eq!(refusal.failure.code, "query_worker_not_ready");
assert!(refusal.failure.message.contains("query_compile"));
}
#[test]
fn identity_memory_and_owner_failures_are_named_from_typed_evidence() {
let limits = settings(Path::new("/worker"));
for (code, stage, expected) in [
(
Refusal::QueryEngineMismatch,
StartupStage::Identity,
"query_worker_identity_mismatch",
),
(
Refusal::QueryMemoryLimit,
StartupStage::ReadReady,
"query_worker_start_memory_limit",
),
(
Refusal::QueryStopped,
StartupStage::Owner,
"query_worker_owner_failed",
),
] {
let error = Error::new(code, "same diagnostic").at_start(stage, Some(123));
assert_eq!(classify(error, &limits).failure.code, expected);
}
}
#[test]
fn service_carries_query_settings_and_ready_without_a_live_child() -> TestResult {
use crate::serve::service::Service;
use crate::serve::storage::Storage;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
let directory = tempfile::tempdir()?;
let storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 1,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let prepared = Prepared {
settings: settings(Path::new("/worker")),
outcome: Ok(Ready {
protocol: 1,
worker: "worker-version".to_owned(),
core_limit: 0,
descriptors_closed: 0,
environment: 0,
engine: "engine".to_owned(),
jaq_core: "core-version".to_owned(),
jaq_std: "std-version".to_owned(),
jaq_json: "json-version".to_owned(),
reading: Some(haematite_query::Reading::Candidate),
}),
};
let service = Service::with_query(storage, NonZeroU32::MIN, Some(prepared))?;
let query = service.query.as_ref().ok_or("query state lost")?;
assert_eq!(query.settings().worker, PathBuf::from("/worker"));
let ready = query.ready().map_err(ToString::to_string)?;
assert_eq!(ready.worker, "worker-version");
assert_eq!(ready.jaq_core, "core-version");
assert_eq!(ready.jaq_std, "std-version");
assert_eq!(ready.jaq_json, "json-version");
Ok(())
}
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}