#!/usr/bin/env python3"""abdiff.py -- differential A/B regression hunter for rsync. Runs the same transfer with two rsync binaries (A = the build under test, e.g../rsync; B = a baseline, e.g. old_versions/rsync_3.4.1) and compares theOUTCOME: exit code, error output, --stats "Literal data", the destination tree(content + full metadata), and the --itemize change list. Core oracle: for a BENIGN input a correctness/behaviour change between the twobuilds must be invisible, so A and B must produce an identical destination treeand both exit 0.  Any divergence is a regression candidate (e.g. a refactor thatsilently changes what a benign `rsync -a` transfers). This is a developer tool, NOT a runtests.py test (it does not end in _test.py andimports nothing from the test harness).  Findings are printed and appended to alog; minimize each into a testsuite/*_test.py. Usage:    testsuite/abdiff.py [--rsync-a ./rsync] [--rsync-b old_versions/rsync_3.4.1]                      [--sweep options|pathshape|all] [--workdir DIR] [--keep]                      [--findings abdiff-findings.txt] [--only NAME] [--list]Exit 0 iff no regression candidates were found.""" from __future__ import annotations import argparseimport hashlibimport itertoolsimport osimport randomimport reimport shutilimport signalimport statimport subprocessimport sysimport threadingimport timefrom concurrent.futures import (FIRST_COMPLETED, ThreadPoolExecutor,                                as_completed, wait)from pathlib import Path # ---------------------------------------------------------------------------# config / globals (set in main)RSYNC_A = "./rsync"RSYNC_B = "old_versions/rsync_3.4.1"RRSYNC_A = None     # rrsync wrapper paired with A/B (None -> in-tree support/rrsync)RRSYNC_B = NoneKEEP = FalseREPEAT = 2   # stability gate: run each binary N times; flaky scenarios quarantinedCMD_TIMEOUT = 120   # per-subprocess wall-clock guard, seconds (0 = unlimited)COST = False        # --cost: also compare peak process-group RSS (resource oracle)SCALE_N = 2000      # --scale: element count for the scale-escalation fixtures_supports_cache: dict = {}_supports_lock = threading.Lock()_tls = threading.local()   # per-worker: .measure (bool) + .rss (peak bytes)_PAGE = os.sysconf("SC_PAGE_SIZE")  def _group_rss(pgid):    """Summed RSS (bytes) of every live process in process group pgid -- catches    rsync's forked generator/receiver/sender (and ssh/daemon children), which a    direct-child measure (/usr/bin/time) would miss."""    total = 0    for pid in os.listdir("/proc"):        if not pid.isdigit():            continue        try:            with open(f"/proc/{pid}/stat") as f:                fields = f.read().split()            if int(fields[4]) != pgid:        # field 5 (0-idx 4) = pgrp                continue            with open(f"/proc/{pid}/statm") as f:                total += int(f.read().split()[1]) * _PAGE   # resident pages        except (OSError, ValueError, IndexError):            continue    return total  def sh(cmd, cwd=None, env=None, timeout=None):    """Run cmd capturing stdout/stderr. Runs in its own process group with a    wall-clock timeout so a wedged rsync (or its ssh/daemon children) can't hang a    worker forever -- on timeout the whole group is killed and rc 124 returned."""    if timeout is None:        timeout = CMD_TIMEOUT    p = subprocess.Popen(cmd, cwd=cwd, env=env, stdout=subprocess.PIPE,                         stderr=subprocess.PIPE, text=True, start_new_session=True)    # cost oracle: while the transfer runs, sample the peak summed RSS of its    # whole process group (set per-worker by one_run via _tls.measure).    measure = getattr(_tls, "measure", False)    stop = peak = poller = None    if measure:        peak = [0]        stop = threading.Event()        pgid = os.getpgid(p.pid)         def _poll():            while not stop.is_set():                peak[0] = max(peak[0], _group_rss(pgid))                stop.wait(0.03)        poller = threading.Thread(target=_poll, daemon=True)        poller.start()    try:        out, err = p.communicate(timeout=timeout or None)        rc = p.returncode    except subprocess.TimeoutExpired:        try:            os.killpg(os.getpgid(p.pid), signal.SIGKILL)        except OSError:            pass        out, err = p.communicate()        rc = 124        err = "[abdiff: TIMEOUT]\n" + (err or "")    if measure:        stop.set()        poller.join(timeout=1)        _tls.rss = peak[0] or None    return subprocess.CompletedProcess(cmd, rc, out or "", err or "")  def supports(binary, opt):    key = (binary, opt)    with _supports_lock:        if key in _supports_cache:            return _supports_cache[key]    r = sh([binary, opt, "--version"])    bad = any(m in (r.stderr or "").lower()              for m in ("unknown option", "unrecognized option", "no such option"))    val = not bad    with _supports_lock:        _supports_cache[key] = val    return val  # ---------------------------------------------------------------------------# tree snapshot + comparison ERR_MARKERS = ("rsync error", "failed to open", "rsync: ", "Invalid argument",               "No such file", "Operation not permitted", "cannot ") # Intentional, documented behaviour-change refusals (A errors where B didn't),# recorded as ALLOW rather than a silent regression. Each entry is# (substring-in-A's-stderr, human note). Populate as deliberate behaviour changes# between the two builds are identified.ALLOWLIST = []  def _xattrs(path):    try:        names = sorted(os.listxattr(path, follow_symlinks=False))    except (OSError, AttributeError):        return {}    out = {}    for n in names:        if n.startswith("system.posix_acl_"):            continue  # captured via getfacl        try:            out[n] = os.getxattr(path, n, follow_symlinks=False).hex()        except OSError:            out[n] = "?"    return out  def _acl(path, is_dir):    r = sh(["getfacl", "-pcEn", path]) if shutil.which("getfacl") else None    if not r or r.returncode != 0:        return None    lines = [ln for ln in r.stdout.splitlines() if ln and not ln.startswith("#")]    return "\n".join(sorted(lines)) or None  def snapshot(root: Path):    """Map rel-path -> attribute dict for every entry under root (root itself    excluded). Symlinks/specials are recorded, never followed."""    root = Path(root)    snap = {}    inode_of = {}  # (dev,ino) -> first rel path, for hardlink grouping    if not root.exists():        return snap    stack = [root]    while stack:        d = stack.pop()        try:            entries = sorted(os.scandir(d), key=lambda e: e.name)        except OSError:            continue        for e in entries:            p = Path(e.path)            rel = str(p.relative_to(root))            try:                st = os.lstat(p)            except OSError:                snap[rel] = {"type": "GONE"}                continue            m = st.st_mode            a = {                "mode": stat.S_IMODE(m),                "uid": st.st_uid, "gid": st.st_gid,                "mtime": int(st.st_mtime),            }            if stat.S_ISDIR(m):                a["type"] = "d"                stack.append(p)            elif stat.S_ISLNK(m):                a["type"] = "l"                a["target"] = os.readlink(p)            elif stat.S_ISREG(m):                a["type"] = "f"                a["size"] = st.st_size                a["blocks"] = st.st_blocks  # sparseness                # "is this file hardlinked at all" (link-dest / -H) -- a robust                # boolean; raw nlink counts are contaminated when A and B share a                # --link-dest basis dir, but "copied(1) vs linked(>1)" still                # catches a real link-dest/hardlink regression.                a["linked"] = st.st_nlink > 1                if st.st_nlink > 1:                    key = (st.st_dev, st.st_ino)                    a["hardlink"] = inode_of.setdefault(key, rel)                h = hashlib.sha256()                try:                    with open(p, "rb") as fh:                        for chunk in iter(lambda: fh.read(1 << 20), b""):                            h.update(chunk)                    a["sha"] = h.hexdigest()                except OSError as ex:                    a["sha"] = f"ERR:{ex.errno}"            elif stat.S_ISFIFO(m):                a["type"] = "p"            elif stat.S_ISSOCK(m):                a["type"] = "s"            elif stat.S_ISBLK(m) or stat.S_ISCHR(m):                a["type"] = "b" if stat.S_ISBLK(m) else "c"                a["rdev"] = (os.major(st.st_rdev), os.minor(st.st_rdev))            else:                a["type"] = "?"            xa = _xattrs(p)            if xa:                a["xattr"] = xa            ac = _acl(p, stat.S_ISDIR(m))            if ac:                a["acl"] = ac            snap[rel] = a    return snap  # which attrs are meaningful depends on the options used; keep it simple and# compare everything, but let callers ignore mtime when -t isn't in play, or# for a type whose times rsync intentionally leaves unmanaged (-O dirs, -J# symlinks) -> those dest mtimes are creation-time and differ between runs.def diff_snapshots(sa, sb, ignore_mtime=False, ignore_mtime_types=()):    diffs = []    for rel in sorted(set(sa) | set(sb)):        a, b = sa.get(rel), sb.get(rel)        if a is None:            diffs.append(f"  only in B(baseline): {rel} ({b.get('type')})")            continue        if b is None:            diffs.append(f"  only in A(under-test): {rel} ({a.get('type')})")            continue        for k in sorted(set(a) | set(b)):            if k == "mtime" and (ignore_mtime                                 or a.get("type") in ignore_mtime_types):                continue            if a.get(k) != b.get(k):                diffs.append(f"  {rel}: {k}  A={a.get(k)!r}  B={b.get(k)!r}")    return diffs  # ---------------------------------------------------------------------------# running a transfer with one binary def run_xfer(binary, workdir, opts, src_args, dest, cwd=None, pre=None):    """Run `binary opts src_args dest` (cwd default=workdir). Returns    (rc, stderr, literal_data, itemize)."""    cwd = cwd or workdir    argv = [binary, "--stats", "-i", *opts, *src_args, dest]    return _parse_out(sh(argv, cwd=cwd))  def _parse_out(r):    literal = None    for ln in r.stdout.splitlines():        if ln.startswith("Literal data:"):            literal = ln.split(":", 1)[1].strip()    itemize = "\n".join(sorted(        ln for ln in r.stdout.splitlines()        if len(ln) > 11 and ln[1] in "fdLDS" and ln[0] in "<>ch.*"        and ln[11:].strip() != "./"))  # bench-dependent top-dir time line    return r.returncode, (r.stderr or "").strip(), literal, itemize, (r.stdout or "")  _NOISE = re.compile(r'bytes/sec|speedup is|^sent .*received |^total size is|'                    r'^Number of |^Total |^Literal data:|^Matched data:|'                    r'^File list |^total:|^created |^deleting ')  def _norm_out(text, wd, dest):    """Normalised stdout for A/B compare: canonicalise the workdir/dest paths and    drop bench-variant stats lines (keeps itemize/listing/warning lines)."""    text = text.replace(str(dest), "DEST").replace(str(wd), "WD")    return "\n".join(ln for ln in text.splitlines() if ln and not _NOISE.search(ln))  def _norm_err(text, wd, dest):    """Normalised stderr TEXT for A/B compare: canonicalise paths, strip the    version-dependent role tag ([sender=3.4.x], [client=VERSION]) and at-FILE(LINE)    source location (line numbers shift between versions) so only the MESSAGE is    compared."""    text = text.replace(str(dest), "DEST").replace(str(wd), "WD")    text = re.sub(r'\[(?:sender|receiver|generator|client|server'                  r'|Sender|Receiver|Generator|Client|Server)=[^\]]*\]',                  '[ROLE]', text)    text = re.sub(r' at [\w./-]+\(\d+\)', ' at LOC', text)    return text.strip()  def _wait_port(port, timeout=10.0):    import socket as _sock    end = time.time() + timeout    while time.time() < end:        try:            _sock.create_connection(("127.0.0.1", port), 0.3).close()            return True        except OSError:            time.sleep(0.05)    return False  def run_daemon_xfer(binary, wd, module_path, opts, src_args, port, chroot="no"):    """Push src_args into a [m] module served by `binary --daemon` over a PRIVATE    STDIO PIPE (RSYNC_CONNECT_PROG) -- no TCP port, so no port-bind/startup race    (the old TCP path was nondeterministic under load).  `port` only uniquifies    the per-invocation config filename.  Same (rc, err, lit, item) as run_xfer."""    conf = Path(wd) / f"rsyncd_{port}.conf"    Path(module_path).mkdir(parents=True, exist_ok=True)    conf.write_text(        f"use chroot = {chroot}\n"        f"[m]\n  path = {module_path}\n  read only = no\n"        f"  hosts allow = 127.0.0.1\n")    env = {**os.environ,           "RSYNC_CONNECT_PROG": f"{binary} --config={conf} --daemon"}    argv = [binary, "--stats", "-i", *opts, *src_args, "rsync://localhost/m/"]    return _parse_out(sh(argv, cwd=str(wd), env=env))  def run_daemon_pull(binary, wd, served, opts, localdest, port, chroot="no"):    """PULL from a read-only [m] module (the daemon SENDER side) over a private    stdio pipe into localdest. `served` is the served directory."""    conf = Path(wd) / f"rsyncd_{port}.conf"    Path(served).mkdir(parents=True, exist_ok=True)    Path(localdest).mkdir(parents=True, exist_ok=True)    conf.write_text(        f"use chroot = {chroot}\n"        f"[m]\n  path = {served}\n  read only = yes\n"        f"  hosts allow = 127.0.0.1\n")    env = {**os.environ,           "RSYNC_CONNECT_PROG": f"{binary} --config={conf} --daemon"}    argv = [binary, "--stats", "-i", *opts, "rsync://localhost/m/",            str(localdest) + "/"]    return _parse_out(sh(argv, cwd=str(wd), env=env))  # sibling helper scripts live in support/ (abdiff.py itself lives in testsuite/)_SUPPORT = Path(__file__).resolve().parent.parent / "support"_LSH = str(_SUPPORT / "lsh.sh")  def run_ssh_xfer(binary, wd, opts, src_args, dest):    """PUSH over a remote-shell split via support/lsh.sh (host 'lh' = no chdir);    the remote side runs the same binary via --rsync-path. Real client+server    processes + protocol, unlike a both-paths-local copy."""    argv = [binary, "--stats", "-i", "-e", f"sh {_LSH}",            f"--rsync-path={binary}", *opts, *src_args, f"lh:{dest}/"]    return _parse_out(sh(argv, cwd=wd))  # --- rrsync lane: route the remote side through the restricted rrsync wrapper --_RRSH = str(_SUPPORT / "rrsh.sh")_RRSYNC_SRC = _SUPPORT / "rrsync"  def _patch_rrsync(binary, rrsync_src, wd):    """A copy of `rrsync_src` (the version's rrsync wrapper) whose RSYNC points at    `binary`, so the rrsync-launched server is the binary under test. rrsync is    SHIPPED PER VERSION, so A and B use their OWN rrsync (the regressions live in    the script, not just the binary) -- keyed per (binary, src) to keep them    distinct in a shared workdir."""    key = abs(hash((binary, str(rrsync_src)))) % 1000000    dst = Path(wd) / f"rrsync-{key}"    if not dst.exists():        txt = Path(rrsync_src).read_text()        txt = re.sub(r"^RSYNC = '[^']*'", f"RSYNC = {binary!r}", txt, count=1,                     flags=re.M)        dst.write_text(txt)        dst.chmod(0o755)    return dst  def run_rrsync_push(binary, rrsync_src, wd, opts, src_args, dest):    """PUSH through `rrsync <restricted>` (the dest's parent is the restricted    root; the client writes into the <dest-name>/ subdir, so rrsync's subdir    restrictions are exercised). Exercises rrsync option/path validation."""    dest = Path(dest)    rr = _patch_rrsync(binary, rrsync_src, wd)    dest.parent.mkdir(parents=True, exist_ok=True)    argv = [binary, "--stats", "-i", "-e", f"sh {_RRSH} {rr} {dest.parent}",            *opts, *src_args, f"lh:{dest.name}/"]    return _parse_out(sh(argv, cwd=str(wd)))  def run_rrsync_pull(binary, rrsync_src, wd, served, opts, localdest):    """PULL through `rrsync <restricted>` (rrsync's --sender side): the served    dir's parent is the restricted root, the client reads the <served-name>/    subdir into localdest."""    served = Path(served)    rr = _patch_rrsync(binary, rrsync_src, wd)    Path(localdest).mkdir(parents=True, exist_ok=True)    argv = [binary, "--stats", "-i", "-e", f"sh {_RRSH} {rr} {served.parent}",            *opts, f"lh:{served.name}/", str(localdest) + "/"]    return _parse_out(sh(argv, cwd=str(wd)))  # --- real-TCP-daemon lane: a genuine `rsync --daemon` on a bound port ----------_port_lock = threading.Lock()_next_port = [40000]  def _alloc_port():    """A free localhost TCP port, bind-probed under a lock so concurrent workers    don't collide (the real-daemon path needs a real port, unlike the stdio one)."""    import socket as _s    with _port_lock:        for _ in range(4000):            p = _next_port[0]            _next_port[0] = 40000 if p >= 60000 else p + 1            s = _s.socket(_s.AF_INET, _s.SOCK_STREAM)            try:                s.setsockopt(_s.SOL_SOCKET, _s.SO_REUSEADDR, 1)                s.bind(("127.0.0.1", p))                return p            except OSError:                continue            finally:                s.close()    return _next_port[0]  def _tcp_daemon(binary, wd, module_path, opts, src_args, localdest=None,                pull=False, chroot="no", auth=False):    """Start a real `binary --daemon` on a bound TCP port and run the client over    rsync://127.0.0.1:PORT/m/ -- exercises the genuine socket path / greeting /    handshake (and, with auth, the challenge-response) that the stdio-pipe daemon    lane bypasses."""    port = _alloc_port()    conf = Path(wd) / f"tcpd_{port}.conf"    Path(module_path).mkdir(parents=True, exist_ok=True)    authlines = ""    env = dict(os.environ)    if auth:        sp = Path(wd) / f"tcpd_{port}.secrets"        sp.write_text("abuser:abpass\n")        sp.chmod(0o600)        authlines = f"  auth users = abuser\n  secrets file = {sp}\n"        env["RSYNC_PASSWORD"] = "abpass"    conf.write_text(        f"use chroot = {chroot}\nport = {port}\n"        f"log file = {wd}/tcpd_{port}.log\npid file = {wd}/tcpd_{port}.pid\n"        f"[m]\n  path = {module_path}\n  read only = {'yes' if pull else 'no'}\n"        f"  hosts allow = 127.0.0.1\n{authlines}")    proc = subprocess.Popen(        [binary, "--daemon", "--no-detach", f"--config={conf}",         f"--port={port}", "--address=127.0.0.1"],        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,        start_new_session=True)    try:        if not _wait_port(port):            return (99, "tcp daemon failed to start", None, "", "")        user = "abuser@" if auth else ""        url = f"rsync://{user}127.0.0.1:{port}/m/"        if pull:            Path(localdest).mkdir(parents=True, exist_ok=True)            argv = [binary, "--stats", "-i", *opts, url, str(localdest) + "/"]        else:            argv = [binary, "--stats", "-i", *opts, *src_args, url]        return _parse_out(sh(argv, cwd=str(wd), env=env))    finally:        proc.terminate()        try:            proc.wait(timeout=5)        except subprocess.TimeoutExpired:            proc.kill()  # ---------------------------------------------------------------------------# fixtures def _write(p: Path, data: bytes):    p.parent.mkdir(parents=True, exist_ok=True)    p.write_bytes(data)

Lines 1–500 of 2824. Load more