"""Phase 4: lifecycle guard + per-profile observability.""" import pytest from gateway.config import GatewayConfig from gateway.restart import GATEWAY_FATAL_CONFIG_EXIT_CODE class TestServedProfilesStatus: def test_write_and_read_served_profiles(self, tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) import importlib import gateway.status as status importlib.reload(status) try: status.write_runtime_status( gateway_state="running", served_profiles=["default", "coder"] ) rec = status.read_runtime_status() assert rec.get("served_profiles") == ["default", "coder"] finally: importlib.reload(status) def test_cron_profile_homes_serve_every_live_profile(tmp_path, monkeypatch): """The helper wired into in-process cron returns default + every live named profile; a tombstoned profile dir is skipped.""" monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path) default_home = tmp_path / ".hermes" monkeypatch.setenv("HERMES_HOME", str(default_home)) for name in ("worker", "guest", "gone"): (default_home / "profiles" / name).mkdir(parents=True) (default_home / "profiles" / name / "config.yaml").write_text("{}\n") # identity marker from hermes_constants import mark_named_profile_deleted mark_named_profile_deleted(default_home / "profiles" / "gone") import gateway.run as gateway_run homes = gateway_run._multiplex_profile_homes(GatewayConfig(multiplex_profiles=True)) assert [name for name, _home in homes] == ["default", "guest", "worker"] def test_cron_tick_homes_include_active_named_host(tmp_path, monkeypatch): """A named-profile host running the multiplexer ticks its own store exactly once: it is part of the served set, and cron must not union it a second time.""" monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path) default_home = tmp_path / ".hermes" for name in ("host", "worker"): (default_home / "profiles" / name).mkdir(parents=True) (default_home / "profiles" / name / "config.yaml").write_text("{}\n") # identity marker monkeypatch.setenv("HERMES_HOME", str(default_home / "profiles" / "host")) import gateway.run as gateway_run cfg = GatewayConfig(multiplex_profiles=True) adapter_names = [name for name, _home in gateway_run._multiplex_profile_homes(cfg)] cron_homes = gateway_run._cron_tick_profile_homes(cfg) cron_names = [name for name, _home in cron_homes] cron_by_name = dict(cron_homes) assert adapter_names == ["default", "host", "worker"] assert cron_names == ["default", "host", "worker"] assert cron_by_name["host"] == default_home / "profiles" / "host" class TestNamedProfileMultiplexerGuard: """_guard_named_profile_under_multiplexer is inert unless all conditions hold.""" def test_inert_when_no_default_gateway_running(self, monkeypatch, tmp_path): from hermes_cli import gateway as gw monkeypatch.setattr(gw, "_profile_suffix", lambda: "coder") monkeypatch.setattr( "hermes_constants.get_default_hermes_root", lambda: tmp_path ) # No gateway.pid in tmp_path => no running default gateway => no raise. gw._guard_named_profile_under_multiplexer(force=False) def _fake_running_default_gateway(self, monkeypatch, tmp_path): """Make the guard believe a live default gateway exists at tmp_path.""" from hermes_cli import gateway as gw import gateway.status as status monkeypatch.setattr(gw, "_profile_suffix", lambda: "coder") monkeypatch.setattr( "hermes_constants.get_default_hermes_root", lambda: tmp_path ) import json import os # Liveness is a verified identity (live PID + gateway command line + home), so this pytest # process stands in for the gateway by wearing a gateway command line. (tmp_path / "gateway.pid").write_text(str(os.getpid()), encoding="utf-8") (tmp_path / "gateway_state.json").write_text(json.dumps( {"pid": os.getpid(), "hermes_home": str(tmp_path), "gateway_state": "running"})) monkeypatch.setattr(status, "_read_process_cmdline", lambda pid: "hermes gateway run") def test_unset_allowlist_preserves_historical_guard(self, monkeypatch, tmp_path): self._fake_running_default_gateway(monkeypatch, tmp_path) (tmp_path / "config.yaml").write_text( "gateway:\n multiplex_profiles: true\n", encoding="utf-8", ) from hermes_cli import gateway as gw with pytest.raises(SystemExit) as excinfo: gw._guard_named_profile_under_multiplexer(force=False) assert excinfo.value.code == GATEWAY_FATAL_CONFIG_EXIT_CODE def test_recorded_served_profiles_win_over_config(self, monkeypatch, tmp_path): """The live gateway's own ``served_profiles`` record is authoritative: a named profile it does not list may run standalone even though the default config multiplexes.""" self._fake_running_default_gateway(monkeypatch, tmp_path) (tmp_path / "config.yaml").write_text( "gateway:\n multiplex_profiles: true\n", encoding="utf-8", ) import json import os (tmp_path / "gateway_state.json").write_text(json.dumps({ "pid": os.getpid(), "hermes_home": str(tmp_path), "gateway_state": "running", "served_profiles": ["default", "worker"]})) from hermes_cli import gateway as gw gw._guard_named_profile_under_multiplexer(force=False) assert gw.named_profile_served_by_running_multiplexer("worker") is True def test_non_multiplexing_default_gateway_lets_named_profile_run(self, monkeypatch, tmp_path): self._fake_running_default_gateway(monkeypatch, tmp_path) (tmp_path / "config.yaml").write_text( "gateway:\n multiplex_profiles: false\n", encoding="utf-8", ) from hermes_cli import gateway as gw gw._guard_named_profile_under_multiplexer(force=False) def test_named_profile_served_probe_matches_the_start_guard(self, monkeypatch, tmp_path): self._fake_running_default_gateway(monkeypatch, tmp_path) (tmp_path / "config.yaml").write_text( "gateway:\n multiplex_profiles: true\n", encoding="utf-8", ) from hermes_cli import gateway as gw assert gw.named_profile_served_by_running_multiplexer() is True monkeypatch.setattr(gw, "_profile_suffix", lambda: "") assert gw.named_profile_served_by_running_multiplexer() is False