TapTap/FastSync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'test: auto-generate test files, verify content, CLI args, test_data dir' (#7) from improve-test-py into main

Browse Source 
Reviewed-on: #7
This commit is contained in:
TapTap
2026-07-04 22:56:58 +02:00
parent 42870ec719 7cacce24e3
commit 1c5099b963
5 changed files with 274 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -1,2 +1,4 @@
build
data_copied
test_data/
__pycache__/
src/client/client.c
+8
View File
@@ -207,6 +207,14 @@ int main(int argc, char *argv[]) {
config->compression_level);
}
}
} else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) {
free(config->send_directory);
config->send_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) {
free(config->receive_root_directory);
config->receive_root_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--save-to-disk") == 0) {
config->save_to_disk = true;
} else {
handle_arg(argv[i], "-m", &config->use_multithreading,
"Enabled Multithreading");
src/client/scanner.c
+63 -28
View File
@@ -7,10 +7,14 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
DirectoryScanner *directory_scanner_create(char *root_directory) {
DirectoryScanner *scanner = malloc(sizeof(DirectoryScanner));
scanner->directories = queue_create(100, free);
scanner->current_dir = NULL;
scanner->current_path = NULL;
queue_enqueue(scanner->directories, str_dup(root_directory));
return scanner;
}
@@ -18,11 +22,16 @@ DirectoryScanner *directory_scanner_create(char *root_directory) {
void directory_scanner_destroy(DirectoryScanner *scanner) {
if (scanner == NULL)
return;
if (scanner->current_dir) {
closedir(scanner->current_dir);
scanner->current_dir = NULL;
}
free(scanner->current_path);
queue_destroy(scanner->directories);
free(scanner);
}
Chunk *chunk_data_to_chunk(ArrayList *chunk_data) {
static Chunk *chunk_data_to_chunk(ArrayList *chunk_data) {
void **chunk_items = array_list_to_array(chunk_data);
Chunk *chunk = chunk_create((File **)chunk_items, chunk_data->size);
free(chunk_items);
@@ -31,40 +40,66 @@ Chunk *chunk_data_to_chunk(ArrayList *chunk_data) {
return chunk;
}
static int open_next_directory(DirectoryScanner *scanner) {
if (scanner->current_dir) {
closedir(scanner->current_dir);
scanner->current_dir = NULL;
}
free(scanner->current_path);
if (queue_is_empty(scanner->directories))
return 0;
scanner->current_path = (char *)queue_dequeue(scanner->directories);
scanner->current_dir = opendir(scanner->current_path);
if (scanner->current_dir == NULL) {
perror("Could not open directory!");
exit(EXIT_FAILURE);
}
return 1;
}
Chunk *directory_scanner_next(DirectoryScanner *scanner) {
ArrayList *chunk_data = array_list_create(file_destroy);
unsigned long long chunk_data_size = 0;
while (!queue_is_empty(scanner->directories)) {
char *path = (char *)queue_dequeue(scanner->directories);
DIR *dir;
struct dirent *entry;
dir = opendir(path);
if (dir == NULL) {
perror("Could not open directory!");
exit(EXIT_FAILURE);
while (1) {
if (scanner->current_dir == NULL) {
if (!open_next_directory(scanner))
break;
}
while ((entry = readdir(dir)) != NULL) {
if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) {
continue;
}
char *cur_path = path_cat(path, entry->d_name);
struct stat stats;
stat(cur_path, &stats);
if (!S_ISREG(stats.st_mode))
queue_enqueue(scanner->directories, (void *)cur_path);
else {
File *file = file_create(cur_path, &stats);
array_list_add(chunk_data, file);
chunk_data_size += file->stats.st_size;
if (chunk_data_size > DESIRED_CHUNK_SIZE)
return chunk_data_to_chunk(chunk_data);
free(cur_path);
}
struct dirent *entry = readdir(scanner->current_dir);
if (entry == NULL) {
closedir(scanner->current_dir);
scanner->current_dir = NULL;
free(scanner->current_path);
scanner->current_path = NULL;
continue;
}
if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
continue;
char *cur_path = path_cat(scanner->current_path, entry->d_name);
struct stat stats;
if (stat(cur_path, &stats) != 0) {
free(cur_path);
continue;
}
if (!S_ISREG(stats.st_mode)) {
queue_enqueue(scanner->directories, (void *)cur_path);
} else {
File *file = file_create(cur_path, &stats);
array_list_add(chunk_data, file);
chunk_data_size += file->stats.st_size;
if (chunk_data_size > DESIRED_CHUNK_SIZE)
return chunk_data_to_chunk(chunk_data);
free(cur_path);
}
closedir(dir);
free(path);
}
if (chunk_data->size > 0)
return chunk_data_to_chunk(chunk_data);
return NULL;
src/client/scanner.h
+4
View File
@@ -3,8 +3,12 @@
#include "chunk.h"
#include "queue.h"
#include <dirent.h>
typedef struct {
Queue *directories;
DIR *current_dir;
char *current_path;
} DirectoryScanner;
DirectoryScanner *directory_scanner_create(char *root_directory);
test.py
+197 -124
View File
@@ -1,27 +1,31 @@
import argparse
import filecmp
import os
import shutil
import subprocess
import sys
import tempfile
import time
# --- Configuration ---
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"]
# --- Resource Limit Configuration ---
# 💾 Disk throttling settings
DISK_DEVICE = (
"/dev/nvme0n1p5" # IMPORTANT: Change this to your disk (e.g., /dev/nvme0n1)
)
READ_BPS_MAX = "15M" # Max read speed (M for megabytes)
WRITE_BPS_MAX = "10M" # Max write speed
DISK_DEVICE = "/dev/nvme0n1p5"
READ_BPS_MAX = "15M"
WRITE_BPS_MAX = "10M"
# 🐢 Network throttling settings (Linux tc)
NET_LIMIT = "100mbit"
NET_DELAY = "100ms"
NETWORK_INTERFACE = "lo"
NET_LIMIT_CMD = f"sudo tc qdisc add dev {NETWORK_INTERFACE} root netem rate {NET_LIMIT} delay {NET_DELAY}".split()
NET_LIMIT_CMD = (
f"sudo tc qdisc add dev {NETWORK_INTERFACE} root netem rate {NET_LIMIT} delay {NET_DELAY}".split()
)
NET_RESET_CMD = f"sudo tc qdisc del dev {NETWORK_INTERFACE} root".split()
# --- Build the client command prefix with throttling ---
CLIENT_CMD_PREFIX = [
"sudo",
"systemd-run",
@@ -32,11 +36,10 @@ CLIENT_CMD_PREFIX = [
f"IOWriteBandwidthMax={DISK_DEVICE} {WRITE_BPS_MAX}",
]
# --- Test Cases ---
TEST_CASES = [
{"name": "Standard (Single-threaded)", "flags": []},
{"name": "Multithreading (-m)", "flags": ["-m"]},
{"name": "Compression (-c 0)", "flags": ["-c 0"]},
{"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"]},
@@ -48,179 +51,249 @@ TEST_CASES = [
]
def run_suite(env_name, apply_limits):
def generate_test_files(source_dir):
if os.path.exists(source_dir):
shutil.rmtree(source_dir)
os.makedirs(source_dir)
target_total = 50 * 1024 * 1024
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)
i = 0
while written < target_total:
chunk_size = min(5 * 1024 * 1024, target_total - written)
rel_path = f"bulk/file_{i}.dat"
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(b"0" * chunk_size)
written += chunk_size
i += 1
total_mb = written / (1024 * 1024)
print(f" Generated {total_mb:.1f} MB of test data in {source_dir}")
def verify_transfer(source_dir, dest_dir):
source_dir = os.path.abspath(source_dir)
dest_dir = os.path.abspath(dest_dir)
received_prefix = os.path.join(dest_dir, source_dir.lstrip(os.sep))
if not os.path.exists(received_prefix):
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_prefix, 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 run_suite(env_name, apply_limits, source_dir, dest_dir):
results = []
print(f"\n{'=' * 60}")
print(f"🚀 Starting Suite: {env_name}")
print(f"Suite: {env_name}")
print(f"{'=' * 60}")
if apply_limits:
print(
f"Applying Disk I/O Limits: Reads <= {READ_BPS_MAX}, Writes <= {WRITE_BPS_MAX}"
)
print(f"Applying Network Limits: {NET_LIMIT}, {NET_DELAY} delay")
print(f" Disk I/O: Reads <= {READ_BPS_MAX}, Writes <= {WRITE_BPS_MAX}")
print(f" Network: {NET_LIMIT}, {NET_DELAY} delay")
client_prefix = CLIENT_CMD_PREFIX
else:
print("Running Baseline (No limits applied)")
client_prefix = [] # Run normally without systemd-run/limits
print(" Baseline (no limits)")
client_prefix = []
try:
# SETUP: Apply or ensure clean network limits
if apply_limits:
subprocess.run(NET_LIMIT_CMD, check=True)
else:
# Silently attempt to clear any leftover rules just to ensure a clean baseline
subprocess.run(NET_RESET_CMD, capture_output=True)
for case in TEST_CASES:
name = case["name"]
flags = case["flags"]
print(f"\n --- {name} ---")
print(f"\n--- Running: {name} ---")
if os.path.exists(dest_dir):
shutil.rmtree(dest_dir)
server_process = None
try:
# 1. Start the server
print(" Starting server...")
server_process = subprocess.Popen(
SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None
)
time.sleep(0.5) # Allow server to bind to port
time.sleep(0.5)
# 2. Build and run the client
client_cmd = client_prefix + base_client_cmd + flags
print(f" Running client: {' '.join(client_cmd)}")
env = os.environ.copy()
client_cmd = (
client_prefix
+ base_client_cmd
+ ["--source-dir", source_dir, "--dest-dir", dest_dir, "--save-to-disk"]
+ flags
)
print(f" Running: {' '.join(client_cmd)}")
start_time = time.monotonic()
client_result = subprocess.run(
client_cmd, text=True, capture_output=True
client_cmd, env=env, text=True, capture_output=True
)
end_time = time.monotonic()
duration = end_time - start_time
if server_process:
try:
server_process.wait(timeout=5)
except subprocess.TimeoutExpired:
server_process.kill()
server_process.wait()
server_process = None
mismatches, missing = [], []
if client_result.returncode == 0:
results.append(
{
"environment": env_name,
"name": name,
"status": "Success",
"time": f"{duration:.4f}s",
"error": "",
}
)
mismatches, missing = verify_transfer(source_dir, dest_dir)
entry = {
"name": name,
"suite": env_name,
"time": f"{duration:.4f}s" if client_result.returncode == 0 else "N/A",
}
if client_result.returncode == 0 and not mismatches and not missing:
entry["status"] = "Success"
entry["error"] = ""
else:
print(f" ⚠️ Failed (code: {client_result.returncode})")
err_msg = (
client_result.stderr.strip().split("\n")[0]
if client_result.stderr
else (
client_result.stdout.strip().split("\n")[0]
if client_result.stdout
else "No output"
entry["status"] = "Failed"
errors = []
if client_result.returncode != 0:
err = (
client_result.stderr.strip().split("\n")[0]
if client_result.stderr
else (
client_result.stdout.strip().split("\n")[0]
if client_result.stdout
else "No output"
)
)
)
results.append(
{
"environment": env_name,
"name": name,
"status": "Failed",
"time": "N/A",
"error": f"Exit code {client_result.returncode}: {err_msg[:40]}",
}
)
errors.append(f"Exit code {client_result.returncode}: {err[: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)
results.append(entry)
except subprocess.TimeoutExpired:
print(" ⚠️ Timeout (exceeded 15s)")
results.append(
{
"environment": env_name,
"name": name,
"status": "Timeout",
"time": "N/A",
"error": "Exceeded 15 seconds",
}
{"name": name, "suite": env_name, "status": "Timeout", "time": "N/A", "error": "Exceeded 15s"}
)
except Exception as e:
print(f" ❌ Error: {e}")
results.append(
{
"environment": env_name,
"name": name,
"status": "Error",
"time": "N/A",
"error": str(e),
}
{"name": name, "suite": env_name, "status": "Error", "time": "N/A", "error": str(e)}
)
finally:
# Clean up the server for this test case
if server_process:
print(" Stopping server...")
try:
server_process.terminate()
server_process.wait(timeout=5)
except subprocess.TimeoutExpired:
server_process.kill()
server_process.wait()
except subprocess.CalledProcessError as e:
print(f"❌ Error running system limit command: {' '.join(e.cmd)}")
print("Are you running this script with 'sudo' privileges?")
print(f" Error running limit command: {' '.join(e.cmd)}")
finally:
# TEARDOWN: Remove network limits if they were applied
if apply_limits:
print("\nCleaning up limits for this suite...")
try:
subprocess.run(NET_RESET_CMD, check=True, capture_output=True)
print("Network limits removed.")
except Exception as e:
print(f"⚠️ Could not reset network settings: {e}")
except Exception:
pass
return results
os.system("cmake -B build -S .")
os.system("cd build && make")
def main():
parser = argparse.ArgumentParser(description="FastSync integration test / benchmark")
parser.add_argument("--source-dir", default=DEFAULT_SOURCE_DIR,
help="Source directory for test files (default: %(default)s)")
parser.add_argument("--dest-dir", default=DEFAULT_DEST_DIR,
help="Destination directory for received files (default: %(default)s)")
parser.add_argument("--keep-data", action="store_true",
help="Keep test_data directory after run")
parser.add_argument("--no-throttled", action="store_true",
help="Skip throttled suite (requires sudo)")
parser.add_argument("--no-unlimited", action="store_true",
help="Skip unlimited suite")
args = parser.parse_args()
# --- Main Execution ---
all_results = []
os.system("cmake -B build -S . > /dev/null 2>&1")
ret = os.system("cd build && make -j$(nproc) 2>&1 | tail -3")
if ret != 0:
print("Build failed")
sys.exit(1)
# 1. Run Baseline (No Limits)
all_results.extend(run_suite("Unlimited", apply_limits=False))
generate_test_files(args.source_dir)
os.makedirs(args.dest_dir, exist_ok=True)
# # 2. Run Throttled (With Limits)
all_results.extend(run_suite("Throttled", apply_limits=True))
try:
all_results = []
# --- Print Comparison Table ---
print("\n" + "=" * 105)
print(f"{'fastSync BENCHMARK RESULTS (COMPARISON)':^105}")
print("=" * 105)
print(
f"{'Configuration':<45} | {'Environment':<12} | {'Status':<10} | {'Time':<10} | {'Details/Error':<20}"
)
print("-" * 105)
if not args.no_unlimited:
all_results.extend(
run_suite("Unlimited", False, args.source_dir, args.dest_dir)
)
# Sort results by test case name first, then environment to easily compare
# This groups the baseline and throttled results for the same test next to each other
# sorted_results = sorted(
# all_results,
# key=lambda x: (
# TEST_CASES.index(
# next(item for item in TEST_CASES if item["name"] == x["name"])
# ),
# x["environment"],
# ),
# )
if not args.no_throttled:
all_results.extend(
run_suite("Throttled", True, args.source_dir, args.dest_dir)
)
for res in all_results:
status_symbol = (
"✅"
if res["status"] == "Success"
else ("⏳" if res["status"] == "Timeout" else "❌")
)
status_str = f"{status_symbol} {res['status']}"
print(
f"{res['name']:<45} | {res['environment']:<12} | {status_str:<10} | {res['time']:<10} | {res['error']:<20}"
)
print("=" * 105)
print("\n" + "=" * 110)
print(f"{'RESULTS':^110}")
print("=" * 110)
print(f"{'Configuration':<45} | {'Suite':<12} | {'Status':<8} | {'Time':<10} | {'Details'}")
print("-" * 110)
for res in all_results:
print(
f"{res['name']:<45} | {res['suite']:<12} | {res['status']:<8} | {res['time']:<10} | {res['error']}"
)
failed = [r for r in all_results if r["status"] != "Success"]
if failed:
print(f"\n {len(failed)} test(s) FAILED")
sys.exit(1)
else:
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()