Nextcloud vs Seafile: upload, download, RAM and disk

#!/usr/bin/env python3
"""Runs in the VM. sync_bench.py nextcloud|seafile <out.json> <project>
Uploads the generated datasets (r1..r3) over the real HTTP API and downloads them again, timing each phase.
Nextcloud: WebDAV (MKCOL, PUT, GET) with basic auth. Seafile: /api2 auth token, library per rep, upload-api multipart POST, download link GET."""
import io, sys, os, json, time, hashlib, http.client, base64, urllib.parse, subprocess, threading, uuid
kind, outp, proj = sys.argv[1:4]; WORKERS = 4
HOST, PORT = "127.0.0.1", 8080; PW = "lab-only-not-a-secret"; root = os.path.expanduser("~/data"); man = json.load(open(f"{root}/manifest.json"))
def conn(host=HOST, port=PORT): return http.client.HTTPConnection(host, port, timeout=600)
def disk():
    r = subprocess.run(f"sudo sh -c 'du -sb /var/lib/docker/volumes/{proj}_*' | awk '{{s+=$1}} END{{print s}}'", shell=True, capture_output=True, text=True); return int(r.stdout.strip())
class Reader:  # streams prefix + file + suffix without loading the file
    def __init__(s, pre, path, post): s.parts = [io.BytesIO(pre), open(path, "rb"), io.BytesIO(post)]; s.i = 0; s.len = len(pre) + os.path.getsize(path) + len(post)
    def read(s, n=1 << 20):
        while s.i < len(s.parts):
            b = s.parts[s.i].read(n)
            if b: return b
            s.i += 1
        return b""
class NC:
    def __init__(s): s.auth = {"Authorization": "Basic " + base64.b64encode(f"admin:{PW}".encode()).decode()}; s.base = "/remote.php/dav/files/admin"
    def prep(s, rep): c = conn(); c.request("MKCOL", f"{s.base}/r{rep}", headers=s.auth); r = c.getresponse(); r.read(); assert r.status in (201, 405), r.status; s.dirs = set()
    def mkdirs(s, rep, dirs):  # sequential MKCOL, counted in the timed upload phase
        c = conn()
        for p in ["small"] + [f"small/{d}" for d in dirs]:
            c.request("MKCOL", f"{s.base}/r{rep}/{p}", headers=s.auth); r = c.getresponse(); r.read(); assert r.status in (201, 405), (p, r.status)
    def put(s, c, rep, rel, path):
        sz = os.path.getsize(path)
        with open(path, "rb") as f:
            c.request("PUT", f"{s.base}/r{rep}/{rel}", body=f, headers={**s.auth, "Content-Length": str(sz)}); r = c.getresponse(); r.read(); assert r.status in (201, 204), r.status
    def get(s, c, rep, rel):
        c.request("GET", f"{s.base}/r{rep}/{rel}", headers=s.auth); r = c.getresponse(); h = hashlib.sha256(); n = 0
        assert r.status == 200, r.status
        while True:
            b = r.read(1 << 20)
            if not b: break
            h.update(b); n += len(b)
        return h.hexdigest(), n
