rsync/runtests.py

#!/usr/bin/env python3

# Copyright (C) 2001, 2002 by Martin Pool <mbp@samba.org>
# Copyright (C) 2003-2022 Wayne Davison
# Copyright (C) 2026 Andrew Tridgell
#
# Rewrite of runtests.sh in Python (runtests.sh is now deprecated).
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version
# 2 as published by the Free Software Foundation.

"""rsync test runner.

Invokes test scripts from testsuite/ and reports results.
Can be called by 'make check' or directly.

Usage:
    ./runtests.py [options] [TEST ...]

Each TEST is a test name (e.g. 'delete') or glob pattern (e.g. 'xattr*').
If no tests are specified, all tests are run.
"""

import argparse
import concurrent.futures
import fnmatch
import glob
import math
import os
import signal
import subprocess
import sys
import threading
import time

# Share the test exit-code enum with the test helpers. exitcodes.py lives in
# testsuite/ (next to this script); it has no import-time side effects.
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'testsuite'))
from exitcodes import Exit


def _race_seconds(text):
    """argparse type for --race-timeout: a finite, strictly positive number.

    A race test loops `while monotonic() < deadline`, so a budget of 0 (or a
    negative, or a NaN, which fails every comparison) runs the body ZERO times
    and the test reports PASS without ever exercising its oracle -- a silently
    disarmed security test, which is worse than a slow one. Infinity would run
    until the unrelated per-test timeout. The old max(RACE_TIMEOUT, 10.0) floor
    used to make this unreachable; validating here restores that guarantee."""
    try:
        secs = float(text)
    except ValueError:
        raise argparse.ArgumentTypeError(f'not a number: {text!r}')
    if not math.isfinite(secs) or secs <= 0:
        raise argparse.ArgumentTypeError(
            f'must be a finite positive number of seconds, got {text!r}; '
            'a zero/negative/NaN budget would make every race test pass '
            'without running its race')
    return secs


def parse_args():
    p = argparse.ArgumentParser(description='Run rsync test suite')
    p.add_argument('tests', nargs='*', metavar='TEST',
                   help='Test names or patterns to run (default: all)')
    p.add_argument('--exclude', default=None, metavar='LIST',
                   help='Comma-separated test names/globs to skip entirely: '
                        'they are not run and not reported as skipped. Useful '
                        'for tests that cannot work in a given build/CI '
                        'environment (e.g. a restricted buildd chroot). '
                        'Falls back to the RSYNC_EXCLUDE environment variable.')
    p.add_argument('-j', '--parallel', type=int, default=1, metavar='N',
                   help='Run up to N tests in parallel (default: 1)')
    p.add_argument('--valgrind', action='store_true',
                   help='Run rsync under valgrind (logs to per-process files)')
    p.add_argument('--valgrind-opts', default='', metavar='OPTS',
                   help='Extra valgrind options (e.g. "--leak-check=full")')
    p.add_argument('--preserve-scratch', action='store_true',
                   help='Keep scratch directories after tests complete')
    p.add_argument('--log-level', type=int, default=1, metavar='N',
                   help='Verbosity level 1-10 (default: 1)')
    p.add_argument('--always-log', action='store_true',
                   help='Show test logs even for passing tests')
    p.add_argument('--stop-on-fail', action='store_true',
                   help='Stop after first test failure')
    p.add_argument('--timing', action='store_true',
                   help='After the run, report each test\'s wall-clock time, '
                        'slowest first. With -j N the report also shows how '
                        'much of the run the slowest test alone accounts for.')
    p.add_argument('--timeout', type=int, default=300, metavar='SECS',
                   help='Per-test timeout in seconds (default: 300)')
    p.add_argument('--race-timeout', type=_race_seconds, default=None, metavar='SECS',
                   help='Budget (seconds) a TOCTOU symlink-race test may spend '
                        'trying to win its race before concluding. Overrides '
                        'every such test\'s own default (5-15s, the suite\'s '
                        'slowest tests: a race test always spends its whole '
                        'budget). Lowering it speeds the suite up but weakens '
                        'the oracle. Unset: each test keeps its default.')
    p.add_argument('--rsync-bin', default=None, metavar='PATH',
                   help='Path to rsync binary (default: ./rsync)')
    p.add_argument('--rsync-bin2', default=None, metavar='PATH',
                   help='Path to a second ("peer") rsync binary used for the '
                        'daemon side and remote-shell --rsync-path. Lets the '
                        'suite mix two rsync versions over the wire. Default: '
                        'same as --rsync-bin (no version mixing).')
    p.add_argument('--tooldir', default=None, metavar='DIR',
                   help='Tool/build directory (default: cwd)')
    p.add_argument('--srcdir', default=None, metavar='DIR',
                   help='Source directory (default: script directory)')
    p.add_argument('--protocol', type=int, default=None, metavar='VER',
                   help='Force protocol version (adds --protocol=VER to rsync)')
    p.add_argument('--expect-skipped', default=None, metavar='LIST',
                   help='Comma-separated list of expected-skipped tests. An '
                        '@FILE entry reads a skip list (one test per line, '
                        '"#" comments); relative paths resolve against srcdir '
                        'and several may be composed, e.g. '
                        '@testsuite/skiplist/linux.txt,@testsuite/skiplist/proto29.txt. '
                        'A -NAME entry removes a name the rest of the spec '
                        'added, for a host that can really run a test its '
                        'platform list expects to skip.')
    p.add_argument('--expect-result', default=None, metavar='FILE',
                   help='Path to an expected-outcome manifest (one '
                        '"<testname> <pass|skip|fail|xfail>" per line). When '
                        'set, ONLY the tests listed in FILE are run, and each '
                        "test's actual outcome is compared against its "
                        'expected one; any mismatch (including an unexpected '
                        'pass) fails the run. Used for version-mixing CI.')
    p.add_argument('--daemon-tests-only', action='store_true',
                   help='Run only the tests that can reach the daemon '
                        'transport. Intended for a --use-tcp pass that follows '
                        'a full default-transport run: the tests this drops '
                        'never call start_test_daemon(), so they cannot observe '
                        '--use-tcp and would just repeat themselves. Disables '
                        'the expected-skip oracle (it describes a full run).')
    p.add_argument('--use-tcp', action='store_true',
                   help='Run daemon tests against a real rsyncd bound to '
                        '127.0.0.1 (non-default). The default is the secure '
                        'stdio-pipe transport, which opens no listening '
                        'socket; --use-tcp exposes a loopback port for the '
                        'duration of each daemon test.')
    return p.parse_args()


