diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index e17f38d4e7..bc7eb4e459 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -25,6 +25,7 @@ hex_to_npub = _buzz_mod.hex_to_npub npub_to_hex = _buzz_mod.npub_to_hex _normalize_user_ref = _buzz_mod._normalize_user_ref _cli_error_message = _buzz_mod._cli_error_message +_MAX_CLI_MESSAGE_CHARS = _buzz_mod._MAX_CLI_MESSAGE_CHARS _resolve_private_key = _buzz_mod._resolve_private_key _resolve_auth_tag = _buzz_mod._resolve_auth_tag _event_reply_parent_id = _buzz_mod._event_reply_parent_id @@ -3062,14 +3063,19 @@ class TestInboundMediaAuthorizationGate: @pytest.mark.asyncio async def test_live_media_redacts_long_path_before_bounding(self, tmp_path): + # Invariant: the host path is redacted BEFORE the 900-char bound is applied, + # so a path long enough to straddle the cut never leaks in fragments. The + # path must therefore exceed the bound, but stay under PATH_MAX (1024 on + # macOS; 4096 on Linux) so the directory can actually be created. parent = tmp_path private_parts = [] - for index in range(6): - part = f"private-{index}-" + ("x" * 150) + while len(str(parent)) < _MAX_CLI_MESSAGE_CHARS: + part = f"private-{len(private_parts)}-" + ("x" * 80) private_parts.append(part) parent = parent / part parent.mkdir() media = parent / "handoff.txt" + assert _MAX_CLI_MESSAGE_CHARS < len(str(media)) < 1024 media.write_text("safe handoff", encoding="utf-8") adapter = _make_adapter() adapter._run_cli = AsyncMock( diff --git a/tests/gateway/test_scale_to_zero.py b/tests/gateway/test_scale_to_zero.py index 7334633f7c..938bf463ff 100644 --- a/tests/gateway/test_scale_to_zero.py +++ b/tests/gateway/test_scale_to_zero.py @@ -108,8 +108,11 @@ def test_idle_exactly_at_threshold(): import os +import shutil import socket as _socket +import tempfile import threading +from pathlib import Path from gateway.scale_to_zero import ( # noqa: E402 - grouped with their section @@ -121,10 +124,39 @@ from gateway.scale_to_zero import ( # noqa: E402 - grouped with their section _FLY_ENV = {FLY_APP_NAME_ENV: "hermes-agent-stg-test", FLY_MACHINE_ID_ENV: "d891234f"} +# sockaddr_un.sun_path is 104 bytes on macOS/BSD and 108 on Linux (incl. NUL). +_SUN_PATH_MAX = 100 -def _fake_flaps(tmp_path, status_line, capture): + +@pytest.fixture() +def short_sock_dir(): + """A directory short enough that ``/fly-api.sock`` fits in sun_path. + + pytest's ``tmp_path`` nests deep enough on macOS and on CI runners that a + socket bound under it fails with ``OSError: AF_UNIX path too long``. The + production path is the fixed ``/.fly/api`` so the length limit never + applies there; only the fake flaps server needs a short home. + """ + candidates = [tempfile.gettempdir(), "/tmp"] + for base in candidates: + try: + path = Path(tempfile.mkdtemp(prefix="s2z-", dir=base)) + except OSError: + continue + if len(str(path / "fly-api.sock").encode()) <= _SUN_PATH_MAX: + break + shutil.rmtree(path, ignore_errors=True) + else: + pytest.skip(f"no temp dir short enough for an AF_UNIX socket (tried {candidates})") + try: + yield path + finally: + shutil.rmtree(path, ignore_errors=True) + + +def _fake_flaps(sock_dir, status_line, capture): """One-shot unix-socket HTTP server standing in for flaps.""" - sock_path = str(tmp_path / "fly-api.sock") + sock_path = str(sock_dir / "fly-api.sock") server = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM) server.bind(sock_path) server.listen(1) @@ -150,9 +182,9 @@ def _fake_flaps(tmp_path, status_line, capture): return sock_path, t -def test_suspend_self_posts_suspend_for_this_machine(tmp_path): +def test_suspend_self_posts_suspend_for_this_machine(short_sock_dir): captured: list[bytes] = [] - sock_path, t = _fake_flaps(tmp_path, "200 OK", captured) + sock_path, t = _fake_flaps(short_sock_dir, "200 OK", captured) assert suspend_self(_FLY_ENV, socket_path=sock_path) is True t.join(timeout=5) request = captured[0].decode() @@ -164,9 +196,9 @@ def test_suspend_self_posts_suspend_for_this_machine(tmp_path): assert "Host: flaps\r\n" in request -def test_suspend_self_non_2xx_is_false_not_raise(tmp_path): +def test_suspend_self_non_2xx_is_false_not_raise(short_sock_dir): captured: list[bytes] = [] - sock_path, t = _fake_flaps(tmp_path, "412 Precondition Failed", captured) + sock_path, t = _fake_flaps(short_sock_dir, "412 Precondition Failed", captured) assert suspend_self(_FLY_ENV, socket_path=sock_path) is False t.join(timeout=5) diff --git a/tests/gateway/test_shutdown_forensics.py b/tests/gateway/test_shutdown_forensics.py index 9b37d53031..75565e26cb 100644 --- a/tests/gateway/test_shutdown_forensics.py +++ b/tests/gateway/test_shutdown_forensics.py @@ -5,7 +5,6 @@ from __future__ import annotations import json import os import signal -import sys import time from pathlib import Path @@ -89,7 +88,12 @@ class TestFormatters: # --------------------------------------------------------------------------- class TestSpawnAsyncDiagnostic: - @pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only diagnostic") + # The diagnostic wraps its script in GNU coreutils ``timeout`` and the script + # body is Linux-only (``ps auxf --sort``, ``/proc/loadavg``, ``dmesg``, + # ``pstree``). On hosts without ``timeout`` (macOS) Popen raises and the + # producer returns None by design (fail-soft), so the spawn can only be + # observed on Linux. + @pytest.mark.linux_only def test_spawns_subprocess_and_writes_output(self, tmp_path): log_path = tmp_path / "diag.log" pid = sf.spawn_async_diagnostic(log_path, "SIGTERM", timeout_seconds=3.0) diff --git a/tests/gateway/test_systemd_notify.py b/tests/gateway/test_systemd_notify.py index b0dea324cb..84ce5ae93b 100644 --- a/tests/gateway/test_systemd_notify.py +++ b/tests/gateway/test_systemd_notify.py @@ -8,9 +8,7 @@ import socket import pytest -@pytest.mark.skipif( - not hasattr(socket, "AF_UNIX"), reason="Unix datagram sockets are unavailable" -) +@pytest.mark.linux_only # abstract (NUL-prefixed) AF_UNIX names are a Linux kernel feature def test_notify_supports_systemd_abstract_socket(monkeypatch): name = "\0hermes-test-notify" receiver = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) diff --git a/tests/gateway/test_wecom_callback.py b/tests/gateway/test_wecom_callback.py index 26ae5dfbc0..78343dfd38 100644 --- a/tests/gateway/test_wecom_callback.py +++ b/tests/gateway/test_wecom_callback.py @@ -6,9 +6,20 @@ from xml.etree import ElementTree as ET import pytest from gateway.config import PlatformConfig +from plugins.platforms.wecom import callback_adapter as _callback_mod from plugins.platforms.wecom.callback_adapter import WecomCallbackAdapter from plugins.platforms.wecom.wecom_crypto import WXBizMsgCrypt +# ``_build_event`` parses inbound XML with defusedxml, which ships with the +# ``wecom`` extra (and transitively with ``youtube``, hence ``[all]`` in CI). A +# dev venv without either extra cannot exercise the parser; the crypto, +# routing, token-refresh and body-size tests below do not touch it and run +# regardless. This is a dependency gate, not a host-OS gate. +requires_defusedxml = pytest.mark.skipif( + not _callback_mod.DEFUSEDXML_AVAILABLE, + reason="defusedxml not installed (uv sync --extra wecom)", +) + def _app(name="test-app", corp_id="ww1234567890", agent_id="1000002"): return { @@ -46,6 +57,7 @@ class TestWecomCrypto: class TestWecomCallbackEventConstruction: + @requires_defusedxml def test_build_event_extracts_text_message(self): adapter = WecomCallbackAdapter(_config()) xml_text = """ @@ -140,6 +152,7 @@ class TestWecomCallbackSendTokenRefresh: class TestWecomCallbackPollLoop: + @requires_defusedxml @pytest.mark.asyncio async def test_poll_loop_dispatches_handle_message(self, monkeypatch): adapter = WecomCallbackAdapter(_config())