class SF:
    def __init__(s):
        c = conn(); c.request("POST", "/api2/auth-token/", urllib.parse.urlencode({"username": "admin@lab.internal", "password": PW}), {"Content-Type": "application/x-www-form-urlencoded"})
        r = c.getresponse(); s.tok = json.loads(r.read())["token"]; s.auth = {"Authorization": "Token " + s.tok}; s.repos = {}
    def api(s, method, path, body=None):
        c = conn(); h = dict(s.auth)
        if body: h["Content-Type"] = "application/x-www-form-urlencoded"; body = urllib.parse.urlencode(body)
        c.request(method, path, body, h); r = c.getresponse(); t = r.read(); assert r.status < 300, (path, r.status, t[:200]); return json.loads(t) if t else None
    def prep(s, rep):
        r = s.api("POST", "/api2/repos/", {"name": f"r{rep}"}); s.repos[rep] = r["repo_id"]
    def mkdirs(s, rep, dirs):
        s.api("POST", f"/api2/repos/{s.repos[rep]}/dir/?p=/small", {"operation": "mkdir", "create_parents": "true"})
        for d in dirs: s.api("POST", f"/api2/repos/{s.repos[rep]}/dir/?p=/small/{d}", {"operation": "mkdir"})
    def link(s, rep): return urllib.parse.urlparse(s.api("GET", f"/api2/repos/{s.repos[rep]}/upload-link/?p=/")).path
    def put(s, c, rep, rel, path):
        d, name = os.path.split(rel); b = uuid.uuid4().hex; sz = os.path.getsize(path)
        pre = (f'--{b}\r\nContent-Disposition: form-data; name="parent_dir"\r\n\r\n/\r\n--{b}\r\nContent-Disposition: form-data; name="relative_path"\r\n\r\n{d}\r\n'
               f'--{b}\r\nContent-Disposition: form-data; name="replace"\r\n\r\n1\r\n--{b}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\nContent-Type: application/octet-stream\r\n\r\n').encode()
        post = f"\r\n--{b}--\r\n".encode(); rd = Reader(pre, path, post)
        c2 = conn(); c2.request("POST", s.link(rep) + "?ret-json=1", body=rd, headers={"Content-Type": f"multipart/form-data; boundary={b}", "Content-Length": str(rd.len)}); r = c2.getresponse(); t = r.read(); assert r.status == 200, (r.status, t[:200]); c2.close()
    def get(s, c, rep, rel):
        u = s.api("GET", f"/api2/repos/{s.repos[rep]}/file/?p=/{urllib.parse.quote(rel)}&reuse=1"); p = urllib.parse.urlparse(u).path
        c2 = conn(); c2.request("GET", p); r = c2.getresponse(); h = hashlib.sha256(); n = 0; assert r.status == 200, r.status
        while True:
            b = r.read(1 << 20)
            if not b: break
            h.update(b); n += len(b)
        c2.close(); return h.hexdigest(), n
srv = NC() if kind == "nextcloud" else SF()
def timed(fn):
    t0 = time.time(); p0 = time.perf_counter(); err = fn(); return {"t_start": t0, "t_end": time.time(), "secs": time.perf_counter() - p0, "errors": err}
def workers(items, fn):
    errs = [0]; lk = threading.Lock(); it = iter(items)
    def w():
        c = conn()
        while True:
            with lk: x = next(it, None)
            if x is None: return
            try: fn(c, x)
            except Exception as e:
                errs[0] += 1; print("ERR", x, repr(e)[:200], file=sys.stderr)
                c = conn()
    th = [threading.Thread(target=w) for _ in range(WORKERS)]; [t.start() for t in th]; [t.join() for t in th]; return errs[0]
out = {"kind": kind, "workers_small": WORKERS, "reps": [], "disk_before": disk(), "t0": time.time()}
for rep in range(1, int(os.environ.get("REPS", 3)) + 1):
    m = man[f"r{rep}"]; res = {"rep": rep, "small_files": len(m["small"]), "small_bytes": m["small_bytes"], "large_bytes": m["large"][1]}
    srv.prep(rep)
    dirs = sorted({os.path.basename(os.path.dirname(x[0])) for x in m["small"]})
    def up_small(): srv.mkdirs(rep, dirs); return workers(m["small"], lambda c, x: srv.put(c, rep, x[0], f"{root}/r{rep}/{x[0]}"))
    res["up_small"] = timed(up_small)
    res["up_large"] = timed(lambda: (srv.put(conn(), rep, "large.bin", f"{root}/r{rep}/large.bin"), 0)[1])
    res["disk_after_upload"] = disk()
    bad = [0]
    def dl_small():
        def f(c, x):
            h, n = srv.get(c, rep, x[0])
            if h != x[2] or n != x[1]: bad[0] += 1
        return workers(m["small"], f)
    res["dl_small"] = timed(dl_small); res["dl_small"]["hash_mismatches"] = bad[0]
    def dl_large():
        h, n = srv.get(conn(), rep, "large.bin"); return 0 if (h == m["large"][2] and n == m["large"][1]) else 1
    res["dl_large"] = timed(dl_large)
    out["reps"].append(res); print("rep", rep, {k: (round(v["secs"], 1) if isinstance(v, dict) else v) for k, v in res.items() if k.startswith(("up", "dl"))}, flush=True)
    json.dump(out, open(outp, "w"), indent=1)
out["t_end"] = time.time(); json.dump(out, open(outp, "w"), indent=1)