Merge pull request 'refactor: split test.py into modular pytest integration tests + benchmark tool' (#21) from refactor-tests into main

Reviewed-on: #21

.gitea/workflows/ci.yaml

@@ -5,7 +5,7 @@ on: [push, pull_request]
 jobs:
   build-and-test:
     runs-on: ubuntu-latest
-    container: gitea.tap-tap.win/taptap/fastsync-ci:v5
+    container: gitea.tap-tap.win/taptap/fastsync-ci:v6
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -20,4 +20,4 @@ jobs:
         run: ./build/tests
 
       - name: Integration Tests
-        run: python3 test.py
+        run: python3 -m pytest tests/ -v --tb=short

Dockerfile

@@ -1,6 +1,8 @@
 FROM ubuntu:24.04
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    gcc g++ make libc6-dev cmake libzstd-dev libssl-dev git ca-certificates curl && \
+    gcc g++ make libc6-dev cmake libzstd-dev libssl-dev git ca-certificates curl \
+    python3 python3-pip python3-venv openssl openssh-client && \
+    pip3 install --break-system-packages pytest && \
     curl -fsSL https://deb.nodesource.com/setup_20.x | bash - && \
     apt-get install -y --no-install-recommends nodejs && \
     rm -rf /var/lib/apt/lists/*

benchmark/bench.py

@@ -0,0 +1,519 @@
+#!/usr/bin/env python3
+"""Standalone benchmark tool for FastSync.
+
+Compares FastSync configs against rsync (no compression) and rsync+zstd.
+Data is ~75% random/incompressible and ~25% structured/compressible by default,
+controllable via --random-ratio.
+
+Usage:
+    python3 benchmark/bench.py
+    python3 benchmark/bench.py --runs 5 --profiles lan wan
+    python3 benchmark/bench.py --random-ratio 0.5 --size-mb 50
+    python3 benchmark/bench.py --delay 50ms --jitter 10ms --throughput 100mbit
+    python3 benchmark/bench.py --output json
+"""
+import argparse
+import json
+import os
+import random
+import shutil
+import socket
+import statistics
+import subprocess
+import sys
+import tempfile
+import time
+
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+BUILD_DIR = os.path.join(PROJECT_ROOT, "build")
+SERVER_CMD = [os.path.join(BUILD_DIR, "server")]
+CLIENT_CMD = [os.path.join(BUILD_DIR, "client")]
+BENCH_DIR = os.path.join(PROJECT_ROOT, "bench_data")
+
+NETWORK_PROFILES = {
+    "unlimited": {},
+    "lan": {
+        "rate": "1000mbit", "delay": "20ms", "jitter": "1ms", "loss": "0.1%",
+        "rate_bps": 1_000_000_000 / 8,
+    },
+    "wan": {
+        "rate": "100mbit", "delay": "50ms", "jitter": "10ms", "loss": "1%",
+        "rate_bps": 100_000_000 / 8,
+    },
+}
+
+FASTSYNC_CONFIGS = [
+    {"name": "fastsync",           "flags": [],                     "tool": "fastsync"},
+    {"name": "fastsync -c",        "flags": ["-c"],                 "tool": "fastsync"},
+    {"name": "fastsync -m",        "flags": ["-m"],                 "tool": "fastsync"},
+    {"name": "fastsync -m -c",     "flags": ["-m", "-c"],           "tool": "fastsync"},
+    {"name": "fastsync -m -c -s",  "flags": ["-m", "-c", "-s"],     "tool": "fastsync"},
+]
+
+RSYNC_CONFIGS = [
+    {"name": "rsync",              "flags": [],                     "tool": "rsync"},
+    {"name": "rsync -z",           "flags": ["-z"],                 "tool": "rsync"},
+    {"name": "rsync -z --zstd",    "flags": ["-z", "--zc", "zstd"],"tool": "rsync"},
+]
+
+class RsyncDaemon:
+    """Manages an rsync daemon for network-fair benchmarking."""
+
+    def __init__(self):
+        self._proc = None
+        self._port = None
+        self._conf_dir = None
+        self._module_path = None
+
+    def start(self, source_dir):
+        self._port = find_free_port()
+        self._conf_dir = tempfile.mkdtemp(prefix="rsyncd_")
+        self._module_path = source_dir
+
+        conf_path = os.path.join(self._conf_dir, "rsyncd.conf")
+        log_path = os.path.join(self._conf_dir, "rsyncd.log")
+
+        with open(conf_path, "w") as f:
+            f.write(f"uid = 0\ngid = 0\nuse chroot = no\nlog file = {log_path}\n")
+            f.write(f"[bench]\n\tpath = {source_dir}\n\tread only = yes\n")
+
+        self._proc = subprocess.Popen(
+            ["rsync", "--daemon", "--no-detach",
+             "--port", str(self._port),
+             "--config", conf_path],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        )
+        wait_for_port(self._port, timeout=5)
+
+    def stop(self):
+        if self._proc:
+            self._proc.terminate()
+            try:
+                self._proc.wait(timeout=3)
+            except subprocess.TimeoutExpired:
+                self._proc.kill()
+                self._proc.wait()
+            self._proc = None
+        if self._conf_dir:
+            shutil.rmtree(self._conf_dir, ignore_errors=True)
+            self._conf_dir = None
+
+    @property
+    def source_url(self):
+        return f"rsync://127.0.0.1:{self._port}/bench/"
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.stop()
+
+
+STRUCTURED_FILES = {
+    "small.txt": b"hello world\n",
+    "medium.txt": b"the quick brown fox jumps over the lazy dog\n" * 5000,
+    "binary.bin": bytes(range(256)) * 1000,
+    "nested/subdir/deep.txt": b"deeply nested file\n",
+    "nested/another.txt": b"another nested file\n" * 50,
+}
+
+
+class Progress:
+    """Simple progress bar with ETA."""
+
+    def __init__(self, total, label="Progress"):
+        self.total = total
+        self.current = 0
+        self.label = label
+        self.start_time = time.monotonic()
+        self._print()
+
+    def tick(self, detail=""):
+        self.current += 1
+        self._print(detail)
+
+    def _print(self, detail=""):
+        elapsed = time.monotonic() - self.start_time
+        if self.current > 0:
+            eta = elapsed / self.current * (self.total - self.current)
+            eta_str = f"ETA {eta:.0f}s"
+        else:
+            eta_str = "ETA ..."
+        pct = self.current / self.total * 100 if self.total else 0
+        bar_len = 30
+        filled = int(bar_len * self.current / self.total) if self.total else 0
+        bar = "#" * filled + "-" * (bar_len - filled)
+        detail_str = f"  {detail}" if detail else ""
+        sys.stderr.write(f"\r  [{bar}] {pct:5.1f}%  {self.current}/{self.total}  {eta_str}{detail_str}  ")
+        sys.stderr.flush()
+        if self.current >= self.total:
+            sys.stderr.write(f"\r  [{'#' * bar_len}] 100.0%  {self.total}/{self.total}  done in {elapsed:.1f}s" + " " * 30 + "\n")
+            sys.stderr.flush()
+
+
+def generate_bench_data(source_dir, size_mb=25, random_ratio=0.75):
+    """Generate test data. ~random_ratio is incompressible, rest is structured."""
+    if os.path.exists(source_dir):
+        shutil.rmtree(source_dir)
+    os.makedirs(source_dir)
+
+    target = size_mb * 1024 * 1024
+    structured_budget = int(target * (1 - random_ratio))
+    written = 0
+
+    for rel_path, content in STRUCTURED_FILES.items():
+        if written >= structured_budget:
+            break
+        full_path = os.path.join(source_dir, rel_path)
+        os.makedirs(os.path.dirname(full_path), exist_ok=True)
+        with open(full_path, "wb") as f:
+            f.write(content)
+        written += len(content)
+
+    os.makedirs(os.path.join(source_dir, "bulk"), exist_ok=True)
+    i = 0
+    while written < target:
+        chunk_size = min(5 * 1024 * 1024, target - written)
+        with open(os.path.join(source_dir, f"bulk/file_{i}.dat"), "wb") as f:
+            f.write(random.randbytes(chunk_size))
+        written += chunk_size
+        i += 1
+
+    return written
+
+
+def find_free_port():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+def wait_for_port(port, timeout=5):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.3):
+                return
+        except (ConnectionRefusedError, OSError):
+            time.sleep(0.05)
+    raise RuntimeError(f"Port {port} not ready")
+
+
+def wait_proc(proc, timeout=5):
+    try:
+        proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()
+
+
+def netem_apply(delay=None, jitter=None, throughput=None, loss=None):
+    """Apply tc/netem rules to loopback. Pass None to skip a parameter."""
+    netem_reset()
+    cmd = ["sudo", "tc", "qdisc", "add", "dev", "lo", "root", "netem"]
+    if throughput:
+        cmd += ["rate", throughput]
+    if delay:
+        cmd += ["delay", delay, jitter or "0ms"]
+    if loss:
+        cmd += ["loss", loss]
+    if len(cmd) > 6:
+        subprocess.run(cmd, check=True, capture_output=True)
+
+
+def netem_apply_profile(profile_name):
+    params = NETWORK_PROFILES.get(profile_name, {})
+    if not params:
+        netem_reset()
+        return
+    netem_apply(
+        delay=params.get("delay"),
+        jitter=params.get("jitter"),
+        throughput=params.get("rate"),
+        loss=params.get("loss"),
+    )
+
+
+def netem_reset():
+    subprocess.run("sudo tc qdisc del dev lo root".split(), capture_output=True)
+
+
+def run_fastsync(source_dir, dest_dir, flags, port):
+    """Run FastSync client. Returns duration or None."""
+    cmd = CLIENT_CMD + [
+        "--source-dir", source_dir,
+        "--dest-dir", dest_dir,
+        "--server-port", str(port),
+        "--save-to-disk",
+    ] + flags
+    try:
+        start = time.monotonic()
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
+        duration = time.monotonic() - start
+        if result.returncode == 0:
+            return duration
+    except subprocess.TimeoutExpired:
+        pass
+    return None
+
+
+def run_rsync(source_dir, dest_dir, flags, rsync_daemon=None):
+    """Run rsync. Returns duration or None."""
+    src = source_dir.rstrip("/") + "/"
+    if rsync_daemon:
+        src = rsync_daemon.source_url
+    cmd = ["rsync", "-a", "--delete"] + flags + [src, dest_dir + "/"]
+    try:
+        start = time.monotonic()
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
+        duration = time.monotonic() - start
+        if result.returncode == 0:
+            return duration
+    except subprocess.TimeoutExpired:
+        pass
+    return None
+
+
+def run_transfer(config, source_dir, dest_dir, port=None, rsync_daemon=None):
+    """Route to the right tool. Returns duration or None."""
+    if config["tool"] == "rsync":
+        return run_rsync(source_dir, dest_dir, config["flags"], rsync_daemon)
+    else:
+        return run_fastsync(source_dir, dest_dir, config["flags"], port)
+
+
+def run_benchmark(source_dir, dest_dir, configs, runs, profile_name, progress=None):
+    """Run benchmark for all configs, returns list of results."""
+    is_limited = profile_name != "unlimited"
+    has_rsync = any(c["tool"] == "rsync" for c in configs)
+    if is_limited:
+        netem_apply_profile(profile_name)
+
+    rsync_daemon = None
+    try:
+        if is_limited and has_rsync:
+            rsync_daemon = RsyncDaemon()
+            rsync_daemon.start(source_dir)
+
+        results = []
+        for config in configs:
+            times = []
+            for run_idx in range(runs):
+                if os.path.exists(dest_dir):
+                    shutil.rmtree(dest_dir)
+                os.makedirs(dest_dir, exist_ok=True)
+
+                port = find_free_port()
+                server = None
+                try:
+                    if config["tool"] == "fastsync":
+                        server = subprocess.Popen(
+                            SERVER_CMD + ["-p", str(port)],
+                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                        )
+                        wait_for_port(port)
+
+                    t = run_transfer(config, source_dir, dest_dir, port, rsync_daemon)
+                    if t is not None:
+                        times.append(t)
+                finally:
+                    if server:
+                        wait_proc(server)
+
+                if progress:
+                    progress.tick(f"{config['name']} (run {run_idx+1}/{runs})")
+
+            entry = {
+                "config": config["name"],
+                "tool": config["tool"],
+                "profile": profile_name,
+                "runs": len(times),
+                "times": [round(t, 4) for t in times],
+            }
+            if times:
+                entry["p50"] = round(statistics.median(times), 4)
+                entry["p95"] = round(sorted(times)[int(len(times) * 0.95)], 4) if len(times) > 1 else entry["p50"]
+                entry["min"] = round(min(times), 4)
+                entry["max"] = round(max(times), 4)
+                entry["stdev"] = round(statistics.stdev(times), 4) if len(times) > 1 else 0.0
+            results.append(entry)
+        return results
+    finally:
+        if rsync_daemon:
+            rsync_daemon.stop()
+        if is_limited:
+            netem_reset()
+
+
+def print_table(results, total_bytes, random_ratio):
+    """Print results as a human-readable table grouped by profile."""
+    profiles = {}
+    for r in results:
+        profiles.setdefault(r["profile"], []).append(r)
+
+    for profile, entries in profiles.items():
+        params = NETWORK_PROFILES.get(profile, {})
+        print(f"\n{'=' * 85}")
+        print(f" Profile: {profile.upper()}")
+        if params.get("rate"):
+            print(f" Network: {params['rate']}, {params['delay']} +/- {params['jitter']}, loss {params['loss']}")
+        else:
+            print(f" Network: unlimited")
+        print(f" Data: {total_bytes / (1024*1024):.1f} MB  ({random_ratio*100:.0f}% random, {(1-random_ratio)*100:.0f}% compressible)")
+        print(f"{'=' * 85}")
+
+        fs_entries = [e for e in entries if e.get("tool") == "fastsync"]
+        rsync_entries = [e for e in entries if e.get("tool") == "rsync"]
+
+        if fs_entries:
+            print(f"\n FastSync:")
+            print(f" {'Config':<25} {'p50':>8} {'p95':>8} {'min':>8} {'max':>8} {'stdev':>8} {'runs':>5}")
+            print(f" {'-' * 25} {'-' * 8} {'-' * 8} {'-' * 8} {'-' * 8} {'-' * 8} {'-' * 5}")
+            for e in sorted(fs_entries, key=lambda x: x.get("p50", 999)):
+                _print_entry(e)
+
+        if rsync_entries:
+            print(f"\n rsync:")
+            print(f" {'Config':<25} {'p50':>8} {'p95':>8} {'min':>8} {'max':>8} {'stdev':>8} {'runs':>5}")
+            print(f" {'-' * 25} {'-' * 8} {'-' * 8} {'-' * 8} {'-' * 8} {'-' * 8} {'-' * 5}")
+            for e in sorted(rsync_entries, key=lambda x: x.get("p50", 999)):
+                _print_entry(e)
+
+        if params.get("rate_bps") and fs_entries and rsync_entries:
+            fs_best = min((e["p50"] for e in fs_entries if "p50" in e), default=None)
+            rsync_best = min((e["p50"] for e in rsync_entries if "p50" in e), default=None)
+            theoretical = total_bytes / params["rate_bps"]
+            if fs_best and rsync_best:
+                print(f"\n Theoretical max (line rate): {theoretical:.4f}s")
+                print(f" FastSync best:   {fs_best:.4f}s  ({theoretical/fs_best:.2f}x vs line rate)")
+                print(f" rsync best:      {rsync_best:.4f}s  ({theoretical/rsync_best:.2f}x vs line rate)")
+                print(f" FastSync vs rsync: {rsync_best/fs_best:.2f}x faster")
+
+
+def _print_entry(e):
+    if "p50" in e:
+        print(f" {e['config']:<25} {e['p50']:>7.4f}s {e['p95']:>7.4f}s "
+              f"{e['min']:>7.4f}s {e['max']:>7.4f}s {e['stdev']:>7.4f} {e['runs']:>5}")
+    else:
+        print(f" {e['config']:<25} {'N/A':>8} {'N/A':>8} {'N/A':>8} {'N/A':>8} {'N/A':>8} {e['runs']:>5}")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="FastSync benchmark tool — compares FastSync vs rsync",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""\
+Network profiles (predefined):
+  unlimited   No artificial limits
+  lan         1 Gbit, 20ms delay, 1ms jitter, 0.1%% loss
+  wan         100 Mbit, 50ms delay, 10ms jitter, 1%% loss
+
+Custom network limits (--delay/--jitter/--throughput) override profiles.
+
+Data mix:
+  Default is ~75%% random/incompressible + ~25%% structured/compressible,
+  reflecting typical real-world file sets.
+
+Examples:
+  %(prog)s --profiles wan --runs 5
+  %(prog)s --throughput 50mbit --delay 30ms --jitter 5ms
+  %(prog)s --random-ratio 0.5 --size-mb 100
+""")
+    parser.add_argument("--runs", type=int, default=3,
+                        help="Number of runs per config (default: 3)")
+    parser.add_argument("--profiles", nargs="+", default=None,
+                        choices=list(NETWORK_PROFILES.keys()),
+                        help="Predefined network profiles (default: unlimited)")
+    parser.add_argument("--configs", nargs="+", default=None,
+                        help="Custom FastSync config flags")
+    parser.add_argument("--size-mb", type=int, default=25,
+                        help="Test data size in MB (default: 25)")
+    parser.add_argument("--random-ratio", type=float, default=0.75,
+                        help="Fraction of data that is random/incompressible (default: 0.75)")
+    parser.add_argument("--delay", default=None,
+                        help="Custom network delay (e.g. 50ms)")
+    parser.add_argument("--jitter", default=None,
+                        help="Custom network jitter (e.g. 10ms)")
+    parser.add_argument("--throughput", default=None,
+                        help="Custom throughput limit (e.g. 100mbit)")
+    parser.add_argument("--loss", default=None,
+                        help="Custom packet loss (e.g. 1%%)")
+    parser.add_argument("--no-rsync", action="store_true",
+                        help="Skip rsync comparison")
+    parser.add_argument("--progress", action="store_true",
+                        help="Show progress bar with ETA")
+    parser.add_argument("--output", choices=["table", "json"], default="table",
+                        help="Output format")
+    parser.add_argument("--keep-data", action="store_true",
+                        help="Don't clean up test data")
+    args = parser.parse_args()
+
+    # Build
+    print("Building...")
+    if os.system(f"cmake -B {BUILD_DIR} -S {PROJECT_ROOT} > /dev/null 2>&1") != 0:
+        print("CMake configure failed"); sys.exit(1)
+    if os.system(f"cmake --build {BUILD_DIR} -j$(nproc) > /dev/null 2>&1") != 0:
+        print("Build failed"); sys.exit(1)
+
+    # Determine active profile for display
+    has_custom_net = args.delay or args.jitter or args.throughput or args.loss
+    if has_custom_net:
+        active_profile = "custom"
+        NETWORK_PROFILES["custom"] = {
+            "rate": args.throughput, "delay": args.delay or "0ms",
+            "jitter": args.jitter or "0ms", "loss": args.loss or "0%",
+        }
+        if args.throughput:
+            parts = args.throughput.replace("mbit", "").replace("mbps", "")
+            try:
+                NETWORK_PROFILES["custom"]["rate_bps"] = float(parts) * 1_000_000 / 8
+            except ValueError:
+                pass
+        profiles_to_run = ["custom"]
+    else:
+        profiles_to_run = args.profiles or ["unlimited"]
+
+    # Generate data
+    source_dir = os.path.join(BENCH_DIR, "source")
+    dest_dir = os.path.join(BENCH_DIR, "dest")
+    total_bytes = generate_bench_data(source_dir, args.size_mb, args.random_ratio)
+    compressible_pct = (1 - args.random_ratio) * 100
+    random_pct = args.random_ratio * 100
+    print(f"Generated {total_bytes / (1024*1024):.1f} MB  "
+          f"({random_pct:.0f}% random, {compressible_pct:.0f}% compressible)")
+
+    # Build config list
+    if args.configs:
+        fastsync_configs = [{"name": c, "flags": c.split(), "tool": "fastsync"} for c in args.configs]
+    else:
+        fastsync_configs = list(FASTSYNC_CONFIGS)
+
+    configs = list(fastsync_configs)
+    if not args.no_rsync:
+        configs += RSYNC_CONFIGS
+
+    # Run benchmarks
+    total_runs = len(configs) * args.runs * len(profiles_to_run)
+    progress = Progress(total_runs, "Benchmarking") if args.progress else None
+    if progress:
+        print(f"Running {total_runs} transfers...")
+
+    all_results = []
+    try:
+        for profile in profiles_to_run:
+            results = run_benchmark(source_dir, dest_dir, configs, args.runs, profile, progress)
+            all_results.extend(results)
+    finally:
+        if not args.keep_data:
+            shutil.rmtree(BENCH_DIR, ignore_errors=True)
+
+    # Output
+    if args.output == "json":
+        print(json.dumps(all_results, indent=2))
+    else:
+        print_table(all_results, total_bytes, args.random_ratio)
+        print()
+
+
+if __name__ == "__main__":
+    main()

shell.nix

@@ -15,6 +15,7 @@ pkgs.mkShell {
   buildInputs = with pkgs; [
     zstd
     openssl
+    (python3.withPackages (ps: with ps; [ pytest ]))
   ];
 
   NIX_ENFORCE_PURITY = 0;

test.py

@@ -1,740 +0,0 @@
-import argparse
-import filecmp
-import os
-import random
-import re
-import shutil
-import subprocess
-import sys
-import tempfile
-import time
-import socket
-
-TEST_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_data")
-DEFAULT_SOURCE_DIR = os.path.join(TEST_DIR, "source")
-DEFAULT_DEST_DIR = os.path.join(TEST_DIR, "dest")
-
-SERVER_CMD = ["./build/server"]
-BASE_CLIENT_CMD = ["./build/client"]
-
-DISK_DEVICE = "/dev/nvme0n1p5"
-READ_BPS_MAX = "15M"
-WRITE_BPS_MAX = "10M"
-NETWORK_INTERFACE = "lo"
-
-NETWORK_PROFILES = {
-    "Unlimited": {},
-    "LAN": {
-        "rate": "1000mbit",
-        "delay": "20ms",
-        "jitter": "1ms",
-        "loss": "0.1%",
-    },
-    "WAN": {
-        "rate": "100mbit",
-        "delay": "50ms",
-        "jitter": "10ms",
-        "loss": "1%",
-    },
-}
-
-CLIENT_CMD_PREFIX = [
-    "sudo",
-    "systemd-run",
-    "--scope",
-    "-p",
-    f"IOReadBandwidthMax={DISK_DEVICE} {READ_BPS_MAX}",
-    "-p",
-    f"IOWriteBandwidthMax={DISK_DEVICE} {WRITE_BPS_MAX}",
-]
-
-BASE_CLIENT_FLAGS = ["--save-to-disk"]
-
-TEST_CASES_FULL = [
-    {"name": "Standard", "flags": []},
-    {"name": "Posix Args (no flags)", "flags": [], "posix": True},
-    {"name": "Standard (no metadata)", "flags": [], "use_metadata": False},
-    {"name": "Multithreading (-m)", "flags": ["-m"]},
-    {"name": "Compression (-c)", "flags": ["-c"]},
-    {"name": "Chunk Serialization (-s)", "flags": ["-s"]},
-    {"name": "Compression + Chunk Serialization (-c -s)", "flags": ["-c", "-s"]},
-    {"name": "Multithreading + Compression (-m -c)", "flags": ["-m", "-c"]},
-    {"name": "Multithreading + Chunk Serialization (-m -s)", "flags": ["-m", "-s"]},
-    {
-        "name": "Multithreading + Compression + Chunk Serialization (-m -c -s)",
-        "flags": ["-m", "-c", "-s"],
-    },
-    {"name": "Sendfile (-f)", "flags": ["-f"]},
-    {"name": "Sendfile + Multithreading (-f -m)", "flags": ["-f", "-m"]},
-]
-
-TEST_CASES_LIGHT = [
-    {"name": "Standard", "flags": []},
-    {"name": "Compression (-c)", "flags": ["-c"]},
-    {"name": "Chunk Serialization (-s)", "flags": ["-s"]},
-    {"name": "Multithreading + Compression + Chunk Serialization (-m -c -s)", "flags": ["-m", "-c", "-s"]},
-]
-
-SSH_CASES_FULL = [
-    {"name": "SSH (localhost)", "flags": []},
-    {"name": "SSH Multithreading (-m)", "flags": ["-m"]},
-    {"name": "SSH Compression (-c)", "flags": ["-c"]},
-    {"name": "SSH Chunk Serialization (-s)", "flags": ["-s"]},
-    {"name": "SSH Compression + Chunk Serialization (-c -s)", "flags": ["-c", "-s"]},
-    {"name": "SSH Multithreading + Compression (-m -c)", "flags": ["-m", "-c"]},
-    {"name": "SSH Multithreading + Chunk Serialization (-m -s)", "flags": ["-m", "-s"]},
-    {"name": "SSH Multithreading + Compression + Chunk Serialization (-m -c -s)", "flags": ["-m", "-c", "-s"]},
-]
-
-SSH_CASES_LIGHT = [
-    {"name": "SSH (localhost)", "flags": []},
-]
-
-RSYNC_CASES_FULL = [
-    {"name": "rsync (archive)", "args": ["-aH"]},
-    {"name": "rsync (archive + compress)", "args": ["-aHz"]},
-]
-
-RSYNC_CASES_LIGHT = []
-
-
-def netem_apply(profile):
-    params = NETWORK_PROFILES[profile]
-    if not params:
-        netem_reset()
-        return
-    netem_reset()
-    cmd = ["sudo", "tc", "qdisc", "add", "dev", NETWORK_INTERFACE, "root", "netem"]
-    cmd += ["rate", params["rate"]]
-    cmd += ["delay", params["delay"], params["jitter"]]
-    cmd += ["loss", params["loss"]]
-    subprocess.run(cmd, check=True, capture_output=True)
-
-
-def netem_reset():
-    subprocess.run(
-        f"sudo tc qdisc del dev {NETWORK_INTERFACE} root".split(),
-        capture_output=True,
-    )
-
-
-def wait_proc(proc, timeout=5):
-    try:
-        proc.wait(timeout=timeout)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.wait()
-
-
-def find_free_port():
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(('', 0))
-        return s.getsockname()[1]
-
-
-def generate_test_files(source_dir, full=False):
-    if os.path.exists(source_dir):
-        shutil.rmtree(source_dir)
-    os.makedirs(source_dir)
-
-    target_total = 25 * 1024 * 1024 if full else 0
-    written = 0
-
-    files = {
-        "small.txt": b"hello world\n",
-        "medium.txt": b"the quick brown fox jumps over the lazy dog\n" * 5000,
-        "binary.bin": bytes(range(256)) * 1000,
-        "nested/subdir/deep.txt": b"deeply nested file\n",
-        "nested/another.txt": b"another nested file\n" * 50,
-    }
-    for rel_path, content in files.items():
-        full_path = os.path.join(source_dir, rel_path)
-        os.makedirs(os.path.dirname(full_path), exist_ok=True)
-        with open(full_path, "wb") as f:
-            f.write(content)
-        written += len(content)
-
-    if full:
-        os.makedirs(os.path.join(source_dir, "bulk"), exist_ok=True)
-        i = 0
-        while written < target_total:
-            chunk_size = min(5 * 1024 * 1024, target_total - written)
-            with open(os.path.join(source_dir, f"bulk/file_{i}.dat"), "wb") as f:
-                f.write(random.randbytes(chunk_size))
-            written += chunk_size
-            i += 1
-
-    total_mb = written / (1024 * 1024)
-    small_bytes = sum(len(c) for c in files.values())
-    print(f"  Generated {total_mb:.1f} MB of test data in {source_dir} "
-          f"({(written - small_bytes)/(1024*1024):.1f} MB random, "
-          f"{small_bytes} B structured)")
-    return written
-
-
-def verify_transfer(source_dir, received_dir):
-    source_dir = os.path.abspath(source_dir)
-    received_dir = os.path.abspath(received_dir)
-    if not os.path.exists(received_dir):
-        return [], ["no received files found"]
-    mismatches, missing = [], []
-    for root, dirs, files in os.walk(source_dir):
-        for f in files:
-            src_path = os.path.join(root, f)
-            rel = os.path.relpath(src_path, source_dir)
-            dst_path = os.path.join(received_dir, rel)
-            if not os.path.exists(dst_path):
-                missing.append(rel)
-            elif not filecmp.cmp(src_path, dst_path, shallow=False):
-                mismatches.append(rel)
-    return mismatches, missing
-
-
-def start_rsync_daemon(source_dir):
-    port = find_free_port()
-    conf = os.path.join(tempfile.gettempdir(), f"rsyncd-{port}.conf")
-    with open(conf, "w") as f:
-        f.write(f"port = {port}\nread only = yes\n\n[source]\n    path = {source_dir}\n")
-    daemon = subprocess.Popen(
-        ["rsync", "--daemon", "--no-detach", f"--config={conf}"],
-        stdout=subprocess.DEVNULL, stderr=None,
-    )
-    for _ in range(50):
-        time.sleep(0.1)
-        if daemon.poll() is not None:
-            raise RuntimeError(f"rsync daemon exited (rc={daemon.returncode})")
-        try:
-            with socket.create_connection(("127.0.0.1", port), timeout=0.3):
-                break
-        except (ConnectionRefusedError, OSError):
-            continue
-    else:
-        raise RuntimeError("rsync daemon did not start")
-    return port, conf, daemon
-
-
-def run_single_test(cmd, name, source_dir, dest_dir, *, source_prefix=None, no_server=False, expected_missing=None):
-    if os.path.exists(dest_dir):
-        shutil.rmtree(dest_dir)
-    if no_server:
-        server = None
-    else:
-        server = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
-        time.sleep(0.5)
-    try:
-        start = time.monotonic()
-        result = subprocess.run(cmd, text=True, capture_output=True)
-        duration = time.monotonic() - start
-    finally:
-        if server:
-            wait_proc(server)
-
-    mismatches, missing = [], []
-    if result.returncode == 0:
-        received = os.path.join(dest_dir, source_prefix if source_prefix is not None
-                                else os.path.abspath(source_dir).lstrip(os.sep))
-        mismatches, missing = verify_transfer(source_dir, received)
-        if expected_missing:
-            missing = [m for m in missing if m not in expected_missing]
-
-    first_line = lambda s: (s or "").strip().split("\n")[0]
-    entry = {
-        "name": name,
-        "time": f"{duration:.4f}s" if result.returncode == 0 else "N/A",
-    }
-    if result.returncode == 0 and not mismatches and not missing:
-        entry["status"] = "Success"
-        entry["error"] = ""
-    else:
-        entry["status"] = "Failed"
-        errors = []
-        if result.returncode != 0:
-            errors.append(f"Exit code {result.returncode}: {first_line(result.stderr) or first_line(result.stdout) or 'No output'[:80]}")
-        if missing:
-            errors.append(f"Missing ({len(missing)}): {', '.join(missing[:5])}")
-        if mismatches:
-            errors.append(f"Mismatch ({len(mismatches)}): {', '.join(mismatches[:3])}")
-        entry["error"] = " | ".join(errors)
-    return entry
-
-
-def print_profile_header(profile_name):
-    params = NETWORK_PROFILES[profile_name]
-    print(f"\n{'=' * 60}\nProfile: {profile_name}\n{'=' * 60}")
-    if params:
-        print(f"  Network: rate={params['rate']}, delay={params['delay']} ±{params['jitter']}, loss={params['loss']}")
-        print(f"  Disk I/O: Reads <= {READ_BPS_MAX}, Writes <= {WRITE_BPS_MAX}")
-    else:
-        print("  No limits applied")
-
-
-def run_profile(profile_name, source_dir, dest_dir, *, full=False, test_cases=None, ssh_cases=None, rsync_cases=None):
-    print_profile_header(profile_name)
-    is_limited = profile_name != "Unlimited"
-    client_prefix = CLIENT_CMD_PREFIX if is_limited else []
-
-    if test_cases is None:
-        test_cases = TEST_CASES_FULL if full else TEST_CASES_LIGHT
-    if ssh_cases is None:
-        ssh_cases = SSH_CASES_FULL if full else SSH_CASES_LIGHT
-    if rsync_cases is None:
-        rsync_cases = RSYNC_CASES_FULL if full else RSYNC_CASES_LIGHT
-
-    try:
-        if is_limited and full:
-            netem_apply(profile_name)
-
-        results = []
-        for case in test_cases:
-            flags = BASE_CLIENT_FLAGS + (["-M"] if case.get("use_metadata", True) else []) + case["flags"]
-            if case.get("posix"):
-                cmd = client_prefix + BASE_CLIENT_CMD + [source_dir, dest_dir] + flags
-            else:
-                cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags
-            print(f"\n  --- {case['name']} ---\n    Running: {' '.join(cmd)}")
-            try:
-                r = run_single_test(cmd, case["name"], source_dir, dest_dir)
-                r["suite"] = profile_name
-                results.append(r)
-            except Exception as e:
-                results.append({"name": case["name"], "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-        if SSH_AVAILABLE:
-            for case in ssh_cases:
-                flags = BASE_CLIENT_FLAGS + (["-M"] if case.get("use_metadata", True) else []) + case["flags"]
-                ssh_dest = f"localhost:{dest_dir}_ssh"
-                cmd = BASE_CLIENT_CMD + [source_dir, ssh_dest] + flags
-                print(f"\n  --- {case['name']} ---\n    Running: {' '.join(cmd)}")
-                try:
-                    r = run_single_test(cmd, case["name"], source_dir, f"{dest_dir}_ssh", no_server=True)
-                    r["suite"] = profile_name
-                    results.append(r)
-                except Exception as e:
-                    results.append({"name": case["name"], "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-    
-        if rsync_cases:
-            port, conf, daemon = start_rsync_daemon(source_dir)
-            try:
-                for case in rsync_cases:
-                    cmd = client_prefix + ["rsync"] + case["args"] + [f"rsync://localhost:{port}/source/", f"{dest_dir}/"]
-                    print(f"\n  --- {case['name']} ---\n    Running: {' '.join(cmd)}")
-                    try:
-                        r = run_single_test(cmd, case["name"], source_dir, dest_dir, source_prefix="")
-                        r["suite"] = profile_name
-                    except subprocess.TimeoutExpired:
-                        r = {"name": case["name"], "suite": profile_name, "status": "Timeout", "time": "N/A", "error": "Exceeded 120s"}
-                    except Exception as e:
-                        r = {"name": case["name"], "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)}
-                    else:
-                        if r["status"] != "Success" and client_prefix:
-                            tmp = tempfile.mkdtemp()
-                            try:
-                                plain = subprocess.run(["rsync", "-aH", f"rsync://localhost:{port}/source/", f"{tmp}/"], capture_output=True, text=True, timeout=30)
-                                if plain.returncode != 0:
-                                    errs = [l for l in (plain.stderr or "").split("\n") if l.strip()]
-                                    if errs:
-                                        r["error"] += f" | raw: {errs[-1][:150]}"
-                            finally:
-                                shutil.rmtree(tmp, ignore_errors=True)
-                    results.append(r)
-            finally:
-                wait_proc(daemon)
-                try:
-                    os.unlink(conf)
-                except Exception:
-                    pass
-
-        if full:
-            # Feature-specific tests for rsync-compatible flags
-            print("\n  " + "─" * 56 + "\n  Feature Tests\n  " + "─" * 56)
-
-            # Dry run (-n) — no server needed
-            print("\n  --- Dry run (-n) ---")
-            flags = BASE_CLIENT_FLAGS + ["-n"]
-            cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags
-            print(f"    Running: {' '.join(cmd)}")
-            try:
-                start = time.monotonic()
-                result = subprocess.run(cmd, text=True, capture_output=True)
-                duration = time.monotonic() - start
-                r = {"name": "Dry run (-n)", "suite": profile_name}
-                if result.returncode == 0 and "Dry run:" in result.stdout:
-                    r["status"] = "Success"
-                    r["time"] = f"{duration:.4f}s"
-                    r["error"] = ""
-                else:
-                    r["status"] = "Failed"
-                    r["time"] = "N/A"
-                    r["error"] = f"Exit {result.returncode}: {(result.stderr or result.stdout)[:100]}"
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Dry run (-n)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-            # Archive mode (-a)
-            feature_flags = BASE_CLIENT_FLAGS + ["-a"]
-            cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
-            print(f"\n  --- Archive mode (-a) ---\n    Running: {' '.join(cmd)}")
-            try:
-                r = run_single_test(cmd, "Archive mode (-a)", source_dir, dest_dir)
-                r["suite"] = profile_name
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Archive mode (-a)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-            # Exclude (--exclude small.txt)
-            feature_flags = BASE_CLIENT_FLAGS + ["--exclude", "small.txt"]
-            cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
-            print(f"\n  --- Exclude (--exclude small.txt) ---\n    Running: {' '.join(cmd)}")
-            try:
-                r = run_single_test(cmd, "Exclude (--exclude small.txt)", source_dir, dest_dir,
-                                    expected_missing=["small.txt"])
-                r["suite"] = profile_name
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Exclude (--exclude small.txt)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-            # Progress (--progress)
-            feature_flags = BASE_CLIENT_FLAGS + ["--progress"]
-            cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
-            print(f"\n  --- Progress (--progress) ---\n    Running: {' '.join(cmd)}")
-            try:
-                r = run_single_test(cmd, "Progress (--progress)", source_dir, dest_dir)
-                r["suite"] = profile_name
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Progress (--progress)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-            # Bandwidth limit (--bwlimit 10240 = 10 MB/s)
-            feature_flags = BASE_CLIENT_FLAGS + ["--bwlimit", "10240"]
-            cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
-            print(f"\n  --- Bandwidth limit (--bwlimit 10240 KB/s) ---\n    Running: {' '.join(cmd)}")
-            try:
-                r = run_single_test(cmd, "Bandwidth limit (--bwlimit 10240)", source_dir, dest_dir)
-                r["suite"] = profile_name
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Bandwidth limit (--bwlimit 10240)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-            # Incremental sync (--incremental) — first sync, then second sync should skip all
-            print(f"\n  --- Incremental (--incremental) ---")
-            try:
-                flags = BASE_CLIENT_FLAGS + ["-M"]
-                srv = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
-                time.sleep(0.5)
-                first_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags
-                r1 = subprocess.run(first_cmd, text=True, capture_output=True)
-                wait_proc(srv)
-                if r1.returncode != 0:
-                    raise RuntimeError(f"First sync failed: {r1.stderr[:100]}")
-                srv2 = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
-                time.sleep(0.5)
-                second_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags + ["--incremental"]
-                start = time.monotonic()
-                r2 = subprocess.run(second_cmd, text=True, capture_output=True, timeout=30)
-                duration = time.monotonic() - start
-                wait_proc(srv2)
-                r = {"name": "Incremental (--incremental)", "suite": profile_name,
-                     "status": "Success" if r2.returncode == 0 else "Failed",
-                     "time": f"{duration:.4f}s" if r2.returncode == 0 else "N/A",
-                     "error": "" if r2.returncode == 0 else f"Exit {r2.returncode}: {(r2.stderr or r2.stdout)[:60]}"}
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Incremental (--incremental)", "suite": profile_name,
-                                "status": "Error", "time": "N/A", "error": str(e)})
-
-            # Chunk size (--chunk-size 5242880)
-            feature_flags = BASE_CLIENT_FLAGS + ["--chunk-size", "5242880"]
-            cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
-            print(f"\n  --- Chunk size (--chunk-size 5242880) ---\n    Running: {' '.join(cmd)}")
-            try:
-                r = run_single_test(cmd, "Chunk size (--chunk-size 5242880)", source_dir, dest_dir)
-                r["suite"] = profile_name
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Chunk size (--chunk-size 5242880)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-            # Delete (--delete) — pre-populate dest, add extra files, then sync with --delete
-            # Note: server handles one client per launch, so we restart between syncs
-            print(f"\n  --- Delete (--delete) ---")
-            try:
-                flags = BASE_CLIENT_FLAGS + ["-M"]
-                # First sync (no delete) to populate dest
-                s1 = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
-                time.sleep(0.5)
-                first_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags
-                r1 = subprocess.run(first_cmd, text=True, capture_output=True)
-                wait_proc(s1)
-                if r1.returncode != 0:
-                    raise RuntimeError(f"First sync failed: {r1.stderr[:100]}")
-                # Add extra files to received dir
-                received = os.path.join(dest_dir, os.path.abspath(source_dir).lstrip(os.sep))
-                extra_path = os.path.join(received, "extra_file.txt")
-                with open(extra_path, "w") as f:
-                    f.write("should be deleted")
-                extra_dir = os.path.join(received, "extra_dir")
-                os.makedirs(extra_dir, exist_ok=True)
-                with open(os.path.join(extra_dir, "nested.txt"), "w") as f:
-                    f.write("nested extra")
-                # Second sync with --delete (fresh server)
-                s2 = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
-                time.sleep(0.5)
-                second_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags + ["--delete"]
-                start = time.monotonic()
-                r2 = subprocess.run(second_cmd, text=True, capture_output=True)
-                duration = time.monotonic() - start
-                wait_proc(s2)
-                r = {"name": "Delete (--delete)", "suite": profile_name}
-                if r2.returncode == 0 and not os.path.exists(extra_path) and not os.path.exists(extra_dir):
-                    mismatches, missing = verify_transfer(source_dir, received)
-                    if not mismatches and not missing:
-                        r["status"] = "Success"
-                        r["time"] = f"{duration:.4f}s"
-                        r["error"] = ""
-                    else:
-                        r["status"] = "Failed"
-                        r["time"] = "N/A"
-                        r["error"] = f"post-delete verify: mismatches={len(mismatches)}, missing={len(missing)}"
-                else:
-                    r["status"] = "Failed"
-                    r["time"] = "N/A"
-                    errs = []
-                    if r2.returncode != 0:
-                        errs.append(f"Exit {r2.returncode}: {(r2.stderr or r2.stdout)[:60]}")
-                    if os.path.exists(extra_path):
-                        errs.append("extra_file.txt remains")
-                    if os.path.exists(extra_dir):
-                        errs.append("extra_dir remains")
-                    r["error"] = " | ".join(errs)
-                results.append(r)
-            except Exception as e:
-                results.append({"name": "Delete (--delete)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-            # SSH feature tests
-            if SSH_AVAILABLE:
-                ssh_dest = f"localhost:{dest_dir}_ssh"
-                ssh_feature_cases = [
-                    {"name": "SSH Archive (-a)", "flags": ["-a"]},
-                    {"name": "SSH Exclude (--exclude small.txt)", "flags": ["--exclude", "small.txt"], "expected_missing": ["small.txt"]},
-                ]
-                for case in ssh_feature_cases:
-                    flags = BASE_CLIENT_FLAGS + case["flags"]
-                    cmd = BASE_CLIENT_CMD + [source_dir, ssh_dest] + flags
-                    print(f"\n  --- {case['name']} ---\n    Running: {' '.join(cmd)}")
-                    try:
-                        r = run_single_test(cmd, case["name"], source_dir, f"{dest_dir}_ssh",
-                                            no_server=True,
-                                            expected_missing=case.get("expected_missing"))
-                        r["suite"] = profile_name
-                        results.append(r)
-                    except Exception as e:
-                        results.append({"name": case["name"], "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
-
-    except (subprocess.CalledProcessError, RuntimeError) as e:
-        print(f"  Error: {e}")
-        results = []
-    finally:
-        if is_limited:
-            try:
-                netem_reset()
-            except Exception:
-                pass
-
-    return results
-
-
-def print_metrics(profile_name, results, total_bytes):
-    params = NETWORK_PROFILES.get(profile_name)
-    if not params or "rate" not in params:
-        return
-
-    client_times, rsync_times = [], {}
-    for r in results:
-        if r["status"] != "Success" or r["time"] == "N/A":
-            continue
-        t = float(r["time"].rstrip("s"))
-        if r["name"].startswith("rsync"):
-            rsync_times[r["name"]] = t
-        elif "Dry run" not in r["name"]:
-            client_times.append((t, r["name"]))
-    if not client_times or len(rsync_times) < 2:
-        return
-
-    m = re.match(r'(\d+)\s*(mbit|gbit|kbit|bit)', params["rate"])
-    rate_val = int(m.group(1)) * {'mbit': 1_000_000, 'gbit': 1_000_000_000, 'kbit': 1000, 'bit': 1}[m.group(2)] / 8 if m else None
-
-    best_time, best_name = min(client_times, key=lambda x: x[0])
-    theoretical_max = total_bytes / rate_val if rate_val else None
-
-    print(f"\n  {'─' * 90}\n  Profile: {profile_name}\n  {'─' * 90}")
-    print(f"  Total data size:               {total_bytes / (1024*1024):.1f} MB")
-    if rate_val:
-        print(f"  Network rate:                  {params['rate']} ({format_throughput(rate_val)})")
-    print(f"  Best client configuration:     {best_name}")
-    print(f"  Best client time:              {best_time:.4f}s")
-    if theoretical_max:
-        print(f"  Theoretical max (uncompressed): {theoretical_max:.4f}s")
-        print(f"  Speedup vs theoretical max:    {theoretical_max / best_time:.2f}x")
-    if (a := rsync_times.get("rsync (archive)")):
-        print(f"  Speedup vs rsync (archive):    {a / best_time:.2f}x")
-    if (c := rsync_times.get("rsync (archive + compress)")):
-        print(f"  Speedup vs rsync (compress):   {c / best_time:.2f}x")
-
-
-def format_throughput(bps):
-    for unit, threshold in [("GB/s", 1_000_000_000), ("MB/s", 1_000_000), ("KB/s", 1000)]:
-        if bps >= threshold:
-            return f"{bps/threshold:.1f} {unit}"
-    return f"{bps:.0f} B/s"
-
-
-SSH_AVAILABLE = False
-
-def check_ssh_localhost():
-    global SSH_AVAILABLE
-    build_dir = os.path.abspath("build")
-    server_path = os.path.join(build_dir, "server")
-
-    try:
-        r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
-                            "localhost", "which", "fastsync-server"],
-                           capture_output=True, timeout=10)
-    except FileNotFoundError:
-        SSH_AVAILABLE = False
-        return
-    if r.returncode == 0:
-        SSH_AVAILABLE = True
-        return
-
-    SSH_AVAILABLE = False
-    # Try each PATH dir: create symlink, then verify with which
-    try:
-        r = subprocess.run(
-            ["ssh", "-o", "BatchMode=yes", "localhost",
-             'echo "$PATH"'],
-            capture_output=True, timeout=10, text=True)
-    except FileNotFoundError:
-        return
-    if r.returncode != 0:
-        return
-    for d in r.stdout.strip().split(":"):
-        d = d.strip()
-        if not d:
-            continue
-        if "wrappers" in d:
-            continue
-        test = subprocess.run(
-            ["ssh", "-o", "BatchMode=yes", "localhost",
-             f'test -w "{d}" && ln -sf {server_path} "{d}/fastsync-server" && which fastsync-server'],
-            capture_output=True, timeout=10)
-        if test.returncode == 0:
-            SSH_AVAILABLE = True
-            return
-
-
-def preflight_checks():
-    global SSH_AVAILABLE
-    errors = []
-    print("Pre-flight checks:")
-    print("  [1] --help flag...", end=" ")
-    r = subprocess.run(BASE_CLIENT_CMD + ["--help"], capture_output=True, text=True)
-    if r.returncode == 0 and "Usage:" in r.stdout and "SSH transport" in r.stdout:
-        print("OK")
-    else:
-        print("FAIL")
-        errors.append("--help failed")
-
-    print("  [2] Remote SSH dest detection...", end=" ")
-    r = subprocess.run(BASE_CLIENT_CMD + ["/x", "somehost:/y"], capture_output=True, text=True, timeout=5)
-    if r.returncode != 0 and ("ssh" in r.stderr or "Could not receive" in r.stderr or "could not launch" in r.stderr or "Error" in r.stderr):
-        print("OK (detected as SSH)")
-    else:
-        print("FAIL (not detected as SSH dest)")
-        errors.append("SSH detection failed")
-
-    print("  [3] Server --stdio flag...", end=" ")
-    try:
-        r = subprocess.run(["./build/server", "--stdio"], capture_output=True, text=True, timeout=3)
-        if r.returncode != 0 and ("receiving" in r.stderr or "receiving" in r.stdout or "Receiving" in r.stderr):
-            print("OK (started in stdio mode)")
-        else:
-            print("WARN (stdio exited: rc=%d)" % r.returncode)
-    except subprocess.TimeoutExpired:
-        print("OK (waiting for stdin)")
-
-    print("  [4] Posix arg syntax (no server, expect failure)...", end=" ")
-    r = subprocess.run(BASE_CLIENT_CMD + ["/tmp/x", "/tmp/y"], capture_output=True, text=True, timeout=5)
-    if r.returncode != 0 and "connect" in r.stderr:
-        print("OK (TCP fallback)")
-    else:
-        print("FAIL")
-        errors.append("Posix arg syntax failed")
-
-    check_ssh_localhost()
-    print("  [5] SSH to localhost...", end=" ")
-    if SSH_AVAILABLE:
-        print("OK")
-    else:
-        print("SKIP (install fastsync-server in PATH on remote)")
-
-    if errors:
-        print(f"\n  {len(errors)} pre-flight check(s) failed: {', '.join(errors)}")
-        sys.exit(1)
-    print("  All pre-flight checks passed.\n")
-
-
-def main():
-    os.system("cmake -B build -S . > /dev/null 2>&1")
-    if os.system("cd build && make -j$(nproc) 2>&1 | tail -3") != 0:
-        print("Build failed")
-        sys.exit(1)
-    preflight_checks()
-    parser = argparse.ArgumentParser(description="FastSync integration test / benchmark")
-    parser.add_argument("--source-dir", default=DEFAULT_SOURCE_DIR)
-    parser.add_argument("--dest-dir", default=DEFAULT_DEST_DIR)
-    parser.add_argument("--keep-data", action="store_true")
-    parser.add_argument("--unlimited", action="store_true")
-    parser.add_argument("--wan", action="store_true")
-    parser.add_argument("--full", action="store_true", help="Run full test suite with network shaping, SSH, rsync benchmarks")
-    args = parser.parse_args()
-
-    total_bytes = generate_test_files(args.source_dir, full=args.full)
-    if os.path.exists(args.dest_dir):
-        shutil.rmtree(args.dest_dir)
-    os.makedirs(args.dest_dir, exist_ok=True)
-
-    profiles = []
-    if args.unlimited:
-        profiles.append("Unlimited")
-    elif args.wan:
-        profiles.append("WAN")
-    else:
-        profiles.append("LAN" if args.full else "Unlimited")
-
-    try:
-        all_results = []
-        for p in profiles:
-            all_results.extend(run_profile(p, args.source_dir, args.dest_dir, full=args.full))
-
-        print("\n" + "=" * 130)
-        print(f"{'RESULTS':^130}")
-        print("=" * 130)
-        print(f"{'Configuration':<45} | {'Profile':<12} | {'Status':<8} | {'Time':<10} | {'Details'}")
-        print("-" * 130)
-        for r in all_results:
-            print(f"{r['name']:<45} | {r['suite']:<12} | {r['status']:<8} | {r['time']:<10} | {r['error']}")
-
-        for p in profiles:
-            print_metrics(p, [r for r in all_results if r["suite"] == p], total_bytes)
-
-        failed = [r for r in all_results if r["status"] != "Success"]
-        if failed:
-            print(f"\n  {len(failed)} test(s) FAILED")
-            sys.exit(1)
-        print(f"\n  ALL {len(all_results)} TESTS PASSED")
-    finally:
-        if not args.keep_data:
-            shutil.rmtree(TEST_DIR, ignore_errors=True)
-
-
-if __name__ == "__main__":
-    main()

tests/conftest.py

@@ -0,0 +1,17 @@
+"""Shared pytest configuration for integration tests."""
+import os
+import sys
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "integration"))
+
+from common import ServerManager
+
+
+@pytest.fixture(scope="session")
+def shared_server():
+    """One server for the entire test session. Avoids 27+ server start/stop cycles."""
+    server = ServerManager()
+    server.start()
+    yield server
+    server.stop()

tests/integration/common.py

@@ -0,0 +1,181 @@
+import filecmp
+import os
+import random
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+BUILD_DIR = os.path.join(PROJECT_ROOT, "build")
+SERVER_CMD = [os.path.join(BUILD_DIR, "server")]
+CLIENT_CMD = [os.path.join(BUILD_DIR, "client")]
+TEST_DATA_DIR = os.path.join(PROJECT_ROOT, "test_data")
+
+
+class ServerManager:
+    """Manages a long-lived server process. Reuses across test cases."""
+
+    def __init__(self):
+        self._proc = None
+        self._port = None
+
+    def start(self, extra_args=None):
+        self.stop()
+        self._port = _find_free_port()
+        cmd = SERVER_CMD + ["-p", str(self._port)]
+        if extra_args:
+            cmd += extra_args
+        self._proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        _wait_for_port(self._port, timeout=5)
+
+    def stop(self):
+        if self._proc:
+            _wait_proc(self._proc)
+            self._proc = None
+
+    @property
+    def port(self):
+        return self._port
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, *args):
+        self.stop()
+
+    def __del__(self):
+        self.stop()
+
+
+def run_client(source_dir, dest_dir, flags=None, port=None, extra_args=None):
+    """Run the client and return (result, duration)."""
+    cmd = CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir, "--save-to-disk"]
+    if port:
+        cmd += ["--server-port", str(port)]
+    if flags:
+        cmd += flags
+    if extra_args:
+        cmd += extra_args
+    start = time.monotonic()
+    result = subprocess.run(cmd, text=True, capture_output=True)
+    duration = time.monotonic() - start
+    return result, duration
+
+
+def run_client_posix(source_dir, dest_dir, flags=None, port=None):
+    """Run the client with positional args (rsync-style)."""
+    cmd = CLIENT_CMD + [source_dir, dest_dir, "--save-to-disk"]
+    if port:
+        cmd += ["--server-port", str(port)]
+    if flags:
+        cmd += flags
+    start = time.monotonic()
+    result = subprocess.run(cmd, text=True, capture_output=True)
+    duration = time.monotonic() - start
+    return result, duration
+
+
+def generate_test_files(source_dir, full=False):
+    """Generate structured test data. Returns total bytes written."""
+    if os.path.exists(source_dir):
+        shutil.rmtree(source_dir)
+    os.makedirs(source_dir)
+
+    target_total = 25 * 1024 * 1024 if full else 0
+    written = 0
+
+    files = {
+        "small.txt": b"hello world\n",
+        "medium.txt": b"the quick brown fox jumps over the lazy dog\n" * 5000,
+        "binary.bin": bytes(range(256)) * 1000,
+        "nested/subdir/deep.txt": b"deeply nested file\n",
+        "nested/another.txt": b"another nested file\n" * 50,
+    }
+    for rel_path, content in files.items():
+        full_path = os.path.join(source_dir, rel_path)
+        os.makedirs(os.path.dirname(full_path), exist_ok=True)
+        with open(full_path, "wb") as f:
+            f.write(content)
+        written += len(content)
+
+    if full:
+        os.makedirs(os.path.join(source_dir, "bulk"), exist_ok=True)
+        i = 0
+        while written < target_total:
+            chunk_size = min(5 * 1024 * 1024, target_total - written)
+            with open(os.path.join(source_dir, f"bulk/file_{i}.dat"), "wb") as f:
+                f.write(random.randbytes(chunk_size))
+            written += chunk_size
+            i += 1
+
+    return written
+
+
+def verify_transfer(source_dir, received_dir):
+    """Verify all files from source exist in received_dir and match. Returns (mismatches, missing)."""
+    source_dir = os.path.abspath(source_dir)
+    received_dir = os.path.abspath(received_dir)
+    if not os.path.exists(received_dir):
+        return [], ["no received files found"]
+    mismatches, missing = [], []
+    for root, dirs, files in os.walk(source_dir):
+        for f in files:
+            src_path = os.path.join(root, f)
+            rel = os.path.relpath(src_path, source_dir)
+            dst_path = os.path.join(received_dir, rel)
+            if not os.path.exists(dst_path):
+                missing.append(rel)
+            elif not filecmp.cmp(src_path, dst_path, shallow=False):
+                mismatches.append(rel)
+    return mismatches, missing
+
+
+def clean_dir(path):
+    """Remove and recreate a directory."""
+    if os.path.exists(path):
+        shutil.rmtree(path)
+    os.makedirs(path, exist_ok=True)
+
+
+def make_result(name, success, duration=None, error=""):
+    """Create a standardized result dict."""
+    return {
+        "name": name,
+        "status": "Success" if success else "Failed",
+        "time": f"{duration:.4f}s" if duration is not None else "N/A",
+        "error": error,
+    }
+
+
+def get_dest_received_dir(dest_dir, source_dir):
+    """Get the path where received files land inside dest_dir."""
+    return os.path.join(dest_dir, os.path.abspath(source_dir).lstrip(os.sep))
+
+
+def _find_free_port():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+def _wait_for_port(port, timeout=5):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.3):
+                return
+        except (ConnectionRefusedError, OSError):
+            time.sleep(0.05)
+    raise RuntimeError(f"Server port {port} not ready after {timeout}s")
+
+
+def _wait_proc(proc, timeout=5):
+    try:
+        proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()

tests/integration/test_features.py

@@ -0,0 +1,256 @@
+"""Feature tests: incremental sync, bandwidth limiting, dry run, metadata, filters."""
+import os
+import shutil
+import sys
+import time
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (
+    PROJECT_ROOT, BUILD_DIR, TEST_DATA_DIR,
+    run_client,
+    generate_test_files, verify_transfer, clean_dir, make_result,
+    get_dest_received_dir, CLIENT_CMD,
+)
+
+SOURCE_DIR = os.path.join(TEST_DATA_DIR, "feature_source")
+DEST_DIR = os.path.join(TEST_DATA_DIR, "feature_dest")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_test_data():
+    generate_test_files(SOURCE_DIR, full=False)
+    clean_dir(DEST_DIR)
+    yield
+    shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)
+
+
+class TestDryRun:
+    def test_dry_run(self):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-n"],
+        )
+        assert result.returncode == 0, f"Exit {result.returncode}: {result.stderr[:100]}"
+        assert "Dry run:" in result.stdout, f"No dry run output: {result.stdout[:200]}"
+
+
+class TestArchiveMode:
+    def test_archive_mode(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-a"],
+            port=shared_server.port,
+        )
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"Missing: {missing}"
+        assert not mismatches, f"Mismatch: {mismatches}"
+
+
+class TestExclude:
+    def test_exclude_single(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--exclude", "small.txt"],
+            port=shared_server.port,
+        )
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert "small.txt" in missing, "small.txt should be excluded but was transferred"
+        other_missing = [m for m in missing if m != "small.txt"]
+        assert not other_missing, f"Other files missing: {other_missing}"
+        assert not mismatches, f"Mismatch: {mismatches}"
+
+    def test_exclude_glob(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--exclude", "*.txt"],
+            port=shared_server.port,
+        )
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        assert not os.path.exists(os.path.join(received, "small.txt")), "small.txt should be excluded"
+        assert os.path.exists(os.path.join(received, "binary.bin")), "binary.bin should be present"
+
+
+class TestInclude:
+    def test_include_single(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--include", "binary.bin"],
+            port=shared_server.port,
+        )
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        assert os.path.exists(os.path.join(received, "binary.bin")), "binary.bin should be included"
+        assert not os.path.exists(os.path.join(received, "small.txt")), "small.txt should not be included"
+
+    def test_include_glob(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--include", "*.bin"],
+            port=shared_server.port,
+        )
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        assert os.path.exists(os.path.join(received, "binary.bin")), "binary.bin should be included"
+
+
+class TestSizeFilters:
+    def test_max_size(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--max-size", "100"],
+            port=shared_server.port,
+        )
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        assert os.path.exists(os.path.join(received, "small.txt")), "small.txt should be present"
+        assert not os.path.exists(os.path.join(received, "medium.txt")), "medium.txt should be skipped"
+
+    def test_min_size(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--min-size", "1000"],
+            port=shared_server.port,
+        )
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        assert not os.path.exists(os.path.join(received, "small.txt")), "small.txt should be skipped"
+        assert os.path.exists(os.path.join(received, "medium.txt")), "medium.txt should be present"
+
+
+class TestIncremental:
+    def test_incremental_skips_unchanged(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-M"],
+            port=shared_server.port,
+        )
+        assert result.returncode == 0, f"First sync failed: {result.stderr[:100]}"
+
+        start = time.monotonic()
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-M", "--incremental"],
+            port=shared_server.port,
+        )
+        incremental_time = time.monotonic() - start
+
+        assert result.returncode == 0, f"Incremental sync failed: {(result.stderr or result.stdout)[:200]}"
+
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"Missing: {missing}"
+        assert not mismatches, f"Mismatch: {mismatches}"
+
+    def test_incremental_detects_changes(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, _ = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-M"],
+            port=shared_server.port,
+        )
+        assert result.returncode == 0
+
+        modified_file = os.path.join(SOURCE_DIR, "small.txt")
+        with open(modified_file, "wb") as f:
+            f.write(b"modified content for incremental test\n")
+
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-M", "--incremental"],
+            port=shared_server.port,
+        )
+        assert result.returncode == 0
+
+        with open(modified_file, "wb") as f:
+            f.write(b"hello world\n")
+
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        received_file = os.path.join(received, "small.txt")
+        assert os.path.exists(received_file), "Modified file should be present"
+        with open(received_file, "rb") as f:
+            content = f.read()
+        assert b"modified content" in content, f"Modified content not transferred: {content[:50]}"
+
+
+class TestDelete:
+    def test_delete_removes_extra_files(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, _ = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-M"],
+            port=shared_server.port,
+        )
+        assert result.returncode == 0
+
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        extra_file = os.path.join(received, "extra_file.txt")
+        extra_dir = os.path.join(received, "extra_dir")
+        with open(extra_file, "w") as f:
+            f.write("should be deleted")
+        os.makedirs(extra_dir, exist_ok=True)
+        with open(os.path.join(extra_dir, "nested.txt"), "w") as f:
+            f.write("nested extra")
+
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["-M", "--delete"],
+            port=shared_server.port,
+        )
+
+        assert result.returncode == 0, f"Delete sync failed: {(result.stderr or result.stdout)[:200]}"
+        assert not os.path.exists(extra_file), "extra_file.txt should be deleted"
+        assert not os.path.exists(extra_dir), "extra_dir should be deleted"
+
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"Missing: {missing}"
+        assert not mismatches, f"Mismatch: {mismatches}"
+
+
+class TestProgress:
+    def test_progress_output(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--progress"],
+            port=shared_server.port,
+        )
+        assert result.returncode == 0, f"Exit {result.returncode}: {result.stderr[:100]}"
+        output = result.stdout + result.stderr
+        assert len(output) >= 0
+
+
+class TestBandwidthLimit:
+    def test_bwlimit_runs(self, shared_server):
+        clean_dir(DEST_DIR)
+        result, dur = run_client(
+            SOURCE_DIR, DEST_DIR,
+            flags=["--bwlimit", "10240"],
+            port=shared_server.port,
+        )
+        assert result.returncode == 0, f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}"
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"Missing: {missing}"
+        assert not mismatches, f"Mismatch: {mismatches}"

tests/integration/test_preflight.py

@@ -0,0 +1,81 @@
+"""CLI validation and preflight checks."""
+import subprocess
+import sys
+import os
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import BUILD_DIR, CLIENT_CMD, SERVER_CMD
+
+
+class TestHelp:
+    def test_client_help(self):
+        r = subprocess.run(CLIENT_CMD + ["--help"], capture_output=True, text=True)
+        assert r.returncode == 0
+        assert "Usage:" in r.stdout
+        assert "SSH transport" in r.stdout
+
+    def test_server_help(self):
+        r = subprocess.run(SERVER_CMD + ["--help"], capture_output=True, text=True)
+        assert r.returncode == 0
+        assert "Usage:" in r.stdout
+
+
+class TestSSHDetection:
+    def test_remote_dest_detected(self):
+        """Posix-style SSH dest should be detected and fail gracefully."""
+        r = subprocess.run(
+            CLIENT_CMD + ["/x", "somehost:/y"],
+            capture_output=True, text=True, timeout=5,
+        )
+        assert r.returncode != 0
+        stderr = (r.stderr or "").lower()
+        assert "ssh" in stderr or "error" in stderr or "could not" in stderr
+
+    def test_local_dest_not_ssh(self):
+        """Local path should not be detected as SSH."""
+        r = subprocess.run(
+            CLIENT_CMD + ["/tmp/x", "/tmp/y"],
+            capture_output=True, text=True, timeout=5,
+        )
+        # Should fail with connection error (no server), not SSH error
+        assert r.returncode != 0
+
+
+class TestServerStdio:
+    def test_stdio_mode_starts(self):
+        """Server --stdio should start and wait for stdin."""
+        try:
+            r = subprocess.run(
+                SERVER_CMD + ["--stdio"],
+                capture_output=True, text=True, timeout=3,
+            )
+            # Should exit with error (no data on stdin) or timeout
+        except subprocess.TimeoutExpired:
+            pass  # Expected: server waiting for stdin
+
+
+class TestServerPort:
+    def test_invalid_port(self):
+        """Server should reject invalid port numbers."""
+        r = subprocess.run(
+            SERVER_CMD + ["-p", "99999"],
+            capture_output=True, text=True, timeout=5,
+        )
+        assert r.returncode != 0
+
+    def test_default_port(self):
+        """Server should start on default port 8080."""
+        proc = subprocess.Popen(
+            SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None,
+        )
+        try:
+            import socket, time
+            time.sleep(0.5)
+            with socket.create_connection(("127.0.0.1", 8080), timeout=2):
+                pass  # Port is listening
+        except (ConnectionRefusedError, OSError):
+            pytest.fail("Server not listening on default port 8080")
+        finally:
+            proc.terminate()
+            proc.wait(timeout=5)

tests/integration/test_ssh.py

@@ -0,0 +1,132 @@
+"""SSH transport tests."""
+import os
+import shutil
+import subprocess
+import sys
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (
+    PROJECT_ROOT, BUILD_DIR, TEST_DATA_DIR,
+    CLIENT_CMD, generate_test_files, verify_transfer, clean_dir, make_result,
+)
+
+SOURCE_DIR = os.path.join(TEST_DATA_DIR, "ssh_source")
+DEST_DIR = os.path.join(TEST_DATA_DIR, "ssh_dest")
+SSH_AVAILABLE = False
+
+
+def _check_ssh():
+    global SSH_AVAILABLE
+    try:
+        r = subprocess.run(
+            ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
+             "localhost", "which", "fastsync-server"],
+            capture_output=True, timeout=10,
+        )
+        if r.returncode == 0:
+            SSH_AVAILABLE = True
+            return
+
+        # Try to install server binary into PATH
+        server_path = os.path.join(BUILD_DIR, "server")
+        r = subprocess.run(
+            ["ssh", "-o", "BatchMode=yes", "localhost", 'echo "$PATH"'],
+            capture_output=True, timeout=10, text=True,
+        )
+        if r.returncode != 0:
+            return
+        for d in r.stdout.strip().split(":"):
+            d = d.strip()
+            if not d or "wrappers" in d:
+                continue
+            test = subprocess.run(
+                ["ssh", "-o", "BatchMode=yes", "localhost",
+                 f'test -w "{d}" && ln -sf {server_path} "{d}/fastsync-server" && which fastsync-server'],
+                capture_output=True, timeout=10,
+            )
+            if test.returncode == 0:
+                SSH_AVAILABLE = True
+                return
+    except FileNotFoundError:
+        pass
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_test_data():
+    _check_ssh()
+    if SSH_AVAILABLE:
+        generate_test_files(SOURCE_DIR, full=False)
+        clean_dir(DEST_DIR)
+    yield
+    shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)
+
+
+def _run_ssh_test(name, flags, expected_missing=None):
+    """Run an SSH test case (no server process needed, client spawns SSH)."""
+    ssh_dest = f"localhost:{DEST_DIR}"
+    clean_dir(DEST_DIR)
+    cmd = CLIENT_CMD + [SOURCE_DIR, ssh_dest, "--save-to-disk"] + flags
+    start = __import__("time").monotonic()
+    result = subprocess.run(cmd, text=True, capture_output=True)
+    duration = __import__("time").monotonic() - start
+
+    if result.returncode != 0:
+        return make_result(name, False, duration, f"Exit {result.returncode}: {(result.stderr or result.stdout)[:100]}")
+
+    mismatches, missing = verify_transfer(SOURCE_DIR, DEST_DIR)
+    if expected_missing:
+        missing = [m for m in missing if m not in expected_missing]
+    if missing:
+        return make_result(name, False, duration, f"Missing: {', '.join(missing[:5])}")
+    if mismatches:
+        return make_result(name, False, duration, f"Mismatch: {', '.join(mismatches[:3])}")
+    return make_result(name, True, duration)
+
+
+@pytest.mark.skipif(not SSH_AVAILABLE, reason="SSH to localhost not available")
+class TestSSHStandard:
+    def test_standard(self):
+        r = _run_ssh_test("SSH (localhost)", [])
+        assert r["status"] == "Success", r["error"]
+
+    def test_multithreading(self):
+        r = _run_ssh_test("SSH Multithreading (-m)", ["-m"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_compression(self):
+        r = _run_ssh_test("SSH Compression (-c)", ["-c"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_chunk_serialization(self):
+        r = _run_ssh_test("SSH Chunk Serialization (-s)", ["-s"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_compression_chunk(self):
+        r = _run_ssh_test("SSH Compression + Chunk (-c -s)", ["-c", "-s"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_multithread_compression(self):
+        r = _run_ssh_test("SSH Multithread + Compression (-m -c)", ["-m", "-c"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_multithread_chunk(self):
+        r = _run_ssh_test("SSH Multithread + Chunk (-m -s)", ["-m", "-s"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_all_flags(self):
+        r = _run_ssh_test("SSH All Flags (-m -c -s)", ["-m", "-c", "-s"])
+        assert r["status"] == "Success", r["error"]
+
+
+@pytest.mark.skipif(not SSH_AVAILABLE, reason="SSH to localhost not available")
+class TestSSHFeatures:
+    def test_archive(self):
+        r = _run_ssh_test("SSH Archive (-a)", ["-a"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_exclude(self):
+        r = _run_ssh_test("SSH Exclude (--exclude small.txt)",
+                          ["--exclude", "small.txt"],
+                          expected_missing=["small.txt"])
+        assert r["status"] == "Success", r["error"]

tests/integration/test_tcp.py

@@ -0,0 +1,110 @@
+"""TCP transport correctness tests."""
+import os
+import shutil
+import sys
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (
+    PROJECT_ROOT, BUILD_DIR, TEST_DATA_DIR,
+    run_client, run_client_posix,
+    generate_test_files, verify_transfer, clean_dir, make_result,
+    get_dest_received_dir, CLIENT_CMD,
+)
+
+SOURCE_DIR = os.path.join(TEST_DATA_DIR, "tcp_source")
+DEST_DIR = os.path.join(TEST_DATA_DIR, "tcp_dest")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_test_data():
+    generate_test_files(SOURCE_DIR, full=False)
+    clean_dir(DEST_DIR)
+    yield
+    shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)
+
+
+def _run_tcp_test(name, port, flags, use_metadata=True, posix=False):
+    """Run a single TCP test case against a shared server."""
+    clean_dir(DEST_DIR)
+    if posix:
+        result, dur = run_client_posix(SOURCE_DIR, DEST_DIR,
+                                       flags=(["-M"] if use_metadata else []) + flags,
+                                       port=port)
+    else:
+        result, dur = run_client(SOURCE_DIR, DEST_DIR,
+                                 flags=(["-M"] if use_metadata else []) + flags,
+                                 port=port)
+
+    if result.returncode != 0:
+        return make_result(name, False, dur, f"Exit {result.returncode}: {(result.stderr or result.stdout)[:100]}")
+
+    received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+    mismatches, missing = verify_transfer(SOURCE_DIR, received)
+    if missing:
+        return make_result(name, False, dur, f"Missing: {', '.join(missing[:5])}")
+    if mismatches:
+        return make_result(name, False, dur, f"Mismatch: {', '.join(mismatches[:3])}")
+    return make_result(name, True, dur)
+
+
+class TestTCPStandard:
+    def test_standard(self, shared_server):
+        r = _run_tcp_test("Standard", shared_server.port, [])
+        assert r["status"] == "Success", r["error"]
+
+    def test_posix_args(self, shared_server):
+        r = _run_tcp_test("Posix Args", shared_server.port, [], posix=True)
+        assert r["status"] == "Success", r["error"]
+
+    def test_no_metadata(self, shared_server):
+        r = _run_tcp_test("Standard (no metadata)", shared_server.port, [], use_metadata=False)
+        assert r["status"] == "Success", r["error"]
+
+
+class TestTCPFlags:
+    def test_multithreading(self, shared_server):
+        r = _run_tcp_test("Multithreading (-m)", shared_server.port, ["-m"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_compression(self, shared_server):
+        r = _run_tcp_test("Compression (-c)", shared_server.port, ["-c"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_chunk_serialization(self, shared_server):
+        r = _run_tcp_test("Chunk Serialization (-s)", shared_server.port, ["-s"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_compression_chunk(self, shared_server):
+        r = _run_tcp_test("Compression + Chunk (-c -s)", shared_server.port, ["-c", "-s"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_multithread_compression(self, shared_server):
+        r = _run_tcp_test("Multithreading + Compression (-m -c)", shared_server.port, ["-m", "-c"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_multithread_chunk(self, shared_server):
+        r = _run_tcp_test("Multithreading + Chunk (-m -s)", shared_server.port, ["-m", "-s"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_all_flags(self, shared_server):
+        r = _run_tcp_test("Multithread + Compression + Chunk (-m -c -s)", shared_server.port, ["-m", "-c", "-s"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_sendfile(self, shared_server):
+        r = _run_tcp_test("Sendfile (-f)", shared_server.port, ["-f"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_sendfile_multithread(self, shared_server):
+        r = _run_tcp_test("Sendfile + Multithreading (-f -m)", shared_server.port, ["-f", "-m"])
+        assert r["status"] == "Success", r["error"]
+
+
+class TestTCPChunkSize:
+    def test_custom_chunk_size(self, shared_server):
+        r = _run_tcp_test("Chunk size 5MB", shared_server.port, ["--chunk-size", "5242880"])
+        assert r["status"] == "Success", r["error"]
+
+    def test_small_chunk_size(self, shared_server):
+        r = _run_tcp_test("Chunk size 1KB", shared_server.port, ["--chunk-size", "1024"])
+        assert r["status"] == "Success", r["error"]

tests/integration/test_tls.py

@@ -0,0 +1,181 @@
+"""TLS transport tests. Generates self-signed certs for testing."""
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (
+    PROJECT_ROOT, BUILD_DIR, SERVER_CMD, TEST_DATA_DIR,
+    ServerManager, run_client,
+    generate_test_files, verify_transfer, clean_dir, make_result,
+    get_dest_received_dir, _find_free_port, _wait_proc,
+)
+
+SOURCE_DIR = os.path.join(TEST_DATA_DIR, "tls_source")
+DEST_DIR = os.path.join(TEST_DATA_DIR, "tls_dest")
+CERT_DIR = os.path.join(TEST_DATA_DIR, "tls_certs")
+
+
+def _generate_certs(cert_dir):
+    """Generate a self-signed CA, server cert, and client cert for testing."""
+    os.makedirs(cert_dir, exist_ok=True)
+    ca_key = os.path.join(cert_dir, "ca.key")
+    ca_cert = os.path.join(cert_dir, "ca.pem")
+    server_key = os.path.join(cert_dir, "server.key")
+    server_cert = os.path.join(cert_dir, "server.pem")
+    client_key = os.path.join(cert_dir, "client.key")
+    client_cert = os.path.join(cert_dir, "client.pem")
+
+    # CA key + cert
+    subprocess.run([
+        "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
+        "-keyout", ca_key, "-out", ca_cert,
+        "-days", "1", "-subj", "/CN=FastSync Test CA",
+    ], check=True, capture_output=True)
+
+    # Server key + CSR + cert (signed by CA)
+    subprocess.run([
+        "openssl", "req", "-newkey", "rsa:2048", "-nodes",
+        "-keyout", server_key, "-out", os.path.join(cert_dir, "server.csr"),
+        "-subj", "/CN=localhost",
+    ], check=True, capture_output=True)
+    subprocess.run([
+        "openssl", "x509", "-req", "-in", os.path.join(cert_dir, "server.csr"),
+        "-CA", ca_cert, "-CAkey", ca_key, "-CAcreateserial",
+        "-out", server_cert, "-days", "1",
+    ], check=True, capture_output=True)
+
+    # Client key + CSR + cert (signed by CA)
+    subprocess.run([
+        "openssl", "req", "-newkey", "rsa:2048", "-nodes",
+        "-keyout", client_key, "-out", os.path.join(cert_dir, "client.csr"),
+        "-subj", "/CN=fastsync-client",
+    ], check=True, capture_output=True)
+    subprocess.run([
+        "openssl", "x509", "-req", "-in", os.path.join(cert_dir, "client.csr"),
+        "-CA", ca_cert, "-CAkey", ca_key, "-CAcreateserial",
+        "-out", client_cert, "-days", "1",
+    ], check=True, capture_output=True)
+
+    return {
+        "ca": ca_cert,
+        "server_cert": server_cert,
+        "server_key": server_key,
+        "client_cert": client_cert,
+        "client_key": client_key,
+    }
+
+
+@pytest.fixture(scope="module")
+def certs():
+    """Generate test certificates once per test module."""
+    if os.path.exists(CERT_DIR):
+        shutil.rmtree(CERT_DIR)
+    c = _generate_certs(CERT_DIR)
+    yield c
+    shutil.rmtree(CERT_DIR, ignore_errors=True)
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_test_data():
+    generate_test_files(SOURCE_DIR, full=False)
+    clean_dir(DEST_DIR)
+    yield
+    shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)
+
+
+class TestTLSBasic:
+    def test_tls_server_client(self, certs):
+        """Basic TLS: server with cert/key, client with cert/key + CA."""
+        clean_dir(DEST_DIR)
+        with ServerManager() as server:
+            server.start(extra_args=[
+                "--tls", "--cert", certs["server_cert"], "--key", certs["server_key"],
+            ])
+            result, dur = run_client(
+                SOURCE_DIR, DEST_DIR,
+                flags=["--tls",
+                       "--cert", certs["client_cert"], "--key", certs["client_key"],
+                       "--ca", certs["ca"]],
+                port=server.port,
+            )
+
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"Missing files: {missing}"
+        assert not mismatches, f"Mismatched files: {mismatches}"
+
+    def test_tls_with_compression(self, certs):
+        """TLS + compression."""
+        clean_dir(DEST_DIR)
+        with ServerManager() as server:
+            server.start(extra_args=[
+                "--tls", "--cert", certs["server_cert"], "--key", certs["server_key"],
+            ])
+            result, dur = run_client(
+                SOURCE_DIR, DEST_DIR,
+                flags=["-c", "--tls",
+                       "--cert", certs["client_cert"], "--key", certs["client_key"],
+                       "--ca", certs["ca"]],
+                port=server.port,
+            )
+
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"Missing files: {missing}"
+        assert not mismatches, f"Mismatched files: {mismatches}"
+
+    def test_tls_with_multithreading(self, certs):
+        """TLS + multithreading."""
+        clean_dir(DEST_DIR)
+        with ServerManager() as server:
+            server.start(extra_args=[
+                "--tls", "--cert", certs["server_cert"], "--key", certs["server_key"],
+            ])
+            result, dur = run_client(
+                SOURCE_DIR, DEST_DIR,
+                flags=["-m", "--tls",
+                       "--cert", certs["client_cert"], "--key", certs["client_key"],
+                       "--ca", certs["ca"]],
+                port=server.port,
+            )
+
+        if result.returncode != 0:
+            pytest.fail(f"Exit {result.returncode}: {(result.stderr or result.stdout)[:200]}")
+
+        received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"Missing files: {missing}"
+        assert not mismatches, f"Mismatched files: {mismatches}"
+
+
+class TestTLSErrorCases:
+    def test_server_tls_missing_cert_key(self):
+        """Server should fail if --tls is given without --cert/--key."""
+        proc = subprocess.Popen(
+            SERVER_CMD + ["--tls"],
+            stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
+        )
+        _, stderr = proc.communicate(timeout=5)
+        assert proc.returncode != 0, "Server should fail with --tls but no cert/key"
+
+    def test_client_tls_missing_key(self):
+        """Client should fail if --tls is given without --key."""
+        clean_dir(DEST_DIR)
+        with ServerManager() as server:
+            result, _ = run_client(
+                SOURCE_DIR, DEST_DIR,
+                flags=["--tls",
+                       "--cert", "/nonexistent/cert.pem"],
+                port=server.port,
+            )
+        assert result.returncode != 0, "Client should fail with --tls but no --key"