From b28bac9f4126bab75a2d662d5090f976d1cbf93e Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 18 Jul 2026 19:19:33 +0200 Subject: [PATCH 1/7] refactor: split test.py into modular pytest integration tests + benchmark tool - tests/integration/common.py: ServerManager (reuses server across tests), run_client, test data generation, verification utilities - tests/integration/test_tcp.py: 14 TCP transport correctness tests - tests/integration/test_ssh.py: 10 SSH transport tests (skip when unavailable) - tests/integration/test_tls.py: 5 TLS encryption tests (new coverage!) - tests/integration/test_features.py: 13 feature tests (incremental, delete, exclude, include, max/min size, bwlimit, dry run, archive, progress) - tests/integration/test_preflight.py: 7 CLI validation/error tests - benchmark/bench.py: standalone benchmark with JSON output, p50/p95, multi-run - Updated Dockerfile with python3-pytest, openssl, openssh-client - Updated CI to use pytest (gitea.tap-tap.win/taptap/fastsync-ci:v6) - Removed old monolithic test.py --- .gitea/workflows/ci.yaml | 4 +- Dockerfile | 4 +- benchmark/bench.py | 284 +++++++++++ test.py | 740 ---------------------------- tests/conftest.py | 5 + tests/integration/__init__.py | 0 tests/integration/common.py | 181 +++++++ tests/integration/test_features.py | 299 +++++++++++ tests/integration/test_preflight.py | 81 +++ tests/integration/test_ssh.py | 132 +++++ tests/integration/test_tcp.py | 111 +++++ tests/integration/test_tls.py | 181 +++++++ 12 files changed, 1279 insertions(+), 743 deletions(-) create mode 100644 benchmark/bench.py delete mode 100755 test.py create mode 100644 tests/conftest.py create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/common.py create mode 100644 tests/integration/test_features.py create mode 100644 tests/integration/test_preflight.py create mode 100644 tests/integration/test_ssh.py create mode 100644 tests/integration/test_tcp.py create mode 100644 tests/integration/test_tls.py diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 92e579b..937c6dd 100644 --- a/.gitea/workflows/ci.yaml +++ b/.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 diff --git a/Dockerfile b/Dockerfile index 81bd44b..40deaf2 100644 --- a/Dockerfile +++ b/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/* diff --git a/benchmark/bench.py b/benchmark/bench.py new file mode 100644 index 0000000..c89fbd1 --- /dev/null +++ b/benchmark/bench.py @@ -0,0 +1,284 @@ +#!/usr/bin/env python3 +"""Standalone benchmark tool for FastSync. + +Usage: + python3 benchmark/bench.py # quick benchmark (unlimited, 3 runs) + python3 benchmark/bench.py --runs 5 --profiles lan wan # thorough + python3 benchmark/bench.py --output json # machine-readable + python3 benchmark/bench.py --configs "-c" "-m" "-m -c" # custom configs +""" +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, + }, +} + +DEFAULT_CONFIGS = [ + {"name": "standard", "flags": []}, + {"name": "-c", "flags": ["-c"]}, + {"name": "-m", "flags": ["-m"]}, + {"name": "-m -c", "flags": ["-m", "-c"]}, + {"name": "-s", "flags": ["-s"]}, + {"name": "-m -c -s", "flags": ["-m", "-c", "-s"]}, +] + + +def generate_bench_data(source_dir, size_mb=25): + """Generate test data for benchmarking.""" + if os.path.exists(source_dir): + shutil.rmtree(source_dir) + os.makedirs(source_dir) + + target = size_mb * 1024 * 1024 + written = 0 + + # Structured files + 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) + + # Fill remaining with random data + 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(profile_name): + params = NETWORK_PROFILES.get(profile_name, {}) + if not params: + netem_reset() + return + netem_reset() + cmd = ["sudo", "tc", "qdisc", "add", "dev", "lo", "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("sudo tc qdisc del dev lo root".split(), capture_output=True) + + +def run_transfer(source_dir, dest_dir, flags, port): + """Run a single transfer. Returns duration in seconds or None on failure.""" + 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_benchmark(source_dir, dest_dir, configs, runs, profile_name): + """Run benchmark for all configs, returns list of results.""" + is_limited = profile_name != "unlimited" + if is_limited: + netem_apply(profile_name) + + try: + results = [] + for config in configs: + times = [] + for run_idx in range(runs): + # Clean dest for each run + if os.path.exists(dest_dir): + shutil.rmtree(dest_dir) + os.makedirs(dest_dir, exist_ok=True) + + # Start fresh server + port = find_free_port() + server = subprocess.Popen( + SERVER_CMD + ["-p", str(port)], + stdout=subprocess.DEVNULL, stderr=None, + ) + try: + wait_for_port(port) + t = run_transfer(source_dir, dest_dir, config["flags"], port) + if t is not None: + times.append(t) + finally: + wait_proc(server) + + entry = { + "config": config["name"], + "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 is_limited: + netem_reset() + + +def print_table(results, total_bytes): + """Print results as a human-readable table.""" + # Group 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{'=' * 80}") + 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") + print(f"{'=' * 80}") + 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(entries, key=lambda x: x.get("p50", 999)): + 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}") + + if params.get("rate_bps"): + best = min((e["p50"] for e in entries if "p50" in e), default=None) + if best: + theoretical = total_bytes / params["rate_bps"] + print(f"\n Best config p50: {best:.4f}s") + print(f" Theoretical max: {theoretical:.4f}s (uncompressed at line rate)") + print(f" Speedup vs max: {theoretical / best:.2f}x") + + +def main(): + parser = argparse.ArgumentParser(description="FastSync benchmark tool") + parser.add_argument("--runs", type=int, default=3, help="Number of runs per config (default: 3)") + parser.add_argument("--profiles", nargs="+", default=["unlimited"], + choices=list(NETWORK_PROFILES.keys()), + help="Network profiles to test") + parser.add_argument("--configs", nargs="+", default=None, + help="Custom config flags (e.g. --configs '-c' '-m' '-m -c')") + parser.add_argument("--size-mb", type=int, default=25, help="Test data size in MB (default: 25)") + 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) + + # 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) + print(f"Generated {total_bytes / (1024*1024):.1f} MB test data") + + # Parse configs + if args.configs: + configs = [{"name": c, "flags": c.split()} for c in args.configs] + else: + configs = DEFAULT_CONFIGS + + # Run benchmarks + all_results = [] + try: + for profile in args.profiles: + results = run_benchmark(source_dir, dest_dir, configs, args.runs, profile) + 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) + print() + + +if __name__ == "__main__": + main() diff --git a/test.py b/test.py deleted file mode 100755 index 9d533a5..0000000 --- a/test.py +++ /dev/null @@ -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() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..7843584 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,5 @@ +"""Shared pytest configuration for integration tests.""" +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "integration")) diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/common.py b/tests/integration/common.py new file mode 100644 index 0000000..6d306dc --- /dev/null +++ b/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=None) + _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() diff --git a/tests/integration/test_features.py b/tests/integration/test_features.py new file mode 100644 index 0000000..e97123b --- /dev/null +++ b/tests/integration/test_features.py @@ -0,0 +1,299 @@ +"""Feature tests: incremental sync, bandwidth limiting, dry run, metadata, filters.""" +import os +import shutil +import subprocess +import sys +import time +import pytest + +sys.path.insert(0, os.path.dirname(__file__)) +from common import ( + PROJECT_ROOT, BUILD_DIR, TEST_DATA_DIR, + ServerManager, 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): + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-a"], + 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: {missing}" + assert not mismatches, f"Mismatch: {mismatches}" + + +class TestExclude: + def test_exclude_single(self): + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--exclude", "small.txt"], + 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) + # small.txt should be missing (excluded) + assert "small.txt" in missing, f"small.txt should be excluded but was transferred" + # all other files should be present + 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): + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--exclude", "*.txt"], + 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) + # Only binary.bin and bulk files should be present + 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): + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--include", "binary.bin"], + 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) + # Only binary.bin should be present + 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): + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--include", "*.bin"], + 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) + assert os.path.exists(os.path.join(received, "binary.bin")), "binary.bin should be included" + + +class TestSizeFilters: + def test_max_size(self): + """Files larger than --max-size should be skipped.""" + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--max-size", "100"], # 100 bytes + 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) + # small.txt (12 bytes) should be present, medium.txt (220k+) should be skipped + 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): + """Files smaller than --min-size should be skipped.""" + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--min-size", "1000"], # 1 KB + 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) + # small.txt (12 bytes) should be skipped, medium.txt should be present + 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): + """Second sync with --incremental should be fast (skips unchanged files).""" + # First sync: populate dest + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M"], + port=server.port, + ) + assert result.returncode == 0, f"First sync failed: {result.stderr[:100]}" + + # Second sync with --incremental (should be near-instant) + with ServerManager() as server: + start = time.monotonic() + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M", "--incremental"], + port=server.port, + ) + incremental_time = time.monotonic() - start + + assert result.returncode == 0, f"Incremental sync failed: {(result.stderr or result.stdout)[:200]}" + + # Verify files are still correct + 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): + """Incremental sync should transfer modified files.""" + # First sync + clean_dir(DEST_DIR) + with ServerManager() as server: + result, _ = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M"], + port=server.port, + ) + assert result.returncode == 0 + + # Modify a file + 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") + + # Second sync with --incremental + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M", "--incremental"], + port=server.port, + ) + assert result.returncode == 0 + + # Restore original content + with open(modified_file, "wb") as f: + f.write(b"hello world\n") + + # Verify the modified content was transferred + 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): + """--delete should remove files on dest that aren't in source.""" + # First sync: populate dest + clean_dir(DEST_DIR) + with ServerManager() as server: + result, _ = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M"], + port=server.port, + ) + assert result.returncode == 0 + + # Add extra files to destination + 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") + + # Second sync with --delete + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M", "--delete"], + port=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" + + # Verify remaining files are correct + 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): + """--progress should produce some output.""" + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--progress"], + port=server.port, + ) + assert result.returncode == 0, f"Exit {result.returncode}: {result.stderr[:100]}" + # Progress output goes to stderr or stdout + output = result.stdout + result.stderr + # Just verify it ran successfully; progress output format may vary + assert len(output) >= 0 # No assertion on specific output format + + +class TestBandwidthLimit: + def test_bwlimit_runs(self): + """--bwlimit should run without error.""" + clean_dir(DEST_DIR) + with ServerManager() as server: + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["--bwlimit", "10240"], + port=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}" diff --git a/tests/integration/test_preflight.py b/tests/integration/test_preflight.py new file mode 100644 index 0000000..6acf868 --- /dev/null +++ b/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) diff --git a/tests/integration/test_ssh.py b/tests/integration/test_ssh.py new file mode 100644 index 0000000..82b86ef --- /dev/null +++ b/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"] diff --git a/tests/integration/test_tcp.py b/tests/integration/test_tcp.py new file mode 100644 index 0000000..04f64a4 --- /dev/null +++ b/tests/integration/test_tcp.py @@ -0,0 +1,111 @@ +"""TCP transport correctness tests.""" +import os +import sys +import pytest + +sys.path.insert(0, os.path.dirname(__file__)) +from common import ( + PROJECT_ROOT, BUILD_DIR, TEST_DATA_DIR, + ServerManager, 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 + import shutil + shutil.rmtree(TEST_DATA_DIR, ignore_errors=True) + + +def _run_tcp_test(name, flags, use_metadata=True, posix=False): + """Run a single TCP test case with a fresh server.""" + clean_dir(DEST_DIR) + with ServerManager() as server: + if posix: + result, dur = run_client_posix(SOURCE_DIR, DEST_DIR, + flags=(["-M"] if use_metadata else []) + flags, + port=server.port) + else: + result, dur = run_client(SOURCE_DIR, DEST_DIR, + flags=(["-M"] if use_metadata else []) + flags, + port=server.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): + r = _run_tcp_test("Standard", []) + assert r["status"] == "Success", r["error"] + + def test_posix_args(self): + r = _run_tcp_test("Posix Args", [], posix=True) + assert r["status"] == "Success", r["error"] + + def test_no_metadata(self): + r = _run_tcp_test("Standard (no metadata)", [], use_metadata=False) + assert r["status"] == "Success", r["error"] + + +class TestTCPFlags: + def test_multithreading(self): + r = _run_tcp_test("Multithreading (-m)", ["-m"]) + assert r["status"] == "Success", r["error"] + + def test_compression(self): + r = _run_tcp_test("Compression (-c)", ["-c"]) + assert r["status"] == "Success", r["error"] + + def test_chunk_serialization(self): + r = _run_tcp_test("Chunk Serialization (-s)", ["-s"]) + assert r["status"] == "Success", r["error"] + + def test_compression_chunk(self): + r = _run_tcp_test("Compression + Chunk (-c -s)", ["-c", "-s"]) + assert r["status"] == "Success", r["error"] + + def test_multithread_compression(self): + r = _run_tcp_test("Multithreading + Compression (-m -c)", ["-m", "-c"]) + assert r["status"] == "Success", r["error"] + + def test_multithread_chunk(self): + r = _run_tcp_test("Multithreading + Chunk (-m -s)", ["-m", "-s"]) + assert r["status"] == "Success", r["error"] + + def test_all_flags(self): + r = _run_tcp_test("Multithread + Compression + Chunk (-m -c -s)", ["-m", "-c", "-s"]) + assert r["status"] == "Success", r["error"] + + def test_sendfile(self): + r = _run_tcp_test("Sendfile (-f)", ["-f"]) + assert r["status"] == "Success", r["error"] + + def test_sendfile_multithread(self): + r = _run_tcp_test("Sendfile + Multithreading (-f -m)", ["-f", "-m"]) + assert r["status"] == "Success", r["error"] + + +class TestTCPChunkSize: + def test_custom_chunk_size(self): + r = _run_tcp_test("Chunk size 5MB", ["--chunk-size", "5242880"]) + assert r["status"] == "Success", r["error"] + + def test_small_chunk_size(self): + r = _run_tcp_test("Chunk size 1KB", ["--chunk-size", "1024"]) + assert r["status"] == "Success", r["error"] diff --git a/tests/integration/test_tls.py b/tests/integration/test_tls.py new file mode 100644 index 0000000..25d31cd --- /dev/null +++ b/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" -- 2.52.0 From 51f20d1c914b0bdaaffcfb79154aa0f56f471eb7 Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 18 Jul 2026 19:32:48 +0200 Subject: [PATCH 2/7] benchmark: suppress server probe noise in output --- benchmark/bench.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/benchmark/bench.py b/benchmark/bench.py index c89fbd1..b139cb6 100644 --- a/benchmark/bench.py +++ b/benchmark/bench.py @@ -165,7 +165,7 @@ def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): port = find_free_port() server = subprocess.Popen( SERVER_CMD + ["-p", str(port)], - stdout=subprocess.DEVNULL, stderr=None, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) try: wait_for_port(port) -- 2.52.0 From 34b46a92dbe112c6f7d722d46e5ffe23a9bdcf88 Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 18 Jul 2026 19:48:19 +0200 Subject: [PATCH 3/7] tests: share server across tests, suppress probe noise - Add session-scoped shared_server fixture to avoid 27+ server start/stop cycles - Refactor test_tcp.py and test_features.py to use shared server - Suppress server health-check probe stderr noise in common.py --- tests/conftest.py | 12 ++ tests/integration/common.py | 2 +- tests/integration/test_features.py | 235 ++++++++++++----------------- tests/integration/test_tcp.py | 81 +++++----- 4 files changed, 149 insertions(+), 181 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 7843584..51d2ded 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +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() diff --git a/tests/integration/common.py b/tests/integration/common.py index 6d306dc..08cc9c9 100644 --- a/tests/integration/common.py +++ b/tests/integration/common.py @@ -28,7 +28,7 @@ class ServerManager: cmd = SERVER_CMD + ["-p", str(self._port)] if extra_args: cmd += extra_args - self._proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=None) + self._proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) _wait_for_port(self._port, timeout=5) def stop(self): diff --git a/tests/integration/test_features.py b/tests/integration/test_features.py index e97123b..63a5bf6 100644 --- a/tests/integration/test_features.py +++ b/tests/integration/test_features.py @@ -1,7 +1,6 @@ """Feature tests: incremental sync, bandwidth limiting, dry run, metadata, filters.""" import os import shutil -import subprocess import sys import time import pytest @@ -9,7 +8,7 @@ import pytest sys.path.insert(0, os.path.dirname(__file__)) from common import ( PROJECT_ROOT, BUILD_DIR, TEST_DATA_DIR, - ServerManager, run_client, + run_client, generate_test_files, verify_transfer, clean_dir, make_result, get_dest_received_dir, CLIENT_CMD, ) @@ -38,14 +37,13 @@ class TestDryRun: class TestArchiveMode: - def test_archive_mode(self): + def test_archive_mode(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["-a"], - port=server.port, - ) + 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) @@ -55,65 +53,57 @@ class TestArchiveMode: class TestExclude: - def test_exclude_single(self): + def test_exclude_single(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--exclude", "small.txt"], - port=server.port, - ) + 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) - # small.txt should be missing (excluded) - assert "small.txt" in missing, f"small.txt should be excluded but was transferred" - # all other files should be present + 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): + def test_exclude_glob(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--exclude", "*.txt"], - port=server.port, - ) + 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) - # Only binary.bin and bulk files should be present 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): + def test_include_single(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--include", "binary.bin"], - port=server.port, - ) + 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) - # Only binary.bin should be present 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): + def test_include_glob(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--include", "*.bin"], - port=server.port, - ) + 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) @@ -121,101 +111,81 @@ class TestInclude: class TestSizeFilters: - def test_max_size(self): - """Files larger than --max-size should be skipped.""" + def test_max_size(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--max-size", "100"], # 100 bytes - port=server.port, - ) + 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) - # small.txt (12 bytes) should be present, medium.txt (220k+) should be skipped 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): - """Files smaller than --min-size should be skipped.""" + def test_min_size(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--min-size", "1000"], # 1 KB - port=server.port, - ) + 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) - # small.txt (12 bytes) should be skipped, medium.txt should be present 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): - """Second sync with --incremental should be fast (skips unchanged files).""" - # First sync: populate dest + def test_incremental_skips_unchanged(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["-M"], - port=server.port, - ) - assert result.returncode == 0, f"First sync failed: {result.stderr[:100]}" + 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]}" - # Second sync with --incremental (should be near-instant) - with ServerManager() as server: - start = time.monotonic() - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["-M", "--incremental"], - port=server.port, - ) - incremental_time = time.monotonic() - start + 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]}" - # Verify files are still correct 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): - """Incremental sync should transfer modified files.""" - # First sync + def test_incremental_detects_changes(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, _ = run_client( - SOURCE_DIR, DEST_DIR, - flags=["-M"], - port=server.port, - ) - assert result.returncode == 0 + result, _ = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M"], + port=shared_server.port, + ) + assert result.returncode == 0 - # Modify a file 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") - # Second sync with --incremental - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["-M", "--incremental"], - port=server.port, - ) - assert result.returncode == 0 + result, dur = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M", "--incremental"], + port=shared_server.port, + ) + assert result.returncode == 0 - # Restore original content with open(modified_file, "wb") as f: f.write(b"hello world\n") - # Verify the modified content was transferred 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" @@ -225,19 +195,15 @@ class TestIncremental: class TestDelete: - def test_delete_removes_extra_files(self): - """--delete should remove files on dest that aren't in source.""" - # First sync: populate dest + def test_delete_removes_extra_files(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, _ = run_client( - SOURCE_DIR, DEST_DIR, - flags=["-M"], - port=server.port, - ) - assert result.returncode == 0 + result, _ = run_client( + SOURCE_DIR, DEST_DIR, + flags=["-M"], + port=shared_server.port, + ) + assert result.returncode == 0 - # Add extra files to destination 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") @@ -247,51 +213,42 @@ class TestDelete: with open(os.path.join(extra_dir, "nested.txt"), "w") as f: f.write("nested extra") - # Second sync with --delete - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["-M", "--delete"], - port=server.port, - ) + 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" - # Verify remaining files are correct 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): - """--progress should produce some output.""" + def test_progress_output(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--progress"], - port=server.port, - ) + 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]}" - # Progress output goes to stderr or stdout output = result.stdout + result.stderr - # Just verify it ran successfully; progress output format may vary - assert len(output) >= 0 # No assertion on specific output format + assert len(output) >= 0 class TestBandwidthLimit: - def test_bwlimit_runs(self): - """--bwlimit should run without error.""" + def test_bwlimit_runs(self, shared_server): clean_dir(DEST_DIR) - with ServerManager() as server: - result, dur = run_client( - SOURCE_DIR, DEST_DIR, - flags=["--bwlimit", "10240"], - port=server.port, - ) + 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) diff --git a/tests/integration/test_tcp.py b/tests/integration/test_tcp.py index 04f64a4..18cc07c 100644 --- a/tests/integration/test_tcp.py +++ b/tests/integration/test_tcp.py @@ -1,12 +1,13 @@ """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, - ServerManager, run_client, run_client_posix, + run_client, run_client_posix, generate_test_files, verify_transfer, clean_dir, make_result, get_dest_received_dir, CLIENT_CMD, ) @@ -20,22 +21,20 @@ def setup_test_data(): generate_test_files(SOURCE_DIR, full=False) clean_dir(DEST_DIR) yield - import shutil shutil.rmtree(TEST_DATA_DIR, ignore_errors=True) -def _run_tcp_test(name, flags, use_metadata=True, posix=False): - """Run a single TCP test case with a fresh server.""" +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) - with ServerManager() as server: - if posix: - result, dur = run_client_posix(SOURCE_DIR, DEST_DIR, - flags=(["-M"] if use_metadata else []) + flags, - port=server.port) - else: - result, dur = run_client(SOURCE_DIR, DEST_DIR, - flags=(["-M"] if use_metadata else []) + flags, - port=server.port) + 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]}") @@ -50,62 +49,62 @@ def _run_tcp_test(name, flags, use_metadata=True, posix=False): class TestTCPStandard: - def test_standard(self): - r = _run_tcp_test("Standard", []) + 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): - r = _run_tcp_test("Posix Args", [], posix=True) + 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): - r = _run_tcp_test("Standard (no metadata)", [], use_metadata=False) + 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): - r = _run_tcp_test("Multithreading (-m)", ["-m"]) + 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): - r = _run_tcp_test("Compression (-c)", ["-c"]) + 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): - r = _run_tcp_test("Chunk Serialization (-s)", ["-s"]) + 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): - r = _run_tcp_test("Compression + Chunk (-c -s)", ["-c", "-s"]) + 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): - r = _run_tcp_test("Multithreading + Compression (-m -c)", ["-m", "-c"]) + 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): - r = _run_tcp_test("Multithreading + Chunk (-m -s)", ["-m", "-s"]) + 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): - r = _run_tcp_test("Multithread + Compression + Chunk (-m -c -s)", ["-m", "-c", "-s"]) + 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): - r = _run_tcp_test("Sendfile (-f)", ["-f"]) + 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): - r = _run_tcp_test("Sendfile + Multithreading (-f -m)", ["-f", "-m"]) + 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): - r = _run_tcp_test("Chunk size 5MB", ["--chunk-size", "5242880"]) + 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): - r = _run_tcp_test("Chunk size 1KB", ["--chunk-size", "1024"]) + 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"] -- 2.52.0 From f0c7bb791a20aeeafe7a2893a18981877f368bb0 Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 18 Jul 2026 19:55:43 +0200 Subject: [PATCH 4/7] benchmark: add rsync comparison, configurable data mix, custom network limits - Add rsync and rsync+zstd as baseline comparisons - --random-ratio controls fraction of incompressible data (default 0.75) - --delay, --jitter, --throughput, --loss for custom network simulation - --no-rsync to skip rsync comparison - Grouped output: FastSync vs rsync with speedup calculations --- benchmark/bench.py | 284 ++++++++++++++++++++++++++++++++------------- 1 file changed, 206 insertions(+), 78 deletions(-) diff --git a/benchmark/bench.py b/benchmark/bench.py index b139cb6..250051c 100644 --- a/benchmark/bench.py +++ b/benchmark/bench.py @@ -1,11 +1,16 @@ #!/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 # quick benchmark (unlimited, 3 runs) - python3 benchmark/bench.py --runs 5 --profiles lan wan # thorough - python3 benchmark/bench.py --output json # machine-readable - python3 benchmark/bench.py --configs "-c" "-m" "-m -c" # custom configs + 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 @@ -16,7 +21,6 @@ import socket import statistics import subprocess import sys -import tempfile import time PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) @@ -37,41 +41,48 @@ NETWORK_PROFILES = { }, } -DEFAULT_CONFIGS = [ - {"name": "standard", "flags": []}, - {"name": "-c", "flags": ["-c"]}, - {"name": "-m", "flags": ["-m"]}, - {"name": "-m -c", "flags": ["-m", "-c"]}, - {"name": "-s", "flags": ["-s"]}, - {"name": "-m -c -s", "flags": ["-m", "-c", "-s"]}, +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"}, +] -def generate_bench_data(source_dir, size_mb=25): - """Generate test data for benchmarking.""" +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, +} + + +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 - # Structured files - 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(): + 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) - # Fill remaining with random data os.makedirs(os.path.join(source_dir, "bulk"), exist_ok=True) i = 0 while written < target: @@ -109,25 +120,39 @@ def wait_proc(proc, timeout=5): proc.wait() -def netem_apply(profile_name): +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_reset() - cmd = ["sudo", "tc", "qdisc", "add", "dev", "lo", "root", "netem"] - cmd += ["rate", params["rate"]] - cmd += ["delay", params["delay"], params["jitter"]] - cmd += ["loss", params["loss"]] - subprocess.run(cmd, check=True, capture_output=True) + 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_transfer(source_dir, dest_dir, flags, port): - """Run a single transfer. Returns duration in seconds or None on failure.""" +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, @@ -145,38 +170,64 @@ def run_transfer(source_dir, dest_dir, flags, port): return None +def run_rsync(source_dir, dest_dir, flags): + """Run rsync. Returns duration or None.""" + src = source_dir.rstrip("/") + "/" + 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): + """Route to the right tool. Returns duration or None.""" + if config["tool"] == "rsync": + return run_rsync(source_dir, dest_dir, config["flags"]) + else: + return run_fastsync(source_dir, dest_dir, config["flags"], port) + + def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): """Run benchmark for all configs, returns list of results.""" is_limited = profile_name != "unlimited" if is_limited: - netem_apply(profile_name) + netem_apply_profile(profile_name) try: results = [] for config in configs: times = [] for run_idx in range(runs): - # Clean dest for each run if os.path.exists(dest_dir): shutil.rmtree(dest_dir) os.makedirs(dest_dir, exist_ok=True) - # Start fresh server port = find_free_port() - server = subprocess.Popen( - SERVER_CMD + ["-p", str(port)], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - ) + server = None try: - wait_for_port(port) - t = run_transfer(source_dir, dest_dir, config["flags"], port) + 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) if t is not None: times.append(t) finally: - wait_proc(server) + if server: + wait_proc(server) entry = { "config": config["name"], + "tool": config["tool"], "profile": profile_name, "runs": len(times), "times": [round(t, 4) for t in times], @@ -194,53 +245,105 @@ def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): netem_reset() -def print_table(results, total_bytes): - """Print results as a human-readable table.""" - # Group by profile +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{'=' * 80}") + 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") - print(f"{'=' * 80}") - 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(entries, key=lambda x: x.get("p50", 999)): - 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}") + print(f" Data: {total_bytes / (1024*1024):.1f} MB ({random_ratio*100:.0f}% random, {(1-random_ratio)*100:.0f}% compressible)") + print(f"{'=' * 85}") - if params.get("rate_bps"): - best = min((e["p50"] for e in entries if "p50" in e), default=None) - if best: - theoretical = total_bytes / params["rate_bps"] - print(f"\n Best config p50: {best:.4f}s") - print(f" Theoretical max: {theoretical:.4f}s (uncompressed at line rate)") - print(f" Speedup vs max: {theoretical / best:.2f}x") + 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") - parser.add_argument("--runs", type=int, default=3, help="Number of runs per config (default: 3)") - parser.add_argument("--profiles", nargs="+", default=["unlimited"], + 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="Network profiles to test") + help="Predefined network profiles (default: unlimited)") parser.add_argument("--configs", nargs="+", default=None, - help="Custom config flags (e.g. --configs '-c' '-m' '-m -c')") - parser.add_argument("--size-mb", type=int, default=25, help="Test data size in MB (default: 25)") + 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("--output", choices=["table", "json"], default="table", help="Output format") - parser.add_argument("--keep-data", action="store_true", help="Don't clean up test data") + parser.add_argument("--keep-data", action="store_true", + help="Don't clean up test data") args = parser.parse_args() # Build @@ -250,22 +353,47 @@ def main(): 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) - print(f"Generated {total_bytes / (1024*1024):.1f} MB test data") + 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)") - # Parse configs + # Build config list if args.configs: - configs = [{"name": c, "flags": c.split()} for c in args.configs] + fastsync_configs = [{"name": c, "flags": c.split(), "tool": "fastsync"} for c in args.configs] else: - configs = DEFAULT_CONFIGS + fastsync_configs = list(FASTSYNC_CONFIGS) + + configs = list(fastsync_configs) + if not args.no_rsync: + configs += RSYNC_CONFIGS # Run benchmarks all_results = [] try: - for profile in args.profiles: + for profile in profiles_to_run: results = run_benchmark(source_dir, dest_dir, configs, args.runs, profile) all_results.extend(results) finally: @@ -276,7 +404,7 @@ def main(): if args.output == "json": print(json.dumps(all_results, indent=2)) else: - print_table(all_results, total_bytes) + print_table(all_results, total_bytes, args.random_ratio) print() -- 2.52.0 From 4ed514ad1bfb96b91eb637aac183b3104db27174 Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 18 Jul 2026 20:08:54 +0200 Subject: [PATCH 5/7] benchmark: fix rsync network throttling via rsync daemon rsync local-to-local copies bypass the network stack entirely, making tc/netem ineffective. Start an rsync daemon on TCP so both tools see the same network conditions. --- benchmark/bench.py | 72 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 68 insertions(+), 4 deletions(-) diff --git a/benchmark/bench.py b/benchmark/bench.py index 250051c..de66bfc 100644 --- a/benchmark/bench.py +++ b/benchmark/bench.py @@ -21,6 +21,7 @@ import socket import statistics import subprocess import sys +import tempfile import time PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) @@ -55,6 +56,59 @@ RSYNC_CONFIGS = [ {"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, @@ -170,9 +224,11 @@ def run_fastsync(source_dir, dest_dir, flags, port): return None -def run_rsync(source_dir, dest_dir, flags): +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() @@ -185,10 +241,10 @@ def run_rsync(source_dir, dest_dir, flags): return None -def run_transfer(config, source_dir, dest_dir, port=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"]) + return run_rsync(source_dir, dest_dir, config["flags"], rsync_daemon) else: return run_fastsync(source_dir, dest_dir, config["flags"], port) @@ -196,10 +252,16 @@ def run_transfer(config, source_dir, dest_dir, port=None): def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): """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 = [] @@ -218,7 +280,7 @@ def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): ) wait_for_port(port) - t = run_transfer(config, source_dir, dest_dir, port) + t = run_transfer(config, source_dir, dest_dir, port, rsync_daemon) if t is not None: times.append(t) finally: @@ -241,6 +303,8 @@ def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): results.append(entry) return results finally: + if rsync_daemon: + rsync_daemon.stop() if is_limited: netem_reset() -- 2.52.0 From 784c57bd13c8ccfd7013208a84f597608ad86975 Mon Sep 17 00:00:00 2001 From: TapTap Date: Sun, 19 Jul 2026 15:03:23 +0200 Subject: [PATCH 6/7] shell: add pytest to nix-shell environment --- shell.nix | 1 + 1 file changed, 1 insertion(+) diff --git a/shell.nix b/shell.nix index b7f96c4..16a2420 100644 --- a/shell.nix +++ b/shell.nix @@ -15,6 +15,7 @@ pkgs.mkShell { buildInputs = with pkgs; [ zstd openssl + (python3.withPackages (ps: with ps; [ pytest ])) ]; NIX_ENFORCE_PURITY = 0; -- 2.52.0 From 045965f74c0be1d3193923072db1682b86c862de Mon Sep 17 00:00:00 2001 From: TapTap Date: Sun, 19 Jul 2026 15:07:53 +0200 Subject: [PATCH 7/7] benchmark: add --progress flag with progress bar and ETA --- benchmark/bench.py | 47 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 45 insertions(+), 2 deletions(-) diff --git a/benchmark/bench.py b/benchmark/bench.py index de66bfc..abb1db8 100644 --- a/benchmark/bench.py +++ b/benchmark/bench.py @@ -118,6 +118,39 @@ STRUCTURED_FILES = { } +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): @@ -249,7 +282,7 @@ def run_transfer(config, source_dir, dest_dir, port=None, rsync_daemon=None): return run_fastsync(source_dir, dest_dir, config["flags"], port) -def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): +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) @@ -287,6 +320,9 @@ def run_benchmark(source_dir, dest_dir, configs, runs, profile_name): if server: wait_proc(server) + if progress: + progress.tick(f"{config['name']} (run {run_idx+1}/{runs})") + entry = { "config": config["name"], "tool": config["tool"], @@ -404,6 +440,8 @@ Examples: 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", @@ -455,10 +493,15 @@ Examples: 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) + results = run_benchmark(source_dir, dest_dir, configs, args.runs, profile, progress) all_results.extend(results) finally: if not args.keep_data: -- 2.52.0