hermes-agent/tests/hermes_cli/test_web_server_messaging_profiles.py

"""Regression tests for profile-scoped dashboard Channels endpoints.

Before the ``profile`` parameter existed, ``/api/messaging/platforms`` always
read/wrote the dashboard process's own (root) ``.env`` via ``load_env()`` /
``save_env_value()`` — so a dashboard switched to a freshly created profile
still displayed and persisted the ROOT install's messaging credentials.
These tests pin the new behavior: reads and writes land in the REQUESTED
profile's HERMES_HOME, and the dashboard's own profile stays untouched.
"""
import pytest
import hermes_yaml as yaml
import gateway.status as _gw_status


_VALID_WORKER_BOT_TOKEN = "123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ_1234"
_VALID_BODY_BOT_TOKEN = "987654321:ZYXWVUTSRQPONMLKJIHGFEDCBA_4321"


@pytest.fixture
def isolated_profiles(tmp_path, monkeypatch, _isolate_hermes_home):
    """Isolated default home + one named profile, each with its own .env."""
    from hermes_constants import get_hermes_home
    from hermes_cli import profiles

    default_home = get_hermes_home()
    profiles_root = default_home / "profiles"
    worker_home = profiles_root / "worker_alpha"
    for home in (default_home, worker_home):
        home.mkdir(parents=True, exist_ok=True)
        (home / "config.yaml").write_text("{}\n", encoding="utf-8")

    (default_home / ".env").write_text(
        "TELEGRAM_BOT_TOKEN=root-token\n", encoding="utf-8"
    )
    (worker_home / ".env").write_text("", encoding="utf-8")

    monkeypatch.setattr(profiles, "_get_default_hermes_home", lambda: default_home)
    monkeypatch.setattr(profiles, "_get_profiles_root", lambda: profiles_root)
    return {"default": default_home, "worker_alpha": worker_home}


@pytest.fixture
def client(monkeypatch, isolated_profiles):
    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")
    # The dashboard process's os.environ may carry root-install credentials;
    # make sure the scoped path never falls back to them.
    monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
    c = TestClient(app)
    c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
    return c


def _telegram(payload):
    return next(p for p in payload["platforms"] if p["id"] == "telegram")


def _env_field(platform, key):
    return next(f for f in platform["env_vars"] if f["key"] == key)


class TestProfileScopedMessagingReads:
    def test_scoped_read_does_not_show_root_credentials(
        self, client, isolated_profiles
    ):
        resp = client.get(
            "/api/messaging/platforms", params={"profile": "worker_alpha"}
        )
        assert resp.status_code == 200
        telegram = _telegram(resp.json())
        token = _env_field(telegram, "TELEGRAM_BOT_TOKEN")
        # The worker profile has an empty .env — the root token must not leak.
        assert token["is_set"] is False
        assert telegram["configured"] is False


    def test_unknown_profile_returns_404(self, client, isolated_profiles):
        resp = client.get(
            "/api/messaging/platforms", params={"profile": "no_such_profile"}
        )
        assert resp.status_code == 404

    def test_scoped_read_returns_profile_path_command_and_startup_failure(
        self, client, isolated_profiles, monkeypatch
    ):
        import hermes_cli.web_server as web_server

        worker_home = isolated_profiles["worker_alpha"]
        (worker_home / ".env").write_text(
            "TELEGRAM_BOT_TOKEN=worker-token\n", encoding="utf-8"
        )
        (worker_home / "config.yaml").write_text(
            yaml.safe_dump({"platforms": {"telegram": {"enabled": True}}}),
            encoding="utf-8",
        )
        monkeypatch.setattr(_gw_status, "get_running_pid", lambda *a, **k: None)
        monkeypatch.setattr(
            _gw_status, "get_running_pid_cached", lambda *a, **k: None
        )
        monkeypatch.setattr(
            _gw_status,
            "read_runtime_status",
            # Accepts path= : the profile-scoped read now passes the
            # profile's own gateway_state.json explicitly rather than
            # relying on process-level HERMES_HOME resolution (#71211).
            lambda *a, **k: {
                "gateway_state": "startup_failed",
                "exit_reason": "all configured messaging platforms failed to connect",
                "platforms": {},
            },
        )

        resp = client.get(
            "/api/messaging/platforms", params={"profile": "worker_alpha"}
        )

        assert resp.status_code == 200
        payload = resp.json()
        assert payload["env_path"] == str(worker_home / ".env")
        assert payload["gateway_start_command"] == (
            "hermes -p worker_alpha gateway start"
        )
        telegram = _telegram(payload)
        assert telegram["state"] == "startup_failed"
        assert telegram["error_code"] == "startup_failed"
        assert telegram["error_message"] == (
            "all configured messaging platforms failed to connect"
        )