def find_setfacl_nodef(scratchbase):
    """Determine the setfacl command to remove default ACLs."""
    for cmd in [
        ['setacl', '-k', 'u::7,g::5,o:5', scratchbase],
        ['setfacl', '-k', scratchbase],
        ['setfacl', '-s', 'u::7,g::5,o:5', scratchbase],
    ]:
        try:
            subprocess.run(cmd, capture_output=True, timeout=5)
            return cmd[:2] if cmd[0] == 'setacl' else cmd[:2]
        except (FileNotFoundError, subprocess.TimeoutExpired):
            continue
    try:
        r = subprocess.run(['setfacl', '--help'], capture_output=True, text=True, timeout=5)
        if '-k,' in r.stdout or '-k,' in r.stderr:
            return ['setfacl', '-k']
    except (FileNotFoundError, subprocess.TimeoutExpired):
        pass
    return None


def get_tls_args(config_h):
    """Determine TLS_ARGS from config.h."""
    args = ''
    try:
        with open(config_h) as f:
            text = f.read()
        if '#define HAVE_LUTIMES 1' in text:
            args += ' -l'
        if '#undef CHOWN_MODIFIES_SYMLINK' in text:
            args += ' -L'
    except FileNotFoundError:
        pass
    return args.strip()


def read_shconfig(path):
    """Read shell config variables from shconfig."""
    env = {}
    try:
        with open(path) as f:
            for line in f:
                line = line.strip()
                if line.startswith('#') or line.startswith('export') or not line:
                    continue
                if '=' in line:
                    k, _, v = line.partition('=')
                    env[k.strip()] = v.strip().strip('"')
    except FileNotFoundError:
        pass
    return env


def get_testuser():
    """Determine the current test user."""
    for cmd in ['/usr/bin/whoami', '/usr/ucb/whoami', '/bin/whoami']:
        if os.path.isfile(cmd):
            try:
                return subprocess.check_output([cmd], text=True).strip()
            except subprocess.CalledProcessError:
                pass
    try:
        return subprocess.check_output(['id', '-un'], text=True).strip()
    except (FileNotFoundError, subprocess.CalledProcessError):
        return os.environ.get('LOGNAME', os.environ.get('USER', 'UNKNOWN'))


def _move_aside(path):
    """Rename an un-removable directory to a unique sibling so its name is free.

    A rename-storm symlink-race test can corrupt a directory on some filesystems
    (OpenBSD FFS soft-updates can leave an "empty" dir that still reports
    ENOTEMPTY/EPERM and only fsck clears). `rm -rf` then can't remove it, but
    renaming the top dir aside succeeds even with a corrupted descendant, freeing
    the original name for a clean scratchdir."""
    n = 0
    while os.path.exists(f"{path}.corrupt.{os.getpid()}.{n}"):
        n += 1
    try:
        os.rename(path, f"{path}.corrupt.{os.getpid()}.{n}")
    except OSError:
        pass


def prep_scratch(scratchdir, srcdir, tooldir, setfacl_nodef):
    """Prepare a scratch directory for a test."""
    if os.path.isdir(scratchdir):
        subprocess.run(['chmod', '-R', 'u+rwX', scratchdir], capture_output=True)
        subprocess.run(['rm', '-rf', scratchdir], capture_output=True)
        if os.path.isdir(scratchdir):
            _move_aside(scratchdir)   # rm -rf left corrupted debris; don't inherit it
    os.makedirs(scratchdir, exist_ok=True)
    if setfacl_nodef:
        subprocess.run(setfacl_nodef + [scratchdir], capture_output=True)
    try:
        os.chmod(scratchdir, os.stat(scratchdir).st_mode & ~0o2000)  # clear setgid
    except OSError:
        pass
    src_link = os.path.join(scratchdir, 'src')
    if not os.path.exists(src_link):
        if os.path.isabs(srcdir):
            os.symlink(srcdir, src_link)
        else:
            os.symlink(os.path.join(tooldir, srcdir), src_link)


