//! Integration tests for `clawsync serve-all` — the universal multi-protocol server.
//!
//! Each test starts a single `serve-all` server and connects to it with
//! different client types (onion push/pull, HDF5-sync, fs-sync) verifying that
//! the correct protocol is dispatched based on the first message.
//!
//! Run with:
//! cargo test -p clawsync-cli --test serve_all -- --nocapture
use std::fs;
use std::io::{BufRead, BufReader};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::Duration;
use clawhdf5::FileBuilder;
use clawhdf5_onion::writer::OnionFile;
use clawsync_hdf5::DatasetManifest;
use tempfile::TempDir;
const BIN: &str = env!("CARGO_BIN_EXE_clawsync");
const H5_MAGIC: &[u8] = b"\x89HDF\r\n\x1a\n";
// ─────────────────────────────────────────────────────────────────────────────
// serve-all harness
// ─────────────────────────────────────────────────────────────────────────────
struct ServeAllServer {
child: Child,
pub addr: SocketAddr,
pub dir: TempDir,
}
impl ServeAllServer {
/// Start `clawsync serve-all
--bind 127.0.0.1:0 [--allow-delete]`
/// and wait for the startup line.
fn start(allow_delete: bool) -> Self {
let dir = TempDir::new().expect("failed to create temp dir");
let mut cmd = Command::new(BIN);
cmd.arg("serve-all")
.arg(dir.path())
.arg("--bind")
.arg("127.0.0.1:0");
if allow_delete {
cmd.arg("--allow-delete");
}
cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());
let mut child = cmd.spawn().expect("failed to spawn clawsync serve-all");
let stdout = child.stdout.take().unwrap();
let mut reader = BufReader::new(stdout);
// Wait for: "ClawSync server listening on 127.0.0.1:PORT"
let deadline = std::time::Instant::now() + Duration::from_secs(10);
let mut line = String::new();
loop {
line.clear();
reader
.read_line(&mut line)
.expect("failed to read startup line");
if line.contains("ClawSync server listening on") {
break;
}
if std::time::Instant::now() > deadline {
panic!("serve-all did not print startup line in time; got: {line:?}");
}
}
let addr_str = line
.trim()
.rsplit("on ")
.next()
.expect("unexpected startup line")
.trim();
let addr: SocketAddr = addr_str
.parse()
.unwrap_or_else(|_| panic!("could not parse bound addr from: {addr_str}"));
// Drain remaining stdout in a background thread.
std::thread::spawn(move || for _ in reader.lines() {});
Self { child, addr, dir }
}
}
impl Drop for ServeAllServer {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Helpers
// ─────────────────────────────────────────────────────────────────────────────
/// Create a temp dir with an HDF5 file containing `n` onion revisions.
fn make_versioned_h5(n: u8) -> (TempDir, PathBuf, OnionFile) {
let dir = TempDir::new().unwrap();
let h5 = dir.path().join("data.h5");
fs::write(&h5, H5_MAGIC).unwrap();
let mut onion = OnionFile::create(&h5, 4096).unwrap();
for i in 0..n {
let mut s = onion.begin_session(None).unwrap();
s.record_page(0, &vec![i; 4096]);
onion.commit_session(s, Some(&format!("rev {i}"))).unwrap();
}
onion.flush().unwrap();
(dir, h5, onion)
}
/// Build an HDF5 file with the given f64 datasets.
fn write_h5(path: &Path, datasets: &[(&str, Vec)]) {
let mut b = FileBuilder::new();
for (name, data) in datasets {
b.create_dataset(name).with_f64_data(data);
}
let bytes = b.finish().unwrap();
fs::write(path, bytes).unwrap();
}
/// Run `clawsync push ` and assert exit 0.
fn run_push(local: &Path, addr: SocketAddr, remote_name: &str) {
let remote = format!("{addr}/{remote_name}");
let status = Command::new(BIN)
.args(["push", local.to_str().unwrap(), &remote])
.status()
.expect("failed to run push");
assert!(status.success(), "push exited with {status}");
}
/// Run `clawsync pull ` and assert exit 0.
fn run_pull(addr: SocketAddr, remote_name: &str, local: &Path) {
let remote = format!("{addr}/{remote_name}");
let status = Command::new(BIN)
.args(["pull", &remote, local.to_str().unwrap()])
.status()
.expect("failed to run pull");
assert!(status.success(), "pull exited with {status}");
}
/// Run `clawsync hdf5-sync ` and assert exit 0.
fn run_hdf5_sync(local: &Path, addr: SocketAddr, rel_path: &str) {
let remote = format!("{addr}/{rel_path}");
let status = Command::new(BIN)
.args(["hdf5-sync", local.to_str().unwrap(), &remote])
.status()
.expect("failed to run hdf5-sync");
assert!(status.success(), "hdf5-sync exited with {status}");
}
/// Run `clawsync sync ` and assert exit 0.
fn run_fs_sync(src: &Path, addr: SocketAddr, delete: bool) {
let remote = format!("{addr}/");
let mut cmd = Command::new(BIN);
cmd.args(["sync", src.to_str().unwrap(), &remote]);
if delete {
cmd.arg("--delete");
}
let status = cmd.status().expect("failed to run sync");
assert!(status.success(), "sync exited with {status}");
}
// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────
/// Push onion revisions to a serve-all server.
///
/// The server should detect IbltRequest as the onion protocol and serve the
/// file at `/.claws`.
#[test]
fn serve_all_routes_onion_push() {
let server = ServeAllServer::start(false);
let (_src_dir, h5, _onion) = make_versioned_h5(3);
run_push(&h5, server.addr, "onion-data");
// Verify the serve-all root has the served file.
let entries: Vec<_> = fs::read_dir(server.dir.path())
.unwrap()
.filter_map(|e| e.ok())
.collect();
assert!(
!entries.is_empty(),
"serve-all root should contain the served file"
);
}
/// Pull revisions via a serve-all server.
#[test]
fn serve_all_routes_onion_pull() {
let server = ServeAllServer::start(false);
let (_src_dir, src_h5, _onion) = make_versioned_h5(4);
// Push first so the server has revisions.
run_push(&src_h5, server.addr, "onion-data");
// Pull to a fresh local file.
let dst_dir = TempDir::new().unwrap();
let dst_h5 = dst_dir.path().join("data.h5");
run_pull(server.addr, "onion-data", &dst_h5);
// Destination should now have the same number of revisions.
let dst_onion = OnionFile::open(&dst_h5).expect("failed to open pulled file");
assert_eq!(
dst_onion.revision_count(),
4,
"pulled file should have 4 revisions"
);
}
/// HDF5 dataset-granular sync through a serve-all server.
#[test]
fn serve_all_routes_hdf5_sync() {
let server = ServeAllServer::start(false);
let src_dir = TempDir::new().unwrap();
let src_h5 = src_dir.path().join("science.h5");
write_h5(
&src_h5,
&[
("temperature", vec![1.0, 2.0, 3.0]),
("pressure", vec![101.0, 102.0]),
],
);
run_hdf5_sync(&src_h5, server.addr, "science.h5");
// Verify the served file now has the same datasets.
let served = server.dir.path().join("science.h5");
assert!(served.exists(), "served h5 file must exist after hdf5-sync");
let src_manifest = DatasetManifest::from_path(&src_h5).expect("failed to read src manifest");
let srv_manifest = DatasetManifest::from_path(&served).expect("failed to read server manifest");
assert_eq!(
src_manifest.len(),
srv_manifest.len(),
"server must have same number of datasets as source"
);
for (path, entry) in &src_manifest.datasets {
let srv = srv_manifest
.get(path)
.unwrap_or_else(|| panic!("dataset {path} missing on server"));
assert_eq!(entry.blake3, srv.blake3, "dataset {path} hash mismatch");
}
}
/// General file sync (any-type) through a serve-all server.
#[test]
fn serve_all_routes_fs_sync() {
let server = ServeAllServer::start(false);
let src_dir = TempDir::new().unwrap();
fs::write(src_dir.path().join("hello.txt"), b"hello, world").unwrap();
fs::write(
src_dir.path().join("data.bin"),
vec![0xDE, 0xAD, 0xBE, 0xEF],
)
.unwrap();
run_fs_sync(src_dir.path(), server.addr, false);
// Both files must exist in the serve-all root.
let hello = server.dir.path().join("hello.txt");
let data = server.dir.path().join("data.bin");
assert!(hello.exists(), "hello.txt must be synced");
assert!(data.exists(), "data.bin must be synced");
assert_eq!(fs::read(&hello).unwrap(), b"hello, world");
assert_eq!(fs::read(&data).unwrap(), [0xDE, 0xAD, 0xBE, 0xEF]);
}
/// Two sequential clients of different protocols connect to the same serve-all
/// port; both must complete successfully without the server crashing.
#[test]
fn serve_all_multi_protocol_sequential() {
let server = ServeAllServer::start(false);
// First: FS sync.
let src_dir = TempDir::new().unwrap();
fs::write(src_dir.path().join("file.txt"), b"content").unwrap();
run_fs_sync(src_dir.path(), server.addr, false);
// Second: HDF5 sync.
let h5_dir = TempDir::new().unwrap();
let h5 = h5_dir.path().join("metrics.h5");
write_h5(&h5, &[("latency", vec![0.1, 0.2, 0.3])]);
run_hdf5_sync(&h5, server.addr, "metrics.h5");
// Both outputs must be present in the serve-all root.
assert!(server.dir.path().join("file.txt").exists());
assert!(server.dir.path().join("metrics.h5").exists());
}
/// A warm no-op FS sync (already in sync) through serve-all succeeds.
#[test]
fn serve_all_fs_warm_noop() {
let server = ServeAllServer::start(false);
let src_dir = TempDir::new().unwrap();
fs::write(src_dir.path().join("stable.txt"), b"unchanged").unwrap();
// First sync.
run_fs_sync(src_dir.path(), server.addr, false);
// Second sync — same content, should be a no-op.
run_fs_sync(src_dir.path(), server.addr, false);
assert_eq!(
fs::read(server.dir.path().join("stable.txt")).unwrap(),
b"unchanged"
);
}
/// Push/pull round-trip through serve-all preserves revision data.
#[test]
fn serve_all_push_pull_round_trip() {
let server = ServeAllServer::start(false);
let (_src_dir, src_h5, src_onion) = make_versioned_h5(5);
let src_rev_count = src_onion.revision_count();
run_push(&src_h5, server.addr, "round-trip");
let dst_dir = TempDir::new().unwrap();
let dst_h5 = dst_dir.path().join("round_trip.h5");
run_pull(server.addr, "round-trip", &dst_h5);
let dst_onion = OnionFile::open(&dst_h5).unwrap();
assert_eq!(
dst_onion.revision_count(),
src_rev_count,
"round-tripped file must have same revision count"
);
}
// ─────────────────────────────────────────────────────────────────────────────
// QUIC transport tests for serve-all
// ─────────────────────────────────────────────────────────────────────────────
/// Harness for `clawsync serve-all --quic`.
struct ServeAllServerQuic {
child: Child,
pub addr: SocketAddr,
pub dir: TempDir,
}
impl ServeAllServerQuic {
fn start(allow_delete: bool) -> Self {
let dir = TempDir::new().expect("failed to create temp dir");
let mut cmd = Command::new(BIN);
cmd.arg("serve-all")
.arg(dir.path())
.arg("--bind")
.arg("127.0.0.1:0")
.arg("--quic");
if allow_delete {
cmd.arg("--allow-delete");
}
cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());
let mut child = cmd
.spawn()
.expect("failed to spawn clawsync serve-all --quic");
let stdout = child.stdout.take().unwrap();
let mut reader = BufReader::new(stdout);
let deadline = std::time::Instant::now() + Duration::from_secs(10);
let mut line = String::new();
loop {
line.clear();
reader
.read_line(&mut line)
.expect("failed to read startup line");
if line.contains("ClawSync server listening on") {
break;
}
if std::time::Instant::now() > deadline {
panic!("serve-all --quic did not print startup line in time; got: {line:?}");
}
}
let addr_str = line.trim().rsplit("on ").next().unwrap().trim();
let addr: SocketAddr = addr_str
.parse()
.unwrap_or_else(|_| panic!("could not parse bound addr from: {addr_str}"));
std::thread::spawn(move || for _ in reader.lines() {});
Self { child, addr, dir }
}
}
impl Drop for ServeAllServerQuic {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
fn run_fs_sync_quic(src: &Path, addr: SocketAddr) {
let remote = format!("{addr}/");
let status = Command::new(BIN)
.args(["sync", src.to_str().unwrap(), &remote, "--quic"])
.status()
.expect("failed to run sync --quic");
assert!(status.success(), "sync --quic exited with {status}");
}
/// Cold copy via QUIC through the universal serve-all server.
#[test]
fn serve_all_quic_fs_cold_copy() {
let server = ServeAllServerQuic::start(false);
let src = TempDir::new().unwrap();
fs::write(src.path().join("quic_file.txt"), b"quic cold copy test").unwrap();
fs::write(src.path().join("binary.bin"), vec![0x42u8; 1024]).unwrap();
run_fs_sync_quic(src.path(), server.addr);
let dst_txt = server.dir.path().join("quic_file.txt");
let dst_bin = server.dir.path().join("binary.bin");
assert!(dst_txt.exists(), "quic_file.txt must be synced via QUIC");
assert!(dst_bin.exists(), "binary.bin must be synced via QUIC");
assert_eq!(fs::read(&dst_txt).unwrap(), b"quic cold copy test");
assert_eq!(fs::read(&dst_bin).unwrap(), vec![0x42u8; 1024]);
}
/// Warm no-op via QUIC: re-syncing identical content transfers nothing
/// and exits 0.
#[test]
fn serve_all_quic_fs_warm_noop() {
let server = ServeAllServerQuic::start(false);
// Seed server directory with the same content as the source.
let src = TempDir::new().unwrap();
let content = b"identical content for noop";
fs::write(src.path().join("noop.txt"), content).unwrap();
fs::write(server.dir.path().join("noop.txt"), content).unwrap();
// Second sync should be a warm no-op.
run_fs_sync_quic(src.path(), server.addr);
assert_eq!(
fs::read(server.dir.path().join("noop.txt")).unwrap(),
content
);
}
/// Incremental sync via QUIC: modifying one file transfers only that file.
#[test]
fn serve_all_quic_fs_incremental() {
let server = ServeAllServerQuic::start(false);
let src = TempDir::new().unwrap();
// Put both files on server already.
fs::write(src.path().join("unchanged.bin"), vec![0x11u8; 512]).unwrap();
fs::write(src.path().join("changed.bin"), vec![0xAAu8; 512]).unwrap();
run_fs_sync_quic(src.path(), server.addr);
// Modify one file; re-sync.
fs::write(src.path().join("changed.bin"), vec![0xBBu8; 512]).unwrap();
run_fs_sync_quic(src.path(), server.addr);
assert_eq!(
fs::read(server.dir.path().join("changed.bin")).unwrap(),
vec![0xBBu8; 512]
);
assert_eq!(
fs::read(server.dir.path().join("unchanged.bin")).unwrap(),
vec![0x11u8; 512]
);
}
/// `serve-all` dispatch: `pull-fs` — the server pushes its files to the client.
///
/// This covers the `FsDirPullRequest` branch that was missing from
/// `handle_any_client` before the fix (it would return an error with
/// "unrecognized first message").
#[test]
fn serve_all_pull_fs() {
let server = ServeAllServer::start(false);
// Seed server root directly with 2 files.
fs::write(server.dir.path().join("server_a.bin"), vec![0x11u8; 1024]).unwrap();
fs::write(server.dir.path().join("server_b.bin"), vec![0x22u8; 2048]).unwrap();
let dst = TempDir::new().unwrap();
let remote = format!("{}/", server.addr);
let out = Command::new(BIN)
.args(["pull-fs", &remote, dst.path().to_str().unwrap()])
.output()
.expect("failed to run pull-fs");
assert!(
out.status.success(),
"pull-fs via serve-all failed:\nstdout: {}\nstderr: {}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr),
);
assert!(
dst.path().join("server_a.bin").exists(),
"server_a.bin not pulled"
);
assert!(
dst.path().join("server_b.bin").exists(),
"server_b.bin not pulled"
);
assert_eq!(
fs::read(dst.path().join("server_a.bin")).unwrap(),
vec![0x11u8; 1024]
);
assert_eq!(
fs::read(dst.path().join("server_b.bin")).unwrap(),
vec![0x22u8; 2048]
);
}