fix(gateway): read every Hermes inline bootstrap's argv as its identity, on all OSes
The #107002 guard keeps an inline ``-c`` program's trailing argv as data. Every Hermes launcher runs the entry point IN the ``-c`` process, so the guard hid real gateways on every OS (#124318, #124588): - the store launcher / Windows updater relaunch (_launchers.runtime_command) - the published launcher script (POSIX shell launcher: every PM-install systemd/launchd gateway) and its Windows .cmd base64 wrapper - the venv_sync re-entry, whose argv is assigned inside the source gateway.status.inline_bootstrap_argv recognises exactly those emitted source shapes, anchored at both ends so a program merely CARRYING one (the restart watcher's respawn argv) still never matches, and rewrites the process to the equivalent ``python -m <entry> <argv>``. /proc, psutil and ``ps`` space-join argv, which splits the source across tokens; the shortest token run ending in a recognised tail is the source whichever reader joined it. Both canonical matchers (looks_like_gateway_command_line and update_cmd_windows._hermes_holder_subcommand) use it. Live on a real PM install (Linux, bwrap): a gateway started through the installed launcher script or runtime_command read "Gateway is not running" on main; with this change both read running, find_gateway_pids and get_running_pid see them. Drops the four #124318 known_failure gates in tests/e2e/core/windows_update. Co-authored-by: Hermes Agent <dmyou@users.noreply.github.com> Co-authored-by: JoaoMarcos44 <joaomarcosdias444@gmail.com> Co-authored-by: DianaBudin <dianabudin0307@gmail.com>
This commit is contained in:
teknium1
@@ -588,30 +588,66 @@ def command_line_runs_inline_source(tokens: list[str]) -> bool:
|
||||
return inline_source_flag_index(tokens) is not None
|
||||
|
||||
|
||||
def _runtime_launcher_entry_argv(tokens: list[str]) -> list[str] | None:
|
||||
"""Trailing argv of a Hermes runtime-launcher process (``hermes_cli._launchers.runtime_command``).
|
||||
# Hermes' own inline bootstraps hand control to a Hermes entry point IN this process, so the argv
|
||||
# they run with is this process's own identity; every other ``-c`` program keeps its trailing argv
|
||||
# as data (#107002). Each pattern is one emitted source shape, anchored at both ends so a program
|
||||
# merely CARRYING a bootstrap command line (the restart watcher's respawn argv) never matches.
|
||||
_Q = r"""['"]?"""
|
||||
_MAIN = rf"{_Q}__main__{_Q}"
|
||||
_RUN_MODULE = rf"runpy\.run_module\(\s*{_Q}(?P<target>[\w.]+){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*,\s*alter_sys\s*=\s*True\s*\)"
|
||||
_BOOTSTRAPS = (
|
||||
# hermes_cli._launchers.runtime_command (store launcher, the Windows updater's relaunch)
|
||||
("module", re.compile(rf"import os, sys, runpy;.*\b{_RUN_MODULE}", re.S)),
|
||||
# hermes_cli.venv_sync.relaunch_command: argv is assigned inside the source
|
||||
("module", re.compile(rf"import sys, runpy; sys\.path\.insert\(.*\b{_RUN_MODULE}", re.S)),
|
||||
("path", re.compile(
|
||||
rf"import sys, runpy; sys\.path\.insert\(.*\brunpy\.run_path\(\s*{_Q}(?P<target>[^'\"]+?){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*\)",
|
||||
re.S)),
|
||||
# hermes_cli._launchers._launcher_script (the published POSIX shell / Windows .cmd launcher)
|
||||
("entry", re.compile(r"import os, re, sys\s.*\bfrom\s+(?P<target>[\w.]+)\s+import\s+(?P<func>\w+)\b.*\bsys\.exit\(\s*(?P=func)\(\)\s*\)", re.S)),
|
||||
# hermes_cli._launchers._write_cmd_launcher: the launcher script, base64-encoded
|
||||
("base64", re.compile(rf"import base64; exec\(base64\.b64decode\({_Q}(?P<target>[A-Za-z0-9+/=]+){_Q}\)\)")),
|
||||
)
|
||||
_ASSIGNED_ARGV = re.compile(r"\bsys\.argv\s*=\s*\[(.*?)\]\s*;")
|
||||
|
||||
The launcher starts Hermes as ``python -I -c <bootstrap> <entry argv…>`` where the bootstrap runs
|
||||
the entry module IN PLACE (``runpy.run_module(…, alter_sys=True)`` after ``import hermes_bootstrap``),
|
||||
so the argv following that program IS this process's own subcommand: ``python -I -c <bootstrap>
|
||||
gateway run --replace`` really is a gateway -- the form the Windows updater's post-update relaunch
|
||||
spawns (``hermes_cli.gateway._gateway_run_args_for_profile``). Every OTHER inline source keeps the
|
||||
#107002 rule: the detached restart watcher (``… -c <watcher> <old_pid> … -m hermes_cli.main gateway
|
||||
run``) hides a FUTURE spawn in its trailing argv, which must never be read as this process's identity.
|
||||
|
||||
Command lines reach us space-joined from argv, so this one-source program arrives split across many
|
||||
tokens, and only the tail it emits (``alter_sys=True)`` last) marks where the entry argv begins.
|
||||
Marker checks against the launcher's own emitted format -- never a loose ``-c`` test.
|
||||
def _bootstrap_entry(source: str, argv: list[str]) -> list[str] | None:
|
||||
"""``[-m, <module>, *argv]`` (or ``[<path>, *argv]``) the inline *source* runs in-process, else None."""
|
||||
source = source.strip()
|
||||
kind, match = next(((k, m) for k, p in _BOOTSTRAPS if (m := p.fullmatch(source))), (None, None))
|
||||
if match is None:
|
||||
return None
|
||||
target = match["target"]
|
||||
if kind == "base64":
|
||||
import base64
|
||||
import binascii
|
||||
try:
|
||||
return _bootstrap_entry(base64.b64decode(target, validate=True).decode("utf-8"), argv)
|
||||
except (binascii.Error, UnicodeDecodeError):
|
||||
return None
|
||||
if kind == "entry": # the launcher script's own ``--run-module <module>`` switch
|
||||
return ["-m", argv[1], *argv[2:]] if argv[:1] == ["--run-module"] and len(argv) > 1 else ["-m", target, *argv]
|
||||
if assigned := _ASSIGNED_ARGV.search(source):
|
||||
argv = [item.strip().strip("'\"") for item in assigned.group(1).split(",")][1:]
|
||||
return [target, *argv] if kind == "path" else ["-m", target, *argv]
|
||||
|
||||
|
||||
def inline_bootstrap_argv(tokens: list[str]) -> list[str] | None:
|
||||
"""*tokens* as the equivalent ``python -m <module> <argv…>`` when this interpreter's ``-c`` source
|
||||
is a Hermes bootstrap running an entry point in-process; None for any other inline source.
|
||||
|
||||
Command lines usually arrive space-joined (``/proc``, psutil, ``ps``), which splits the source
|
||||
across tokens; the shortest token run that ends in a recognised tail is the source, whatever
|
||||
joined it, and the tokens after it are the entry point's argv.
|
||||
"""
|
||||
index = inline_source_flag_index(tokens)
|
||||
if index is None:
|
||||
return None
|
||||
for position in range(index, len(tokens)):
|
||||
if "alter_sys=True" not in tokens[position]:
|
||||
continue
|
||||
program = " ".join(tokens[index : position + 1])
|
||||
if "runpy.run_module(" in program and "hermes_bootstrap" in program:
|
||||
return tokens[position + 1 :]
|
||||
for end in range(index + 1, len(tokens)):
|
||||
if tokens[end].rstrip().endswith(")"):
|
||||
entry = _bootstrap_entry(" ".join(tokens[index + 1 : end + 1]), tokens[end + 1 :])
|
||||
if entry is not None:
|
||||
return [tokens[0], *entry]
|
||||
return None
|
||||
|
||||
|
||||
@@ -636,23 +672,14 @@ def _gateway_command_subcommand(command: str | None) -> str | None:
|
||||
# ``python -c <src> … -m hermes_cli.main gateway run``: the trailing argv belongs to the program
|
||||
# the inline source will spawn later, not to this process (#107002). Case-preserving tokens:
|
||||
# the operand-taking ``-X``/``-W``/``-Q`` must not be conflated with ``-q``/``-b``.
|
||||
launcher_bootstrap = False
|
||||
if command_line_runs_inline_source(cased_tokens):
|
||||
# …with ONE exception: Hermes's own runtime launcher starts Hermes as ``python -I -c
|
||||
# <bootstrap> <entry argv…>`` and the bootstrap runs the entry module IN PLACE, so the
|
||||
# trailing argv IS this process's own subcommand. That is the form the Windows updater's
|
||||
# post-update relaunch spawns (``gateway._gateway_run_args_for_profile`` →
|
||||
# ``gateway run --replace``); refusing it made the updater blind to the gateway it had just
|
||||
# respawned -- its liveness probe "found no stable gateway process" while the respawned
|
||||
# gateway was alive and serving, and the next update's preflight then failed on the lock.
|
||||
source_index = inline_source_flag_index(cased_tokens)
|
||||
launcher_argv = None if source_index is None else _runtime_launcher_entry_argv(cased_tokens)
|
||||
if launcher_argv is None:
|
||||
# …unless the source is a Hermes bootstrap running the entry point in THIS process (store
|
||||
# launcher, launcher script, venv_sync re-entry): then its argv is this process's (#124318).
|
||||
cased_tokens = inline_bootstrap_argv(cased_tokens)
|
||||
if cased_tokens is None:
|
||||
return None
|
||||
cased_tokens = [cased_tokens[0], *launcher_argv]
|
||||
tokens = [t.lower() for t in cased_tokens]
|
||||
basenames = [t.rsplit("/", 1)[-1] for t in tokens]
|
||||
launcher_bootstrap = True
|
||||
# The launchd job's osascript wrapper (gateway_launchd.launchd_program_arguments) carries the gateway argv
|
||||
# inside one JXA script string; the gateway itself is its child and is matched on its own command line.
|
||||
if basenames[0] == "osascript":
|
||||
@@ -669,11 +696,8 @@ def _gateway_command_subcommand(command: str | None) -> str | None:
|
||||
if any(b in ("hermes-gateway", "hermes-gateway.exe") for b in basenames):
|
||||
return "run"
|
||||
joined = " ".join(tokens)
|
||||
# A runtime-launcher process names its entry point in the bootstrap source, not in argv.
|
||||
if not launcher_bootstrap and (
|
||||
"hermes_cli.main" not in joined and "hermes_cli/main.py" not in joined and not any(
|
||||
b in ("hermes", "hermes.exe") for b in basenames
|
||||
)
|
||||
if "hermes_cli.main" not in joined and "hermes_cli/main.py" not in joined and not any(
|
||||
b in ("hermes", "hermes.exe") for b in basenames
|
||||
):
|
||||
return None
|
||||
# Drop --profile X / -p X / --profile=X / -p=X (consumes a VALUE of "gateway" too).
|
||||
|
||||
@@ -240,6 +240,15 @@ def _hermes_holder_subcommand(cmdline: str) -> str | None:
|
||||
tokens = shlex.split(cmdline, posix=False)
|
||||
except Exception:
|
||||
tokens = cmdline.split()
|
||||
# ``python -c <src> … -m hermes_cli.main <subcommand>``: the entry token belongs to the argv the
|
||||
# inline source carries for a LATER spawn, not to this holder (#107002) -- unless the source is a
|
||||
# Hermes bootstrap running the entry point in this process (#124318).
|
||||
from gateway.status import command_line_runs_inline_source, inline_bootstrap_argv
|
||||
normalized = [t.strip('"').replace("\\", "/") for t in tokens]
|
||||
if command_line_runs_inline_source(normalized):
|
||||
tokens = inline_bootstrap_argv(normalized)
|
||||
if tokens is None:
|
||||
return None
|
||||
|
||||
def _is_entry(i: int, token: str) -> bool:
|
||||
low = token.lower().strip('"')
|
||||
@@ -249,11 +258,6 @@ def _hermes_holder_subcommand(cmdline: str) -> str | None:
|
||||
entry_idx = next((i for i, token in enumerate(tokens) if _is_entry(i, token)), None)
|
||||
if entry_idx is None:
|
||||
return None
|
||||
# ``python -c <src> … -m hermes_cli.main <subcommand>``: the entry token belongs to the argv the
|
||||
# inline source carries for a LATER spawn, not to this holder (#107002).
|
||||
from gateway.status import command_line_runs_inline_source
|
||||
if command_line_runs_inline_source([t.strip('"').replace("\\", "/") for t in tokens]):
|
||||
return None
|
||||
value_flags = _holder_value_flags()
|
||||
i = entry_idx + 1
|
||||
while i < len(tokens):
|
||||
|
||||
@@ -14,7 +14,6 @@ import re
|
||||
import psutil
|
||||
import pytest
|
||||
|
||||
from tests.e2e.core._pending_fixes import known_failure
|
||||
from tests.e2e.core.windows._helpers import wait_until
|
||||
from tests.e2e.core.windows_update._machine import (
|
||||
REQUIRES_OPT_IN,
|
||||
@@ -27,8 +26,6 @@ from tests.fakes.fake_llm_provider import FakeLLMServer
|
||||
pytestmark = [pytest.mark.platforms("windows"), pytest.mark.integration,
|
||||
pytest.mark.live_system_guard_bypass, REQUIRES_OPT_IN]
|
||||
|
||||
_INVISIBLE = ("gated on #124318: the gateway the updater relaunches holds its lock without gateway.pid "
|
||||
"metadata, so `hermes gateway status`/`stop` and the next update's pause cannot see it")
|
||||
_RESTART_FAILURE = re.compile(
|
||||
r"^.*(recovery failed|restart could not be verified|restart incomplete|not verified alive).*$", re.M)
|
||||
_STATUS_LINE = re.compile(r"^.*[✓✗].*[Gg]ateway.*$", re.M)
|
||||
@@ -98,12 +95,9 @@ def test_update_with_running_gateway_succeeds(journey: Journey) -> None:
|
||||
m, run = journey.machine, journey["update"]
|
||||
journey["state_before"] # the precondition: a gateway was running when the update began
|
||||
failure = _RESTART_FAILURE.search(run.stdout)
|
||||
with known_failure(r"^hermes update with a running gateway reported a failed gateway restart "
|
||||
r"\(rc=\d+\): ⚠ Windows gateway restart could not be verified",
|
||||
_INVISIBLE):
|
||||
assert run.returncode == 0 and failure is None, fail_with(
|
||||
m, f"hermes update with a running gateway reported a failed gateway restart "
|
||||
f"(rc={run.returncode}): {failure.group(0).strip() if failure else '<no restart message>'}", run)
|
||||
assert run.returncode == 0 and failure is None, fail_with(
|
||||
m, f"hermes update with a running gateway reported a failed gateway restart "
|
||||
f"(rc={run.returncode}): {failure.group(0).strip() if failure else '<no restart message>'}", run)
|
||||
|
||||
|
||||
def test_gateway_serves_next_after_update(journey: Journey) -> None:
|
||||
@@ -117,31 +111,24 @@ def test_gateway_discoverable_after_update(journey: Journey) -> None:
|
||||
m, state, status = journey.machine, journey["state_after"], journey["status_after"]
|
||||
pid, pidfile = int(state.get("pid") or 0), journey["pidfile_after"]
|
||||
line = _status_line(status.stdout)
|
||||
with known_failure(r"^after update the serving gateway \(pid \d+\) is invisible",
|
||||
_INVISIBLE):
|
||||
assert status.returncode == 0 and str(pid) in line and pidfile, fail_with(
|
||||
m, f"after update the serving gateway (pid {pid}) is invisible: `hermes gateway status` says "
|
||||
f"{line!r}; gateway.pid present={pidfile}", status)
|
||||
assert status.returncode == 0 and str(pid) in line and pidfile, fail_with(
|
||||
m, f"after update the serving gateway (pid {pid}) is invisible: `hermes gateway status` says "
|
||||
f"{line!r}; gateway.pid present={pidfile}", status)
|
||||
|
||||
|
||||
def test_next_update_is_not_blocked(journey: Journey) -> None:
|
||||
m, run = journey.machine, journey["update_again"]
|
||||
journey["state_after"] # a gateway was running (the relaunched one)
|
||||
blocked = "Could not map Windows gateway PIDs" in run.stdout
|
||||
with known_failure(r"^the next hermes update is blocked at the gateway pause step \(rc=1, "
|
||||
r"'Could not map Windows gateway PIDs' printed=True\)",
|
||||
_INVISIBLE):
|
||||
assert run.returncode == 0 and not blocked, fail_with(
|
||||
m, f"the next hermes update is blocked at the gateway pause step (rc={run.returncode}, "
|
||||
f"'Could not map Windows gateway PIDs' printed={blocked})", run)
|
||||
assert run.returncode == 0 and not blocked, fail_with(
|
||||
m, f"the next hermes update is blocked at the gateway pause step (rc={run.returncode}, "
|
||||
f"'Could not map Windows gateway PIDs' printed={blocked})", run)
|
||||
|
||||
|
||||
def test_gateway_stop_after_update(journey: Journey) -> None:
|
||||
m, stop, state = journey.machine, journey["stop"], journey["state_before_stop"]
|
||||
pid = int(state.get("pid") or 0)
|
||||
assert pid, fail_with(m, f"no gateway recorded before stop: {state}")
|
||||
with known_failure(r"^after update `hermes gateway stop` left the serving gateway \(pid \d+\) running",
|
||||
_INVISIBLE):
|
||||
assert journey["stopped"], fail_with(
|
||||
m, f"after update `hermes gateway stop` left the serving gateway (pid {pid}) running", stop)
|
||||
assert journey["stopped"], fail_with(
|
||||
m, f"after update `hermes gateway stop` left the serving gateway (pid {pid}) running", stop)
|
||||
assert stop.returncode == 0, fail_with(m, f"hermes gateway stop exited {stop.returncode}", stop)
|
||||
|
||||
55
tests/gateway/test_gateway_inline_bootstrap_identity.py
Normal file
55
tests/gateway/test_gateway_inline_bootstrap_identity.py
Normal file
@@ -0,0 +1,55 @@
|
||||
"""A gateway started through one of Hermes' own inline bootstraps is a gateway on every OS (#124318).
|
||||
|
||||
The store launcher (``_launchers.runtime_command``, also the Windows updater's relaunch), the
|
||||
published launcher script (POSIX shell launcher and the Windows ``.cmd`` base64 wrapper) and the
|
||||
``venv_sync`` re-entry all run ``python -I -c <source> …`` with the entry point IN that process.
|
||||
The #107002 guard that keeps an inline program's trailing argv as data made all of them invisible.
|
||||
The readers hand the matchers different strings: ``/proc``, psutil and ``ps`` space-join argv (the
|
||||
source splits across tokens); Windows CIM reports the CreateProcess line (``list2cmdline``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.status import looks_like_gateway_command_line
|
||||
from hermes_cli import _launchers, venv_sync
|
||||
from hermes_cli.update_cmd_windows import _hermes_holder_subcommand
|
||||
|
||||
ROOT = Path("/opt/Hermes Agent/hermes-agent")
|
||||
PY = "/opt/venv/bin/python3"
|
||||
_SCRIPT = _launchers._launcher_script("hermes", ROOT, None)
|
||||
_JOINS = {"space-joined": " ".join, "windows": subprocess.list2cmdline}
|
||||
|
||||
|
||||
def _forms(argv: list[str]) -> dict[str, list[str]]:
|
||||
return {
|
||||
"store-launcher": _launchers.runtime_command(ROOT, argv, python=Path(PY)),
|
||||
"launcher-script": [PY, "-I", "-c", _SCRIPT, *argv],
|
||||
"cmd-launcher": [PY, "-I", "-c", f"import base64; exec(base64.b64decode('{base64.b64encode(_SCRIPT.encode()).decode()}'))", *argv],
|
||||
"venv-reentry": venv_sync.relaunch_command(
|
||||
Path(PY), ROOT, [str(ROOT / "hermes_cli" / "main.py"), *argv], ["/old/python", "-m", "hermes_cli.main", *argv],
|
||||
"hermes_cli.main"),
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("join", _JOINS)
|
||||
@pytest.mark.parametrize("form", _forms([]))
|
||||
def test_bootstrap_launched_gateway_is_a_gateway(form: str, join: str) -> None:
|
||||
command_line = _JOINS[join]([str(t) for t in _forms(["gateway", "run", "--replace"])[form]])
|
||||
assert looks_like_gateway_command_line(command_line)
|
||||
assert _hermes_holder_subcommand(command_line) == "gateway"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("join", _JOINS)
|
||||
def test_bootstrap_argv_is_identity_only_for_the_process_running_it(join: str) -> None:
|
||||
store = [str(t) for t in _launchers.runtime_command(ROOT, ["gateway", "run"], python=Path(PY))]
|
||||
chat = _JOINS[join]([str(t) for t in _launchers.runtime_command(ROOT, ["chat"], python=Path(PY))])
|
||||
# The restart watcher CARRIES a store-launcher gateway command it spawns later (#107002).
|
||||
watcher = _JOINS[join]([PY, "-c", "import os, sys, time\npid = int(sys.argv[1]); cmd = sys.argv[2:]\n", "1234", *store])
|
||||
assert not looks_like_gateway_command_line(chat) and _hermes_holder_subcommand(chat) == "chat"
|
||||
assert not looks_like_gateway_command_line(watcher) and _hermes_holder_subcommand(watcher) is None
|
||||
@@ -1,77 +0,0 @@
|
||||
"""The Windows launcher gateway form must be recognised as a gateway.
|
||||
|
||||
Windows spawns gateways two ways:
|
||||
|
||||
- ``hermes gateway start`` -> ``<python> -m hermes_cli.main gateway run``
|
||||
- launcher scripts / the updater's relaunch (``_launchers.runtime_command``) ->
|
||||
``<python> -I -c "<bootstrap; runpy.run_module('hermes_cli.main', alter_sys=True)>" gateway run --replace``
|
||||
|
||||
``_gateway_command_subcommand`` used to answer ``None`` for *every* ``-c`` form — a guard
|
||||
(#107002) written for the restart watcher, whose ``-c`` really does carry a command it will
|
||||
spawn later. That made the shim-launched gateway invisible to ``find_gateway_pids``,
|
||||
``_wait_for_gateway_ready`` and the update preflight, so a live gateway read as "no gateway
|
||||
process detected" and the updater refused to continue.
|
||||
|
||||
The fix recognises only *our own* launcher bootstrap (the ``alter_sys=True)`` tail marker)
|
||||
and keeps rejecting foreign inline source, so #107002 stays intact.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.status import _gateway_command_subcommand
|
||||
|
||||
# Verbatim shape of the injected launcher bootstrap (hermes_cli/_launchers.py::runtime_command).
|
||||
LAUNCHER_BOOTSTRAP = (
|
||||
"import os, sys, runpy; "
|
||||
"os.environ.pop('PYTHONHOME', None); "
|
||||
"sys.path.insert(0, 'C:\\\\hermes\\\\hermes-agent'); "
|
||||
"import hermes_bootstrap; "
|
||||
"runpy.run_module('hermes_cli.main', run_name='__main__', alter_sys=True)"
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
sys.platform != "win32", reason="the launcher form only exists on Windows"
|
||||
)
|
||||
|
||||
|
||||
def _launcher_cmdline(*args: str) -> str:
|
||||
"""The cmdline Windows reports for a launcher-spawned gateway.
|
||||
|
||||
``_read_process_cmdline`` returns an unquoted, space-joined argv, so the bootstrap source
|
||||
arrives split across many tokens — which is exactly why the matcher scans for the entry
|
||||
marker instead of trusting a single token.
|
||||
"""
|
||||
return " ".join([r"C:\tools\python.exe", "-I", "-c", LAUNCHER_BOOTSTRAP, *args])
|
||||
|
||||
|
||||
def _console_script_cmdline(*args: str) -> str:
|
||||
return " ".join([r"C:\tools\python.exe", "-m", "hermes_cli.main", *args])
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "win32", reason="windows launcher form")
|
||||
@pytest.mark.parametrize("subcommand", ["run", "status", "restart"])
|
||||
def test_launcher_form_is_recognised(subcommand):
|
||||
argv = ["gateway", subcommand]
|
||||
if subcommand == "run":
|
||||
argv.append("--replace")
|
||||
assert _gateway_command_subcommand(_launcher_cmdline(*argv)) == subcommand
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "win32", reason="windows launcher form")
|
||||
def test_launcher_form_served_by_legacy_module_form():
|
||||
"""The plain console-script form keeps working (regression guard)."""
|
||||
cmdline = r"C:\tools\python.exe -m hermes_cli.main gateway run"
|
||||
assert _gateway_command_subcommand(cmdline) == "run"
|
||||
|
||||
|
||||
def test_watcher_inline_source_still_rejected():
|
||||
"""#107002: the restart watcher's own ``-c`` must never look like a gateway."""
|
||||
watcher = 'python -c "import os, sys, time; pass" 40688 gateway run --replace'
|
||||
assert _gateway_command_subcommand(watcher) is None
|
||||
|
||||
|
||||
def test_foreign_inline_source_still_rejected():
|
||||
"""Any other third-party ``-c`` payload stays unrecognised."""
|
||||
assert _gateway_command_subcommand('python -c "print(1)" gateway run') is None
|
||||
Reference in New Issue
Block a user