diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 9670a1d83b..595330f0a2 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -2368,7 +2368,11 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False): # Exit with code 1 if gateway fails to connect any platform, # so systemd Restart=on-failure will retry on transient errors verbosity = None if quiet else verbose - success = asyncio.run(start_gateway(replace=replace, verbosity=verbosity)) + try: + success = asyncio.run(start_gateway(replace=replace, verbosity=verbosity)) + except KeyboardInterrupt: + print("\nGateway stopped.") + return if not success: sys.exit(1) diff --git a/tests/hermes_cli/test_gateway.py b/tests/hermes_cli/test_gateway.py index 9dea51987d..0a44ac9532 100644 --- a/tests/hermes_cli/test_gateway.py +++ b/tests/hermes_cli/test_gateway.py @@ -1,11 +1,58 @@ """Tests for hermes_cli.gateway.""" -from types import SimpleNamespace +import sys +from types import ModuleType, SimpleNamespace from unittest.mock import patch, call +import pytest + import hermes_cli.gateway as gateway +def _install_fake_gateway_run(monkeypatch, start_gateway): + module = ModuleType("gateway.run") + module.start_gateway = start_gateway + monkeypatch.setitem(sys.modules, "gateway.run", module) + + +def test_run_gateway_exits_cleanly_on_keyboard_interrupt(monkeypatch, capsys): + calls = [] + + def fake_start_gateway(*, replace, verbosity): + calls.append((replace, verbosity)) + return object() + + def fake_asyncio_run(coro): + raise KeyboardInterrupt + + _install_fake_gateway_run(monkeypatch, fake_start_gateway) + monkeypatch.setattr(gateway.asyncio, "run", fake_asyncio_run) + + gateway.run_gateway() + + out = capsys.readouterr().out + assert calls == [(False, 0)] + assert "Press Ctrl+C to stop" in out + assert "Gateway stopped." in out + + +def test_run_gateway_exits_nonzero_when_start_gateway_reports_failure(monkeypatch): + calls = [] + + def fake_start_gateway(*, replace, verbosity): + calls.append((replace, verbosity)) + return object() + + _install_fake_gateway_run(monkeypatch, fake_start_gateway) + monkeypatch.setattr(gateway.asyncio, "run", lambda coro: False) + + with pytest.raises(SystemExit) as exc_info: + gateway.run_gateway(verbose=1, quiet=True, replace=True) + + assert exc_info.value.code == 1 + assert calls == [(True, None)] + + class TestSystemdLingerStatus: def test_reports_enabled(self, monkeypatch): monkeypatch.setattr(gateway, "is_linux", lambda: True)