# Python tests are identified by a positive "_test.py" suffix so that
# helper modules (e.g. rsyncfns.py) sit in testsuite/ without being mistaken
# for tests.
_PY_TEST_SUFFIX = '_test.py'


def _is_test_path(path):
    return os.path.basename(path).endswith(_PY_TEST_SUFFIX)


def _testbase(path):
    """Strip the test extension to get the canonical test name."""
    base = os.path.basename(path)
    if base.endswith(_PY_TEST_SUFFIX):
        return base[:-len(_PY_TEST_SUFFIX)]
    return base


def collect_tests(suitedir, patterns):
    """Collect test scripts (_test.py) matching the given patterns."""
    if not patterns:
        candidates = glob.glob(os.path.join(suitedir, '*' + _PY_TEST_SUFFIX))
        tests = sorted(p for p in candidates if _is_test_path(p))
    else:
        seen = set()
        tests = []
        for pat in patterns:
            # Accept either bare name ("mkpath"), explicit extension, or glob.
            if pat.endswith('.py'):
                pats = [pat]
            else:
                pats = [pat + _PY_TEST_SUFFIX]
            for p in pats:
                for m in sorted(glob.glob(os.path.join(suitedir, p))):
                    if _is_test_path(m) and m not in seen:
                        seen.add(m)
                        tests.append(m)
    return tests


# Tokens through which a test can reach the daemon transport. --use-tcp works by
# setting RSYNC_TEST_USE_TCP, which is read in exactly one place (rsyncfns
# USE_TCP) and acted on in exactly one function (start_test_daemon): a test whose
# source mentions none of these never gets there, so it behaves identically with
# and without --use-tcp and running it a second time under TCP buys no coverage.
#
# The list is the closure of every rsyncfns helper that reaches USE_TCP,
# start_rsyncd or claim_ports, plus the helper modules that open a daemon
# connection themselves and the bare literals a test might use directly. It is
# deliberately over-broad: a false positive only costs runtime, while a false
# negative would silently drop real coverage.
_DAEMON_API = (
    'USE_TCP', 'require_tcp', 'start_test_daemon', 'start_rsyncd',
    'claim_ports', 'claim_free_port', 'setup_chroot_inner',
    'stdio_daemon', 'rsync_proto', 'DaemonClient',
    'rsync://', '--daemon', 'rsyncd',
)


def select_daemon_tests(tests):
    """Split `tests` into (daemon-transport tests, the rest).

    Used by --daemon-tests-only so a TCP pass need not re-run the whole suite.
    A test we cannot read is kept, not dropped -- the failure mode of this
    filter must always be "ran too much"."""
    keep, dropped = [], []
    for path in tests:
        try:
            with open(path, errors='replace') as f:
                text = f.read()
        except OSError:
            keep.append(path)
            continue
        (keep if any(tok in text for tok in _DAEMON_API) else dropped).append(path)
    return keep, dropped


_VALID_OUTCOMES = ('pass', 'skip', 'fail', 'xfail')


def parse_expect_result(path):
    """Parse an expected-outcome manifest into {testbase: outcome}.

    One "<testname> <outcome>" entry per line; '#' comments and blank lines
    are ignored. outcome is one of pass|skip|fail|xfail. The set of listed
    tests doubles as the run set (see main()). Exits 2 on a malformed file.
    """
    expect = {}
    with open(path) as f:
        for lineno, raw in enumerate(f, 1):
            line = raw.split('#', 1)[0].strip()
            if not line:
                continue
            fields = line.split()
            if len(fields) != 2 or fields[1] not in _VALID_OUTCOMES:
                sys.stderr.write(
                    f"{path}:{lineno}: expected '<testname> "
                    f"<{'|'.join(_VALID_OUTCOMES)}>', got: {raw.rstrip()}\n"
                )
                sys.exit(Exit.ERROR)
            expect[fields[0]] = fields[1]
    return expect


