3proxy/tests/run.py
Vladimir Dubrovin 6428c4423f tests: reload cases that hold on slow runners
The downloads spanning a reload asked httpsrv for 64 KB blocks 60 ms
apart, meant as two seconds; but httpsrv sends at most 8 KB a block, so
they took 245 waits, some 15 seconds, and on the CI runners, whose
timers stretch, more than the 20 the cases waited. They now ask for
400 KB in 8 KB blocks 40 ms apart, two seconds indeed, and are waited
for up to a minute.

The UDP part of the takeover case slept 1.5 s after the reload adding
its udppm and then counted every datagram unanswered as lost; it now
asks until the service answers first, and wants more than 20
datagrams through the reloads rather than 100, what a slow runner can
send. That none is lost is the check.

On a failure, run.py prints every log line with a result code other
than 0, not only the last lines, so the reason a connection ended is
shown wherever it was logged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 11:25:07 +03:00

160 lines
5.7 KiB
Python

#!/usr/bin/env python3
"""Run the 3proxy regression tests.
python3 tests/run.py every case
python3 tests/run.py httpsrv cases whose name matches
python3 tests/run.py --bin build/bin/3proxy
python3 tests/run.py --keep leave the temporary files behind
Each case under tests/cases/ defines the configurations it needs and the
positive and negative scenarios expected from them.
"""
import argparse
import importlib.util
import os
import re
import shutil
import sys
import tempfile
import traceback
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from harness import Failure, Tester # noqa: E402
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def default_binary():
"""Find a built 3proxy: the Makefiles put it in bin/, CMake in build/bin/,
and multi-configuration generators one level below that again."""
name = "3proxy.exe" if os.name == "nt" else "3proxy"
candidates = [os.path.join(ROOT, "bin", name),
os.path.join(ROOT, "build", "bin", name)]
for config in ("Release", "Debug", "RelWithDebInfo", "MinSizeRel"):
candidates.append(os.path.join(ROOT, "build", "bin", config, name))
for candidate in candidates:
if os.path.isfile(candidate):
return candidate
return candidates[0]
def load_case(path):
name = os.path.splitext(os.path.basename(path))[0]
spec = importlib.util.spec_from_file_location("case_" + name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return name, module
def main():
parser = argparse.ArgumentParser()
parser.add_argument("pattern", nargs="?", default="",
help="only run cases whose name contains this")
parser.add_argument("--bin", dest="binary", default=None,
help="the 3proxy binary to test")
parser.add_argument("--keep", action="store_true",
help="keep the temporary directory")
parser.add_argument("-v", "--verbose", action="store_true",
help="print every check, not just the failures")
args = parser.parse_args()
binary = args.binary or os.environ.get("BIN") or default_binary()
binary = os.path.abspath(binary)
if not os.path.isfile(binary):
print(f"no 3proxy binary at {binary} (build first, or pass --bin)",
file=sys.stderr)
return 2
case_dir = os.path.join(ROOT, "tests", "cases")
paths = sorted(os.path.join(case_dir, f) for f in os.listdir(case_dir)
if f.endswith(".py") and not f.startswith("_"))
paths = [p for p in paths if args.pattern in os.path.basename(p)]
if not paths:
print(f"no cases matched {args.pattern!r}", file=sys.stderr)
return 2
tmpdir = tempfile.mkdtemp(prefix="3proxy-tests.")
print(f"3proxy tests: {binary}")
print(f"working in: {tmpdir}\n")
passed = failed = skipped = 0
failures = []
try:
for path in paths:
name, module = load_case(path)
print(f" {name}")
tester = Tester(binary, tmpdir, name)
error = None
try:
module.run(tester)
except Failure as exc:
error = str(exc)
except Exception:
error = traceback.format_exc()
finally:
tester.stop_all()
for status, label, expected, actual in tester.checks:
if status is None:
skipped += 1
print(f" skip {label}")
elif status:
passed += 1
if args.verbose:
print(f" ok {label}")
else:
failed += 1
failures.append(f"{name}: {label}")
print(f" FAIL {label}")
if expected is not None:
print(f" expected: {expected}")
if actual is not None:
print(f" actual: {actual}")
if tester.checks and any(status is False for status, _, _, _ in tester.checks):
for name, text in tester.logs:
lines = [line for line in text.splitlines() if line.strip()]
if not lines:
continue
print(f" --- {name} said ---")
# why connections ended: the lines with a result code,
# wherever they are, then the last few
ended = [line for line in lines
if re.match(r"\d+\.\d+ \S+ (?!00000 )\d{5} ", line)]
for line in ended[-20:]:
print(f" {line}")
if ended:
print(" ...")
for line in lines[-12:]:
print(f" {line}")
if error:
failed += 1
failures.append(f"{name}: case aborted")
print(" ERROR the case could not finish:")
for line in error.rstrip().splitlines():
print(f" {line}")
print()
finally:
if args.keep:
print(f"temporary files left in {tmpdir}")
else:
shutil.rmtree(tmpdir, ignore_errors=True)
print("-" * 41)
total = passed + failed
summary = f"cases: {len(paths)} checks: {total} passed: {passed} failed: {failed}"
if skipped:
summary += f" skipped: {skipped}"
print(summary)
for item in failures:
print(f" FAIL {item}")
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())