use std::path::Path;
use std::process::{Child, Command, Output, Stdio};
use std::time::{Duration, Instant};
/// a hydrant on a database in `dir` that never touches the network
fn offline_hydrant(dir: &Path) -> Command {
offline(Command::new(env!("CARGO_BIN_EXE_hydrant")), dir)
}
fn offline(mut cmd: Command, dir: &Path) -> Command {
cmd.env("HYDRANT_DATABASE_PATH", dir.join("db"))
.env(
"HYDRANT_API_BIND",
format!("unix:{}", dir.join("api.sock").display()),
)
.env("HYDRANT_ENABLE_FIREHOSE", "false")
.env("HYDRANT_ENABLE_CRAWLER", "false")
.env("HYDRANT_ENABLE_BACKFILL", "false")
.env("HYDRANT_RELAY_HOSTS", "")
.env("HYDRANT_SEED_HOSTS", "")
.env("HYDRANT_CRAWLER_URLS", "")
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
cmd
}
// config is env only, so anything on argv has to be refused before hydrant boots.
// if it does boot, it gets a throwaway database and no network, and we kill it.
fn hydrant(args: &[&str]) -> Output {
let tmp = tempfile::tempdir().unwrap();
let mut child = offline_hydrant(tmp.path()).args(args).spawn().unwrap();
let deadline = Instant::now() + Duration::from_secs(30);
while child.try_wait().unwrap().is_none() {
if Instant::now() > deadline {
child.kill().unwrap();
let out = child.wait_with_output().unwrap();
panic!(
"hydrant {args:?} booted instead of exiting:\n{}",
String::from_utf8_lossy(&out.stderr)
);
}
std::thread::sleep(Duration::from_millis(50));
}
child.wait_with_output().unwrap()
}
#[test]
fn version_prints_and_exits() {
for flag in ["--version", "-V"] {
let out = hydrant(&[flag]);
assert!(out.status.success(), "{flag}: {out:?}");
assert_eq!(
String::from_utf8_lossy(&out.stdout),
format!("hydrant {}\n", env!("CARGO_PKG_VERSION"))
);
}
}
#[test]
fn help_prints_usage_and_exits() {
for flag in ["--help", "-h"] {
let out = hydrant(&[flag]);
assert!(out.status.success(), "{flag}: {out:?}");
assert!(String::from_utf8_lossy(&out.stdout).starts_with("usage: hydrant"));
}
}
#[test]
fn any_other_argument_is_an_error() {
for args in [&["--port=3000"][..], &["serve"], &["--version", "extra"]] {
let out = hydrant(args);
assert!(!out.status.success(), "{args:?}: {out:?}");
assert!(out.stdout.is_empty(), "{args:?}: {out:?}");
assert!(
String::from_utf8_lossy(&out.stderr).contains("unexpected argument"),
"{args:?}: {out:?}"
);
}
}
/// waits for `child` to exit on its own, killing it after `timeout`
fn wait_for_exit(mut child: Child, timeout: Duration) -> Option<Output> {
let deadline = Instant::now() + timeout;
while child.try_wait().unwrap().is_none() {
if Instant::now() > deadline {
child.kill().unwrap();
return None;
}
std::thread::sleep(Duration::from_millis(50));
}
Some(child.wait_with_output().unwrap())
}
fn send_signal(child: &Child, signal: &str) {
let status = Command::new("kill")
.args([signal, &child.id().to_string()])
.status()
.unwrap();
assert!(status.success(), "kill {signal} failed");
}
/// starts hydrant on `dir` and returns once its api socket is up, so it's fully running
fn start_and_wait_for_api(dir: &Path) -> Child {
let socket = dir.join("api.sock");
let _ = std::fs::remove_file(&socket);
let mut child = offline_hydrant(dir).spawn().unwrap();
let deadline = Instant::now() + Duration::from_secs(60);
while !socket.exists() {
if let Some(status) = child.try_wait().unwrap() {
let out = child.wait_with_output().unwrap();
panic!(
"hydrant exited with {status} before serving:\n{}",
String::from_utf8_lossy(&out.stderr)
);
}
assert!(
Instant::now() < deadline,
"hydrant never bound its api socket"
);
std::thread::sleep(Duration::from_millis(50));
}
child
}
#[test]
fn sigterm_and_sigint_stop_cleanly_and_the_database_opens_again() {
let tmp = tempfile::tempdir().unwrap();
// the second and third start only get past opening the database if the one before let go
for signal in ["-TERM", "-INT", "-TERM"] {
let child = start_and_wait_for_api(tmp.path());
send_signal(&child, signal);
let out = wait_for_exit(child, Duration::from_secs(45))
.unwrap_or_else(|| panic!("hydrant didn't stop within 45s of {signal}"));
let stdout = String::from_utf8_lossy(&out.stdout);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
out.status.success(),
"{signal}: {:?}\n{stdout}{stderr}",
out.status
);
assert!(stdout.contains("stopped"), "{signal}: {stdout}{stderr}");
}
}
#[test]
fn config_warnings_are_logged() {
let tmp = tempfile::tempdir().unwrap();
// the bad history ttl makes it exit right after reading the config
let child = offline_hydrant(tmp.path())
.env("HYDRANT_RELAY_HOSTS", "not a url")
.env("HYDRANT_HISTORY_TTL", "soon")
.spawn()
.unwrap();
let out = wait_for_exit(child, Duration::from_secs(30)).expect("hydrant booted");
assert!(!out.status.success(), "{out:?}");
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
stdout.contains("invalid relay host URL: not a url"),
"{stdout}{}",
String::from_utf8_lossy(&out.stderr)
);
}
#[cfg(feature = "indexer_stream")]
fn socket_activated(dir: &Path, listener: &std::net::TcpListener) -> Child {
use std::os::fd::AsRawFd;
use std::os::unix::process::CommandExt;
// exec keeps the shell's pid, which makes $$ hydrant's pid
let mut sh = Command::new("sh");
sh.args(["-c", "LISTEN_PID=$$ exec \"$0\"", env!("CARGO_BIN_EXE_hydrant")]);
let mut cmd = offline(sh, dir);
cmd.env("HYDRANT_API_BIND", listener.local_addr().unwrap().to_string())
.env("LISTEN_FDS", "1");
let fd = listener.as_raw_fd();
// safety: dup2 and fcntl are async-signal-safe
unsafe {
cmd.pre_exec(move || {
if fd == 3 {
libc::fcntl(3, libc::F_SETFD, 0);
} else if libc::dup2(fd, 3) < 0 {
return Err(std::io::Error::last_os_error());
}
Ok(())
});
}
cmd.spawn().unwrap()
}
/// connect succeeds off the backlog and only the read waits
#[cfg(feature = "indexer_stream")]
fn get_health(addr: std::net::SocketAddr) -> String {
use std::io::{Read, Write};
let mut conn = std::net::TcpStream::connect(addr).expect("the listener is never closed");
conn.set_read_timeout(Some(Duration::from_secs(60))).unwrap();
conn.write_all(b"GET /health HTTP/1.1\r\nhost: hydrant\r\nconnection: close\r\n\r\n")
.unwrap();
let mut response = String::new();
conn.read_to_string(&mut response).unwrap();
response
}
#[cfg(feature = "indexer_stream")]
#[test]
fn a_socket_activated_restart_closes_streams_and_keeps_new_clients_waiting() {
use futures::StreamExt;
use tokio_websockets::{ClientBuilder, CloseCode};
let tmp = tempfile::tempdir().unwrap();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let first = socket_activated(tmp.path(), &listener);
assert!(get_health(addr).starts_with("HTTP/1.1 200"));
let uri = format!("ws://{addr}/stream");
let (mut subscriber, _) = rt
.block_on(ClientBuilder::new().uri(&uri).unwrap().connect())
.unwrap();
send_signal(&first, "-TERM");
let closed = rt
.block_on(async { tokio::time::timeout(Duration::from_secs(10), subscriber.next()).await })
.expect("the stream must be closed when hydrant stops")
.expect("a close frame, not the socket just going")
.unwrap();
assert_eq!(
closed.as_close().map(|(code, _)| code),
Some(CloseCode::SERVICE_RESTART)
);
let out = wait_for_exit(first, Duration::from_secs(45)).expect("hydrant didn't stop");
assert!(out.status.success(), "{out:?}");
let waiting = std::thread::spawn(move || get_health(addr));
std::thread::sleep(Duration::from_millis(200));
let second = socket_activated(tmp.path(), &listener);
let response = waiting.join().unwrap();
assert!(response.starts_with("HTTP/1.1 200"), "{response}");
send_signal(&second, "-TERM");
let out = wait_for_exit(second, Duration::from_secs(45)).expect("hydrant didn't stop");
assert!(out.status.success(), "{out:?}");
}