def expand_skip_spec(spec, srcdir, suitedir):
    """Expand an RSYNC_EXPECT_SKIPPED spec into a normalised csv.

    The spec is a comma-separated list of test names, '@FILE' skip-list
    references, and '-name' removals.  A skip-list file holds one test name per line ('#' starts a
    comment; blank lines are ignored), which is what keeps two branches from
    colliding: adding a test edits one line of one file rather than a shared
    3 KB csv.  Several may be composed, e.g.
        RSYNC_EXPECT_SKIPPED=@testsuite/skiplist/linux.txt,@.../proto29.txt
    Relative paths resolve against srcdir (not the cwd) so out-of-tree builds
    and `make installcheck` work.  A '-name' entry removes a name the rest of
    the spec added, for a host that can genuinely run a test its platform list
    expects to skip; it is applied last, and must actually remove something.

    Entries must name a real test, and each file must be non-empty, sorted and
    free of duplicates: unsorted files defeat the point (everyone appends to
    the same last line), and a stale name would otherwise fail as a skip
    mismatch far from its cause.  Exits 2 on any of those -- nothing malformed
    may quietly shrink the expected set, which would disarm the oracle.
    """
    def die(msg):
        sys.stderr.write(msg + '\n')
        sys.exit(Exit.ERROR)

    # An entirely empty spec is the legitimate "expect no skips at all".  An
    # empty entry *within* a spec is not: it is what an unset shell variable
    # expands to, and silently dropping it would quietly shrink the expected
    # set.
    if not spec.strip():
        return ''

    names = []
    drop = []
    for tok in (t.strip() for t in spec.split(',')):
        if not tok:
            die('RSYNC_EXPECT_SKIPPED: empty entry (an unset variable?): '
                f'{spec!r}')
        if tok.startswith('-'):
            # '-name' removes a name a composed list added, for a host that
            # really can run a test its platform list expects to skip (e.g. a
            # scratch dir on a second filesystem makes a cross-device copy
            # work).  Subtraction cannot be done by whoever composes the spec,
            # because the name lives inside an @FILE that is only expanded
            # here.  Applied after every addition, so order does not matter.
            drop.append((tok[1:], tok))
            continue
        if not tok.startswith('@'):
            names.append((tok, 'RSYNC_EXPECT_SKIPPED'))
            continue
        path = tok[1:]
        if not os.path.isabs(path):
            path = os.path.join(srcdir, path)
        try:
            with open(path) as f:
                lines = f.readlines()
        except (OSError, UnicodeDecodeError) as e:
            die(f'{tok}: cannot read skip list: {e}')
        prev = None
        found = 0
        for lineno, raw in enumerate(lines, 1):
            name = raw.split('#', 1)[0].strip()
            if not name:
                continue
            where = f'{path}:{lineno}'
            if len(name.split()) != 1:
                die(f'{where}: expected one test name per line, got: {raw.rstrip()}')
            if prev is not None and name <= prev:
                die(f'{where}: skip lists must be sorted and duplicate-free '
                    f'({name!r} follows {prev!r})')
            prev = name
            found += 1
            names.append((name, where))
        # A truncated or emptied list must not read as "expect no skips".
        if not found:
            die(f'{path}: skip list contains no test names')

    seen = {}
    for name, where in names:
        if name in seen:
            continue
        seen[name] = where
        # A plain test name: no path, and no comma (which the expanded csv,
        # and the summary the fleet parses back, use as the separator).
        if ',' in name or '/' in name or os.sep in name or name in ('.', '..'):
            die(f'{where}: not a test name: {name!r}')
        if not os.path.isfile(os.path.join(suitedir, name + '_test.py')):
            die(f'{where}: no such test: {name}')
    for name, tok in drop:
        # Every removal must remove something.  A name the spec never added is
        # stale (the list stopped expecting that skip, or it is misspelled), and
        # a repeated removal is the same no-op written twice.  Shrinking the
        # expected set is precisely what must not happen quietly, so neither is
        # allowed to sit in a config unnoticed.
        if name not in seen:
            die(f'RSYNC_EXPECT_SKIPPED: {tok!r} removes a name that nothing '
                f'added: {name}')
        del seen[name]
    return ','.join(sorted(seen))


_TIMING_TOP = 25


def print_timing_report(durations, outcomes, run_wall, parallel):
    """Report per-test wall-clock, slowest first (--timing).

    With -j N the run cannot finish sooner than its slowest single test, so the
    tail matters as much as the total: a 300s test pins the whole suite to 300s
    no matter how many workers there are. The footer gives both bounds -- the
    serial sum (what one worker would take) and that floor -- so it is obvious
    whether a slow run wants more parallelism or a faster individual test."""
    if not durations:
        return
    ranked = sorted(durations.items(), key=lambda kv: kv[1], reverse=True)
    total = sum(durations.values())
    print(f'----- slowest tests (of {len(ranked)}, wall-clock each):')
    for name, secs in ranked[:_TIMING_TOP]:
        print(f'      {secs:7.1f}s  {name:<32} {outcomes.get(name, "?")}')
    print(f'      serial sum {total:.0f}s over {len(ranked)} tests; '
          f'run took {run_wall:.0f}s with -j{parallel}')
    slowest, slowest_secs = ranked[0]
    if parallel > 1:
        # The floor: even with unlimited workers the suite cannot beat its
        # longest single test.
        print(f'      floor {slowest_secs:.0f}s ({slowest}) = '
              f'{100.0 * slowest_secs / run_wall:.0f}% of this run; '
              f'ideal at -j{parallel} is {total / parallel:.0f}s')


def outcome_of(result):
    """Map a per-test exit code to an outcome string."""
    if result == Exit.PASS:
        return 'pass'
    if result == Exit.SKIP:
        return 'skip'
    if result == Exit.XFAIL:
        return 'xfail'
    return 'fail'