class TestProfileScopedMessagingWrites:
    def test_scoped_write_lands_in_target_profile_env(
        self, client, isolated_profiles
    ):
        resp = client.put(
            "/api/messaging/platforms/telegram",
            params={"profile": "worker_alpha"},
            json={
                "enabled": True,
                "env": {"TELEGRAM_BOT_TOKEN": _VALID_WORKER_BOT_TOKEN},
            },
        )
        assert resp.status_code == 200

        worker_env = (
            isolated_profiles["worker_alpha"] / ".env"
        ).read_text(encoding="utf-8")
        assert f"TELEGRAM_BOT_TOKEN={_VALID_WORKER_BOT_TOKEN}" in worker_env

        # The dashboard's own .env must stay untouched — this was the bug.
        root_env = (isolated_profiles["default"] / ".env").read_text(
            encoding="utf-8"
        )
        assert _VALID_WORKER_BOT_TOKEN not in root_env
        assert "TELEGRAM_BOT_TOKEN=root-token" in root_env

        # Enablement lands in the target profile's config.yaml.
        worker_cfg = yaml.safe_load(
            (isolated_profiles["worker_alpha"] / "config.yaml").read_text(encoding="utf-8")
        ) or {}
        assert worker_cfg.get("platforms", {}).get("telegram", {}).get("enabled") is True
        root_cfg = yaml.safe_load(
            (isolated_profiles["default"] / "config.yaml").read_text(encoding="utf-8")
        ) or {}
        assert "telegram" not in (root_cfg.get("platforms") or {})


    def test_scoped_read_after_scoped_write_round_trips(
        self, client, isolated_profiles
    ):
        client.put(
            "/api/messaging/platforms/telegram",
            params={"profile": "worker_alpha"},
            json={
                "enabled": True,
                "env": {"TELEGRAM_BOT_TOKEN": _VALID_WORKER_BOT_TOKEN},
            },
        )
        resp = client.get(
            "/api/messaging/platforms", params={"profile": "worker_alpha"}
        )
        telegram = _telegram(resp.json())
        assert telegram["enabled"] is True
        assert _env_field(telegram, "TELEGRAM_BOT_TOKEN")["is_set"] is True
        assert telegram["configured"] is True



def _enable_multiplex(default_home):
    (default_home / "config.yaml").write_text(
        yaml.safe_dump({"gateway": {"multiplex_profiles": True}}),
        encoding="utf-8",
    )


class TestMultiplexPortBindingGuard:
    """Enabling api_server/webhook on a secondary multiplexed profile is rejected BEFORE anything
    is persisted: the default profile's listener already mirrors them at ``/p/<profile>/`` (#62791).
    Every other inbound-port platform is allowed — the gateway serves it on the shared listener.
    """

    @pytest.fixture(autouse=True)
    def _no_multiplex_env_override(self, monkeypatch):
        # The operator env override must not leak into these tests: the
        # multiplex flag under test comes from the default profile's config.
        monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False)

    def test_rejects_only_mirrored_listeners_on_secondary(
        self, client, isolated_profiles
    ):
        from gateway.config import PORT_BINDING_PLATFORM_VALUES, SHARED_LISTENER_MIRROR_PLATFORMS

        _enable_multiplex(isolated_profiles["default"])
        assert SHARED_LISTENER_MIRROR_PLATFORMS  # guard set must not be empty
        catalog = {p["id"] for p in client.get("/api/messaging/platforms").json()["platforms"]}
        for platform_id in sorted(PORT_BINDING_PLATFORM_VALUES & catalog):
            resp = client.put(
                f"/api/messaging/platforms/{platform_id}",
                params={"profile": "worker_alpha"},
                json={"enabled": True},
            )
            if platform_id in SHARED_LISTENER_MIRROR_PLATFORMS:
                assert resp.status_code == 409, platform_id
            else:  # served at /p/worker_alpha/<path> on the shared listener
                assert resp.status_code == 200, (platform_id, resp.text)





    def test_secondary_can_disable_and_clear_invalid_config(
        self, client, isolated_profiles
    ):
        _enable_multiplex(isolated_profiles["default"])
        worker_home = isolated_profiles["worker_alpha"]
        (worker_home / "config.yaml").write_text(
            yaml.safe_dump({"platforms": {"api_server": {"enabled": True}}}),
            encoding="utf-8",
        )

        resp = client.put(
            "/api/messaging/platforms/api_server",
            params={"profile": "worker_alpha"},
            json={"enabled": False},
        )
        assert resp.status_code == 200
        cfg = yaml.safe_load((worker_home / "config.yaml").read_text(encoding="utf-8"))
        assert cfg["platforms"]["api_server"]["enabled"] is False

        catalog = client.get(
            "/api/messaging/platforms", params={"profile": "worker_alpha"}
        ).json()
        api_server = next(p for p in catalog["platforms"] if p["id"] == "api_server")
        if api_server["env_vars"]:
            resp = client.put(
                "/api/messaging/platforms/api_server",
                params={"profile": "worker_alpha"},
                json={"clear_env": [api_server["env_vars"][0]["key"]]},
            )
            assert resp.status_code == 200

