import sys
import threading
+# 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 parse_args():
p = argparse.ArgumentParser(description='Run rsync test suite')
help='Stop after first test failure')
p.add_argument('--timeout', type=int, default=300, metavar='SECS',
help='Per-test timeout in seconds (default: 300)')
+ p.add_argument('--race-timeout', type=float, default=5.0, metavar='SECS',
+ help='Budget (seconds) a TOCTOU symlink-race test may spend '
+ 'trying to win its race before concluding (default: 5)')
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',
f"{path}:{lineno}: expected '<testname> "
f"<{'|'.join(_VALID_OUTCOMES)}>', got: {raw.rstrip()}\n"
)
- sys.exit(2)
+ sys.exit(Exit.ERROR)
expect[fields[0]] = fields[1]
return expect
def outcome_of(result):
"""Map a per-test exit code to an outcome string."""
- if result == 0:
+ if result == Exit.PASS:
return 'pass'
- if result == 77:
+ if result == Exit.SKIP:
return 'skip'
- if result == 78:
+ if result == Exit.XFAIL:
return 'xfail'
return 'fail'
# Build output text
output_parts = []
- show_log = always_log or (result not in (0, 77, 78))
+ 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:
output_parts.append(f'----- {testbase} rsyncd.log ends')
skipped_reason = ''
- if result == 0:
+ if result == Exit.PASS:
output_parts.append(f'PASS {testbase}')
- elif result == 77:
+ elif result == Exit.SKIP:
whyfile = os.path.join(scratchdir, 'whyskipped')
try:
with open(whyfile) as f:
except FileNotFoundError:
pass
output_parts.append(f'SKIP {testbase} ({skipped_reason})')
- elif result == 78:
+ elif result == Exit.XFAIL:
output_parts.append(f'XFAIL {testbase}')
else:
output_parts.append(f'FAIL {testbase}')
if not os.path.isfile(rsync_bin):
sys.stderr.write(f"rsync_bin {rsync_bin} is not a file\n")
- sys.exit(2)
+ 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(2)
+ sys.exit(Exit.ERROR)
if not os.path.isdir(srcdir):
sys.stderr.write(f"srcdir {srcdir} is not a directory\n")
- sys.exit(2)
+ 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
f"Build them with: make {' '.join(missing)}\n"
f"or run the full test target: make check\n"
)
- sys.exit(2)
+ sys.exit(Exit.ERROR)
testuser = get_testuser()
'scratchbase': scratchbase,
'suitedir': suitedir,
'TESTRUN_TIMEOUT': str(args.timeout),
+ 'race_timeout': str(args.race_timeout),
'HOME': scratchbase,
'PYTHONPATH': pythonpath,
})
passed = 0
failed = 0
skipped = 0
+ xfailed = 0
skipped_list = []
outcomes = {} # testbase -> actual outcome string ('pass'/'skip'/'fail'/'xfail')
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
+ 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
- if tr.result == 0:
+ if tr.result == Exit.PASS:
passed += 1
- elif tr.result == 77:
+ 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 (0, 77) and not args.preserve_scratch \
+ 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 result is a failure.
+ # 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 (0, 77)
+ return tr.result not in (Exit.PASS, Exit.SKIP, Exit.XFAIL)
if args.parallel > 1:
# Parallel execution
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:
ported.
Conventions matching the shell harness:
- * Exit 0 = pass, 1 = fail, 77 = skip, 78 = xfail.
+ * Exit codes (see the Exit enum): 0=pass, 1=fail, 2=error, 77=skip, 78=xfail.
* The runner sets these environment variables before invoking each test:
scratchdir per-test scratch directory
srcdir rsync source directory
import time
from pathlib import Path
+from exitcodes import Exit # re-exported: tests may `from rsyncfns import Exit`
+
# --- environment -----------------------------------------------------------
f"rsyncfns: required environment variable {name} is not set; "
"run this test via runtests.py rather than directly.\n"
)
- sys.exit(2)
+ sys.exit(Exit.ERROR)
return v
def test_fail(msg: str) -> 'None':
sys.stderr.write(msg.rstrip() + '\n')
- sys.exit(1)
+ sys.exit(Exit.FAIL)
def test_skipped(msg: str) -> 'None':
sys.stderr.write(msg.rstrip() + '\n')
(TMPDIR / 'whyskipped').write_text(msg.rstrip() + '\n')
- sys.exit(77)
+ sys.exit(Exit.SKIP)
def test_xfail(msg: str) -> 'None':
sys.stderr.write(msg.rstrip() + '\n')
- sys.exit(78)
+ sys.exit(Exit.XFAIL)
# --- rsync invocation ------------------------------------------------------