From 53dd0e8448e843156ce0abe24b71457573f9ddf3 Mon Sep 17 00:00:00 2001 From: Yiteng Niu <6831097+niukuo@users.noreply.github.com> Date: Wed, 22 Jul 2026 21:55:20 +0800 Subject: [PATCH 1/3] [TRTLLMINF-191][infra] Use native pytest capture for S3 logs Signed-off-by: Yiteng Niu <6831097+niukuo@users.noreply.github.com> --- jenkins/L0_Test.groovy | 19 +- jenkins/scripts/slurm_run.sh | 2 + tests/integration/defs/test_unittests.py | 11 +- tests/test_common/s3_output.py | 1278 ++++++----------- tests/test_common/s3_output_hooks.py | 93 +- tests/unittest/test_s3_output.py | 804 ++++++----- .../tools/test_test_to_stage_mapping.py | 35 +- 7 files changed, 919 insertions(+), 1323 deletions(-) diff --git a/jenkins/L0_Test.groovy b/jenkins/L0_Test.groovy index c647275e6de6..a2bf5e02359b 100644 --- a/jenkins/L0_Test.groovy +++ b/jenkins/L0_Test.groovy @@ -47,7 +47,6 @@ ARTIFACT_PATH = env.artifactPath ? env.artifactPath : "sw-tensorrt-generic/llm-a UPLOAD_PATH = env.uploadPath ? env.uploadPath : "sw-tensorrt-generic/llm-artifacts/${JOB_NAME}/${BUILD_NUMBER}" URM_ARTIFACTORY_BASE = "https://urm.nvidia.com/artifactory" ENABLE_UPLOAD_TEST_RESULTS = params.enableUploadTestResults != null ? params.enableUploadTestResults : true -ENABLE_S3_ECHO_STDOUT = params.enableS3EchoStdout != null ? params.enableS3EchoStdout : false X86_64_TRIPLE = "x86_64-linux-gnu" AARCH64_TRIPLE = "aarch64-linux-gnu" @@ -1480,15 +1479,10 @@ def runLLMTestlistWithSbatch(pipeline, platform, testList, config=VANILLA_CONFIG ] if (ENABLE_UPLOAD_TEST_RESULTS) { extraArgs += [ - "-s", + "--capture=fd", "--s3-upload-path=${uploadPath}/${stageName}", + "--s3-upload-mode=deferred", ] - if (ENABLE_S3_ECHO_STDOUT) { - extraArgs += [ - "--s3-echo-stdout", - "--s3-capture-mode=timestamped", - ] - } } def pytestCommand = getPytestBaseCommandLine( llmSrcNode, @@ -4012,15 +4006,10 @@ def runLLMTestlistOnPlatformImpl(pipeline, platform, testList, config=VANILLA_CO def extraArgs = [*clusterDurationsArgs] if (ENABLE_UPLOAD_TEST_RESULTS) { extraArgs += [ - "-s", + "--capture=fd", "--s3-upload-path=${uploadPath}/${stageName}", + "--s3-upload-mode=deferred", ] - if (ENABLE_S3_ECHO_STDOUT) { - extraArgs += [ - "--s3-echo-stdout", - "--s3-capture-mode=timestamped", - ] - } } def pytestCommand = getPytestBaseCommandLine( llmSrc, diff --git a/jenkins/scripts/slurm_run.sh b/jenkins/scripts/slurm_run.sh index e7d5f1a56d2c..90bb972b117d 100755 --- a/jenkins/scripts/slurm_run.sh +++ b/jenkins/scripts/slurm_run.sh @@ -103,6 +103,8 @@ perf_report_exit_code=0 eval $pytestCommand pytest_exit_code=$? echo "Rank${SLURM_PROCID} Pytest finished execution with exit code $pytest_exit_code" +python3 "$llmSrcNode/tests/test_common/s3_output.py" \ + --drain-spool "$jobWorkspace" || true # DEBUG: Diagnose intermittent "unrecognized arguments" failure (Exit Code 4) # Remove this after the issue is resolved diff --git a/tests/integration/defs/test_unittests.py b/tests/integration/defs/test_unittests.py index badcde1f44b3..f51aa32cc6f5 100644 --- a/tests/integration/defs/test_unittests.py +++ b/tests/integration/defs/test_unittests.py @@ -175,12 +175,15 @@ def test_unittests_v2(llm_root, llm_venv, case: str, output_dir, request): command += ["-m", unittest_markexpr] s3_secret_key = None + s3_output_module = None s3_upload_path = request.config.getoption("--s3-upload-path", default=None) if s3_upload_path: + from test_common import s3_output as s3_output_module + inner_output_dir = os.path.join(output_dir, "inner-s3", case_fn) inner_upload_path = os.path.join(s3_upload_path, "inner", case_fn) command += [ - "-s", + "--capture=fd", f"--output-dir={inner_output_dir}", f"--s3-upload-path={inner_upload_path}", "--s3-upload-mode=deferred", @@ -257,6 +260,12 @@ def run_command(cmd, num_workers=1): ) print(f"{'='*60}\n") return False + finally: + if s3_output_module is not None: + s3_output_module.drain_pending_uploads( + inner_output_dir, + secret_key=s3_secret_key, + ) return True if num_workers == 1: diff --git a/tests/test_common/s3_output.py b/tests/test_common/s3_output.py index 4f3f8f272770..6995c1e568fb 100644 --- a/tests/test_common/s3_output.py +++ b/tests/test_common/s3_output.py @@ -13,942 +13,565 @@ # See the License for the specific language governing permissions and # limitations under the License. import argparse -import io +import fcntl +import hashlib +import json import logging import os +import posixpath import re +import socket import sys -import threading import time -from concurrent.futures import ThreadPoolExecutor, as_completed +from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, as_completed, wait from dataclasses import dataclass -from datetime import datetime +from pathlib import Path import pytest -from _pytest.logging import catching_logs logger = logging.getLogger(__name__) +_CAPTURE_SECTION_PATTERN = re.compile(r"^Captured (stdout|stderr|log)(?: (setup|call|teardown))?$") +_STREAM_FILENAMES = { + "stdout": "stdout", + "stderr": "stderr", + "log": "logging", +} +_SPOOL_ROOT_NAME = ".s3-spool" +_SPOOL_CONFIG_NAME = "upload-config.json" +_SPOOL_WRITE_CHARS = 1024 * 1024 +_FAILED_OUTPUT_MAX_LINES = 200 +_FAILED_OUTPUT_MAX_BYTES = 65536 + + +def _spool_root(output_path: str) -> Path: + output_root = Path(os.path.abspath(output_path)) + return output_root.parent / f"{_SPOOL_ROOT_NAME}-{output_root.name}" + class EnvDefault(argparse.Action): def __init__(self, envvar, required=True, default=None, **kwargs): - if envvar: - if envvar in os.environ: - default = os.environ[envvar] + if envvar and envvar in os.environ: + default = os.environ[envvar] if required and default: required = False - super(EnvDefault, self).__init__(default=default, required=required, **kwargs) + super().__init__(default=default, required=required, **kwargs) def __call__(self, parser, namespace, values, option_string=None): setattr(namespace, self.dest, values) @dataclass(frozen=True) -class FileSlice: - path: str - offset: int - size: int - +class PendingUpload: + source_path: str + object_key: str + test_name: str + filename: str + + +def _create_s3_client( + endpoint_url: str, + aws_access_key_id: str, + aws_secret_access_key: str, +): + try: + import boto3 + except ModuleNotFoundError as exc: + raise RuntimeError("boto3 is required to upload test logs") from exc + + return boto3.client( + "s3", + endpoint_url=endpoint_url, + aws_access_key_id=aws_access_key_id, + aws_secret_access_key=aws_secret_access_key, + ) -class FileSliceReader(io.RawIOBase): - def __init__(self, file_slice): - self._slice = file_slice - self._file = open(file_slice.path, "rb", buffering=0) - self._position = 0 - self._file.seek(file_slice.offset) - def readable(self): - return True - - def seekable(self): +def _process_is_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: return True + return True - def tell(self): - return self._position - - def seek(self, offset, whence=os.SEEK_SET): - if whence == os.SEEK_SET: - position = offset - elif whence == os.SEEK_CUR: - position = self._position + offset - elif whence == os.SEEK_END: - position = self._slice.size + offset - else: - raise ValueError(f"Unsupported whence: {whence}") - if position < 0: - raise ValueError("Negative seek position") - self._file.seek(self._slice.offset + position) - self._position = position - return position - - def readinto(self, buffer): - remaining = self._slice.size - self._position - if remaining <= 0: - return 0 - view = memoryview(buffer)[: min(len(buffer), remaining)] - read_size = self._file.readinto(view) - if read_size is None: - return 0 - self._position += read_size - return read_size - - def close(self): - if not self.closed: - self._file.close() - super().close() - - -class SessionFDSpool: - def __init__(self, target_fd, path): - self.target_fd = target_fd - self.path = path - self._saved_fd = None - self._spool_fd = None - self._attached = False - - def _flush_target_stream(self): - stream = sys.stdout if self.target_fd == 1 else sys.stderr - try: - stream.flush() - except (OSError, ValueError): - pass - def start(self): - self._flush_target_stream() - self._saved_fd = os.dup(self.target_fd) +def _remove_empty_parents(path: Path, stop: Path) -> None: + parent = path.parent + while parent != stop.parent: try: - self._spool_fd = os.open( - self.path, - os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_APPEND, - 0o600, - ) - os.dup2(self._spool_fd, self.target_fd, inheritable=True) - self._attached = True + parent.rmdir() except OSError: - if self._spool_fd is not None: - os.close(self._spool_fd) - self._spool_fd = None - os.close(self._saved_fd) - self._saved_fd = None - raise - - def suspend_parent(self): - if not self._attached: - return - self._flush_target_stream() - os.dup2(self._saved_fd, self.target_fd, inheritable=True) - self._attached = False - - def resume_parent(self): - if self._attached or self._spool_fd is None: return - self._flush_target_stream() - os.dup2(self._spool_fd, self.target_fd, inheritable=True) - self._attached = True - - def snapshot(self): - self._flush_target_stream() - return os.fstat(self._spool_fd).st_size - - def stop(self): - self.suspend_parent() - if self._spool_fd is not None: - os.close(self._spool_fd) - self._spool_fd = None - if self._saved_fd is not None: - os.close(self._saved_fd) - self._saved_fd = None - - -class SessionCapture: - def __init__(self, output_path): - spool_dir = os.path.join(output_path, ".s3-spool") - os.makedirs(spool_dir, exist_ok=True) - suffix = f"{os.getpid()}-{time.time_ns()}" - self._spools = { - "stdout.log": SessionFDSpool(1, os.path.join(spool_dir, f"stdout-{suffix}.log")), - "stderr.log": SessionFDSpool(2, os.path.join(spool_dir, f"stderr-{suffix}.log")), - } - self._suspend_depth = 0 - self._started = False - - def start(self): - started = [] - try: - for spool in self._spools.values(): - spool.start() - started.append(spool) - except Exception: - for spool in reversed(started): - spool.stop() - raise - self._started = True - - def snapshot(self): - return {filename: spool.snapshot() for filename, spool in self._spools.items()} - - def slices_since(self, offsets): - current = self.snapshot() - return { - filename: FileSlice( - path=spool.path, - offset=offsets[filename], - size=max(0, current[filename] - offsets[filename]), - ) - for filename, spool in self._spools.items() - } - - def suspend_parent(self): - if not self._started: + if parent == stop: return - self._suspend_depth += 1 - if self._suspend_depth == 1: - for spool in self._spools.values(): - spool.suspend_parent() + parent = parent.parent - def resume_parent(self): - if not self._started or self._suspend_depth == 0: - return - self._suspend_depth -= 1 - if self._suspend_depth == 0: - for spool in self._spools.values(): - spool.resume_parent() - def stop(self): - if not self._started: - return - self._suspend_depth = 0 - for spool in self._spools.values(): - spool.stop() - self._started = False - - def remove_files(self): - for spool in self._spools.values(): - try: - os.remove(spool.path) - except FileNotFoundError: - pass - try: - os.rmdir(os.path.dirname(next(iter(self._spools.values())).path)) - except OSError: - pass - - -class FDRedirector: - def __init__( - self, - target_fd, - log_file_path, - echo_to_original=False, - timestamp_format="%(asctime)s.%(msecs)03d", - date_format="%Y-%m-%d %H:%M:%S", - ): - self.target_fd = target_fd - self.log_file_path = log_file_path - self.echo_to_original = echo_to_original - self.timestamp_format = timestamp_format - self.date_format = date_format - - self.saved_fd = None - self._reader_thread = None - - def _flush_target_stream(self): - streams = [] - if self.target_fd == 1: - streams = [sys.stdout] - elif self.target_fd == 2: - streams = [sys.stderr] - else: - streams = [sys.stdout, sys.stderr] - - for stream in streams: - try: - stream.flush() - except Exception: - pass - - def __enter__(self): - # Drain any pending buffered writes to the current target fd before we - # redirect it. sys.stdout/sys.stderr are block-buffered when connected - # to a pipe; without this, late flushes (e.g. of a previous test's - # post-teardown print) would land in this test's file after we swap - # the underlying fd. - for stream in (sys.stdout, sys.stderr): - try: - stream.flush() - except Exception: - pass - - log_file = open(self.log_file_path, "w", encoding="utf-8", buffering=1) - pipe_read, pipe_write = os.pipe() - self.saved_fd = os.dup(self.target_fd) - os.dup2(pipe_write, self.target_fd) - os.close(pipe_write) - - pipe_stream = os.fdopen(pipe_read, "r", encoding="utf-8", errors="replace", buffering=1) - - # Child processes may inherit the redirected fd and keep the pipe open - # after capture is restored. Do not let that block pytest shutdown. - self._reader_thread = threading.Thread( - target=self._reader_loop, args=(pipe_stream, log_file), daemon=True - ) - self._reader_thread.start() - - return self - - def _reader_loop(self, pipe_stream, log_file): - need_timestamp = True +def drain_pending_uploads(output_path: str, secret_key: str | None = None) -> bool: + """Upload files left by pytest processes that exited before session finish.""" + spool_root = _spool_root(output_path) + if not spool_root.exists(): + return True - try: - while True: - chunk = pipe_stream.read(4096) - if not chunk: - break - - # Build the timestamp prefix once per chunk and reuse it for every - # line in that chunk. The previous implementation walked the chunk - # char-by-char in Python, which was the bottleneck under high-volume - # output (e.g. tqdm progress bars, repetitive kernel info logs). - # Within a single 4 KiB chunk the timestamp would have been - # identical anyway (the reader processes the chunk under the GIL), - # so reusing one prefix is semantics-preserving. - now = datetime.now() - ts_str = self.timestamp_format % { - "asctime": now.strftime(self.date_format), - "msecs": now.microsecond // 1000, - } - prefix = f"[{ts_str}] " - - parts = [] - for line in chunk.splitlines(keepends=True): - if need_timestamp: - parts.append(prefix) - parts.append(line) - need_timestamp = line.endswith("\n") - log_file.write("".join(parts)) - log_file.flush() - - if self.echo_to_original and self.saved_fd is not None: - ret = os.write(self.saved_fd, chunk.encode("utf-8")) - if ret != len(chunk): - logger.warning( - f"Partial write to original FD {self.target_fd}: {ret} != {len(chunk)}" - ) - - except Exception as e: - logger.error(f"Error reading from pipe: {e}") - finally: - log_file.close() - pipe_stream.close() - - def __exit__(self, exc_type, exc_val, exc_tb): - if self.saved_fd is not None: - self._flush_target_stream() - # Restore the original fd binding. This also closes the previous - # binding (our pipe's write end), so the reader thread will see - # EOF and finish draining. - os.dup2(self.saved_fd, self.target_fd) - # NB: keep self.saved_fd valid until the reader thread joins; - # the reader may still drain pending bytes and want to echo them - # through saved_fd. Closing it here would race with that write. - - if self._reader_thread: - self._reader_thread.join(timeout=5.0) - if self._reader_thread.is_alive(): - logger.warning(f"Reader thread for FD {self.target_fd} did not exit in time") - - if self.saved_fd is not None: - os.close(self.saved_fd) - self.saved_fd = None + config_paths = list(spool_root.rglob(_SPOOL_CONFIG_NAME)) + if not config_paths: + return True + secret_key = secret_key or os.environ.get("S3_SECRET_KEY") + if not secret_key: + logger.warning("Cannot drain S3 test logs without S3_SECRET_KEY") return False + success = True + current_host = socket.gethostname() + for config_path in config_paths: + try: + with config_path.open("r", encoding="utf-8") as config_file: + try: + fcntl.flock(config_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + continue + + config = json.load(config_file) + owner_host = config.get("hostname") + owner_pid = int(config.get("pid", 0)) + if owner_host != current_host or (owner_pid and _process_is_alive(owner_pid)): + continue + + spool_dir = config_path.parent + source_paths = [ + path for path in spool_dir.rglob("*") if path.is_file() and path != config_path + ] + config_success = True + if source_paths: + client = _create_s3_client( + config["endpoint_url"], + config["aws_access_key_id"], + secret_key, + ) + uploads = {} + worker_count = min( + max(1, int(config.get("upload_workers", 8))), + len(source_paths), + ) + with ThreadPoolExecutor(max_workers=worker_count) as executor: + for source_path in source_paths: + relative_path = source_path.relative_to(spool_dir) + object_key = posixpath.join(config["upload_path"], *relative_path.parts) + future = executor.submit( + client.upload_file, + str(source_path), + config["bucket"], + object_key, + ExtraArgs={"ContentType": "text/plain"}, + ) + uploads[future] = (source_path, object_key) + + for future in as_completed(uploads): + source_path, object_key = uploads[future] + try: + future.result() + except Exception as exc: + config_success = False + success = False + logger.warning( + "Failed to drain S3 test log %s to %s: %s", + source_path, + object_key, + exc, + ) + else: + source_path.unlink(missing_ok=True) + _remove_empty_parents(source_path, spool_dir) + + if config_success: + config_path.unlink(missing_ok=True) + _remove_empty_parents(config_path, spool_dir) + try: + spool_dir.parent.rmdir() + except OSError: + pass + except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc: + success = False + logger.warning("Failed to read S3 spool config %s: %s", config_path, exc) -class DirectFDRedirector: - def __init__(self, target_fd, log_file_path): - self.target_fd = target_fd - self.log_file_path = log_file_path - - self.saved_fd = None - self._log_file = None - - def _flush_target_stream(self): - streams = [] - if self.target_fd == 1: - streams = [sys.stdout] - elif self.target_fd == 2: - streams = [sys.stderr] - else: - streams = [sys.stdout, sys.stderr] - - for stream in streams: - try: - stream.flush() - except Exception: - pass - - def __enter__(self): - for stream in (sys.stdout, sys.stderr): - try: - stream.flush() - except Exception: - pass - - self._log_file = open(self.log_file_path, "w", encoding="utf-8", buffering=1) - self.saved_fd = os.dup(self.target_fd) - os.dup2(self._log_file.fileno(), self.target_fd) - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self._flush_target_stream() - - if self.saved_fd is not None: - os.dup2(self.saved_fd, self.target_fd) - os.close(self.saved_fd) - self.saved_fd = None - - if self._log_file is not None: - self._log_file.close() - self._log_file = None - - return False + return success class UploadLogPlugin: def __init__( self, - endpoint_url, - aws_access_key_id, - aws_secret_access_key, - bucket, - upload_path, - output_path, - echo_to_stdout=False, - skip_upload=False, - capture_mode="session", - upload_mode="sync", - upload_workers=8, - inline_output_max_bytes=256, - session_capture=None, - ): - self.upload_path = upload_path - self.output_path = output_path - self.bucket = bucket + endpoint_url: str, + aws_access_key_id: str, + aws_secret_access_key: str | None, + bucket: str, + upload_path: str, + output_path: str, + skip_upload: bool = False, + upload_mode: str = "sync", + upload_workers: int = 8, + inline_output_max_bytes: int = 256, + ) -> None: self.endpoint_url = endpoint_url self.aws_access_key_id = aws_access_key_id - self.echo_to_stdout = echo_to_stdout + self.bucket = bucket + self.upload_path = upload_path + self.output_path = output_path self.skip_upload = skip_upload - self.capture_mode = capture_mode self.upload_mode = upload_mode self.upload_workers = max(1, upload_workers) self.inline_output_max_bytes = inline_output_max_bytes if self.inline_output_max_bytes < 0: raise ValueError("--s3-inline-output-max-bytes must be >= 0") - if self.capture_mode in ("session", "direct") and self.echo_to_stdout: - raise ValueError( - f"--s3-capture-mode={self.capture_mode} cannot be used with --s3-echo-stdout" - ) + if self.upload_mode not in ("sync", "deferred"): + raise ValueError("--s3-upload-mode must be 'sync' or 'deferred'") + self.s3 = None if not self.skip_upload: - try: - import boto3 - except ModuleNotFoundError as exc: - raise RuntimeError( - "boto3 is required when --s3-upload-path is set without --s3-skip-upload" - ) from exc - self.s3 = boto3.client( - "s3", - endpoint_url=endpoint_url, - aws_access_key_id=aws_access_key_id, - aws_secret_access_key=aws_secret_access_key, + if not aws_secret_access_key: + raise ValueError("S3 secret key is required to upload test logs") + self.s3 = _create_s3_client( + endpoint_url, + aws_access_key_id, + aws_secret_access_key, ) - # nodeid -> dict of open capture state + logging handler. Session mode - # spans setup, call, and teardown; legacy modes close after the call. - self._active_capture: dict = {} - self._test_names: dict = {} - self._deferred_uploads = [] - self._captured_slices = {} - self._session_capture = session_capture - self._upload_failed = False - def normalize_test_name(self, nodeid): - import hashlib + suffix = f"{socket.gethostname()}-{os.getpid()}-{time.time_ns()}" + self._spool_dir = str(_spool_root(output_path) / suffix) + self._spool_config_path = os.path.join(self._spool_dir, _SPOOL_CONFIG_NAME) + self._test_names: dict[str, str] = {} + self._used_test_names: set[str] = set() + self._upload_failed = False + self._executor = None + self._pending_uploads: dict[Future, PendingUpload] = {} + self._max_pending_uploads = self.upload_workers * 2 + if not self.skip_upload: + self._write_spool_config() + if self.upload_mode == "deferred": + self._executor = ThreadPoolExecutor( + max_workers=self.upload_workers, + thread_name_prefix="s3-test-log-upload", + ) + def normalize_test_name(self, nodeid: str) -> str: test_name = re.sub(r"[^\w\-]", "_", nodeid) suffix = hashlib.md5(nodeid.encode()).hexdigest()[:8] timestamp = int(time.time()) - # Linux limits a single path component to 255 bytes. if len(test_name) > 200: test_name = test_name[:200] return f"{test_name}-{suffix}-{timestamp}" - def _open_capture(self, item): - """Open stdout/stderr and logging capture for ``item``. - - Returns a state dict on success, or ``None`` if setup failed (in which - case the test runs uncaptured). Never propagates exceptions. - """ - state = {} - try: - test_name = self.normalize_test_name(item.nodeid) - state["test_name"] = test_name - output_path = os.path.join(self.output_path, test_name) - os.makedirs(output_path, exist_ok=True) - - log_file = os.path.join(output_path, "logging.log") - - log_date_format = item.config.getini("log_date_format") - log_format = item.config.getini("log_format") - - timestamp_format = None - if log_format: - match = re.search(r"\[([^\]]*%\(asctime\)s[^\]]*)\]", log_format) - if match: - timestamp_format = match.group(1) - - handler = logging.FileHandler(log_file) - logging_plugin = item.config.pluginmanager.getplugin("logging-plugin") - if logging_plugin is not None: - handler.setFormatter(logging_plugin.formatter) - elif log_format: - formatter = logging.Formatter(log_format, datefmt=log_date_format) - handler.setFormatter(formatter) - state["handler"] = handler - - if self.capture_mode == "session": - if self._session_capture is None: - raise RuntimeError("Session capture has not started") - state["session_offsets"] = self._session_capture.snapshot() - else: - stdout_file = os.path.join(output_path, "stdout.log") - stderr_file = os.path.join(output_path, "stderr.log") - fd_kwargs = {} - if log_date_format: - fd_kwargs["date_format"] = log_date_format - if timestamp_format: - fd_kwargs["timestamp_format"] = timestamp_format - - if self.capture_mode == "direct": - state["stdout_redir"] = DirectFDRedirector(1, stdout_file) - else: - state["stdout_redir"] = FDRedirector( - 1, stdout_file, echo_to_original=self.echo_to_stdout, **fd_kwargs - ) - state["stdout_redir"].__enter__() - if self.capture_mode == "direct": - state["stderr_redir"] = DirectFDRedirector(2, stderr_file) - else: - state["stderr_redir"] = FDRedirector( - 2, stderr_file, echo_to_original=self.echo_to_stdout, **fd_kwargs - ) - state["stderr_redir"].__enter__() - state["log_cm"] = catching_logs(handler) - state["log_cm"].__enter__() - return state - except Exception as e: - logger.warning( - "S3 capture setup failed for %r: %s; running without capture", item.nodeid, e - ) - self._close_capture(state) - return None - - def _close_capture(self, state): - """Close capture state opened by ``_open_capture``. Never raises.""" - if not state: - return - session_offsets = state.get("session_offsets") - if session_offsets is not None and self._session_capture is not None: - try: - slices = self._session_capture.slices_since(session_offsets) - for filename, file_slice in slices.items(): - self._captured_slices[(state["test_name"], filename)] = file_slice - except (OSError, ValueError) as e: - logger.warning("Error finalizing session capture: %s", e) - # Close in reverse order so fd 1/2 are restored before logging stops. - for key in ("log_cm", "stderr_redir", "stdout_redir"): - cm = state.get(key) - if cm is None: - continue - try: - cm.__exit__(None, None, None) - except Exception as e: - logger.warning("Error closing %s capture: %s", key, e) - handler = state.get("handler") - if handler is not None: - try: - handler.close() - except Exception: - pass - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_setup(self, item): - state = self._open_capture(item) - if state is not None: - self._active_capture[item.nodeid] = state - self._test_names[item.nodeid] = state["test_name"] - yield - - @pytest.hookimpl(wrapper=True) - def pytest_sessionstart(self, session): - result = yield - if self.capture_mode == "session": - if self._session_capture is None: - self._session_capture = SessionCapture(self.output_path) - self._session_capture.start() - else: - self._session_capture.resume_parent() - return result - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_makereport(self, item, call): - report = yield - if self.capture_mode != "session" and ( - report.when == "call" or (report.when == "setup" and report.outcome != "passed") - ): - self._close_capture(self._active_capture.pop(item.nodeid, None)) - return report - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_teardown(self, item, nextitem): - if self.capture_mode == "session": - try: - return (yield) - finally: - self._close_capture(self._active_capture.pop(item.nodeid, None)) - - # Fallback for custom or interrupted runtest protocols that did not - # produce a call report. Normal execution closes capture earlier. - state = self._active_capture.pop(item.nodeid, None) - self._close_capture(state) - if state is not None: - logger.warning( - "S3 capture for %r remained active until teardown; " - "pytest result/progress output may have been captured", - item.nodeid, - ) - yield - - def _suspend_session_capture(self): - if self._session_capture is not None: - self._session_capture.suspend_parent() - - def _resume_session_capture(self): - if self._session_capture is not None: - self._session_capture.resume_parent() - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_logstart(self, nodeid, location): - self._suspend_session_capture() - try: - return (yield) - finally: - self._resume_session_capture() - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_logfinish(self, nodeid, location): - self._suspend_session_capture() - try: - return (yield) - finally: - self._resume_session_capture() - - def get_file_size(self, path): - try: - return os.path.getsize(path) - except FileNotFoundError: - return None - - def _capture_source(self, test_name, filename): - file_slice = self._captured_slices.pop((test_name, filename), None) - if file_slice is not None: - return file_slice - return os.path.join(self.output_path, test_name, filename) - - def _source_exists(self, source): - if isinstance(source, FileSlice): - return os.path.exists(source.path) - return os.path.exists(source) - - def _source_size(self, source): - if isinstance(source, FileSlice): - return source.size - return os.path.getsize(source) - - def _open_source(self, source): - if isinstance(source, FileSlice): - return io.BufferedReader(FileSliceReader(source)) - return open(source, "rb") - - def _remove_source(self, source): - if not isinstance(source, FileSlice): - self._remove_local_log_file(source) - - def _object_key(self, test_name, filename): - return os.path.join(self.upload_path, test_name, filename) - - def _file_url(self, object_key): - return os.path.join( + def _write_spool_config(self) -> None: + os.makedirs(self._spool_dir, exist_ok=True) + config = { + "endpoint_url": self.endpoint_url, + "aws_access_key_id": self.aws_access_key_id, + "bucket": self.bucket, + "upload_path": self.upload_path, + "hostname": socket.gethostname(), + "pid": os.getpid(), + "upload_workers": self.upload_workers, + } + temporary_path = f"{self._spool_config_path}.{os.getpid()}.tmp" + with open(temporary_path, "w", encoding="utf-8") as config_file: + json.dump(config, config_file) + os.replace(temporary_path, self._spool_config_path) + + def _test_name(self, nodeid: str) -> str: + test_name = self._test_names.get(nodeid) + if test_name is None: + test_name = self.normalize_test_name(nodeid) + base_name = test_name + collision_index = 1 + while test_name in self._used_test_names: + test_name = f"{base_name}-{collision_index}" + collision_index += 1 + self._used_test_names.add(test_name) + self._test_names[nodeid] = test_name + return test_name + + def _object_key(self, test_name: str, filename: str) -> str: + return posixpath.join(self.upload_path, test_name, filename) + + def _file_url(self, object_key: str) -> str: + return posixpath.join( self.endpoint_url, "v1/AUTH_" + self.aws_access_key_id, self.bucket, object_key, ) - def _upload_source(self, source, object_key): - extra_args = {"ContentType": "text/plain"} - if isinstance(source, FileSlice): - with self._open_source(source) as source_file: - self.s3.upload_fileobj( - source_file, - self.bucket, - object_key, - ExtraArgs=extra_args, - ) - else: - self.s3.upload_file( - source, - self.bucket, - object_key, - ExtraArgs=extra_args, - ) - - def _append_upload_failed(self, report, section_name, source, filesize, error): - with self._open_source(source) as source_file: - limit = 65536 - # Limit content to 64k (65536 bytes) - trail_content = "... [truncated]" - content = source_file.read(limit + 1).decode("utf-8", errors="replace") - if len(content) > limit: - content = content[: limit - len(trail_content)] + trail_content - report.sections.append( - ( - section_name, - f"""upload failed: {error}\nsize: {filesize} bytes\ncontent: {content}""", - ) + def _should_inline(self, stream_name: str, content: str) -> bool: + if stream_name not in ("stdout", "stderr"): + return False + if self.inline_output_max_bytes == 0: + return False + if len(content) >= self.inline_output_max_bytes: + return False + return len(content.encode("utf-8")) < self.inline_output_max_bytes + + def _write_spool_file(self, test_name: str, filename: str, content: str) -> str: + if not self.skip_upload and not os.path.exists(self._spool_config_path): + self._write_spool_config() + test_dir = os.path.join(self._spool_dir, test_name) + os.makedirs(test_dir, exist_ok=True) + source_path = os.path.join(test_dir, filename) + file_descriptor = os.open( + source_path, + os.O_WRONLY | os.O_CREAT | os.O_TRUNC, + 0o600, ) + with os.fdopen(file_descriptor, "w", encoding="utf-8", errors="replace") as output: + for offset in range(0, len(content), _SPOOL_WRITE_CHARS): + output.write(content[offset : offset + _SPOOL_WRITE_CHARS]) + return source_path - def _should_inline_output(self, filename, filesize): - return filename in ("stdout.log", "stderr.log") and filesize < self.inline_output_max_bytes - - def _append_inline_output(self, report, section_name, source): - with self._open_source(source) as source_file: - content = source_file.read().decode("utf-8", errors="replace") - report.sections.append((section_name, content)) - - def _remove_local_log_file(self, filepath): + def _remove_source(self, source_path: str) -> None: + path = Path(source_path) try: - os.remove(filepath) + path.unlink() except FileNotFoundError: return - except OSError as e: - logger.warning("Failed to remove local S3 log file %s: %s", filepath, e) + except OSError as exc: + logger.warning("Failed to remove local S3 log file %s: %s", source_path, exc) return + _remove_empty_parents(path, Path(self._spool_dir)) + + def _upload_source(self, source_path: str, object_key: str) -> None: + self.s3.upload_file( + source_path, + self.bucket, + object_key, + ExtraArgs={"ContentType": "text/plain"}, + ) - output_root = os.path.abspath(self.output_path) - parent = os.path.abspath(os.path.dirname(filepath)) - while parent != output_root and os.path.commonpath([output_root, parent]) == output_root: + def _finish_uploads(self, futures: set[Future]) -> None: + for future in futures: + upload = self._pending_uploads.pop(future) try: - os.rmdir(parent) - except OSError: - break - parent = os.path.dirname(parent) + future.result() + except Exception as exc: + self._upload_failed = True + logger.warning( + "Deferred upload failed. test_name: %s, filename: %s, " + "object_key: %s, source: %s, error: %s", + upload.test_name, + upload.filename, + upload.object_key, + upload.source_path, + exc, + ) + else: + self._remove_source(upload.source_path) - def upload_and_report(self, report, test_name, filename, section_name): - source = self._capture_source(test_name, filename) - if not self._source_exists(source): - report.sections.append((section_name, "")) - return - filesize = self._source_size(source) - if filesize == 0: - report.sections.append((section_name, "")) - self._remove_source(source) - return - if self._should_inline_output(filename, filesize): - self._append_inline_output(report, section_name, source) - self._remove_source(source) - return + def _reap_uploads(self, block_when_full: bool = False) -> None: + completed = {future for future in self._pending_uploads if future.done()} + if completed: + self._finish_uploads(completed) + if block_when_full and len(self._pending_uploads) >= self._max_pending_uploads: + completed, _ = wait(self._pending_uploads, return_when=FIRST_COMPLETED) + self._finish_uploads(completed) + + def _schedule_upload( + self, + source_path: str, + object_key: str, + test_name: str, + filename: str, + ) -> None: + self._reap_uploads(block_when_full=True) + future = self._executor.submit(self._upload_source, source_path, object_key) + self._pending_uploads[future] = PendingUpload( + source_path=source_path, + object_key=object_key, + test_name=test_name, + filename=filename, + ) + + def _tail(self, content: str) -> str: + position = len(content) + lines = 0 + while position > 0 and lines < _FAILED_OUTPUT_MAX_LINES: + position = content.rfind("\n", 0, position) + if position < 0: + position = 0 + break + lines += 1 + tail = content[position:] + truncated = len(tail) > _FAILED_OUTPUT_MAX_BYTES + if truncated: + tail = tail[-_FAILED_OUTPUT_MAX_BYTES:] + encoded = tail.encode("utf-8") + if len(encoded) > _FAILED_OUTPUT_MAX_BYTES: + tail = encoded[-_FAILED_OUTPUT_MAX_BYTES:].decode("utf-8", errors="replace") + truncated = True + if truncated: + tail = "... [truncated]\n" + tail + return tail + + def _format_report_content(self, message: str, content: str, failed: bool) -> str: + if not failed: + return message + return f"{message}\n\nLast {_FAILED_OUTPUT_MAX_LINES} lines:\n{self._tail(content)}" + + def _transform_section( + self, + section_name: str, + content: str, + stream_name: str, + phase: str, + test_name: str, + failed: bool, + duplicate_index: int, + ) -> tuple[str, str]: + if self._should_inline(stream_name, content): + return section_name, content + + filename = f"{_STREAM_FILENAMES[stream_name]}-{phase}.log" + if duplicate_index: + filename = f"{_STREAM_FILENAMES[stream_name]}-{phase}-{duplicate_index}.log" + source_path = self._write_spool_file(test_name, filename, content) + filesize = os.path.getsize(source_path) object_key = self._object_key(test_name, filename) - fileurl = self._file_url(object_key) + file_url = self._file_url(object_key) + if self.skip_upload: - # Experiment knob: skip the actual S3 upload to measure how much - # of this plugin's per-test overhead comes from boto3 network IO. - report.sections.append( - ( - section_name, - f"{filesize} bytes (upload skipped, would upload to {fileurl})", - ) - ) - return + message = f"{filesize} bytes (upload skipped, would upload to {file_url})" + return section_name, self._format_report_content(message, content, failed) + if self.upload_mode == "deferred": - self._deferred_uploads.append((source, object_key, test_name, filename)) - report.sections.append( - ( - section_name, - f"{filesize} bytes scheduled for upload to {fileurl}", - ) - ) - return + self._schedule_upload(source_path, object_key, test_name, filename) + message = f"{filesize} bytes scheduled for upload to {file_url}" + return section_name, self._format_report_content(message, content, failed) + try: - self._upload_source(source, object_key) - report.sections.append( - ( - section_name, - f"{filesize} bytes uploaded to {fileurl}", - ) - ) - self._remove_source(source) - except Exception as e: + self._upload_source(source_path, object_key) + except Exception as exc: self._upload_failed = True logger.warning( - f"Upload failed. test_name: {test_name}, filename: {filename}, error: {e}" + "Upload failed. test_name: %s, filename: %s, error: %s", + test_name, + filename, + exc, ) - self._append_upload_failed(report, section_name, source, filesize, e) + message = f"upload failed: {exc}\nsize: {filesize} bytes" + return section_name, self._format_report_content(message, content, True) + + self._remove_source(source_path) + message = f"{filesize} bytes uploaded to {file_url}" + return section_name, self._format_report_content(message, content, failed) + + def _process_report(self, report, default_phase: str) -> None: + nodeid = getattr(report, "nodeid", "unknown") + test_name = self._test_name(nodeid) + failed = getattr(report, "outcome", None) == "failed" + duplicate_counts: dict[tuple[str, str], int] = {} + transformed_sections = [] + for section_name, content in report.sections: + match = _CAPTURE_SECTION_PATTERN.fullmatch(section_name) + if match is None: + transformed_sections.append((section_name, content)) + continue + stream_name = match.group(1) + phase = match.group(2) or default_phase + duplicate_key = (stream_name, phase) + duplicate_index = duplicate_counts.get(duplicate_key, 0) + duplicate_counts[duplicate_key] = duplicate_index + 1 + transformed_sections.append( + self._transform_section( + section_name, + content, + stream_name, + phase, + test_name, + failed, + duplicate_index, + ) + ) + report.sections[:] = transformed_sections + + def pytest_runtest_logstart(self, nodeid: str, location) -> None: + self._test_name(nodeid) - @pytest.hookimpl(wrapper=True) + @pytest.hookimpl(wrapper=True, tryfirst=True) def pytest_runtest_logreport(self, report): - self._suspend_session_capture() - try: - if report.when == "teardown": - test_name = self._test_names.pop(report.nodeid, None) - if test_name is None: - test_name = self.normalize_test_name(report.nodeid) - # Add S3 report sections before the junitxml plugin handles - # this report, while pytest's terminal output bypasses capture. - self.upload_and_report(report, test_name, "stdout.log", "Captured stdout") - self.upload_and_report(report, test_name, "stderr.log", "Captured stderr") - self.upload_and_report(report, test_name, "logging.log", "Captured log") - return (yield) - finally: - self._resume_session_capture() + self._process_report(report, getattr(report, "when", "call")) + return (yield) @pytest.hookimpl(tryfirst=True) - def pytest_sessionfinish(self, session, exitstatus): - for state in self._active_capture.values(): - self._close_capture(state) - self._active_capture.clear() - - if self._session_capture is not None: - self._session_capture.stop() - - if not self.skip_upload and self._deferred_uploads: - workers = min(self.upload_workers, len(self._deferred_uploads)) - logger.info( - "Uploading %d deferred S3 test log files with %d workers", - len(self._deferred_uploads), - workers, - ) - with ThreadPoolExecutor(max_workers=workers) as executor: - futures = { - executor.submit(self._upload_source, source, object_key): ( - source, - object_key, - test_name, - filename, - ) - for source, object_key, test_name, filename in self._deferred_uploads - } - for future in as_completed(futures): - source, object_key, test_name, filename = futures[future] - try: - future.result() - self._remove_source(source) - except Exception as e: - self._upload_failed = True - logger.warning( - "Deferred upload failed. test_name: %s, filename: %s, " - "object_key: %s, source: %s, error: %s", - test_name, - filename, - object_key, - source, - e, - ) - self._deferred_uploads.clear() - - if ( - self._session_capture is not None - and not self.skip_upload - and not self._upload_failed - and not self._captured_slices - ): - self._session_capture.remove_files() - - -def add_options(parser): - """Register S3 CLI options. Call from pytest_addoption in any conftest that needs S3 upload.""" + def pytest_collectreport(self, report) -> None: + self._process_report(report, "collect") + self._test_names.pop(getattr(report, "nodeid", "unknown"), None) + + def pytest_runtest_logfinish(self, nodeid: str, location) -> None: + self._test_names.pop(nodeid, None) + + @pytest.hookimpl(tryfirst=True) + def pytest_sessionfinish(self, session, exitstatus) -> None: + if self._executor is not None: + self._executor.shutdown(wait=True) + self._executor = None + self._finish_uploads(set(self._pending_uploads)) + if not self.skip_upload and not self._upload_failed: + try: + Path(self._spool_config_path).unlink(missing_ok=True) + Path(self._spool_dir).rmdir() + Path(self._spool_dir).parent.rmdir() + except OSError: + pass + + +def add_options(parser) -> None: + """Register S3 options on a pytest parser.""" parser.addoption( "--s3-endpoint", action=EnvDefault, envvar="S3_ENDPOINT", default="https://pbss.s8k.io", - help="s3 endpoint", + help="S3 endpoint", ) parser.addoption( "--s3-username", action=EnvDefault, envvar="S3_USERNAME", default="svc_tensorrt", - help="s3 username", + help="S3 username", ) parser.addoption( "--s3-secret-key", action=EnvDefault, envvar="S3_SECRET_KEY", required=False, - help="s3 secret key", + help="S3 secret key", ) parser.addoption( "--s3-bucket", action=EnvDefault, envvar="S3_BUCKET", default="trtllm-ci-logs", - help="s3 bucket name", + help="S3 bucket name", ) parser.addoption( "--s3-upload-path", action=EnvDefault, envvar="S3_UPLOAD_PATH", required=False, - help="s3 upload path", - ) - parser.addoption( - "--s3-echo-stdout", - action="store_true", - default=False, - help="Besides capturing stdout/stderr to per-test log files, also echo " - "them through to the original stdout/stderr for live debugging. " - "This requires --s3-capture-mode=timestamped and should be set on the outer pytest " - "invocation; nested pytest invocations spawned by individual tests " - "should NOT set this, to avoid duplicating their output back through " - "the outer pipe.", + help="S3 upload path", ) parser.addoption( "--s3-skip-upload", action="store_true", default=False, - help="Experiment knob: still capture stdout/stderr/log per test and " - "append URLs to report sections, but skip the actual s3.upload_file " - "call. Used to measure how much of the plugin's per-test overhead " - "comes from boto3/network IO vs. local capture machinery.", - ) - parser.addoption( - "--s3-capture-mode", - action="store", - choices=("session", "timestamped", "direct"), - default="session", - help="Capture each stream through a session-scoped append-only file and " - "split it into per-test ranges. Timestamped and direct retain the legacy " - "per-test FD redirection modes.", + help="Transform captured output into S3 report links without uploading it.", ) parser.addoption( "--s3-upload-mode", action="store", choices=("sync", "deferred"), default="sync", - help="Upload each per-test log file synchronously in the test teardown, " - "or defer uploads until pytest session finish. Deferred mode keeps " - "per-test files and report URLs while reducing teardown latency for " - "nested pytest runs with many cases.", + help="Upload report sections synchronously or with a bounded background pool.", ) parser.addoption( "--s3-upload-workers", @@ -962,42 +585,55 @@ def add_options(parser): action="store", type=int, default=256, - help="Inline captured stdout/stderr into the test report and skip S3 " - "upload when each file is smaller than this many bytes. Set to 0 to " - "disable inlining for non-empty output.", + help="Inline stdout/stderr smaller than this size instead of uploading it.", ) -def register_plugin(config, session_capture=None): - """Register UploadLogPlugin if --s3-upload-path and --output-dir are both set.""" +def register_plugin(config): + """Register the report transformer when S3 output is configured.""" + existing_plugin = config.pluginmanager.getplugin("upload_log_plugin") + if existing_plugin is not None: + return existing_plugin + s3_upload_path = config.getoption("--s3-upload-path", default=None) output_dir = config.getoption("--output-dir", default=None) if not (s3_upload_path and output_dir): return None - pytest_capture_mode = config.getoption("capture", default="no") - if pytest_capture_mode != "no": - raise ValueError("capture mode must be 'no' when upload path is specified") - s3_secret_key = config.getoption("--s3-secret-key") + capture_mode = config.getoption("capture", default="fd") + if capture_mode != "fd": + raise ValueError("--s3-upload-path requires pytest --capture=fd") + skip_upload = config.getoption("--s3-skip-upload", default=False) + s3_secret_key = config.getoption("--s3-secret-key", default=None) if not skip_upload and not s3_secret_key: raise ValueError( "--s3-secret-key (or S3_SECRET_KEY env var) is required when --s3-upload-path is set" ) - s3_capture_mode = config.getoption("--s3-capture-mode", default="session") + plugin = UploadLogPlugin( endpoint_url=config.getoption("--s3-endpoint"), aws_access_key_id=config.getoption("--s3-username"), aws_secret_access_key=s3_secret_key, bucket=config.getoption("--s3-bucket"), upload_path=s3_upload_path, - output_path=output_dir, - echo_to_stdout=config.getoption("--s3-echo-stdout", default=False), + output_path=os.path.abspath(output_dir), skip_upload=skip_upload, - capture_mode=s3_capture_mode, upload_mode=config.getoption("--s3-upload-mode", default="sync"), upload_workers=config.getoption("--s3-upload-workers", default=8), inline_output_max_bytes=config.getoption("--s3-inline-output-max-bytes", default=256), - session_capture=session_capture if s3_capture_mode == "session" else None, ) config.pluginmanager.register(plugin, "upload_log_plugin") return plugin + + +def _main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--drain-spool", metavar="OUTPUT_PATH") + args = parser.parse_args() + if args.drain_spool: + return 0 if drain_pending_uploads(args.drain_spool) else 1 + return 0 + + +if __name__ == "__main__": + sys.exit(_main()) diff --git a/tests/test_common/s3_output_hooks.py b/tests/test_common/s3_output_hooks.py index ccaf34f6ea02..39df46e8e932 100644 --- a/tests/test_common/s3_output_hooks.py +++ b/tests/test_common/s3_output_hooks.py @@ -12,31 +12,14 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Start S3 FD capture before pytest imports the initial conftests.""" +"""Register the S3 report transformer on pytest workers.""" -import argparse import os -from collections.abc import Generator -from dataclasses import dataclass -from typing import cast import pytest from test_common import s3_output -_CAPTURE_STATE_ATTRIBUTE = "_s3_output_early_capture_state" - - -@dataclass -class _EarlyCaptureState: - capture: s3_output.SessionCapture - claimed: bool = False - - def cleanup(self) -> None: - self.capture.stop() - if not self.claimed: - self.capture.remove_files() - def _is_xdist_worker(config: pytest.Config) -> bool: return bool(os.environ.get("PYTEST_XDIST_WORKER")) or hasattr(config, "workerinput") @@ -52,77 +35,7 @@ def _is_xdist_controller(config: pytest.Config) -> bool: return num_processes not in (None, 0, "0") -def _parse_early_capture_options( - early_config: pytest.Config, args: list[str] -) -> tuple[str | None, str | None, str, str]: - parser = argparse.ArgumentParser(add_help=False, allow_abbrev=False) - parser.add_argument("--output-dir", "-O") - parser.add_argument("--s3-upload-path") - parser.add_argument("--s3-capture-mode") - parsed, _ = parser.parse_known_args(args) - - namespace = early_config.known_args_namespace - output_path = cast( - str | None, - getattr(namespace, "output_dir", None) or parsed.output_dir, - ) - upload_path = cast( - str | None, - getattr(namespace, "s3_upload_path", None) - or parsed.s3_upload_path - or os.environ.get("S3_UPLOAD_PATH"), - ) - capture_mode = cast( - str, - getattr(namespace, "s3_capture_mode", None) or parsed.s3_capture_mode or "session", - ) - pytest_capture = cast(str, getattr(namespace, "capture", "fd")) - return output_path, upload_path, capture_mode, pytest_capture - - -def _capture_state(config: pytest.Config) -> _EarlyCaptureState | None: - return cast( - _EarlyCaptureState | None, - getattr(config, _CAPTURE_STATE_ATTRIBUTE, None), - ) - - -@pytest.hookimpl(wrapper=True, tryfirst=True) -def pytest_load_initial_conftests( - early_config: pytest.Config, args: list[str] -) -> Generator[None, object, object]: - output_path, upload_path, capture_mode, pytest_capture = _parse_early_capture_options( - early_config, args - ) - if ( - _capture_state(early_config) is not None - or _is_xdist_controller(early_config) - or not output_path - or not upload_path - or capture_mode != "session" - or pytest_capture != "no" - ): - return (yield) - - capture = s3_output.SessionCapture(output_path) - capture.start() - state = _EarlyCaptureState(capture) - setattr(early_config, _CAPTURE_STATE_ATTRIBUTE, state) - early_config.add_cleanup(state.cleanup) - try: - return (yield) - finally: - # MPI runtimes initialized while conftests load retain the spool FD. - # Restore the parent so pytest configuration output stays visible. - capture.suspend_parent() - - @pytest.hookimpl(trylast=True) def pytest_configure(config: pytest.Config) -> None: - if _is_xdist_controller(config): - return - state = _capture_state(config) - session_capture = state.capture if state is not None else None - plugin = s3_output.register_plugin(config, session_capture=session_capture) - if plugin is not None and state is not None: - state.claimed = True + if not _is_xdist_controller(config): + s3_output.register_plugin(config) diff --git a/tests/unittest/test_s3_output.py b/tests/unittest/test_s3_output.py index 899ad5a8fe27..9660b56220e9 100644 --- a/tests/unittest/test_s3_output.py +++ b/tests/unittest/test_s3_output.py @@ -13,461 +13,515 @@ # See the License for the specific language governing permissions and # limitations under the License. -import io -import logging +import json import os import subprocess import sys +import threading +from pathlib import Path from types import SimpleNamespace import pytest -from test_common import s3_output_hooks -from test_common.s3_output import ( - FDRedirector, - FileSlice, - FileSliceReader, - SessionCapture, - UploadLogPlugin, -) +from test_common import s3_output, s3_output_hooks +from test_common.s3_output import UploadLogPlugin class Report: - def __init__(self, when=None, outcome=None): - self.sections = [] + def __init__( + self, + sections, + nodeid="test_module.py::test_case", + when="call", + outcome="passed", + ): + self.sections = sections + self.nodeid = nodeid self.when = when self.outcome = outcome -class CaptureContext: +class RecordingS3Client: def __init__(self): - self.closed = False + self.uploads = [] + + def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): + content = Path(filepath).read_bytes() + self.uploads.append((content, bucket, object_key, ExtraArgs)) - def __exit__(self, exc_type, exc_val, exc_tb): - self.closed = True +class FailingS3Client: + def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): + raise RuntimeError("upload error") -class RecordingS3Client: + +class BlockingS3Client(RecordingS3Client): def __init__(self): - self.uploads = [] - self.fileobj_uploads = [] + super().__init__() + self.started = threading.Event() + self.release = threading.Event() + + def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): + self.started.set() + assert self.release.wait(timeout=10) + super().upload_file(filepath, bucket, object_key, ExtraArgs) + + +class ConcurrentS3Client(RecordingS3Client): + def __init__(self, workers): + super().__init__() + self.barrier = threading.Barrier(workers) def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): - self.uploads.append((filepath, bucket, object_key, ExtraArgs)) + self.barrier.wait(timeout=5) + super().upload_file(filepath, bucket, object_key, ExtraArgs) - def upload_fileobj(self, fileobj, bucket, object_key, ExtraArgs=None): - self.fileobj_uploads.append((fileobj.read(), bucket, object_key, ExtraArgs)) + +class PluginManager: + def __init__(self): + self.plugins = {} + + def getplugin(self, name): + return self.plugins.get(name) + + def register(self, plugin, name): + self.plugins[name] = plugin + + +class Config: + def __init__(self, **options): + self.options = options + self.option = SimpleNamespace(numprocesses=options.get("numprocesses")) + self.known_args_namespace = self.option + self.pluginmanager = PluginManager() + + def getoption(self, name, default=None): + return self.options.get(name, default) def make_plugin( tmp_path, - inline_output_max_bytes, - capture_mode="timestamped", - session_capture=None, + inline_output_max_bytes=256, + skip_upload=True, + upload_mode="sync", + upload_workers=8, ): return UploadLogPlugin( endpoint_url="https://example.com", aws_access_key_id="user", - aws_secret_access_key=None, + aws_secret_access_key=None if skip_upload else "secret", bucket="bucket", upload_path="logs", output_path=str(tmp_path), - skip_upload=True, - capture_mode=capture_mode, + skip_upload=skip_upload, + upload_mode=upload_mode, + upload_workers=upload_workers, inline_output_max_bytes=inline_output_max_bytes, - session_capture=session_capture, ) -def write_log(tmp_path, test_name, filename, content): - test_dir = tmp_path / test_name - test_dir.mkdir() - (test_dir / filename).write_text(content, encoding="utf-8") +def make_uploading_plugin(tmp_path, monkeypatch, client, **kwargs): + monkeypatch.setattr(s3_output, "_create_s3_client", lambda *args: client) + return make_plugin(tmp_path, skip_upload=False, **kwargs) -def complete_makereport_hook(plugin, nodeid, report): - item = type("Item", (), {"nodeid": nodeid})() - hook = plugin.pytest_runtest_makereport(item, call=None) +def process_report(plugin, report): + hook = plugin.pytest_runtest_logreport(report) next(hook) - with pytest.raises(StopIteration) as stop: - hook.send(report) - assert stop.value.value is report + with pytest.raises(StopIteration): + next(hook) -def test_capture_stays_open_after_successful_setup_report(tmp_path): - nodeid = "test_module.py::test_case" - plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} - - complete_makereport_hook(plugin, nodeid, Report(when="setup", outcome="passed")) - - assert not capture.closed - assert nodeid in plugin._active_capture - plugin._close_capture(plugin._active_capture.pop(nodeid)) - - -@pytest.mark.parametrize( - ("when", "outcome"), - [ - ("call", "passed"), - ("call", "failed"), - ("setup", "failed"), - ("setup", "skipped"), - ], -) -def test_capture_closes_before_terminal_report(tmp_path, when, outcome): - nodeid = "test_module.py::test_case" - plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} +def test_small_stdout_remains_inline(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=4) + report = Report([("Captured stdout call", "ok\n")]) - complete_makereport_hook(plugin, nodeid, Report(when=when, outcome=outcome)) + process_report(plugin, report) - assert capture.closed - assert nodeid not in plugin._active_capture + assert report.sections == [("Captured stdout call", "ok\n")] + assert not s3_output._spool_root(str(tmp_path)).exists() -def test_teardown_fallback_warns_when_capture_is_still_active(tmp_path, caplog): - nodeid = "test_module.py::test_case" +def test_stdout_at_threshold_is_replaced_with_url(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=4) + report = Report([("Captured stdout call", "four")]) + + process_report(plugin, report) + + section_name, section_content = report.sections[0] + assert section_name == "Captured stdout call" + assert "4 bytes (upload skipped" in section_content + assert "/stdout-call.log" in section_content + assert "four" not in section_content + + +def test_logging_section_is_uploaded_even_when_small(tmp_path): plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} - item = type("Item", (), {"nodeid": nodeid})() + report = Report([("Captured log call", "log\n")]) - caplog.set_level(logging.WARNING, logger="test_common.s3_output") - hook = plugin.pytest_runtest_teardown(item, nextitem=None) - next(hook) + process_report(plugin, report) - assert capture.closed - assert nodeid not in plugin._active_capture - assert "remained active until teardown" in caplog.text + assert "upload skipped" in report.sections[0][1] + assert "/logging-call.log" in report.sections[0][1] - with pytest.raises(StopIteration): - next(hook) +def test_sync_upload_transforms_native_sections(tmp_path, monkeypatch): + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + ) + report = Report( + [ + ("Captured stdout setup", "setup output\n"), + ("custom", "keep me"), + ("Captured stderr call", "call error\n"), + ] + ) -def test_session_capture_closes_after_teardown(tmp_path): - nodeid = "test_module.py::test_case" - plugin = make_plugin(tmp_path, inline_output_max_bytes=256, capture_mode="session") - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} - item = type("Item", (), {"nodeid": nodeid})() + process_report(plugin, report) + plugin.pytest_sessionfinish(None, 0) - hook = plugin.pytest_runtest_teardown(item, nextitem=None) - next(hook) - assert not capture.closed + assert [upload[0] for upload in client.uploads] == [ + b"setup output\n", + b"call error\n", + ] + assert client.uploads[0][2].endswith("/stdout-setup.log") + assert client.uploads[1][2].endswith("/stderr-call.log") + assert report.sections[1] == ("custom", "keep me") + assert all("uploaded to" in report.sections[index][1] for index in (0, 2)) + assert not s3_output._spool_root(str(tmp_path)).exists() - with pytest.raises(StopIteration): - next(hook) - assert capture.closed - assert nodeid not in plugin._active_capture +def test_duplicate_capture_sections_get_distinct_objects(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=0) + report = Report( + [ + ("Captured stdout call", "first"), + ("Captured stdout call", "second"), + ] + ) -def test_small_stdout_is_inlined_without_upload(tmp_path): - test_name = "test-small" - write_log(tmp_path, test_name, "stdout.log", "ok\n") - plugin = make_plugin(tmp_path, inline_output_max_bytes=4) - report = Report() + process_report(plugin, report) - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + assert "/stdout-call.log" in report.sections[0][1] + assert "/stdout-call-1.log" in report.sections[1][1] - assert report.sections == [("Captured stdout", "ok\n")] - assert plugin._deferred_uploads == [] - assert not (tmp_path / test_name).exists() +def test_same_nodeid_rerun_gets_distinct_test_path(tmp_path, monkeypatch): + monkeypatch.setattr(s3_output.time, "time", lambda: 1234) + plugin = make_plugin(tmp_path, inline_output_max_bytes=0) + nodeid = "test_module.py::test_case" -def test_empty_stdout_is_removed_after_report(tmp_path): - test_name = "test-empty" - write_log(tmp_path, test_name, "stdout.log", "") - plugin = make_plugin(tmp_path, inline_output_max_bytes=4) - report = Report() + plugin.pytest_runtest_logstart(nodeid, None) + first_name = plugin._test_names[nodeid] + plugin.pytest_runtest_logfinish(nodeid, None) + plugin.pytest_runtest_logstart(nodeid, None) + second_name = plugin._test_names[nodeid] - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + assert second_name == f"{first_name}-1" - assert report.sections == [("Captured stdout", "")] - assert plugin._deferred_uploads == [] - assert not (tmp_path / test_name).exists() +def test_failed_report_keeps_only_recent_bounded_output(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=0) + content = "".join(f"line-{index:03d}\n" for index in range(250)) + report = Report( + [("Captured stdout call", content)], + outcome="failed", + ) -def test_large_stdout_keeps_existing_upload_report(tmp_path): - test_name = "test-large" - write_log(tmp_path, test_name, "stdout.log", "large output") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3) - report = Report() + process_report(plugin, report) - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + section_content = report.sections[0][1] + assert "Last 200 lines:" in section_content + assert "line-000" not in section_content + assert "line-249" in section_content - assert len(report.sections) == 1 - section_name, section_content = report.sections[0] - assert section_name == "Captured stdout" - assert "upload skipped" in section_content - assert "large output" not in section_content - assert (tmp_path / test_name / "stdout.log").exists() - - -def test_uploaded_stdout_is_removed_after_sync_upload(tmp_path): - test_name = "test-uploaded" - write_log(tmp_path, test_name, "stdout.log", "large output") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3) - plugin.skip_upload = False - plugin.s3 = RecordingS3Client() - report = Report() - - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") - - assert len(plugin.s3.uploads) == 1 - assert plugin.s3.uploads[0][1:] == ( - "bucket", - "logs/test-uploaded/stdout.log", - {"ContentType": "text/plain"}, + large_line = "x" * 70000 + report = Report( + [("Captured stderr call", large_line)], + nodeid="test_module.py::test_other", + outcome="failed", ) - assert len(report.sections) == 1 - assert "uploaded to" in report.sections[0][1] - assert not (tmp_path / test_name).exists() - - -def test_deferred_stdout_is_removed_after_upload_finishes(tmp_path): - test_name = "test-deferred" - write_log(tmp_path, test_name, "stdout.log", "large output") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3) - plugin.skip_upload = False - plugin.upload_mode = "deferred" - plugin.s3 = RecordingS3Client() - report = Report() - - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") - - assert (tmp_path / test_name / "stdout.log").exists() - assert len(plugin._deferred_uploads) == 1 - - plugin.pytest_sessionfinish(session=None, exitstatus=0) - - assert len(plugin.s3.uploads) == 1 - assert plugin._deferred_uploads == [] - assert not (tmp_path / test_name).exists() - - -def test_file_slice_upload_reads_only_test_range(tmp_path): - test_name = "test-slice" - spool = tmp_path / "stdout-spool.log" - prefix = b"previous test\n" - content = b"parent\nchild\nparent again\n" - spool.write_bytes(prefix + content + b"next test\n") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3, capture_mode="session") - plugin.skip_upload = False - plugin.s3 = RecordingS3Client() - plugin._captured_slices[(test_name, "stdout.log")] = FileSlice( - str(spool), len(prefix), len(content) + process_report(plugin, report) + assert "... [truncated]" in report.sections[0][1] + assert len(report.sections[0][1].encode()) < 66000 + + +def test_deferred_upload_starts_before_session_finish(tmp_path, monkeypatch): + client = BlockingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + upload_mode="deferred", + upload_workers=1, ) - report = Report() + report = Report([("Captured stdout call", "background output")]) - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + process_report(plugin, report) - assert plugin.s3.fileobj_uploads == [ - ( - content, - "bucket", - "logs/test-slice/stdout.log", - {"ContentType": "text/plain"}, - ) - ] - assert "uploaded to" in report.sections[0][1] - - -def test_file_slice_reader_supports_upload_size_probe(tmp_path): - spool = tmp_path / "stdout-spool.log" - spool.write_bytes(b"prefix\ntest output\nsuffix\n") - file_slice = FileSlice(str(spool), len(b"prefix\n"), len(b"test output\n")) - - with io.BufferedReader(FileSliceReader(file_slice)) as source_file: - assert source_file.tell() == 0 - assert source_file.seek(0, os.SEEK_END) == file_slice.size - assert source_file.tell() == file_slice.size - assert source_file.seek(0) == 0 - assert source_file.read() == b"test output\n" - - -def test_session_capture_preserves_parent_and_child_stdout_order(tmp_path): - capture = SessionCapture(str(tmp_path)) - capture.start() - child = None - try: - child = subprocess.Popen( - [ - sys.executable, - "-c", - "import os, sys; sys.stdin.buffer.read(1); " - "os.write(1, b'child stdout\\n'); os.write(2, b'child stderr\\n')", - ], - stdin=subprocess.PIPE, - ) - offsets = capture.snapshot() - os.write(1, b"parent before\n") - child.stdin.write(b"x") - child.stdin.close() - assert child.wait(timeout=10) == 0 - os.write(1, b"parent after\n") - slices = capture.slices_since(offsets) - finally: - if child is not None and child.poll() is None: - child.kill() - child.wait() - capture.stop() - - with io.BufferedReader(FileSliceReader(slices["stdout.log"])) as stdout_file: - assert stdout_file.read() == b"parent before\nchild stdout\nparent after\n" - with io.BufferedReader(FileSliceReader(slices["stderr.log"])) as stderr_file: - assert stderr_file.read() == b"child stderr\n" - capture.remove_files() - - -class _EarlyConfig: - def __init__(self, capture="no", numprocesses=None): - self.option = SimpleNamespace(numprocesses=numprocesses) - self.known_args_namespace = SimpleNamespace( - capture=capture, - numprocesses=numprocesses, - ) - self.cleanups = [] - - def add_cleanup(self, cleanup): - self.cleanups.append(cleanup) - - -def test_early_capture_preserves_conftest_child_output_for_case(tmp_path, monkeypatch): + assert client.started.wait(timeout=5) + assert "scheduled for upload" in report.sections[0][1] + client.release.set() + plugin.pytest_sessionfinish(None, 0) + assert client.uploads[0][0] == b"background output" + assert not s3_output._spool_root(str(tmp_path)).exists() + + +def test_parent_drain_retries_upload_left_by_failed_process(tmp_path, monkeypatch): + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + FailingS3Client(), + inline_output_max_bytes=0, + ) + report = Report([("Captured stdout call", "recover me")]) + process_report(plugin, report) + assert "upload failed" in report.sections[0][1] + + config_path = Path(plugin._spool_config_path) + config = json.loads(config_path.read_text(encoding="utf-8")) + config["pid"] = 99999999 + config_path.write_text(json.dumps(config), encoding="utf-8") + + client = RecordingS3Client() + monkeypatch.setattr(s3_output, "_create_s3_client", lambda *args: client) + assert s3_output.drain_pending_uploads(str(tmp_path), secret_key="secret") + assert client.uploads[0][0] == b"recover me" + assert client.uploads[0][2].endswith("/stdout-call.log") + assert not s3_output._spool_root(str(tmp_path)).exists() + + +def test_parent_drain_uses_configured_upload_workers(tmp_path, monkeypatch): + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + RecordingS3Client(), + upload_workers=2, + ) + plugin._write_spool_file("test", "stdout-call.log", "stdout") + plugin._write_spool_file("test", "stderr-call.log", "stderr") + config_path = Path(plugin._spool_config_path) + config = json.loads(config_path.read_text(encoding="utf-8")) + config["pid"] = 99999999 + config_path.write_text(json.dumps(config), encoding="utf-8") + + client = ConcurrentS3Client(workers=2) + monkeypatch.setattr(s3_output, "_create_s3_client", lambda *args: client) + + assert s3_output.drain_pending_uploads(str(tmp_path), secret_key="secret") + assert sorted(upload[0] for upload in client.uploads) == [b"stderr", b"stdout"] + assert not s3_output._spool_root(str(tmp_path)).exists() + + +def test_parent_drain_skips_live_pytest_process(tmp_path, monkeypatch): + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + upload_mode="deferred", + ) + plugin._write_spool_file("test", "stdout-call.log", "still active") + + assert s3_output.drain_pending_uploads(str(tmp_path), secret_key="secret") + assert client.uploads == [] + assert Path(plugin._spool_config_path).exists() + plugin.pytest_sessionfinish(None, 0) + + +def test_register_plugin_requires_native_fd_capture(tmp_path): + config = Config( + **{ + "--s3-upload-path": "logs", + "--output-dir": str(tmp_path), + "--s3-skip-upload": True, + "capture": "no", + } + ) + + with pytest.raises(ValueError, match="requires pytest --capture=fd"): + s3_output.register_plugin(config) + + +def test_register_plugin_uses_report_transformer(tmp_path): + config = Config( + **{ + "--s3-upload-path": "logs", + "--output-dir": str(tmp_path), + "--s3-skip-upload": True, + "--s3-endpoint": "https://example.com", + "--s3-username": "user", + "--s3-bucket": "bucket", + "capture": "fd", + } + ) + + plugin = s3_output.register_plugin(config) + + assert isinstance(plugin, UploadLogPlugin) + assert config.pluginmanager.getplugin("upload_log_plugin") is plugin + + +def test_s3_hook_skips_xdist_controller(monkeypatch): + registered = [] + monkeypatch.setattr(s3_output_hooks.s3_output, "register_plugin", registered.append) monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False) - early_config = _EarlyConfig() - initial_hook = s3_output_hooks.pytest_load_initial_conftests( - early_config, + + controller = Config(numprocesses=2) + s3_output_hooks.pytest_configure(controller) + assert registered == [] + + monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw0") + worker = Config(numprocesses=2) + s3_output_hooks.pytest_configure(worker) + assert registered == [worker] + + +def _write_nested_pytest_plugin(tmp_path, register_plugin=True): + tests_root = Path(s3_output.__file__).parents[1] + conftest = ( + "from test_common import s3_output\n" + "def pytest_addoption(parser):\n" + " parser.addoption('--output-dir')\n" + " s3_output.add_options(parser)\n" + ) + if register_plugin: + conftest += "def pytest_configure(config):\n s3_output.register_plugin(config)\n" + (tmp_path / "conftest.py").write_text(conftest, encoding="utf-8") + env = os.environ.copy() + env["PYTHONPATH"] = os.pathsep.join([str(tests_root), env.get("PYTHONPATH", "")]) + return env + + +def test_native_capture_is_replaced_before_junit_consumes_report(tmp_path): + env = _write_nested_pytest_plugin(tmp_path) + (tmp_path / "test_native.py").write_text( + "import os\n" + "import subprocess\n" + "import sys\n" + "def test_output():\n" + " os.write(1, b'parent before\\n')\n" + " subprocess.run([sys.executable, '-c', " + "\"import os; os.write(1, b'child capture ' + b'x' * 300 + b'\\\\n')\"], check=True)\n" + " os.write(1, b'parent after\\n')\n", + encoding="utf-8", + ) + output_dir = tmp_path / "output" + xml_path = tmp_path / "results.xml" + + result = subprocess.run( [ - "-s", + sys.executable, + "-m", + "pytest", + "-q", + "--capture=fd", "--s3-upload-path=logs", - f"--output-dir={tmp_path}", + f"--output-dir={output_dir}", + "--s3-skip-upload", + f"--junitxml={xml_path}", + "-o", + "junit_logging=all", + str(tmp_path / "test_native.py"), ], + capture_output=True, + text=True, + env=env, + timeout=30, ) - child = None - capture = None - try: - next(initial_hook) - state = s3_output_hooks._capture_state(early_config) - assert state is not None - capture = state.capture - assert capture is not None - child = subprocess.Popen( - [ - sys.executable, - "-c", - "import os, sys; sys.stdin.buffer.read(1); " - "os.write(1, b'child stdout\\n'); os.write(2, b'child stderr\\n')", - ], - stdin=subprocess.PIPE, - ) - - with pytest.raises(StopIteration): - next(initial_hook) - - plugin = make_plugin( - tmp_path, - inline_output_max_bytes=0, - capture_mode="session", - session_capture=capture, - ) - session = SimpleNamespace(config=SimpleNamespace()) - session_hook = plugin.pytest_sessionstart(session) - next(session_hook) - with pytest.raises(StopIteration): - next(session_hook) - - offsets = capture.snapshot() - os.write(1, b"parent before\n") - child.stdin.write(b"x") - child.stdin.close() - assert child.wait(timeout=10) == 0 - os.write(1, b"parent after\n") - slices = capture.slices_since(offsets) - capture.stop() - - with io.BufferedReader(FileSliceReader(slices["stdout.log"])) as stdout_file: - assert stdout_file.read() == b"parent before\nchild stdout\nparent after\n" - with io.BufferedReader(FileSliceReader(slices["stderr.log"])) as stderr_file: - assert stderr_file.read() == b"child stderr\n" - finally: - if child is not None and child.poll() is None: - child.kill() - child.wait() - if capture is not None: - capture.stop() - for cleanup in early_config.cleanups: - cleanup() - - -def test_early_capture_skips_xdist_controller(tmp_path, monkeypatch): - monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False) - early_config = _EarlyConfig(numprocesses=2) - initial_hook = s3_output_hooks.pytest_load_initial_conftests( - early_config, + + assert result.returncode == 0, result.stdout + result.stderr + xml = xml_path.read_text(encoding="utf-8") + assert "upload skipped" in xml + assert "child capture xxxxx" not in xml + assert "stdout-call.log" in xml + stdout_files = list(s3_output._spool_root(str(output_dir)).rglob("stdout-call.log")) + assert len(stdout_files) == 1 + assert stdout_files[0].read_bytes() == ( + b"parent before\n" + b"child capture " + b"x" * 300 + b"\nparent after\n" + ) + + +def test_xdist_worker_transforms_report_before_controller_junit(tmp_path): + env = _write_nested_pytest_plugin(tmp_path, register_plugin=False) + (tmp_path / "test_xdist.py").write_text( + "import os\ndef test_output():\n os.write(1, b'xdist capture ' + b'x' * 300)\n", + encoding="utf-8", + ) + output_dir = tmp_path / "output" + xml_path = tmp_path / "results.xml" + + result = subprocess.run( [ - "-s", + sys.executable, + "-m", + "pytest", + "-q", "-n", "2", + "-p", + "test_common.s3_output_hooks", + "--capture=fd", "--s3-upload-path=logs", - f"--output-dir={tmp_path}", + f"--output-dir={output_dir}", + "--s3-skip-upload", + f"--junitxml={xml_path}", + "-o", + "junit_logging=all", + str(tmp_path / "test_xdist.py"), ], + capture_output=True, + text=True, + env=env, + timeout=30, ) - next(initial_hook) - assert s3_output_hooks._capture_state(early_config) is None - with pytest.raises(StopIteration): - next(initial_hook) - - -def test_s3_plugin_registration_runs_on_xdist_worker_only(monkeypatch): - monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False) - registered_configs = [] - - def register_plugin(config, session_capture=None): - registered_configs.append(config) - return object() - - monkeypatch.setattr(s3_output_hooks.s3_output, "register_plugin", register_plugin) - controller_config = _EarlyConfig(numprocesses=2) - s3_output_hooks.pytest_configure(controller_config) - assert registered_configs == [] - - monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw0") - worker_config = _EarlyConfig(numprocesses=2) - s3_output_hooks.pytest_configure(worker_config) - assert registered_configs == [worker_config] - - -def test_small_log_file_is_not_inlined(tmp_path): - test_name = "test-log" - write_log(tmp_path, test_name, "logging.log", "ok\n") - plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - report = Report() + assert result.returncode == 0, result.stdout + result.stderr + xml = xml_path.read_text(encoding="utf-8") + assert "upload skipped" in xml + assert "xdist capture xxxxx" not in xml + assert "stdout-call.log" in xml + + +def test_native_capture_restores_timeout_output_to_console(tmp_path): + env = _write_nested_pytest_plugin(tmp_path) + (tmp_path / "test_timeout.py").write_text( + "import os\n" + "import time\n" + "import pytest\n" + "@pytest.mark.timeout(0.2, method='thread')\n" + "def test_timeout():\n" + " os.write(1, b'timeout stdout\\n')\n" + " os.write(2, b'timeout stderr\\n')\n" + " time.sleep(30)\n", + encoding="utf-8", + ) - plugin.upload_and_report(report, test_name, "logging.log", "Captured log") + result = subprocess.run( + [ + sys.executable, + "-m", + "pytest", + "-q", + "--capture=fd", + "--s3-upload-path=logs", + f"--output-dir={tmp_path / 'output'}", + "--s3-skip-upload", + str(tmp_path / "test_timeout.py"), + ], + capture_output=True, + text=True, + env=env, + timeout=10, + ) - assert len(report.sections) == 1 - section_name, section_content = report.sections[0] - assert section_name == "Captured log" - assert "upload skipped" in section_content - - -def test_fd_redirector_reader_thread_is_daemon_when_pipe_writer_lingers(tmp_path): - redir = FDRedirector(1, str(tmp_path / "stdout.log")) - proc = None - try: - redir.__enter__() - proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) - redir.__exit__(None, None, None) - - assert redir._reader_thread is not None - assert redir._reader_thread.is_alive() - assert redir._reader_thread.daemon - finally: - if proc is not None: - proc.kill() - proc.wait() - if redir._reader_thread is not None: - redir._reader_thread.join(timeout=2.0) + console = result.stdout + result.stderr + assert result.returncode == 1 + assert "timeout stdout" in console + assert "timeout stderr" in console + assert "Timeout" in console diff --git a/tests/unittest/tools/test_test_to_stage_mapping.py b/tests/unittest/tools/test_test_to_stage_mapping.py index f8689ff12398..29ded249dd91 100644 --- a/tests/unittest/tools/test_test_to_stage_mapping.py +++ b/tests/unittest/tools/test_test_to_stage_mapping.py @@ -79,35 +79,28 @@ def test_data_availability(stage_query): print(f"Max samples configured: {MAX_SAMPLES}") -def test_s3_stdout_echo_requires_explicit_opt_in(): - """Keep Jenkins pytest progress readable unless live log echo is requested.""" +def test_s3_output_uses_native_fd_capture(): + """Keep S3 collection on pytest's native FD capture path.""" with open(GROOVY, 'r') as f: lines = f.readlines() - assert any( - 'ENABLE_S3_ECHO_STDOUT = params.enableS3EchoStdout != null ? params.enableS3EchoStdout : false' - in line for line in lines) - - echo_lines = [ - idx for idx, line in enumerate(lines) if '"--s3-echo-stdout"' in line + capture_lines = [ + idx for idx, line in enumerate(lines) if '"--capture=fd"' in line ] - assert echo_lines, 'Expected at least one opt-in --s3-echo-stdout path' - - for idx in echo_lines: - context = lines[max(0, idx - 3):idx] - assert any('if (ENABLE_S3_ECHO_STDOUT)' in line for line in context) - - progress_lines = [ - idx for idx, line in enumerate(lines) - if 'console_output_style=progress-even-when-capture-no' in line - ] - assert progress_lines, 'Expected upload-only pytest progress configuration' - - for idx in progress_lines: + assert capture_lines, 'Expected S3 pytest paths to enable native FD capture' + for idx in capture_lines: context = lines[max(0, idx - 3):idx] assert any('if (ENABLE_UPLOAD_TEST_RESULTS)' in line for line in context) + source = ''.join(lines) + assert 'ENABLE_S3_ECHO_STDOUT' not in source + assert '--s3-echo-stdout' not in source + assert 'console_output_style=progress-even-when-capture-no' in source + assert 'testFilter[(DETAILED_LOG)] ? "-s"' in source + assert source.count('tar -czvf results-') == 2 + assert "tar --exclude='*/.s3-spool'" not in source + @pytest.mark.skip(reason="https://nvbugs/5547275") @pytest.mark.parametrize("direction", From ba8cca90bdc353f7ddea7603555427ddd2ee7170 Mon Sep 17 00:00:00 2001 From: Yiteng Niu <6831097+niukuo@users.noreply.github.com> Date: Thu, 23 Jul 2026 11:19:40 +0800 Subject: [PATCH 2/3] [TRTLLMINF-191][infra] Deduplicate S3 capture uploads across reports Signed-off-by: Yiteng Niu <6831097+niukuo@users.noreply.github.com> --- tests/test_common/s3_output.py | 116 ++++++++++++++---- tests/unittest/test_s3_output.py | 204 +++++++++++++++++++++++++++++++ 2 files changed, 295 insertions(+), 25 deletions(-) diff --git a/tests/test_common/s3_output.py b/tests/test_common/s3_output.py index 6995c1e568fb..59fc5ead9f6a 100644 --- a/tests/test_common/s3_output.py +++ b/tests/test_common/s3_output.py @@ -69,6 +69,13 @@ class PendingUpload: filename: str +@dataclass(frozen=True) +class CapturedSection: + content: str + message: str + force_output: bool = False + + def _create_s3_client( endpoint_url: str, aws_access_key_id: str, @@ -243,6 +250,9 @@ def __init__( self._spool_config_path = os.path.join(self._spool_dir, _SPOOL_CONFIG_NAME) self._test_names: dict[str, str] = {} self._used_test_names: set[str] = set() + self._captured_sections: dict[str, dict[tuple[str, str, int], CapturedSection]] = {} + self._capture_filename_counts: dict[str, dict[tuple[int, str, str], int]] = {} + self._pending_reruns: set[str] = set() self._upload_failed = False self._executor = None self._pending_uploads: dict[Future, PendingUpload] = {} @@ -413,25 +423,41 @@ def _tail(self, content: str) -> str: def _format_report_content(self, message: str, content: str, failed: bool) -> str: if not failed: - return message - return f"{message}\n\nLast {_FAILED_OUTPUT_MAX_LINES} lines:\n{self._tail(content)}" + return f"{message}\n" + formatted = f"{message}\n\nLast {_FAILED_OUTPUT_MAX_LINES} lines:\n{self._tail(content)}" + if not formatted.endswith("\n"): + formatted += "\n" + return formatted - def _transform_section( + def _next_filename( self, - section_name: str, - content: str, + nodeid: str, stream_name: str, phase: str, - test_name: str, - failed: bool, - duplicate_index: int, - ) -> tuple[str, str]: - if self._should_inline(stream_name, content): - return section_name, content - - filename = f"{_STREAM_FILENAMES[stream_name]}-{phase}.log" + attempt: int, + ) -> str: + counts = self._capture_filename_counts.setdefault(nodeid, {}) + count_key = (attempt, stream_name, phase) + duplicate_index = counts.get(count_key, 0) + counts[count_key] = duplicate_index + 1 + + suffixes = [] + if attempt > 1: + suffixes.append(f"attempt-{attempt}") if duplicate_index: - filename = f"{_STREAM_FILENAMES[stream_name]}-{phase}-{duplicate_index}.log" + suffixes.append(str(duplicate_index)) + + filename = f"{_STREAM_FILENAMES[stream_name]}-{phase}" + if suffixes: + filename += f"-{'-'.join(suffixes)}" + return f"{filename}.log" + + def _upload_section( + self, + content: str, + test_name: str, + filename: str, + ) -> CapturedSection: source_path = self._write_spool_file(test_name, filename, content) filesize = os.path.getsize(source_path) object_key = self._object_key(test_name, filename) @@ -439,12 +465,12 @@ def _transform_section( if self.skip_upload: message = f"{filesize} bytes (upload skipped, would upload to {file_url})" - return section_name, self._format_report_content(message, content, failed) + return CapturedSection(content, message) if self.upload_mode == "deferred": self._schedule_upload(source_path, object_key, test_name, filename) message = f"{filesize} bytes scheduled for upload to {file_url}" - return section_name, self._format_report_content(message, content, failed) + return CapturedSection(content, message) try: self._upload_source(source_path, object_key) @@ -457,16 +483,25 @@ def _transform_section( exc, ) message = f"upload failed: {exc}\nsize: {filesize} bytes" - return section_name, self._format_report_content(message, content, True) + return CapturedSection(content, message, force_output=True) self._remove_source(source_path) message = f"{filesize} bytes uploaded to {file_url}" - return section_name, self._format_report_content(message, content, failed) + return CapturedSection(content, message) + + @staticmethod + def _attempt(report) -> int: + rerun = getattr(report, "rerun", 0) + return rerun + 1 if isinstance(rerun, int) and rerun >= 0 else 1 def _process_report(self, report, default_phase: str) -> None: nodeid = getattr(report, "nodeid", "unknown") test_name = self._test_name(nodeid) failed = getattr(report, "outcome", None) == "failed" + if getattr(report, "outcome", None) == "rerun": + self._pending_reruns.add(nodeid) + attempt = self._attempt(report) + captured_sections = self._captured_sections.setdefault(nodeid, {}) duplicate_counts: dict[tuple[str, str], int] = {} transformed_sections = [] for section_name, content in report.sections: @@ -479,15 +514,34 @@ def _process_report(self, report, default_phase: str) -> None: duplicate_key = (stream_name, phase) duplicate_index = duplicate_counts.get(duplicate_key, 0) duplicate_counts[duplicate_key] = duplicate_index + 1 - transformed_sections.append( - self._transform_section( - section_name, - content, + if self._should_inline(stream_name, content): + transformed_sections.append((section_name, content)) + continue + + section_key = (stream_name, phase, duplicate_index) + captured_section = captured_sections.get(section_key) + if captured_section is None or captured_section.content != content: + filename = self._next_filename( + nodeid, stream_name, phase, + attempt, + ) + captured_section = self._upload_section( + content, test_name, - failed, - duplicate_index, + filename, + ) + captured_sections[section_key] = captured_section + + transformed_sections.append( + ( + section_name, + self._format_report_content( + captured_section.message, + content, + failed or captured_section.force_output, + ), ) ) report.sections[:] = transformed_sections @@ -503,10 +557,19 @@ def pytest_runtest_logreport(self, report): @pytest.hookimpl(tryfirst=True) def pytest_collectreport(self, report) -> None: self._process_report(report, "collect") - self._test_names.pop(getattr(report, "nodeid", "unknown"), None) + nodeid = getattr(report, "nodeid", "unknown") + self._test_names.pop(nodeid, None) + self._captured_sections.pop(nodeid, None) + self._capture_filename_counts.pop(nodeid, None) + self._pending_reruns.discard(nodeid) def pytest_runtest_logfinish(self, nodeid: str, location) -> None: self._test_names.pop(nodeid, None) + if nodeid in self._pending_reruns: + self._pending_reruns.remove(nodeid) + return + self._captured_sections.pop(nodeid, None) + self._capture_filename_counts.pop(nodeid, None) @pytest.hookimpl(tryfirst=True) def pytest_sessionfinish(self, session, exitstatus) -> None: @@ -521,6 +584,9 @@ def pytest_sessionfinish(self, session, exitstatus) -> None: Path(self._spool_dir).parent.rmdir() except OSError: pass + self._captured_sections.clear() + self._capture_filename_counts.clear() + self._pending_reruns.clear() def add_options(parser) -> None: diff --git a/tests/unittest/test_s3_output.py b/tests/unittest/test_s3_output.py index 9660b56220e9..1b91e9375ddb 100644 --- a/tests/unittest/test_s3_output.py +++ b/tests/unittest/test_s3_output.py @@ -18,6 +18,7 @@ import subprocess import sys import threading +import xml.etree.ElementTree as ET from pathlib import Path from types import SimpleNamespace @@ -33,11 +34,14 @@ def __init__( nodeid="test_module.py::test_case", when="call", outcome="passed", + rerun=None, ): self.sections = sections self.nodeid = nodeid self.when = when self.outcome = outcome + if rerun is not None: + self.rerun = rerun class RecordingS3Client: @@ -207,6 +211,55 @@ def test_duplicate_capture_sections_get_distinct_objects(tmp_path): assert "/stdout-call.log" in report.sections[0][1] assert "/stdout-call-1.log" in report.sections[1][1] + assert all(content.endswith("\n") for _, content in report.sections) + + +def test_cumulative_capture_sections_are_uploaded_once(tmp_path, monkeypatch): + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + ) + setup_section = ("Captured stdout setup", "setup output\n") + call_section = ("Captured stdout call", "call output\n") + teardown_section = ("Captured stderr teardown", "teardown error\n") + + setup_report = Report([setup_section], when="setup") + call_report = Report( + [setup_section, call_section], + when="call", + outcome="failed", + ) + teardown_report = Report( + [setup_section, call_section, teardown_section], + when="teardown", + ) + + process_report(plugin, setup_report) + process_report(plugin, call_report) + process_report(plugin, teardown_report) + plugin.pytest_runtest_logfinish(setup_report.nodeid, None) + plugin.pytest_sessionfinish(None, 0) + + assert [upload[0] for upload in client.uploads] == [ + b"setup output\n", + b"call output\n", + b"teardown error\n", + ] + assert [upload[2].rsplit("/", 1)[-1] for upload in client.uploads] == [ + "stdout-setup.log", + "stdout-call.log", + "stderr-teardown.log", + ] + assert "Last 200 lines:" in call_report.sections[0][1] + assert "Last 200 lines:" not in teardown_report.sections[0][1] + assert ( + setup_report.sections[0][1].splitlines()[0] + == (teardown_report.sections[0][1].splitlines()[0]) + ) + assert all(content.endswith("\n") for _, content in teardown_report.sections) def test_same_nodeid_rerun_gets_distinct_test_path(tmp_path, monkeypatch): @@ -271,6 +324,97 @@ def test_deferred_upload_starts_before_session_finish(tmp_path, monkeypatch): assert not s3_output._spool_root(str(tmp_path)).exists() +def test_deferred_upload_reuses_cumulative_section(tmp_path, monkeypatch): + client = BlockingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + upload_mode="deferred", + upload_workers=1, + ) + section = ("Captured stdout call", "background output") + call_report = Report([section], outcome="failed") + teardown_report = Report([section], when="teardown") + + process_report(plugin, call_report) + assert client.started.wait(timeout=5) + process_report(plugin, teardown_report) + + assert len(plugin._pending_uploads) == 1 + assert ( + call_report.sections[0][1].splitlines()[0] + == (teardown_report.sections[0][1].splitlines()[0]) + ) + client.release.set() + plugin.pytest_sessionfinish(None, 0) + assert [upload[0] for upload in client.uploads] == [b"background output"] + + +def test_rerun_keeps_one_url_per_attempt(tmp_path, monkeypatch): + monkeypatch.setattr(s3_output.time, "time", lambda: 1234) + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + ) + nodeid = "test_module.py::test_case" + first_sections = [ + ("Captured stdout call", "first stdout\n"), + ("Captured stderr call", "first stderr\n"), + ] + all_sections = first_sections + [ + ("Captured stdout call", "second stdout\n"), + ("Captured stderr call", "second stderr\n"), + ] + + plugin.pytest_runtest_logstart(nodeid, None) + first_report = Report( + first_sections, + nodeid=nodeid, + outcome="rerun", + rerun=0, + ) + process_report(plugin, first_report) + plugin.pytest_runtest_logfinish(nodeid, None) + + plugin.pytest_runtest_logstart(nodeid, None) + second_report = Report( + list(all_sections), + nodeid=nodeid, + outcome="failed", + rerun=1, + ) + process_report(plugin, second_report) + teardown_report = Report( + list(all_sections), + nodeid=nodeid, + when="teardown", + rerun=1, + ) + process_report(plugin, teardown_report) + plugin.pytest_runtest_logfinish(nodeid, None) + plugin.pytest_sessionfinish(None, 0) + + assert [upload[0] for upload in client.uploads] == [ + b"first stdout\n", + b"first stderr\n", + b"second stdout\n", + b"second stderr\n", + ] + object_keys = [upload[2] for upload in client.uploads] + assert object_keys[0].endswith("/stdout-call.log") + assert object_keys[1].endswith("/stderr-call.log") + assert object_keys[2].endswith("/stdout-call-attempt-2.log") + assert object_keys[3].endswith("/stderr-call-attempt-2.log") + assert object_keys[0].rsplit("/", 1)[0] != object_keys[2].rsplit("/", 1)[0] + assert all(content.endswith("\n") for _, content in teardown_report.sections) + assert "Last 200 lines:" not in teardown_report.sections[0][1] + + def test_parent_drain_retries_upload_left_by_failed_process(tmp_path, monkeypatch): plugin = make_uploading_plugin( tmp_path, @@ -447,6 +591,66 @@ def test_native_capture_is_replaced_before_junit_consumes_report(tmp_path): ) +def test_rerun_urls_are_distinct_and_line_delimited_in_junit(tmp_path): + env = _write_nested_pytest_plugin(tmp_path) + (tmp_path / "test_rerun.py").write_text( + "import os\n" + "attempt = 0\n" + "def test_output():\n" + " global attempt\n" + " attempt += 1\n" + " os.write(1, f'attempt {attempt} '.encode() + b'x' * 300 + b'\\n')\n" + " assert False\n", + encoding="utf-8", + ) + output_dir = tmp_path / "output" + xml_path = tmp_path / "results.xml" + + result = subprocess.run( + [ + sys.executable, + "-m", + "pytest", + "-q", + "--capture=fd", + "--reruns=1", + "--s3-upload-path=logs", + f"--output-dir={output_dir}", + "--s3-skip-upload", + f"--junitxml={xml_path}", + "-o", + "junit_logging=all", + str(tmp_path / "test_rerun.py"), + ], + capture_output=True, + text=True, + env=env, + timeout=30, + ) + + assert result.returncode == 1, result.stdout + result.stderr + system_output = "\n".join( + element.text or "" for element in ET.parse(xml_path).findall(".//system-out") + ) + url_lines = [ + line + for line in system_output.splitlines() + if "upload skipped" in line and "stdout-call" in line + ] + assert len(url_lines) == 2, system_output + assert any("/stdout-call.log)" in line for line in url_lines) + assert any("/stdout-call-attempt-2.log)" in line for line in url_lines) + + stdout_files = list(s3_output._spool_root(str(output_dir)).rglob("stdout-call*.log")) + assert {path.name for path in stdout_files} == { + "stdout-call.log", + "stdout-call-attempt-2.log", + } + contents = {path.name: path.read_bytes() for path in stdout_files} + assert contents["stdout-call.log"].startswith(b"attempt 1 ") + assert contents["stdout-call-attempt-2.log"].startswith(b"attempt 2 ") + + def test_xdist_worker_transforms_report_before_controller_junit(tmp_path): env = _write_nested_pytest_plugin(tmp_path, register_plugin=False) (tmp_path / "test_xdist.py").write_text( From e764c3c6f1bf7d459da4015093549c5228549e46 Mon Sep 17 00:00:00 2001 From: Yiteng Niu <6831097+niukuo@users.noreply.github.com> Date: Fri, 24 Jul 2026 00:47:06 +0800 Subject: [PATCH 3/3] [TRTLLMINF-191][infra] Merge pytest capture streams before upload Signed-off-by: Yiteng Niu <6831097+niukuo@users.noreply.github.com> --- tests/test_common/s3_output.py | 279 ++++++++++++++++++++----------- tests/unittest/test_s3_output.py | 132 ++++++++++----- 2 files changed, 274 insertions(+), 137 deletions(-) diff --git a/tests/test_common/s3_output.py b/tests/test_common/s3_output.py index 59fc5ead9f6a..86ec4b51421f 100644 --- a/tests/test_common/s3_output.py +++ b/tests/test_common/s3_output.py @@ -72,8 +72,18 @@ class PendingUpload: @dataclass(frozen=True) class CapturedSection: content: str - message: str + attempt: int + + +@dataclass +class CapturedStream: + test_name: str + filename: str + source_path: str + filesize: int = 0 + message: str = "" force_output: bool = False + finalized: bool = False def _create_s3_client( @@ -251,7 +261,7 @@ def __init__( self._test_names: dict[str, str] = {} self._used_test_names: set[str] = set() self._captured_sections: dict[str, dict[tuple[str, str, int], CapturedSection]] = {} - self._capture_filename_counts: dict[str, dict[tuple[int, str, str], int]] = {} + self._captured_streams: dict[str, dict[tuple[int, str], CapturedStream]] = {} self._pending_reruns: set[str] = set() self._upload_failed = False self._executor = None @@ -267,7 +277,7 @@ def __init__( def normalize_test_name(self, nodeid: str) -> str: test_name = re.sub(r"[^\w\-]", "_", nodeid) - suffix = hashlib.md5(nodeid.encode()).hexdigest()[:8] + suffix = hashlib.md5(nodeid.encode(), usedforsecurity=False).hexdigest()[:8] timestamp = int(time.time()) if len(test_name) > 200: test_name = test_name[:200] @@ -313,15 +323,6 @@ def _file_url(self, object_key: str) -> str: object_key, ) - def _should_inline(self, stream_name: str, content: str) -> bool: - if stream_name not in ("stdout", "stderr"): - return False - if self.inline_output_max_bytes == 0: - return False - if len(content) >= self.inline_output_max_bytes: - return False - return len(content.encode("utf-8")) < self.inline_output_max_bytes - def _write_spool_file(self, test_name: str, filename: str, content: str) -> str: if not self.skip_upload and not os.path.exists(self._spool_config_path): self._write_spool_config() @@ -338,6 +339,22 @@ def _write_spool_file(self, test_name: str, filename: str, content: str) -> str: output.write(content[offset : offset + _SPOOL_WRITE_CHARS]) return source_path + def _append_spool_file(self, test_name: str, filename: str, content: str) -> str: + if not self.skip_upload and not os.path.exists(self._spool_config_path): + self._write_spool_config() + test_dir = os.path.join(self._spool_dir, test_name) + os.makedirs(test_dir, exist_ok=True) + source_path = os.path.join(test_dir, filename) + file_descriptor = os.open( + source_path, + os.O_WRONLY | os.O_CREAT | os.O_APPEND, + 0o600, + ) + with os.fdopen(file_descriptor, "a", encoding="utf-8", errors="replace") as output: + for offset in range(0, len(content), _SPOOL_WRITE_CHARS): + output.write(content[offset : offset + _SPOOL_WRITE_CHARS]) + return source_path + def _remove_source(self, source_path: str) -> None: path = Path(source_path) try: @@ -348,6 +365,10 @@ def _remove_source(self, source_path: str) -> None: logger.warning("Failed to remove local S3 log file %s: %s", source_path, exc) return _remove_empty_parents(path, Path(self._spool_dir)) + try: + Path(self._spool_dir).parent.rmdir() + except OSError: + pass def _upload_source(self, source_path: str, object_key: str) -> None: self.s3.upload_file( @@ -400,94 +421,131 @@ def _schedule_upload( filename=filename, ) - def _tail(self, content: str) -> str: - position = len(content) - lines = 0 - while position > 0 and lines < _FAILED_OUTPUT_MAX_LINES: - position = content.rfind("\n", 0, position) - if position < 0: - position = 0 - break - lines += 1 - tail = content[position:] - truncated = len(tail) > _FAILED_OUTPUT_MAX_BYTES - if truncated: - tail = tail[-_FAILED_OUTPUT_MAX_BYTES:] - encoded = tail.encode("utf-8") - if len(encoded) > _FAILED_OUTPUT_MAX_BYTES: - tail = encoded[-_FAILED_OUTPUT_MAX_BYTES:].decode("utf-8", errors="replace") + def _tail(self, source_path: str) -> str: + filesize = os.path.getsize(source_path) + read_size = min(filesize, _FAILED_OUTPUT_MAX_BYTES) + with open(source_path, "rb") as source: + source.seek(filesize - read_size) + tail = source.read(read_size).decode("utf-8", errors="replace") + + lines = tail.splitlines(keepends=True) + truncated = filesize > read_size + if len(lines) > _FAILED_OUTPUT_MAX_LINES: + lines = lines[-_FAILED_OUTPUT_MAX_LINES:] truncated = True + tail = "".join(lines) if truncated: tail = "... [truncated]\n" + tail return tail - def _format_report_content(self, message: str, content: str, failed: bool) -> str: - if not failed: + @staticmethod + def _format_report_content(message: str, tail: str | None = None) -> str: + if tail is None: return f"{message}\n" - formatted = f"{message}\n\nLast {_FAILED_OUTPUT_MAX_LINES} lines:\n{self._tail(content)}" + formatted = f"{message}\n\nLast {_FAILED_OUTPUT_MAX_LINES} lines:\n{tail}" if not formatted.endswith("\n"): formatted += "\n" return formatted - def _next_filename( - self, - nodeid: str, - stream_name: str, - phase: str, - attempt: int, - ) -> str: - counts = self._capture_filename_counts.setdefault(nodeid, {}) - count_key = (attempt, stream_name, phase) - duplicate_index = counts.get(count_key, 0) - counts[count_key] = duplicate_index + 1 - - suffixes = [] + @staticmethod + def _stream_filename(stream_name: str, attempt: int) -> str: + filename = _STREAM_FILENAMES[stream_name] if attempt > 1: - suffixes.append(f"attempt-{attempt}") - if duplicate_index: - suffixes.append(str(duplicate_index)) - - filename = f"{_STREAM_FILENAMES[stream_name]}-{phase}" - if suffixes: - filename += f"-{'-'.join(suffixes)}" + filename += f"-attempt-{attempt}" return f"{filename}.log" - def _upload_section( + def _stream_message(self, stream: CapturedStream) -> str: + object_key = self._object_key(stream.test_name, stream.filename) + file_url = self._file_url(object_key) + if self.skip_upload: + return f"{stream.filesize} bytes (upload skipped, would upload to {file_url})" + return f"{stream.filesize} bytes scheduled for upload to {file_url}" + + def _append_capture( self, + nodeid: str, + attempt: int, + stream_name: str, content: str, - test_name: str, - filename: str, - ) -> CapturedSection: - source_path = self._write_spool_file(test_name, filename, content) - filesize = os.path.getsize(source_path) - object_key = self._object_key(test_name, filename) - file_url = self._file_url(object_key) + ) -> CapturedStream: + streams = self._captured_streams.setdefault(nodeid, {}) + stream_key = (attempt, stream_name) + stream = streams.get(stream_key) + if stream is None: + test_name = self._test_name(nodeid) + filename = self._stream_filename(stream_name, attempt) + source_path = self._append_spool_file(test_name, filename, content) + stream = CapturedStream( + test_name=test_name, + filename=filename, + source_path=source_path, + ) + streams[stream_key] = stream + else: + if stream.finalized: + raise RuntimeError(f"Captured output arrived after {stream.filename} was finalized") + self._append_spool_file(stream.test_name, stream.filename, content) + + stream.filesize = os.path.getsize(stream.source_path) + stream.message = self._stream_message(stream) + return stream + + def _should_inline_stream(self, stream_name: str, stream: CapturedStream) -> bool: + return ( + stream_name in ("stdout", "stderr") + and self.inline_output_max_bytes > 0 + and stream.filesize < self.inline_output_max_bytes + ) - if self.skip_upload: - message = f"{filesize} bytes (upload skipped, would upload to {file_url})" - return CapturedSection(content, message) + def _finalize_stream(self, stream_name: str, stream: CapturedStream) -> None: + if stream.finalized: + return + stream.finalized = True + if self._should_inline_stream(stream_name, stream): + self._remove_source(stream.source_path) + return + object_key = self._object_key(stream.test_name, stream.filename) + file_url = self._file_url(object_key) + if self.skip_upload: + stream.message = self._stream_message(stream) + return if self.upload_mode == "deferred": - self._schedule_upload(source_path, object_key, test_name, filename) - message = f"{filesize} bytes scheduled for upload to {file_url}" - return CapturedSection(content, message) + self._schedule_upload( + stream.source_path, + object_key, + stream.test_name, + stream.filename, + ) + stream.message = f"{stream.filesize} bytes scheduled for upload to {file_url}" + return try: - self._upload_source(source_path, object_key) + self._upload_source(stream.source_path, object_key) except Exception as exc: self._upload_failed = True logger.warning( "Upload failed. test_name: %s, filename: %s, error: %s", - test_name, - filename, + stream.test_name, + stream.filename, exc, ) - message = f"upload failed: {exc}\nsize: {filesize} bytes" - return CapturedSection(content, message, force_output=True) + stream.message = f"upload failed: {exc}\nsize: {stream.filesize} bytes" + stream.force_output = True + return - self._remove_source(source_path) - message = f"{filesize} bytes uploaded to {file_url}" - return CapturedSection(content, message) + self._remove_source(stream.source_path) + stream.message = f"{stream.filesize} bytes uploaded to {file_url}" + + def _finalize_attempt(self, nodeid: str, attempt: int) -> None: + streams = self._captured_streams.get(nodeid, {}) + for (stream_attempt, stream_name), stream in streams.items(): + if stream_attempt == attempt: + self._finalize_stream(stream_name, stream) + + def _finalize_node(self, nodeid: str) -> None: + for (_, stream_name), stream in self._captured_streams.get(nodeid, {}).items(): + self._finalize_stream(stream_name, stream) @staticmethod def _attempt(report) -> int: @@ -496,52 +554,80 @@ def _attempt(report) -> int: def _process_report(self, report, default_phase: str) -> None: nodeid = getattr(report, "nodeid", "unknown") - test_name = self._test_name(nodeid) failed = getattr(report, "outcome", None) == "failed" if getattr(report, "outcome", None) == "rerun": self._pending_reruns.add(nodeid) attempt = self._attempt(report) captured_sections = self._captured_sections.setdefault(nodeid, {}) duplicate_counts: dict[tuple[str, str], int] = {} - transformed_sections = [] + section_entries = [] + represented_streams = [] + represented_stream_set = set() for section_name, content in report.sections: match = _CAPTURE_SECTION_PATTERN.fullmatch(section_name) if match is None: - transformed_sections.append((section_name, content)) + section_entries.append((section_name, content, None)) continue stream_name = match.group(1) phase = match.group(2) or default_phase duplicate_key = (stream_name, phase) duplicate_index = duplicate_counts.get(duplicate_key, 0) duplicate_counts[duplicate_key] = duplicate_index + 1 - if self._should_inline(stream_name, content): - transformed_sections.append((section_name, content)) - continue section_key = (stream_name, phase, duplicate_index) captured_section = captured_sections.get(section_key) if captured_section is None or captured_section.content != content: - filename = self._next_filename( + self._append_capture( nodeid, - stream_name, - phase, attempt, - ) - captured_section = self._upload_section( + stream_name, content, - test_name, - filename, ) + captured_section = CapturedSection(content, attempt) captured_sections[section_key] = captured_section + stream_key = (captured_section.attempt, stream_name) + section_entries.append((section_name, content, stream_key)) + if stream_key not in represented_stream_set: + represented_stream_set.add(stream_key) + represented_streams.append(stream_key) + + failure_tails = {} + streams = self._captured_streams.get(nodeid, {}) + if failed: + for stream_key in represented_streams: + stream_attempt, stream_name = stream_key + stream = streams[stream_key] + if stream_attempt == attempt and not self._should_inline_stream( + stream_name, stream + ): + failure_tails[stream_key] = self._tail(stream.source_path) + + if default_phase in ("teardown", "collect"): + self._finalize_attempt(nodeid, attempt) + + transformed_sections = [] + reported_streams = set() + for section_name, content, stream_key in section_entries: + if stream_key is None: + transformed_sections.append((section_name, content)) + continue + _, stream_name = stream_key + stream = streams[stream_key] + if self._should_inline_stream(stream_name, stream): + transformed_sections.append((section_name, content)) + continue + if stream_key in reported_streams: + continue + reported_streams.add(stream_key) + + tail = failure_tails.get(stream_key) + if tail is None and stream.force_output: + tail = self._tail(stream.source_path) transformed_sections.append( ( - section_name, - self._format_report_content( - captured_section.message, - content, - failed or captured_section.force_output, - ), + f"Captured {stream_name}", + self._format_report_content(stream.message, tail), ) ) report.sections[:] = transformed_sections @@ -560,19 +646,22 @@ def pytest_collectreport(self, report) -> None: nodeid = getattr(report, "nodeid", "unknown") self._test_names.pop(nodeid, None) self._captured_sections.pop(nodeid, None) - self._capture_filename_counts.pop(nodeid, None) + self._captured_streams.pop(nodeid, None) self._pending_reruns.discard(nodeid) def pytest_runtest_logfinish(self, nodeid: str, location) -> None: + self._finalize_node(nodeid) self._test_names.pop(nodeid, None) if nodeid in self._pending_reruns: self._pending_reruns.remove(nodeid) return self._captured_sections.pop(nodeid, None) - self._capture_filename_counts.pop(nodeid, None) + self._captured_streams.pop(nodeid, None) @pytest.hookimpl(tryfirst=True) def pytest_sessionfinish(self, session, exitstatus) -> None: + for nodeid in tuple(self._captured_streams): + self._finalize_node(nodeid) if self._executor is not None: self._executor.shutdown(wait=True) self._executor = None @@ -585,7 +674,7 @@ def pytest_sessionfinish(self, session, exitstatus) -> None: except OSError: pass self._captured_sections.clear() - self._capture_filename_counts.clear() + self._captured_streams.clear() self._pending_reruns.clear() diff --git a/tests/unittest/test_s3_output.py b/tests/unittest/test_s3_output.py index 1b91e9375ddb..67eb00e56911 100644 --- a/tests/unittest/test_s3_output.py +++ b/tests/unittest/test_s3_output.py @@ -140,6 +140,7 @@ def test_small_stdout_remains_inline(tmp_path): report = Report([("Captured stdout call", "ok\n")]) process_report(plugin, report) + plugin.pytest_runtest_logfinish(report.nodeid, None) assert report.sections == [("Captured stdout call", "ok\n")] assert not s3_output._spool_root(str(tmp_path)).exists() @@ -152,12 +153,32 @@ def test_stdout_at_threshold_is_replaced_with_url(tmp_path): process_report(plugin, report) section_name, section_content = report.sections[0] - assert section_name == "Captured stdout call" + assert section_name == "Captured stdout" assert "4 bytes (upload skipped" in section_content - assert "/stdout-call.log" in section_content + assert "/stdout.log" in section_content assert "four" not in section_content +def test_inline_threshold_applies_to_combined_stream(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=8) + setup_section = ("Captured stdout setup", "abc") + call_section = ("Captured stdout call", "defgh") + setup_report = Report([setup_section], when="setup") + call_report = Report([setup_section, call_section]) + teardown_report = Report([setup_section, call_section], when="teardown") + + process_report(plugin, setup_report) + process_report(plugin, call_report) + process_report(plugin, teardown_report) + + assert setup_report.sections == [setup_section] + assert len(call_report.sections) == 1 + assert "/stdout.log" in call_report.sections[0][1] + assert len(teardown_report.sections) == 1 + stdout_file = next(s3_output._spool_root(str(tmp_path)).rglob("stdout.log")) + assert stdout_file.read_text(encoding="utf-8") == "abcdefgh" + + def test_logging_section_is_uploaded_even_when_small(tmp_path): plugin = make_plugin(tmp_path, inline_output_max_bytes=256) report = Report([("Captured log call", "log\n")]) @@ -165,7 +186,7 @@ def test_logging_section_is_uploaded_even_when_small(tmp_path): process_report(plugin, report) assert "upload skipped" in report.sections[0][1] - assert "/logging-call.log" in report.sections[0][1] + assert "/logging.log" in report.sections[0][1] def test_sync_upload_transforms_native_sections(tmp_path, monkeypatch): @@ -181,7 +202,8 @@ def test_sync_upload_transforms_native_sections(tmp_path, monkeypatch): ("Captured stdout setup", "setup output\n"), ("custom", "keep me"), ("Captured stderr call", "call error\n"), - ] + ], + when="teardown", ) process_report(plugin, report) @@ -191,14 +213,14 @@ def test_sync_upload_transforms_native_sections(tmp_path, monkeypatch): b"setup output\n", b"call error\n", ] - assert client.uploads[0][2].endswith("/stdout-setup.log") - assert client.uploads[1][2].endswith("/stderr-call.log") + assert client.uploads[0][2].endswith("/stdout.log") + assert client.uploads[1][2].endswith("/stderr.log") assert report.sections[1] == ("custom", "keep me") assert all("uploaded to" in report.sections[index][1] for index in (0, 2)) assert not s3_output._spool_root(str(tmp_path)).exists() -def test_duplicate_capture_sections_get_distinct_objects(tmp_path): +def test_duplicate_capture_sections_share_one_object(tmp_path): plugin = make_plugin(tmp_path, inline_output_max_bytes=0) report = Report( [ @@ -209,9 +231,10 @@ def test_duplicate_capture_sections_get_distinct_objects(tmp_path): process_report(plugin, report) - assert "/stdout-call.log" in report.sections[0][1] - assert "/stdout-call-1.log" in report.sections[1][1] - assert all(content.endswith("\n") for _, content in report.sections) + assert len(report.sections) == 1 + assert "/stdout.log" in report.sections[0][1] + stdout_file = next(s3_output._spool_root(str(tmp_path)).rglob("stdout.log")) + assert stdout_file.read_text(encoding="utf-8") == "firstsecond" def test_cumulative_capture_sections_are_uploaded_once(tmp_path, monkeypatch): @@ -239,26 +262,26 @@ def test_cumulative_capture_sections_are_uploaded_once(tmp_path, monkeypatch): process_report(plugin, setup_report) process_report(plugin, call_report) + assert client.uploads == [] process_report(plugin, teardown_report) plugin.pytest_runtest_logfinish(setup_report.nodeid, None) plugin.pytest_sessionfinish(None, 0) assert [upload[0] for upload in client.uploads] == [ - b"setup output\n", - b"call output\n", + b"setup output\ncall output\n", b"teardown error\n", ] assert [upload[2].rsplit("/", 1)[-1] for upload in client.uploads] == [ - "stdout-setup.log", - "stdout-call.log", - "stderr-teardown.log", + "stdout.log", + "stderr.log", ] assert "Last 200 lines:" in call_report.sections[0][1] + assert "setup output" in call_report.sections[0][1] + assert "call output" in call_report.sections[0][1] assert "Last 200 lines:" not in teardown_report.sections[0][1] - assert ( - setup_report.sections[0][1].splitlines()[0] - == (teardown_report.sections[0][1].splitlines()[0]) - ) + assert sum("/stdout.log" in content for _, content in teardown_report.sections) == 1 + assert sum("/stderr.log" in content for _, content in teardown_report.sections) == 1 + assert len(teardown_report.sections) == 2 assert all(content.endswith("\n") for _, content in teardown_report.sections) @@ -315,6 +338,7 @@ def test_deferred_upload_starts_before_session_finish(tmp_path, monkeypatch): report = Report([("Captured stdout call", "background output")]) process_report(plugin, report) + plugin.pytest_runtest_logfinish(report.nodeid, None) assert client.started.wait(timeout=5) assert "scheduled for upload" in report.sections[0][1] @@ -339,9 +363,9 @@ def test_deferred_upload_reuses_cumulative_section(tmp_path, monkeypatch): teardown_report = Report([section], when="teardown") process_report(plugin, call_report) - assert client.started.wait(timeout=5) process_report(plugin, teardown_report) + assert client.started.wait(timeout=5) assert len(plugin._pending_uploads) == 1 assert ( call_report.sections[0][1].splitlines()[0] @@ -406,10 +430,10 @@ def test_rerun_keeps_one_url_per_attempt(tmp_path, monkeypatch): b"second stderr\n", ] object_keys = [upload[2] for upload in client.uploads] - assert object_keys[0].endswith("/stdout-call.log") - assert object_keys[1].endswith("/stderr-call.log") - assert object_keys[2].endswith("/stdout-call-attempt-2.log") - assert object_keys[3].endswith("/stderr-call-attempt-2.log") + assert object_keys[0].endswith("/stdout.log") + assert object_keys[1].endswith("/stderr.log") + assert object_keys[2].endswith("/stdout-attempt-2.log") + assert object_keys[3].endswith("/stderr-attempt-2.log") assert object_keys[0].rsplit("/", 1)[0] != object_keys[2].rsplit("/", 1)[0] assert all(content.endswith("\n") for _, content in teardown_report.sections) assert "Last 200 lines:" not in teardown_report.sections[0][1] @@ -422,7 +446,7 @@ def test_parent_drain_retries_upload_left_by_failed_process(tmp_path, monkeypatc FailingS3Client(), inline_output_max_bytes=0, ) - report = Report([("Captured stdout call", "recover me")]) + report = Report([("Captured stdout call", "recover me")], when="teardown") process_report(plugin, report) assert "upload failed" in report.sections[0][1] @@ -435,7 +459,7 @@ def test_parent_drain_retries_upload_left_by_failed_process(tmp_path, monkeypatc monkeypatch.setattr(s3_output, "_create_s3_client", lambda *args: client) assert s3_output.drain_pending_uploads(str(tmp_path), secret_key="secret") assert client.uploads[0][0] == b"recover me" - assert client.uploads[0][2].endswith("/stdout-call.log") + assert client.uploads[0][2].endswith("/stdout.log") assert not s3_output._spool_root(str(tmp_path)).exists() @@ -548,11 +572,20 @@ def test_native_capture_is_replaced_before_junit_consumes_report(tmp_path): "import os\n" "import subprocess\n" "import sys\n" - "def test_output():\n" + "import pytest\n" + "@pytest.fixture\n" + "def output_fixture():\n" + " os.write(1, b'setup stdout\\n')\n" + " os.write(2, b'setup stderr ' + b'y' * 300 + b'\\n')\n" + " yield\n" + " os.write(1, b'teardown stdout\\n')\n" + " os.write(2, b'teardown stderr ' + b'y' * 300 + b'\\n')\n" + "def test_output(output_fixture):\n" " os.write(1, b'parent before\\n')\n" " subprocess.run([sys.executable, '-c', " "\"import os; os.write(1, b'child capture ' + b'x' * 300 + b'\\\\n')\"], check=True)\n" - " os.write(1, b'parent after\\n')\n", + " os.write(1, b'parent after\\n')\n" + " os.write(2, b'call stderr ' + b'y' * 300 + b'\\n')\n", encoding="utf-8", ) output_dir = tmp_path / "output" @@ -583,11 +616,28 @@ def test_native_capture_is_replaced_before_junit_consumes_report(tmp_path): xml = xml_path.read_text(encoding="utf-8") assert "upload skipped" in xml assert "child capture xxxxx" not in xml - assert "stdout-call.log" in xml - stdout_files = list(s3_output._spool_root(str(output_dir)).rglob("stdout-call.log")) + assert xml.count("stdout.log") == 1 + assert xml.count("stderr.log") == 1 + stdout_files = list(s3_output._spool_root(str(output_dir)).rglob("stdout.log")) assert len(stdout_files) == 1 assert stdout_files[0].read_bytes() == ( - b"parent before\n" + b"child capture " + b"x" * 300 + b"\nparent after\n" + b"setup stdout\n" + + b"parent before\n" + + b"child capture " + + b"x" * 300 + + b"\nparent after\n" + + b"teardown stdout\n" + ) + stderr_files = list(s3_output._spool_root(str(output_dir)).rglob("stderr.log")) + assert len(stderr_files) == 1 + assert stderr_files[0].read_bytes() == ( + b"setup stderr " + + b"y" * 300 + + b"\ncall stderr " + + b"y" * 300 + + b"\nteardown stderr " + + b"y" * 300 + + b"\n" ) @@ -633,22 +683,20 @@ def test_rerun_urls_are_distinct_and_line_delimited_in_junit(tmp_path): element.text or "" for element in ET.parse(xml_path).findall(".//system-out") ) url_lines = [ - line - for line in system_output.splitlines() - if "upload skipped" in line and "stdout-call" in line + line for line in system_output.splitlines() if "upload skipped" in line and "stdout" in line ] assert len(url_lines) == 2, system_output - assert any("/stdout-call.log)" in line for line in url_lines) - assert any("/stdout-call-attempt-2.log)" in line for line in url_lines) + assert any("/stdout.log)" in line for line in url_lines) + assert any("/stdout-attempt-2.log)" in line for line in url_lines) - stdout_files = list(s3_output._spool_root(str(output_dir)).rglob("stdout-call*.log")) + stdout_files = list(s3_output._spool_root(str(output_dir)).rglob("stdout*.log")) assert {path.name for path in stdout_files} == { - "stdout-call.log", - "stdout-call-attempt-2.log", + "stdout.log", + "stdout-attempt-2.log", } contents = {path.name: path.read_bytes() for path in stdout_files} - assert contents["stdout-call.log"].startswith(b"attempt 1 ") - assert contents["stdout-call-attempt-2.log"].startswith(b"attempt 2 ") + assert contents["stdout.log"].startswith(b"attempt 1 ") + assert contents["stdout-attempt-2.log"].startswith(b"attempt 2 ") def test_xdist_worker_transforms_report_before_controller_junit(tmp_path): @@ -689,7 +737,7 @@ def test_xdist_worker_transforms_report_before_controller_junit(tmp_path): xml = xml_path.read_text(encoding="utf-8") assert "upload skipped" in xml assert "xdist capture xxxxx" not in xml - assert "stdout-call.log" in xml + assert "stdout.log" in xml def test_native_capture_restores_timeout_output_to_console(tmp_path):