use crate::serve::query::stages_tests_support as support;
use std::num::NonZeroU64;
use serde_json::{Value, json};
use support::{Fixture, TestResult, patients, request, values};
const FOUR: NonZeroU64 = NonZeroU64::MIN.saturating_add(3);
#[test]
fn rg_pipeline_keeps_only_matching_whole_records_in_source_order() -> TestResult {
let fixture = Fixture::new(None, &patients())?;
let answer = fixture.run(&request(r#"patients | rg("4B")"#, FOUR), &[])?;
assert_eq!(
values(&answer)?,
vec![
patients()[0].clone(),
patients()[2].clone(),
patients()[3].clone()
]
);
assert_eq!(answer["records_read"], 4);
assert_eq!(answer["stages"], json!(["rg"]));
assert!(answer["next"].is_null());
Ok(())
}
#[test]
fn rg_pipeline_field_case_and_nonstring_matching_are_exact() -> TestResult {
let documents = vec![
json!({"notes":"CHEST", "other":"clear", "n":[4,true]}),
json!({"notes":"clear", "other":"chest", "n":[4,false]}),
json!({"notes":"chest", "other":"clear", "n":[4,true]}),
];
let fixture = Fixture::new(None, &documents)?;
let answer = fixture.run(
&request(r#"patients | rg(.notes; "chest"; "-i")"#, FOUR),
&[],
)?;
assert_eq!(
values(&answer)?,
vec![documents[0].clone(), documents[2].clone()]
);
let numbers = fixture.run(
&request(r#"patients | rg(.n; "[4,true]"; "-F")"#, FOUR),
&[],
)?;
assert_eq!(values(&numbers)?, values(&answer)?);
Ok(())
}
#[test]
fn rg_pipeline_refusals_name_bad_flag_and_regex_error() -> TestResult {
let fixture = Fixture::new(None, &patients())?;
for (query, code, word) in [
(
r#"patients | rg("ward"; "--mystery")"#,
"query_flag",
"--mystery",
),
(r#"patients | rg("[")"#, "query_pattern", "regex"),
] {
let error = fixture
.run(&request(query, FOUR), &[])
.err()
.ok_or("invalid rg accepted")?;
assert_eq!(error.code, code);
assert!(error.message.contains(word), "{}", error.message);
}
Ok(())
}
#[test]
fn rg_pipeline_pages_and_charges_nonmatches_without_exposing_denied_records() -> TestResult {
let fixture = Fixture::new(None, &patients())?;
let mut ask = request(r#"patients | rg("4B") | limit(2)"#, NonZeroU64::MIN);
let first = fixture.run(&ask, &[("patients", "p2")])?;
assert_eq!(values(&first)?, vec![patients()[0].clone()]);
ask.continuation = Some(
first
.get("next")
.and_then(Value::as_str)
.ok_or("missing rg continuation")?
.to_owned(),
);
let second = fixture.run(&ask, &[("patients", "p2")])?;
assert_eq!(values(&second)?, vec![patients()[3].clone()]);
assert_eq!(second["records_read"], 3);
assert!(second["next"].is_null());
assert_eq!(first["selection"], second["selection"]);
Ok(())
}
#[test]
fn rg_pipeline_after_jq_preserves_output_and_limit_order() -> TestResult {
let fixture = Fixture::new(None, &patients())?;
let answer = fixture.run(
&request(r#"patients | '.name' | rg("Cy") | limit(1)"#, FOUR),
&[],
)?;
assert_eq!(values(&answer)?, vec![json!("Cy")]);
assert_eq!(answer["records_read"], 3);
assert_eq!(answer["stages"], json!(["jq", "rg", "limit"]));
assert!(answer["next"].is_null());
let limited = fixture.run(
&request(r#"patients | '.name' | limit(2) | rg("Cy")"#, FOUR),
&[],
)?;
assert!(values(&limited)?.is_empty());
assert_eq!(limited["records_read"], 2);
assert!(limited["next"].is_null());
Ok(())
}
#[test]
fn rg_inverted_pipeline_keeps_exact_nonmatching_records_in_order() -> TestResult {
let fixture = Fixture::new(None, &patients())?;
for stage in ["rg", "grep"] {
let query = format!("patients | {stage}(\"Ada|Cy\"; \"-v\")");
let answer = fixture.run(&request(&query, FOUR), &[])?;
assert_eq!(
values(&answer)?,
vec![patients()[1].clone(), patients()[3].clone()]
);
assert_eq!(answer["records_read"], 4);
assert_eq!(answer["stages"], json!(["rg"]));
assert!(answer["next"].is_null());
}
Ok(())
}
#[test]
fn rg_inverted_field_pipeline_keeps_missing_fields_in_order() -> TestResult {
let documents = vec![
json!({"ward":"4B", "name":"Ada"}),
json!({"name":"Bea"}),
json!({"ward":"3A", "name":"Cy"}),
json!({"ward":"4B", "name":"Dee"}),
];
let fixture = Fixture::new(None, &documents)?;
let answer = fixture.run(&request(r#"patients | rg(.ward; "4B"; "-v")"#, FOUR), &[])?;
assert_eq!(
values(&answer)?,
vec![documents[1].clone(), documents[2].clone()]
);
assert_eq!(answer["records_read"], 4);
assert!(answer["next"].is_null());
Ok(())
}
#[test]
fn quoted_grep_reaches_the_worker_program_unchanged() -> TestResult {
use std::collections::BTreeMap;
use std::num::{NonZeroU32, NonZeroUsize};
use std::os::unix::fs::PermissionsExt;
use crate::serve::query::{Settings, parser, start};
let directory = tempfile::tempdir()?;
let worker = directory.path().join("worker");
std::fs::write(
&worker,
r"#!/usr/bin/python3
import json, struct, sys
header = sys.stdin.buffer.read(4)
size = struct.unpack('>I', header)[0]
frame = sys.stdin.buffer.read(size)
if len(frame) != size or frame[0] != 1:
raise ValueError('expected one complete Start frame')
start = json.loads(frame[1:])
reply = bytes([6]) + json.dumps(dict(
code='query_function_refused', message=start['alternate']
)).encode()
sys.stdout.buffer.write(struct.pack('>I', len(reply)) + reply)
sys.stdout.buffer.flush()
sys.exit(1)
",
)?;
std::fs::set_permissions(&worker, std::fs::Permissions::from_mode(0o700))?;
let settings = Settings {
worker,
time_ms: NonZeroU64::MIN.saturating_add(59_999),
memory_bytes: NonZeroU64::MIN.saturating_add((1 << 20) - 1),
outputs_per_record: NonZeroU64::MIN,
scan_records: FOUR,
max_running: NonZeroUsize::MIN,
};
let query = r#"patients | 'grep("x")'"#;
let parsed = parser::pipeline(query);
let plan = parsed.as_ref().map_err(ToString::to_string)?;
let error = start::open(
&settings,
NonZeroU32::MIN.saturating_add(8191),
super::super::shape::program(query, &parsed),
Some(super::super::plan::jq(plan).to_owned()),
BTreeMap::new(),
)
.err()
.ok_or("fixture worker did not echo its program refusal")?;
assert_eq!(
error.refusal.code,
haematite_query::Refusal::QueryFunctionRefused
);
assert_eq!(error.refusal.message, r#"grep("x")"#);
assert_eq!(start::caller_failure(&error).code, "query_function_refused");
Ok(())
}