071ccfbcdf
Four origin/main merges brought back `linux_only` / `macos_only` / `windows_only` marks in 41 test files, along with the pre-platforms() versions of scripts/ci/list_os_marked_tests.py and check_os_marker_fakes.py. Because the legacy names are no longer registered, pytest treated them as unknown marks — a warning — so every Windows- or macOS-only test RAN on Linux (test_local_runtime_recovery.py tripped the live-system kill guard). Rewrite the marks, restore the platforms()-aware CI scripts (keeping main's os.walk fix for vanishing __pycache__ dirs), drop the stale _BASELINE entries, and make the conftest reject the retired marks outright so the next merge cannot resurrect them silently.
96 lines
4.1 KiB
Python
96 lines
4.1 KiB
Python
"""Real child output must drain even when no interactive console is consuming it."""
|
|
|
|
import logging
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from hermes_cli.desktop_console import desktop_console_output, desktop_launch_notice
|
|
|
|
|
|
@pytest.mark.platforms("windows")
|
|
def test_packaged_console_output_drains_both_streams(caplog, monkeypatch):
|
|
caplog.set_level(logging.INFO, logger="hermes_cli.desktop")
|
|
|
|
def blocked_console(*args, **kwargs):
|
|
raise AssertionError("packaged startup attempted a synchronous console write")
|
|
|
|
monkeypatch.setattr("builtins.print", blocked_console)
|
|
desktop_launch_notice("Starting Hermes")
|
|
with desktop_console_output(source_mode=False) as streams:
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", "import sys; sys.stdout.write('x'*131072); "
|
|
"sys.stdout.flush(); sys.stderr.write('diagnostic\\n'); sys.exit(7)"],
|
|
timeout=15, check=False, **streams,
|
|
)
|
|
assert result.returncode == 7
|
|
messages = [record.getMessage() for record in caplog.records]
|
|
assert sum(message.count("x") for message in messages) == 131072
|
|
assert any("diagnostic" in message for message in messages)
|
|
|
|
|
|
@pytest.mark.platforms("windows")
|
|
@pytest.mark.parametrize("level", [logging.WARNING, logging.ERROR])
|
|
def test_stderr_survives_logging_threshold(caplog, level):
|
|
caplog.set_level(level, logger="hermes_cli.desktop")
|
|
with desktop_console_output(source_mode=False) as streams:
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", "import sys; print('ordinary-output'); "
|
|
"sys.stderr.write('fatal-diagnostic\\n'); sys.exit(7)"],
|
|
timeout=15, check=False, **streams,
|
|
)
|
|
assert result.returncode == 7
|
|
messages = [record.getMessage() for record in caplog.records]
|
|
assert any("fatal-diagnostic" in message for message in messages)
|
|
assert not any("ordinary-output" in message for message in messages)
|
|
|
|
|
|
@pytest.mark.platforms("windows")
|
|
@pytest.mark.parametrize("stream", ["stdout", "stderr"])
|
|
def test_complete_records_reach_redacted_logs(tmp_path, monkeypatch, caplog, stream):
|
|
import hermes_logging
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
hermes_logging._reset_queued_handlers()
|
|
caplog.set_level(logging.INFO)
|
|
caplog.set_level(logging.INFO, logger="hermes_cli.desktop")
|
|
secret = "synthetic-boundary-credential"
|
|
payload = (
|
|
b" " * (8192 - len(b"API_KEY=")) + b"API_KEY=" + secret.encode() + b"\n"
|
|
+ b"u" * 8191 + "\u20ac\n".encode()
|
|
+ b"oversized-start API_KEY=" + secret.encode() + b"z" * 262144
|
|
+ b"oversized-end\nrecovered\ntail " + "\u20ac".encode()
|
|
)
|
|
payload_path = tmp_path / "child-output.bin"
|
|
payload_path.write_bytes(payload)
|
|
try:
|
|
log_dir = hermes_logging.setup_logging(hermes_home=tmp_path, log_level="INFO", mode="gui")
|
|
with desktop_console_output(source_mode=False) as streams:
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", "import pathlib, sys; "
|
|
"data = pathlib.Path(sys.argv[1]).read_bytes(); "
|
|
"out = getattr(sys, sys.argv[2]).buffer; "
|
|
"[(out.write(data[i:i+997]), out.flush()) for i in range(0, len(data), 997)]",
|
|
str(payload_path), stream],
|
|
timeout=15, check=False, **streams,
|
|
)
|
|
assert result.returncode == 0
|
|
hermes_logging.flush_log_queue()
|
|
text = (log_dir / "agent.log").read_text(encoding="utf-8")
|
|
assert secret not in text
|
|
assert "u" * 8191 + "\u20ac" in text
|
|
assert "\ufffd" not in text
|
|
assert "oversized-start" not in text and "oversized-end" not in text
|
|
assert "recovered" in text and "tail \u20ac" in text
|
|
assert "omitted" in text.lower()
|
|
gui_text = (log_dir / "gui.log").read_text(encoding="utf-8")
|
|
assert "recovered" in gui_text and secret not in gui_text
|
|
finally:
|
|
hermes_logging._reset_queued_handlers()
|
|
|
|
|
|
def test_source_launch_keeps_interactive_streams():
|
|
with desktop_console_output(source_mode=True) as streams:
|
|
assert streams == {}
|