def build_rsync_cmd(rsync_bin, args, scratchbase):
    """Build the RSYNC command string for tests."""
    parts = []
    if args.valgrind:
        # Logs go in a world-writable+sticky subdir so that rsync children
        # which drop privileges (the setpriv cap-drop in partial_nowrite, a
        # daemon dropping to the module's uid) can still create their log file
        # even when scratchbase itself is root-owned.
        vgdir = os.path.join(scratchbase, 'valgrind-logs')
        os.makedirs(vgdir, exist_ok=True)
        os.chmod(vgdir, 0o1777)
        vlog = os.path.join(vgdir, 'valgrind.%p.log')
        vopts = f'--log-file={vlog}'
        supp = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                            'testsuite', 'valgrind.supp')
        if os.path.exists(supp):
            vopts += f' --suppressions={supp}'
        if args.valgrind_opts:
            vopts += ' ' + args.valgrind_opts
        parts.append(f'valgrind {vopts}')
    parts.append(rsync_bin)
    if args.protocol is not None:
        parts.append(f'--protocol={args.protocol}')
    return ' '.join(parts)


class TestResult:
    """Result of a single test execution."""
    __slots__ = ('testbase', 'result', 'output', 'skipped_reason', 'duration')

    def __init__(self, testbase, result, output='', skipped_reason='',
                 duration=0.0):
        self.testbase = testbase
        self.result = result
        self.output = output
        self.skipped_reason = skipped_reason
        self.duration = duration


def run_one_test(testscript, testbase, scratchdir, base_env, timeout,
                 srcdir, tooldir, setfacl_nodef, always_log):
    """Run a single test. Returns a TestResult.

    This function is safe to call from multiple threads — it uses only
    per-test state (unique scratchdir, copy of env).
    """
    started = time.monotonic()
    prep_scratch(scratchdir, srcdir, tooldir, setfacl_nodef)

    env = base_env.copy()
    env['scratchdir'] = scratchdir

    # Dispatch by extension: shell tests via /bin/sh -e, Python tests via
    # the same python3 that's running this runner.
    if testscript.endswith('.py'):
        cmd = [sys.executable, testscript]
    else:
        cmd = ['sh', '-e', testscript]

    logfile = os.path.join(scratchdir, 'test.log')
    with open(logfile, 'w') as log:
        # start_new_session: run the test driver as its own session/group leader
        # so the daemon, clients and flipper it spawns inherit that group. A
        # timeout then killpg's the whole tree (not just the driver), and the
        # lock-file sweep can reap a SIGKILLed run's stranded group the same way.
        proc = subprocess.Popen(
            cmd,
            stdout=log, stderr=subprocess.STDOUT,
            env=env, cwd=env.get('TOOLDIR', '.'),
            start_new_session=True,
        )
        try:
            result = proc.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            # Reap the whole session group, but only if the driver really is its
            # own group leader (start_new_session took) and that group isn't ours
            # -- killpg of our own group would take down the runner.
            try:
                pgid = os.getpgid(proc.pid)
            except OSError:
                pgid = -1
            if pgid == proc.pid and pgid != os.getpgrp():
                try:
                    os.killpg(pgid, signal.SIGKILL)
                except OSError:
                    proc.kill()
            else:
                proc.kill()
            proc.wait()
            result = 1
            log.write(f"\nTIMEOUT: test took over {timeout} seconds\n")

    # Build output text
    output_parts = []

    show_log = always_log or (result not in (Exit.PASS, Exit.SKIP, Exit.XFAIL))
    if show_log:
        output_parts.append(f'----- {testbase} log follows')
        try:
            with open(logfile) as f:
                output_parts.append(f.read().rstrip())
        except FileNotFoundError:
            pass
        output_parts.append(f'----- {testbase} log ends')
        rsyncd_log = os.path.join(scratchdir, 'rsyncd.log')
        if os.path.isfile(rsyncd_log):
            output_parts.append(f'----- {testbase} rsyncd.log follows')
            with open(rsyncd_log) as f:
                output_parts.append(f.read().rstrip())
            output_parts.append(f'----- {testbase} rsyncd.log ends')

    skipped_reason = ''
    if result == Exit.PASS:
        output_parts.append(f'PASS    {testbase}')
    elif result == Exit.SKIP:
        whyfile = os.path.join(scratchdir, 'whyskipped')
        try:
            with open(whyfile) as f:
                skipped_reason = f.read().strip()
        except FileNotFoundError:
            pass
        output_parts.append(f'SKIP    {testbase} ({skipped_reason})')
    elif result == Exit.XFAIL:
        output_parts.append(f'XFAIL   {testbase}')
    else:
        output_parts.append(f'FAIL    {testbase}')

    return TestResult(testbase, result, '\n'.join(output_parts), skipped_reason,
                      time.monotonic() - started)


# Lock for serializing output in parallel mode
_print_lock = threading.Lock()


