diff --git a/jenkins/L0_Test.groovy b/jenkins/L0_Test.groovy index ee9654166f9c..af5ca6b818bb 100644 --- a/jenkins/L0_Test.groovy +++ b/jenkins/L0_Test.groovy @@ -48,6 +48,8 @@ UPLOAD_PATH = env.uploadPath ? env.uploadPath : "sw-tensorrt-generic/llm-artifac 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 +// Kill switch for the scoped post-merge multi-GPU echo (shouldEchoTestOutputToConsole). +DISABLE_POST_MERGE_STDOUT_ECHO = params.disableStdoutEchoOnPostMerge != null ? params.disableStdoutEchoOnPostMerge : false X86_64_TRIPLE = "x86_64-linux-gnu" AARCH64_TRIPLE = "aarch64-linux-gnu" @@ -1249,6 +1251,10 @@ def getPytestBaseCommandLine( testCmdLine += ["--unittest-markexpr='${unittestMarkExpr}'"] if (ENABLE_UPLOAD_TEST_RESULTS) { testCmdLine += ["-o console_output_style=progress-even-when-capture-no"] + // ENABLE_S3_ECHO_STDOUT already appends this at the call site; don't duplicate. + if (!ENABLE_S3_ECHO_STDOUT && shouldEchoTestOutputToConsole(stageName)) { + testCmdLine += ["--s3-echo-stdout"] + } } if (extraArgs) { testCmdLine += extraArgs @@ -1256,6 +1262,41 @@ def getPytestBaseCommandLine( return testCmdLine as String[] } +// Whether a stage should echo per-test stdout/stderr to the console as well as +// capturing it for S3. +// +// pytest capture is already off (-s), but the S3 log plugin then spools every +// byte a test -- and every MPI worker rank that inherited its fds -- writes +// into a session file, and publishes it per test on completion. A test that +// wedges never completes: pytest's --timeout hard-kills the process with +// os._exit, so that test's output is never published and the stage log holds +// no trace of the wedge. The output is not destroyed (the spool ships inside +// results-.tar.gz), but recovering it means knowing to download and +// unpack an artifact, which is not how a stage failure gets triaged. +// +// What echoing recovers, precisely: everything written up to roughly a quarter +// second before the process dies, which covers a HangDetector report. It does +// not recover pytest-timeout's own stack dump, which is written immediately +// before os._exit -- that still only reaches the spool, as it does today. +// +// Echoing costs log volume, so it is limited to where the evidence is worth +// most: post-merge multi-GPU stages. Those are the stages whose timeouts burn +// the most GPU-hours per occurrence, and the wedges there are the ones that +// leave nothing behind today. PerfSanity stages are excluded so that timing +// runs keep their current, quieter console. Set the disableStdoutEchoOnPostMerge +// build parameter to turn this off without a code change; the plugin also caps +// the echoed bytes per session. +def shouldEchoTestOutputToConsole(String stageName) { + if (DISABLE_POST_MERGE_STDOUT_ECHO) { + return false + } + if (!stageName.contains("Post-Merge") || stageName.contains("PerfSanity")) { + return false + } + def taskConfig = parseTaskConfigFromStageName(stageName) + return taskConfig != null && (taskConfig.system_gpu_count as Integer) > 1 +} + def getMountListForSlurmTest(SlurmCluster cluster, boolean useSbatch = false) { def mounts = [] diff --git a/tests/test_common/s3_output.py b/tests/test_common/s3_output.py index 4f3f8f272770..398d39137a2e 100644 --- a/tests/test_common/s3_output.py +++ b/tests/test_common/s3_output.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import argparse +import codecs import io import logging import os @@ -106,6 +107,11 @@ def __init__(self, target_fd, path): self._spool_fd = None self._attached = False + @property + def console_fd(self): + """The pre-capture duplicate of the target fd, i.e. the real console.""" + return self._saved_fd + def _flush_target_stream(self): stream = sys.stdout if self.target_fd == 1 else sys.stderr try: @@ -160,8 +166,178 @@ def stop(self): self._saved_fd = None +@dataclass +class _SpoolReader: + path: str + stream: io.RawIOBase + console_fd: int + + +class SessionSpoolEchoer: + """Mirror the session spool files onto the real console as they grow. + + Session capture points fd 1 and 2 at append-only spool files, so nothing a + test -- or any MPI worker rank that inherited those fds -- writes reaches + the console; it is published only once the test finishes and its slice is + uploaded. A test that wedges never finishes, so its output is never + published and the stage log holds no trace of it. + + Tailing the spool puts that output on the console while it is produced, + without changing how capture, per-test slicing, or uploading work. + + What this does and does not recover, precisely: everything written more + than one poll interval before the process dies reaches the console, which + covers a HangDetector report (the detector logs, dumps stacks, and only + then hard-kills). It does *not* recover output written in the last instants + before an abrupt exit -- notably pytest-timeout's own stack dump, which it + writes and then immediately calls ``os._exit``. That dump still reaches the + spool file, and so the ``results-.tar.gz`` artifact, exactly as it + does today; it is simply not on the console. + + The console copy is strictly best effort. The spool is written + synchronously by the producer and is never affected by anything here: a + reader that cannot be read or written is dropped, and echoing stops after + ``max_bytes`` so a pathological test cannot flood the console. + """ + + POLL_INTERVAL_SECONDS = 0.25 + # Console output is bounded so one runaway test cannot flood the Jenkins + # log. A post-merge multi-GPU stage captures single-digit MB in total, so + # this only trips on pathological output, and says where the rest lives. + DEFAULT_MAX_BYTES = 128 * 1024 * 1024 + _READ_CHUNK_BYTES = 1 << 20 + # A write of at most PIPE_BUF bytes is atomic on a pipe, so lines never + # interleave with pytest's own progress output or with another writer. + _WRITE_CHUNK_BYTES = 4096 + # A graceful exit needs one poll interval plus a drain. Anything longer + # means the thread is stuck writing to a blocked console, and stalling + # session teardown on it buys nothing: the spool is already complete. + _JOIN_TIMEOUT_SECONDS = 2.0 + + def __init__(self, spools, poll_interval=POLL_INTERVAL_SECONDS, max_bytes=DEFAULT_MAX_BYTES): + self._spools = list(spools) + self._poll_interval = poll_interval + self._max_bytes = max_bytes + self._echoed_bytes = 0 + self._readers = [] + self._stop = threading.Event() + self._closed = False + self._thread = None + + def start(self): + for spool in self._spools: + if spool.console_fd is None: + continue + try: + stream = open(spool.path, "rb", buffering=0) + except OSError as e: + logger.warning("Cannot tail spool %s for console echo: %s", spool.path, e) + continue + self._readers.append(_SpoolReader(spool.path, stream, spool.console_fd)) + if not self._readers: + return + self._thread = threading.Thread(target=self._loop, daemon=True, name="s3-spool-echo") + self._thread.start() + + def _loop(self): + while not self._stop.is_set() and self._readers: + self._drain() + self._stop.wait(self._poll_interval) + # Final pass so a graceful stop does not truncate the console copy. + self._drain() + + def _drain(self): + for reader in list(self._readers): + if self._closed: + return + try: + while True: + data = reader.stream.read(self._READ_CHUNK_BYTES) + if not data: + break + if not self._echo(reader, data): + break + except (OSError, ValueError) as e: + # ValueError: stop() closed the stream underneath us. + self._drop(reader, f"cannot read spool: {e}") + + def _echo(self, reader, data): + """Write ``data`` to the console. Returns False once this reader is done.""" + if self._max_bytes is not None: + remaining = self._max_bytes - self._echoed_bytes + if remaining <= 0: + self._exhaust_budget() + return False + data = data[:remaining] + try: + self._write_all(reader.console_fd, data) + except (OSError, ValueError) as e: + # A broken console (EPIPE) never heals. Drop the reader instead of + # retrying every poll: the warning would go to stderr, i.e. back + # into the spool we are reading, and feed itself. + self._drop(reader, f"console write failed: {e}") + return False + self._echoed_bytes += len(data) + if self._max_bytes is not None and self._echoed_bytes >= self._max_bytes: + self._exhaust_budget() + return False + return True + + def _exhaust_budget(self): + if not self._readers: + return + logger.warning( + "Console echo capped after %d bytes; the full output is in the stage's " + "results-.tar.gz artifact", + self._echoed_bytes, + ) + for reader in list(self._readers): + self._close_reader(reader) + self._readers = [] + + def _drop(self, reader, reason): + logger.warning("Console echo of %s stopped: %s", reader.path, reason) + self._close_reader(reader) + try: + self._readers.remove(reader) + except ValueError: + pass + + @staticmethod + def _close_reader(reader): + try: + reader.stream.close() + except (OSError, ValueError): + pass + + @classmethod + def _write_all(cls, fd, data): + view = memoryview(data) + while view: + written = os.write(fd, view[: cls._WRITE_CHUNK_BYTES]) + view = view[written:] + + def stop(self): + self._stop.set() + thread, self._thread = self._thread, None + if thread is not None: + thread.join(timeout=self._JOIN_TIMEOUT_SECONDS) + if thread.is_alive(): + # Do not close the streams the thread may still be reading: + # that turns into a ValueError on a closed file, which escapes + # into threading.excepthook and prints a traceback on the real + # stderr -- the one place a triager is looking. The process is + # ending anyway, so leaking two descriptors is the cheaper bug. + logger.warning("Spool echo thread still running; leaving its readers open") + return + self._closed = True + for reader in self._readers: + self._close_reader(reader) + self._readers = [] + + class SessionCapture: - def __init__(self, output_path): + def __init__(self, output_path, echo_to_console=False): spool_dir = os.path.join(output_path, ".s3-spool") os.makedirs(spool_dir, exist_ok=True) suffix = f"{os.getpid()}-{time.time_ns()}" @@ -171,6 +347,8 @@ def __init__(self, output_path): } self._suspend_depth = 0 self._started = False + self._echo_to_console = echo_to_console + self._echoer = None def start(self): started = [] @@ -183,6 +361,19 @@ def start(self): spool.stop() raise self._started = True + if self._echo_to_console: + # Console echo is a diagnostic nicety layered on top of capture. If + # it cannot start -- a thread limit on a many-rank stage, say -- + # swallow it: raising here would escape pytest_load_initial_conftests + # before the capture cleanup is registered, leaving fd 1 and 2 pinned + # to the spool and pytest's own traceback inside it. + try: + echoer = SessionSpoolEchoer(self._spools.values()) + echoer.start() + self._echoer = echoer + except Exception as e: # noqa: BLE001 - echo must never fail the session + logger.warning("Console echo disabled: %s", e) + self._echoer = None def snapshot(self): return {filename: spool.snapshot() for filename, spool in self._spools.items()} @@ -218,6 +409,10 @@ def stop(self): if not self._started: return self._suspend_depth = 0 + # Drain the console copy before the spool fds go away. + if self._echoer is not None: + self._echoer.stop() + self._echoer = None for spool in self._spools.values(): spool.stop() self._started = False @@ -285,7 +480,13 @@ def __enter__(self): 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) + # Unbuffered binary: FileIO.read(n) issues a single read() syscall and + # returns whatever is already in the pipe. A buffered *text* stream + # would instead block until it has n characters or sees EOF, so the + # trailing partial chunk -- exactly where a hang dump lands, since the + # writer then stops producing -- would never be drained before the + # process is killed. + pipe_stream = os.fdopen(pipe_read, "rb", buffering=0) # Child processes may inherit the redirected fd and keep the pipe open # after capture is restored. Do not let that block pytest shutdown. @@ -298,13 +499,29 @@ def __enter__(self): def _reader_loop(self, pipe_stream, log_file): need_timestamp = True + # A raw pipe read can split a multi-byte UTF-8 sequence, so decode + # incrementally instead of per chunk. + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") try: while True: - chunk = pipe_stream.read(4096) - if not chunk: + raw = pipe_stream.read(4096) + if not raw: break + # Echo first: the whole point of echoing is live visibility, and + # the writer may be killed at any moment. + if self.echo_to_original and self.saved_fd is not None: + ret = os.write(self.saved_fd, raw) + if ret != len(raw): + logger.warning( + f"Partial write to original FD {self.target_fd}: {ret} != {len(raw)}" + ) + + chunk = decoder.decode(raw) + if not chunk: + continue + # 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 @@ -328,13 +545,6 @@ def _reader_loop(self, pipe_stream, log_file): 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: @@ -444,7 +654,9 @@ def __init__( 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: + # "direct" dup2s the target fd straight onto a log file, so there is + # nothing in the process that could copy the bytes on to the console. + if self.capture_mode == "direct" and self.echo_to_stdout: raise ValueError( f"--s3-capture-mode={self.capture_mode} cannot be used with --s3-echo-stdout" ) @@ -593,7 +805,9 @@ 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 = SessionCapture( + self.output_path, echo_to_console=self.echo_to_stdout + ) self._session_capture.start() else: self._session_capture.resume_parent() @@ -916,11 +1130,13 @@ def add_options(parser): 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.", + "them through to the original stdout/stderr for live debugging, so that " + "output survives a test that is hard-killed before its capture is " + "published. Supported by --s3-capture-mode=session (the default, echoed " + "by tailing the spool) and =timestamped; not by =direct. Set this on the " + "outer pytest invocation only; nested pytest invocations spawned by " + "individual tests should NOT set it, to avoid duplicating their output " + "back through the outer stream.", ) parser.addoption( "--s3-skip-upload", diff --git a/tests/test_common/s3_output_hooks.py b/tests/test_common/s3_output_hooks.py index ccaf34f6ea02..351f5eadc339 100644 --- a/tests/test_common/s3_output_hooks.py +++ b/tests/test_common/s3_output_hooks.py @@ -54,11 +54,12 @@ def _is_xdist_controller(config: pytest.Config) -> bool: def _parse_early_capture_options( early_config: pytest.Config, args: list[str] -) -> tuple[str | None, str | None, str, str]: +) -> tuple[str | None, str | None, str, str, bool]: 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") + parser.add_argument("--s3-echo-stdout", action="store_true", default=False) parsed, _ = parser.parse_known_args(args) namespace = early_config.known_args_namespace @@ -76,8 +77,9 @@ def _parse_early_capture_options( str, getattr(namespace, "s3_capture_mode", None) or parsed.s3_capture_mode or "session", ) + echo_to_console = bool(getattr(namespace, "s3_echo_stdout", False) or parsed.s3_echo_stdout) pytest_capture = cast(str, getattr(namespace, "capture", "fd")) - return output_path, upload_path, capture_mode, pytest_capture + return output_path, upload_path, capture_mode, pytest_capture, echo_to_console def _capture_state(config: pytest.Config) -> _EarlyCaptureState | None: @@ -91,9 +93,13 @@ def _capture_state(config: pytest.Config) -> _EarlyCaptureState | None: 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 - ) + ( + output_path, + upload_path, + capture_mode, + pytest_capture, + echo_to_console, + ) = _parse_early_capture_options(early_config, args) if ( _capture_state(early_config) is not None or _is_xdist_controller(early_config) @@ -104,7 +110,7 @@ def pytest_load_initial_conftests( ): return (yield) - capture = s3_output.SessionCapture(output_path) + capture = s3_output.SessionCapture(output_path, echo_to_console=echo_to_console) capture.start() state = _EarlyCaptureState(capture) setattr(early_config, _CAPTURE_STATE_ATTRIBUTE, state) diff --git a/tests/unittest/test_s3_output.py b/tests/unittest/test_s3_output.py index 899ad5a8fe27..653fe04506a9 100644 --- a/tests/unittest/test_s3_output.py +++ b/tests/unittest/test_s3_output.py @@ -18,10 +18,11 @@ import os import subprocess import sys +import time from types import SimpleNamespace import pytest -from test_common import s3_output_hooks +from test_common import s3_output, s3_output_hooks from test_common.s3_output import ( FDRedirector, FileSlice, @@ -454,6 +455,209 @@ def test_small_log_file_is_not_inlined(tmp_path): assert "upload skipped" in section_content +def _drain_console_pipe(read_fd, expected, timeout=10.0): + """Read from ``read_fd`` until ``expected`` is seen or ``timeout`` elapses.""" + deadline = time.monotonic() + timeout + seen = b"" + os.set_blocking(read_fd, False) + while time.monotonic() < deadline: + try: + chunk = os.read(read_fd, 65536) + except BlockingIOError: + chunk = b"" + if chunk: + seen += chunk + if expected in seen: + break + else: + time.sleep(0.05) + return seen + + +def test_session_capture_echoes_output_to_console_before_the_test_ends(tmp_path): + """Echo must appear while output is produced, not at teardown. + + A wedged test is killed before its capture is ever published. + """ + console_read, console_write = os.pipe() + saved_stdout = os.dup(1) + capture = SessionCapture(str(tmp_path), echo_to_console=True) + child = None + try: + os.dup2(console_write, 1) + capture.start() + # A child that inherits fd 1 and then never exits: the MPI worker case. + child = subprocess.Popen( + [ + sys.executable, + "-c", + "import os, time; os.write(1, b'Hang detected after 5 seconds.\\n'); " + "time.sleep(300)", + ], + ) + seen = _drain_console_pipe(console_read, b"Hang detected after 5 seconds.") + finally: + if child is not None: + child.kill() + child.wait() + capture.stop() + os.dup2(saved_stdout, 1) + os.close(saved_stdout) + os.close(console_write) + os.close(console_read) + + assert b"Hang detected after 5 seconds." in seen + capture.remove_files() + + +def test_session_capture_does_not_echo_to_console_by_default(tmp_path): + console_read, console_write = os.pipe() + saved_stdout = os.dup(1) + capture = SessionCapture(str(tmp_path)) + try: + os.dup2(console_write, 1) + capture.start() + os.write(1, b"spooled only\n") + capture.stop() + finally: + os.dup2(saved_stdout, 1) + os.close(saved_stdout) + os.close(console_write) + try: + os.set_blocking(console_read, False) + try: + leaked = os.read(console_read, 65536) + except BlockingIOError: + leaked = b"" + finally: + os.close(console_read) + + assert b"spooled only" not in leaked + capture.remove_files() + + +def test_session_echo_drops_a_reader_whose_console_is_gone(tmp_path): + """A broken console must not be retried every poll. + + The warning would go to stderr, into the spool being read, and feed itself. + """ + console_read, console_write = os.pipe() + saved_stdout = os.dup(1) + capture = SessionCapture(str(tmp_path), echo_to_console=True) + try: + os.dup2(console_write, 1) + capture.start() + echoer = capture._echoer + os.close(console_read) # reader gone -> EPIPE on the next write + os.write(1, b"first\n") + deadline = time.monotonic() + 10.0 + while echoer._readers and time.monotonic() < deadline: + time.sleep(0.05) + for _ in range(5): + os.write(1, b"more\n") + time.sleep(0.1) + finally: + capture.stop() + os.dup2(saved_stdout, 1) + os.close(saved_stdout) + os.close(console_write) + + # Both readers dropped, so the loop exits instead of warning every poll. + assert echoer._readers == [] + assert not echoer._thread + capture.remove_files() + + +def test_session_echo_stops_after_the_byte_budget(tmp_path): + console_read, console_write = os.pipe() + saved_stdout = os.dup(1) + spool = s3_output.SessionFDSpool(1, str(tmp_path / "stdout.log")) + payload = b"x" * 512 + try: + os.dup2(console_write, 1) + spool.start() + echoer = s3_output.SessionSpoolEchoer([spool], poll_interval=0.02, max_bytes=1024) + echoer.start() + for _ in range(20): + os.write(1, payload) + deadline = time.monotonic() + 10.0 + while echoer._readers and time.monotonic() < deadline: + time.sleep(0.02) + echoer.stop() + spool.stop() + finally: + os.dup2(saved_stdout, 1) + os.close(saved_stdout) + os.close(console_write) + + seen = _drain_console_pipe(console_read, b"\0", timeout=0.5) + os.close(console_read) + assert echoer._echoed_bytes <= 1024 + assert len(seen) <= 1024 + # The spool itself is untouched by the cap. + assert (tmp_path / "stdout.log").stat().st_size == 20 * len(payload) + + +def test_session_capture_survives_an_echo_thread_that_cannot_start(tmp_path, monkeypatch): + """Echo is best effort; failing to start it must not abort the session.""" + + def boom(self): + raise RuntimeError("can't start new thread") + + monkeypatch.setattr(s3_output.SessionSpoolEchoer, "start", boom) + capture = SessionCapture(str(tmp_path), echo_to_console=True) + capture.start() + try: + assert capture._started + assert capture._echoer is None + os.write(1, b"still spooled\n") + finally: + capture.stop() + assert "still spooled" in (open(capture._spools["stdout.log"].path, errors="replace").read()) + capture.remove_files() + + +def test_session_echo_writes_are_atomic_sized(tmp_path): + """Writes over PIPE_BUF are not atomic on a pipe and splice other writers.""" + assert s3_output.SessionSpoolEchoer._WRITE_CHUNK_BYTES <= 4096 + + written = [] + fake_fd = object() + monkey = lambda fd, buf: (written.append(len(buf)), len(buf))[1] # noqa: E731 + real_write = os.write + os.write = monkey + try: + s3_output.SessionSpoolEchoer._write_all(fake_fd, b"y" * 10000) + finally: + os.write = real_write + assert max(written) <= 4096 + assert sum(written) == 10000 + + +def test_fd_redirector_echoes_a_partial_chunk_without_waiting_for_more(tmp_path): + """The reader must not wait to fill a buffer before echoing. + + The bytes a hang leaves behind are a short final chunk with nothing to follow. + """ + console_read, console_write = os.pipe() + saved_stdout = os.dup(1) + redir = FDRedirector(1, str(tmp_path / "stdout.log"), echo_to_original=True) + try: + os.dup2(console_write, 1) + redir.__enter__() + os.write(1, b"Hang detected after 5 seconds.\n") # far below one 4 KiB chunk + seen = _drain_console_pipe(console_read, b"Hang detected after 5 seconds.", timeout=5.0) + redir.__exit__(None, None, None) + finally: + os.dup2(saved_stdout, 1) + os.close(saved_stdout) + os.close(console_write) + os.close(console_read) + + assert b"Hang detected after 5 seconds." in seen + assert "Hang detected after 5 seconds." in (tmp_path / "stdout.log").read_text() + + def test_fd_redirector_reader_thread_is_daemon_when_pipe_writer_lingers(tmp_path): redir = FDRedirector(1, str(tmp_path / "stdout.log")) proc = None diff --git a/tests/unittest/tools/test_test_to_stage_mapping.py b/tests/unittest/tools/test_test_to_stage_mapping.py index f8689ff12398..b5671b303b9c 100644 --- a/tests/unittest/tools/test_test_to_stage_mapping.py +++ b/tests/unittest/tools/test_test_to_stage_mapping.py @@ -79,8 +79,18 @@ 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_stdout_echo_is_opt_in_or_scoped_to_post_merge_multi_gpu(): + """Keep Jenkins pytest progress readable: echo only where it is justified. + + Live echo used to be reachable only through the ``enableS3EchoStdout`` + build parameter. It is now also enabled for post-merge multi-GPU stages, + because a test that wedges there is hard-killed before the S3 capture + plugin publishes its output, leaving the stage log with no trace of the + wedge. That is a deliberate widening, so this test no longer requires the + parameter guard -- but every ``--s3-echo-stdout`` path must still be + guarded by one of exactly two things, so echo cannot spread to pre-merge + stages unnoticed. + """ with open(GROOVY, 'r') as f: lines = f.readlines() @@ -93,9 +103,30 @@ def test_s3_stdout_echo_requires_explicit_opt_in(): ] assert echo_lines, 'Expected at least one opt-in --s3-echo-stdout path' + allowed_guards = ('ENABLE_S3_ECHO_STDOUT', + 'shouldEchoTestOutputToConsole(stageName)') 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) + assert any(guard in line for guard in allowed_guards + for line in context), ( + f'{GROOVY}:{idx + 1} enables --s3-echo-stdout without ' + 'the enableS3EchoStdout parameter or the ' + 'shouldEchoTestOutputToConsole() stage gate') + + # The stage gate itself must stay narrow: post-merge only, multi-GPU only, + # not perf stages, and disableable without a code change. + groovy = ''.join(lines) + gate = groovy.split('def shouldEchoTestOutputToConsole(')[1] + gate = gate.split('\ndef ')[0] + for required in ('DISABLE_POST_MERGE_STDOUT_ECHO', '"Post-Merge"', + '"PerfSanity"', 'system_gpu_count'): + assert required in gate, ( + f'shouldEchoTestOutputToConsole() no longer checks {required}; ' + 'console echo must stay scoped to post-merge multi-GPU stages') + assert any( + 'DISABLE_POST_MERGE_STDOUT_ECHO = params.disableStdoutEchoOnPostMerge' + in line for line in lines), ( + 'the post-merge echo needs a build-parameter kill switch') progress_lines = [ idx for idx, line in enumerate(lines)