"""Own disposable Wine test files, saved output and prefix teardown.""" from contextlib import contextmanager from datetime import datetime, timezone import os from pathlib import Path import selectors import shutil import signal import subprocess import sys import tempfile import time def add_test_options(parser): parser.add_argument("--keep", action="store_true", default=os.environ.get("WINE_ACCESS_TEST_KEEP") == "1", help="retain temporary files even after success") parser.add_argument("--clean-on-failure", dest="keepOnFailure", action="store_false", default=os.environ.get("WINE_ACCESS_TEST_KEEP_ON_FAILURE", "1") != "0", help="remove temporary files after failure too; logs remain") @contextmanager def wine_test_directory(prefix): # The outer runner owns deletion, after all child processes have stopped. # In particular, an inner finally/TemporaryDirectory must not erase the # failed fixture before the runner can retain it for inspection. yield tempfile.mkdtemp(prefix=prefix, dir=os.environ["WINE_ACCESS_TEST_ROOT"]) @contextmanager def test_signals(): def interrupted(signum, frame): raise KeyboardInterrupt(signum) handlers = {signum: signal.signal(signum, interrupted) for signum in (signal.SIGINT, signal.SIGTERM)} try: yield finally: for signum, handler in handlers.items(): signal.signal(signum, handler) def stop_wine_prefixes(testRoot, env, log): prefixes = set() if (testRoot / "prefix").exists(): prefixes.add(testRoot / "prefix") for current, dirs, files in os.walk(testRoot, followlinks=False): if "system.reg" in files or "drive_c" in dirs: prefixes.add(Path(current)) dirs.clear() stopped = True for prefix in sorted(prefixes): # These paths originate in this run's private directory. Never stop a # server selected by the caller's WINEPREFIX or a symlink outside it. if not prefix.resolve().is_relative_to(testRoot): stopped = False continue prefixEnv = dict(env, WINEPREFIX=str(prefix)) try: # -k also returns 1 when the fixture already stopped its server. # -w confirms the final state without treating that as a failure. subprocess.run(["wineserver", "-k"], env=prefixEnv, check=False, timeout=5, stdout=log, stderr=subprocess.STDOUT) subprocess.run(["wineserver", "-w"], env=prefixEnv, check=True, timeout=5, stdout=log, stderr=subprocess.STDOUT) except (OSError, subprocess.SubprocessError) as error: log.write(f"Wine prefix teardown failed: {prefix}: {error}\n".encode()) stopped = False return stopped def run_wine_test(command, buildDir, name, *, timeout=120, keep=False, keepOnFailure=True, minFreeMiB=None): if not name or any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-" for char in name): raise ValueError("test name must contain only letters, digits, underscores or hyphens") if timeout <= 0: raise ValueError("test timeout must be positive") tempBase = Path(tempfile.gettempdir()) if minFreeMiB is None: minFreeMiB = int(os.environ.get("WINE_ACCESS_TEST_MIN_FREE_MIB", "1024")) if minFreeMiB < 0: raise ValueError("minimum free space must not be negative") freeMiB = shutil.disk_usage(tempBase).free // (1024 * 1024) if freeMiB < minFreeMiB: print(f"Wine test {name}: {tempBase} has {freeMiB} MiB free; " f"at least {minFreeMiB} MiB is required. Clean up retained test directories first.", file=sys.stderr, flush=True) return 2 logDir = Path(buildDir).resolve() / "Testing" / "WineAccessLogs" logDir.mkdir(parents=True, exist_ok=True) stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ") logPath = logDir / f"{name}-{stamp}-{os.getpid()}.log" testRoot = Path(tempfile.mkdtemp(prefix=f"wine-access-{name}-", dir=tempBase)).resolve() env = os.environ.copy() env.update(WINE_ACCESS_TEST_ROOT=str(testRoot), TMPDIR=str(testRoot), WINEPREFIX=str(testRoot / "prefix"), CTHULHU_WINE_ACCESS_MANAGED="1", WINEDEBUG=env.get("WINE_ACCESS_TEST_DEBUG", "-all"), WINEDLLOVERRIDES="mscoree,mshtml=", PYTHONUNBUFFERED="1") process = None returnCode = 1 with logPath.open("wb", buffering=0) as log, test_signals(): def report(message): print(message, flush=True) log.write((message + "\n").encode()) report(f"Wine test directory: {testRoot}") report(f"Wine test log: {logPath}") try: process = subprocess.Popen(command, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=True) deadline = time.monotonic() + timeout exitedAt = None with selectors.DefaultSelector() as selector: selector.register(process.stdout, selectors.EVENT_READ) while selector.get_map(): now = time.monotonic() if process.poll() is None and now >= deadline: raise subprocess.TimeoutExpired(command, timeout) if process.poll() is not None: if exitedAt is None: exitedAt = now # Detached Wine services may retain the output pipe. # Prefix teardown below closes them; do not wait forever. if now - exitedAt >= 1: break for key, _ in selector.select(.1): data = os.read(key.fd, 32768) if not data: selector.unregister(key.fileobj) continue log.write(data) if hasattr(sys.stdout, "buffer"): sys.stdout.buffer.write(data) sys.stdout.buffer.flush() else: sys.stdout.write(data.decode(errors="replace")) returnCode = process.wait(timeout=max(.1, deadline - time.monotonic())) if returnCode < 0: returnCode = 128 - returnCode except subprocess.TimeoutExpired: returnCode = 124 report(f"Wine test timed out after {timeout} seconds") except KeyboardInterrupt as error: returnCode = 128 + (error.args[0] if error.args else signal.SIGINT) report("Wine test interrupted") except OSError as error: report(f"Wine test could not run: {error}") finally: # Ignore repeated interrupts during the bounded teardown, then # restore the caller's handlers. Never signal another test's group. for signum in (signal.SIGINT, signal.SIGTERM): signal.signal(signum, signal.SIG_IGN) if process is not None: try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: pass try: process.wait(timeout=1) except subprocess.TimeoutExpired: os.killpg(process.pid, signal.SIGKILL) process.wait(timeout=2) stopped = stop_wine_prefixes(testRoot, env, log) if process is not None: # Save any last diagnostics from services stopped above. os.set_blocking(process.stdout.fileno(), False) while True: try: data = os.read(process.stdout.fileno(), 32768) except BlockingIOError: break if not data: break log.write(data) process.stdout.close() if not stopped: returnCode = returnCode or 1 report("Wine teardown did not finish; retaining temporary files") evidenceSaved = True try: for current, _, files in os.walk(testRoot, followlinks=False): for name in files: source = Path(current) / name if source.suffix != ".log" or source.is_symlink(): continue destination = logDir / logPath.stem / source.relative_to(testRoot) destination.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, destination) except OSError as error: evidenceSaved = False returnCode = returnCode or 1 report(f"Could not save fixture logs; retaining temporary files: {error}") retain = keep or (returnCode != 0 and keepOnFailure) or not stopped or not evidenceSaved if retain: report(f"Retained Wine test directory: {testRoot}") else: try: shutil.rmtree(testRoot) report(f"Removed Wine test directory: {testRoot}") except OSError as error: returnCode = returnCode or 1 report(f"Could not remove Wine test directory: {testRoot}: {error}") report(f"Wine test exit status: {returnCode}; log saved: {logPath}") return returnCode