"""Tests for the /api/status profile + gateway topology readout. Covers the loopback-only ``profiles`` / ``gateway_mode`` / ``gateways`` fields added to ``/api/status``: profile enumeration, single vs multiplex vs multiple gateway detection, and per-platform port resolution. """ import pytest from hermes_cli import web_server from hermes_cli.web_server import ( _collect_profile_gateway_topology, _profile_platform_ports, ) # --------------------------------------------------------------------------- # _profile_platform_ports # --------------------------------------------------------------------------- class TestProfilePlatformPorts: def test_no_runtime_platforms_returns_empty(self, tmp_path): assert _profile_platform_ports(tmp_path, None) == {} assert _profile_platform_ports(tmp_path, {"platforms": {}}) == {} def test_top_level_platforms_wins_over_gateway_block(self, tmp_path): (tmp_path / "config.yaml").write_text( "gateway:\n platforms:\n webhook:\n port: 1111\n" "platforms:\n webhook:\n port: 2222\n", encoding="utf-8", ) runtime = {"platforms": {"webhook": {"state": "connected"}}} assert _profile_platform_ports(tmp_path, runtime) == {"webhook": 2222} def test_dead_platform_states_excluded(self, tmp_path): runtime = { "platforms": { "webhook": {"state": "fatal"}, "api_server": {"state": "disconnected"}, "msgraph_webhook": {"state": "connected"}, } } assert _profile_platform_ports(tmp_path, runtime) == {"msgraph_webhook": 8646} # --------------------------------------------------------------------------- # _collect_profile_gateway_topology # --------------------------------------------------------------------------- def _patch_topology(monkeypatch, homes, running, runtimes): """Patch the topology collector's collaborators. ``homes``: list of (name, Path); ``running``: set of profile names with a live gateway; ``runtimes``: {name: runtime dict}. """ import hermes_cli.profiles as profiles_mod import gateway.status as status_mod monkeypatch.setattr(profiles_mod, "profiles_to_serve", lambda multiplex: homes) monkeypatch.setattr( profiles_mod, "_check_gateway_running", lambda home: next(n for n, h in homes if h == home) in running, ) by_path = {home / "gateway_state.json": runtimes.get(name) for name, home in homes} monkeypatch.setattr( status_mod, "read_runtime_status", lambda path=None: by_path.get(path) ) class TestCollectProfileGatewayTopology: def test_no_gateways_running(self, tmp_path, monkeypatch): homes = [("default", tmp_path / "d"), ("coder", tmp_path / "c")] _patch_topology(monkeypatch, homes, running=set(), runtimes={}) topo = _collect_profile_gateway_topology() assert topo["profiles"] == ["default", "coder"] assert topo["gateway_mode"] == "none" assert topo["gateways"] == [] def test_enumeration_failure_degrades_gracefully(self, monkeypatch): import hermes_cli.profiles as profiles_mod def _boom(multiplex): raise RuntimeError("no profiles root") monkeypatch.setattr(profiles_mod, "profiles_to_serve", _boom) topo = _collect_profile_gateway_topology() assert topo == {"profiles": [], "gateway_mode": "unknown", "gateways": []} # --------------------------------------------------------------------------- # /api/status wiring # --------------------------------------------------------------------------- class TestStatusEndpointTopology: @pytest.fixture(autouse=True) def _setup_client(self, monkeypatch, _isolate_hermes_home): try: from starlette.testclient import TestClient except ImportError: pytest.skip("fastapi/starlette not installed") import hermes_state from hermes_constants import get_hermes_home from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN monkeypatch.setattr( hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db" ) self.client = TestClient(app) self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN def test_status_includes_full_topology_on_loopback(self, monkeypatch): monkeypatch.setattr( web_server, "_collect_profile_gateway_topology", lambda: { "profiles": ["default", "coder"], "gateway_mode": "single", "gateways": [{"profile": "default", "ports": {}}], }, ) resp = self.client.get("/api/status") assert resp.status_code == 200 data = resp.json() assert data["profiles"] == ["default", "coder"] assert data["gateway_mode"] == "single" # The per-gateway detail (host ports) is loopback-only recon. assert data["gateways"] == [{"profile": "default", "ports": {}}] def test_profile_names_and_mode_public_when_auth_gated(self, monkeypatch): # Profile NAMES + gateway_mode are low-sensitivity product surface: the # Hermes Cloud Portal reads /api/status over the network (a gated bind) # to render the profile list, so they must survive the auth gate. monkeypatch.setattr( web_server, "_collect_profile_gateway_topology", lambda: { "profiles": ["default", "coder"], "gateway_mode": "multiplex", "gateways": [{"profile": "default", "ports": {"webhook": 8644}}], }, ) monkeypatch.setattr(web_server.app.state, "auth_required", True, raising=False) try: resp = self.client.get("/api/status") assert resp.status_code == 200 data = resp.json() assert data["profiles"] == ["default", "coder"] assert data["gateway_mode"] == "multiplex" # But the per-gateway detail (host ports = recon) stays gated, # alongside hermes_home / gateway_pid. assert "gateways" not in data assert "hermes_home" not in data assert "gateway_pid" not in data finally: monkeypatch.setattr( web_server.app.state, "auth_required", False, raising=False )