def test_named_current_home_matches_unscoped(client, isolated_profiles, monkeypatch):
    from hermes_cli.web_server_profiles import _config_profile_scope, _hermes_home_scope
    from hermes_constants import get_hermes_home

    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "root-token")
    for scope in (None, "current", "default"):
        response = client.get("/api/messaging/platforms", params={"profile": scope} if scope else {})
        assert response.status_code == 200
        assert _telegram(response.json())["enabled"] is True
    with _hermes_home_scope(isolated_profiles["worker_alpha"]):
        with _config_profile_scope("default") as scoped:
            assert scoped is None
            assert get_hermes_home() == isolated_profiles["default"]


def test_scoped_enablement_uses_only_own_credentials(client, isolated_profiles, monkeypatch):
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "root-token")
    worker = isolated_profiles["worker_alpha"]
    params = {"profile": "worker_alpha"}
    assert _telegram(client.get("/api/messaging/platforms", params=params).json())["enabled"] is False
    (worker / ".env").write_text("TELEGRAM_BOT_TOKEN=worker-token\n", encoding="utf-8")
    payload = client.get("/api/messaging/platforms", params=params).json()
    assert _telegram(payload)["enabled"] is True
    assert _telegram(payload)["configured"] is True
    assert _telegram(payload)["state"] != "disabled"
    from hermes_cli.web_server_messaging import _messaging_platform_catalog
    empty = {entry["id"] for entry in _messaging_platform_catalog() if not entry["required_env"]}
    for platform in payload["platforms"]:
        if platform["id"] in empty:
            assert platform["enabled"] is False
            assert platform["configured"] is False
    for enabled in (False, True):
        (worker / "config.yaml").write_text(yaml.safe_dump({"platforms": {"telegram": {"enabled": enabled}}}), encoding="utf-8")
        platform = _telegram(client.get("/api/messaging/platforms", params=params).json())
        assert platform["enabled"] is enabled
        assert platform["configured"] is True
    assert "root-token" in (isolated_profiles["default"] / ".env").read_text(encoding="utf-8")


@pytest.mark.parametrize("topology", ["scoped_query", "pooled_unscoped"])
def test_credential_write_hot_serves_a_multiplexed_profile(client, isolated_profiles, monkeypatch, topology):
    """A token saved for a profile the live multiplexer serves is handed to the multiplexer right
    away (``hot_served``), so the UI skips its restart banner. Both Desktop topologies: the dashboard's
    ``?profile=`` and a pooled ``hermes --profile X serve`` that receives the PUT unscoped (#109088)."""
    import hermes_cli.gateway as gateway_cli
    import hermes_cli.gateway_multiplex_served as served_mod
    notified = []
    monkeypatch.setattr(gateway_cli, "named_profile_served_by_running_multiplexer", lambda name=None: name == "worker_alpha")
    monkeypatch.setattr(served_mod, "notify_multiplexer_profiles_changed", lambda name, **kw: notified.append(name) or ["default", name])
    if topology == "pooled_unscoped":
        monkeypatch.setattr(gateway_cli, "_current_profile_name", lambda: "worker_alpha")
        params = {}
    else:
        params = {"profile": "worker_alpha"}
    resp = client.put("/api/messaging/platforms/telegram", params=params,
                      json={"enabled": True, "env": {"TELEGRAM_BOT_TOKEN": _VALID_WORKER_BOT_TOKEN}})
    assert resp.status_code == 200
    assert resp.json()["hot_served"] is True
    assert notified == ["worker_alpha"]


def test_credential_write_on_default_profile_is_not_hot_served(client, isolated_profiles, monkeypatch):
    """The default profile is the multiplexer itself (its own adapters are restart-managed): never
    claim a hot serve for it."""
    import hermes_cli.gateway_multiplex_served as served_mod
    monkeypatch.setattr(served_mod, "notify_multiplexer_profiles_changed",
                        lambda name, **kw: pytest.fail("default profile must not ping the multiplexer"))
    resp = client.put("/api/messaging/platforms/telegram",
                      json={"enabled": True, "env": {"TELEGRAM_BOT_TOKEN": _VALID_WORKER_BOT_TOKEN}})
    assert resp.status_code == 200
    assert resp.json()["hot_served"] is False