//! 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] ); }