def main():
    args = parse_args()

    # Also accept legacy environment variables
    if args.preserve_scratch or os.environ.get('preserve_scratch') == 'yes':
        args.preserve_scratch = True
    if args.log_level == 1:
        args.log_level = int(os.environ.get('loglevel', '1'))
    if args.expect_skipped is None:
        args.expect_skipped = os.environ.get('RSYNC_EXPECT_SKIPPED', 'IGNORE')
    if args.exclude is None:
        args.exclude = os.environ.get('RSYNC_EXCLUDE', '')
    if os.environ.get('whichtests'):
        args.tests = [os.environ['whichtests']]

    # Determine directories
    tooldir = args.tooldir or os.environ.get('TOOLDIR') or os.getcwd()
    script_path = os.path.dirname(os.path.abspath(__file__))
    srcdir = args.srcdir or script_path
    if not srcdir or srcdir == '.':
        srcdir = tooldir
    rsync_bin = args.rsync_bin or os.environ.get('rsync_bin') or os.path.join(tooldir, 'rsync')
    # Absolutize: tests run with subprocess(cwd=TOOLDIR) below, so a relative
    # argv[0] would re-resolve against TOOLDIR rather than the runner's
    # invocation cwd, breaking --rsync-bin=../foo/rsync forms.  abspath()
    # captures os.getcwd() now, which is what the operator intended.
    if rsync_bin and not os.path.isabs(rsync_bin):
        rsync_bin = os.path.abspath(rsync_bin)

    # Optional second ("peer") binary for the daemon / remote-shell side, so a
    # run can mix two rsync versions. Defaults to rsync_bin -> no mixing.
    rsync_bin2 = args.rsync_bin2 or os.environ.get('rsync_bin2') or rsync_bin
    if rsync_bin2 and not os.path.isabs(rsync_bin2):
        rsync_bin2 = os.path.abspath(rsync_bin2)

    suitedir = os.path.join(srcdir, 'testsuite')
    if args.expect_skipped != 'IGNORE':
        args.expect_skipped = expand_skip_spec(args.expect_skipped, srcdir, suitedir)
    scratchbase = os.path.join(os.environ.get('scratchbase', tooldir), 'testtmp')
    os.makedirs(scratchbase, exist_ok=True)

    shconfig = read_shconfig(os.path.join(tooldir, 'shconfig'))
    tls_args = get_tls_args(os.path.join(tooldir, 'config.h'))
    setfacl_nodef = find_setfacl_nodef(scratchbase)
    rsync_cmd = build_rsync_cmd(rsync_bin, args, scratchbase)
    rsync_peer_cmd = build_rsync_cmd(rsync_bin2, args, scratchbase)

    if not os.path.isfile(rsync_bin):
        sys.stderr.write(f"rsync_bin {rsync_bin} is not a file\n")
        sys.exit(Exit.ERROR)
    if not os.path.isfile(rsync_bin2):
        sys.stderr.write(f"rsync_bin2 {rsync_bin2} is not a file\n")
        sys.exit(Exit.ERROR)
    if not os.path.isdir(srcdir):
        sys.stderr.write(f"srcdir {srcdir} is not a directory\n")
        sys.exit(Exit.ERROR)

    # Helper programs the test scripts invoke directly. Missing any of these
    # would cause many tests to fail with confusing "not found" errors, so
    # check up front and point the user at the make target that builds them.
    required_helpers = ['tls', 'trimslash', 't_unsafe', 't_chmod_secure',
                        't_secure_relpath',
                        'wildtest', 'getgroups', 'getfsdev']
    missing = [h for h in required_helpers
               if not os.path.isfile(os.path.join(tooldir, h))]
    if missing:
        sys.stderr.write(
            f"runtests.py: missing test helper program(s) in {tooldir}: "
            f"{', '.join(missing)}\n"
            f"Build them with: make {' '.join(missing)}\n"
            f"or run the full test target: make check\n"
        )
        sys.exit(Exit.ERROR)

    testuser = get_testuser()

    # Print header
    print('=' * 60)
    print(f'{sys.argv[0]} running in {tooldir}')
    print(f'    rsync_bin={rsync_cmd}')
    if rsync_peer_cmd != rsync_cmd:
        print(f'    rsync_peer={rsync_peer_cmd}')
    print(f'    srcdir={srcdir}')
    print(f'    TLS_ARGS={tls_args}')
    print(f'    testuser={testuser}')
    print(f'    os={subprocess.check_output(["uname", "-a"], text=True).strip()}')
    print(f'    preserve_scratch={"yes" if args.preserve_scratch else "no"}')
    if args.valgrind:
        print(f'    valgrind=enabled (logs in valgrind-logs/valgrind.*.log)')
    if args.parallel > 1:
        print(f'    parallel={args.parallel}')
    print(f'    daemon_transport={"tcp (loopback)" if args.use_tcp else "pipe (secure default)"}')
    print(f'    scratchbase={scratchbase}')

    # Build base environment for test scripts
    path = os.environ.get('PATH', '')
    if os.path.isdir('/usr/xpg4/bin'):
        path = '/usr/xpg4/bin:' + path

    # Make the testsuite/ directory importable so Python tests can `import rsyncfns`.
    pythonpath = suitedir
    if os.environ.get('PYTHONPATH'):
        pythonpath = suitedir + os.pathsep + os.environ['PYTHONPATH']

    base_env = os.environ.copy()
    base_env.update({
        'PATH': path,
        'POSIXLY_CORRECT': '1',
        'TOOLDIR': tooldir,
        'srcdir': srcdir,
        'RSYNC': rsync_cmd,
        'RSYNC_PEER': rsync_peer_cmd,
        'TLS_ARGS': tls_args,
        'RUNSHFLAGS': '-e',
        'scratchbase': scratchbase,
        'suitedir': suitedir,
        'TESTRUN_TIMEOUT': str(args.timeout),
        'HOME': scratchbase,
        'PYTHONPATH': pythonpath,
    })
    if args.use_tcp:
        # Opt-in: daemon tests start a real rsyncd on a claimed loopback port.
        # Default (unset) keeps the secure stdio-pipe transport.
        base_env['RSYNC_TEST_USE_TCP'] = '1'
    if args.race_timeout is not None:
        # Only exported when the operator actually passed --race-timeout: its
        # mere presence is what tells a race test to override its own default.
        base_env['race_timeout'] = str(args.race_timeout)
    else:
        # A stale value inherited from the environment would silently override
        # every test's default; the flag is the only way to set this.
        base_env.pop('race_timeout', None)
    for k, v in shconfig.items():
        if v:
            base_env[k] = v
    if setfacl_nodef:
        base_env['setfacl_nodef'] = ' '.join(setfacl_nodef)
    else:
        base_env['setfacl_nodef'] = 'true'
    if args.log_level > 8:
        base_env['RUNSHFLAGS'] = '-e -x'

    # Collect tests
    tests = collect_tests(suitedir, args.tests)
    full_run = len(args.tests) == 0

    # Drop excluded tests entirely (matched by basename against name/glob).
    excl = [e.strip() for e in args.exclude.split(',') if e.strip()]
    if excl:
        before = len(tests)
        tests = [t for t in tests
                 if not any(fnmatch.fnmatch(_testbase(t), pat) for pat in excl)]
        if before != len(tests):
            print(f"Excluding {before - len(tests)} test(s) matching: "
                  f"{', '.join(excl)}")

    # Narrow to the daemon-transport tests. The dropped count is always printed:
    # a pass that silently ran a third of the suite would read in the report as
    # if it had run all of it.
    if args.daemon_tests_only:
        tests, dropped = select_daemon_tests(tests)
        print(f"Daemon-transport tests only: running {len(tests)}, skipping "
              f"{len(dropped)} test(s) that cannot observe the transport")
        # The expected-skip list describes a full run, so it cannot be enforced
        # against a subset -- same rule as naming tests explicitly.
        full_run = False

    # An expected-result manifest defines BOTH the run set (its keys) and the
    # expected per-test outcome (its values). Used for version-mixing runs.
    expect = parse_expect_result(args.expect_result) if args.expect_result else None
    if expect is not None:
        have = {_testbase(t) for t in tests}
        unknown = sorted(k for k in expect if k not in have)
        if unknown:
            sys.stderr.write(
                "runtests.py: --expect-result lists test(s) with no matching "
                f"test file (ignored): {', '.join(unknown)}\n"
            )
        tests = [t for t in tests if _testbase(t) in expect]
        full_run = False

    def _cls(outcome):
        """Equivalence class for outcome comparison: fail and xfail both just
        mean 'broke', so a manifest 'fail' matches an actual fail OR xfail."""
        return 'broken' if outcome in ('fail', 'xfail') else outcome

    def mismatch(testbase, actual):
        """True if actual outcome disagrees with the manifest expectation."""
        return expect is not None and _cls(expect[testbase]) != _cls(actual)

    # Record test order for consistent skipped-list output
    test_order = {_testbase(t): i for i, t in enumerate(tests)}

    passed = 0
    failed = 0
    skipped = 0
    xfailed = 0
    skipped_list = []
    outcomes = {}  # testbase -> actual outcome string ('pass'/'skip'/'fail'/'xfail')
    durations = {}  # testbase -> wall-clock seconds (for --timing)

    def process_result(tr):
        """Process a TestResult and update counters. Returns True if the test
        should count as a failure for --stop-on-fail purposes."""
        nonlocal passed, failed, skipped, xfailed
        with _print_lock:
            if tr.output:
                print(tr.output)
        scratchdir = os.path.join(scratchbase, tr.testbase)
        oc = outcome_of(tr.result)
        outcomes[tr.testbase] = oc
        durations[tr.testbase] = tr.duration
        if tr.result == Exit.PASS:
            passed += 1
        elif tr.result == Exit.SKIP:
            skipped_list.append(tr.testbase)
            skipped += 1
        elif tr.result == Exit.XFAIL:
            # XFAIL: an expected failure (a known, documented residual the test
            # asserts against). Reported distinctly but does NOT fail the suite;
            # when the underlying issue is fixed the test returns 0 instead.
            xfailed += 1
        else:
            failed += 1
        if tr.result in (Exit.PASS, Exit.SKIP, Exit.XFAIL) and not args.preserve_scratch \
                and os.path.isdir(scratchdir):
            subprocess.run(['rm', '-rf', scratchdir], capture_output=True)
        # With a manifest, only a mismatch is a "failure" (an expected fail is
        # fine); without one, any non-pass/non-skip/non-xfail result is a failure.
        if expect is not None:
            return mismatch(tr.testbase, oc)
        return tr.result not in (Exit.PASS, Exit.SKIP, Exit.XFAIL)

    run_started = time.monotonic()

    if args.parallel > 1:
        # Parallel execution
        with concurrent.futures.ThreadPoolExecutor(max_workers=args.parallel) as executor:
            futures = {}
            for testscript in tests:
                testbase = _testbase(testscript)
                scratchdir = os.path.join(scratchbase, testbase)
                timeout = 600 if ('hardlinks' in testbase or testbase == 'variety') else args.timeout
                f = executor.submit(
                    run_one_test, testscript, testbase, scratchdir,
                    base_env, timeout, srcdir, tooldir, setfacl_nodef,
                    args.always_log
                )
                futures[f] = testbase

            for f in concurrent.futures.as_completed(futures):
                tr = f.result()
                is_fail = process_result(tr)
                if is_fail and args.stop_on_fail:
                    # Cancel pending futures
                    for pending in futures:
                        pending.cancel()
                    break
    else:
        # Sequential execution
        for testscript in tests:
            testbase = _testbase(testscript)
            scratchdir = os.path.join(scratchbase, testbase)
            timeout = 600 if ('hardlinks' in testbase or testbase == 'variety') else args.timeout
            tr = run_one_test(
                testscript, testbase, scratchdir,
                base_env, timeout, srcdir, tooldir, setfacl_nodef,
                args.always_log
            )
            is_fail = process_result(tr)
            if is_fail and args.stop_on_fail:
                break

    run_wall = time.monotonic() - run_started

    # Check valgrind logs for errors
    vg_errors = 0
    if args.valgrind:
        for vlog in sorted(glob.glob(os.path.join(scratchbase, 'valgrind-logs', 'valgrind.*.log'))):
            try:
                with open(vlog) as f:
                    content = f.read()
                for line in content.splitlines():
                    if 'ERROR SUMMARY:' in line and 'ERROR SUMMARY: 0 errors' not in line:
                        vg_errors += 1
                        print(f'----- valgrind errors in {os.path.basename(vlog)}:')
                        print(content)
                        break
            except FileNotFoundError:
                pass

    # Summary
    print('-' * 60)
    print('----- overall results:')
    print(f'      {passed} passed')
    if failed > 0:
        print(f'      {failed} failed')
    if xfailed > 0:
        print(f'      {xfailed} xfailed (expected)')
    if skipped > 0:
        print(f'      {skipped} skipped')
    if vg_errors > 0:
        print(f'      {vg_errors} valgrind error(s) found (see logs in {os.path.join(scratchbase, "valgrind-logs")})')

    if args.timing:
        print_timing_report(durations, outcomes, run_wall, args.parallel)

    if expect is not None:
        # Version-mixing mode: the run is judged purely on whether each test's
        # actual outcome matched its manifest expectation. An expected 'fail'
        # is fine; an UNEXPECTED pass (xpass) or any other divergence is not.
        mismatches = []
        for tb in sorted(expect, key=lambda x: test_order.get(x, 1 << 30)):
            actual = outcomes.get(tb, 'notrun')
            if actual == 'notrun' or mismatch(tb, actual):
                mismatches.append((tb, expect[tb], actual))
        if mismatches:
            print('----- expected-result mismatches:')
            for tb, want, got in mismatches:
                tag = ' (xpass)' if _cls(want) == 'broken' and got == 'pass' else ''
                print(f'      {tb}: expected {want}, got {got}{tag}')
        print('-' * 60)
        exit_code = len(mismatches) + vg_errors
        print(f'overall result is {exit_code}')
        sys.exit(exit_code)

    skipped_str = ','.join(sorted(skipped_list, key=lambda x: test_order.get(x, 0)))
    if full_run and args.expect_skipped != 'IGNORE':
        print('----- skipped results:')
        print(f'      expected: {args.expect_skipped}')
        print(f'      got:      {skipped_str}')
    else:
        skipped_str = ''
        args.expect_skipped = ''

    print('-' * 60)

    exit_code = failed + vg_errors
    if exit_code == 0:
        # Compare the skipped set order-insensitively: which tests skipped is
        # what matters, not the order runtests happened to collect them in
        # (that order is just sorted filenames -- an easy thing to get subtly
        # wrong when maintaining the per-platform expected lists).
        got = set(s for s in skipped_str.split(',') if s)
        want = set(s for s in args.expect_skipped.split(',') if s)
        if got != want:
            exit_code = 1

    print(f'overall result is {exit_code}')
    sys.exit(exit_code)


if __name__ == '__main__':
    main()