hermes-agent/tests/plugins/test_google_meet_node.py

"""Tests for the google_meet node primitive.

Covers protocol helpers, the file-backed registry, the server's
token-and-dispatch machinery, a mocked client, and the CLI plumbing.
We never open a real socket — websockets.serve / websockets.sync.client
are fully mocked.
"""

from __future__ import annotations

import argparse
import asyncio
import json
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest


@pytest.fixture(autouse=True)
def _isolate_home(tmp_path, monkeypatch):
    hermes_home = tmp_path / ".hermes"
    hermes_home.mkdir()
    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
    yield hermes_home


# ---------------------------------------------------------------------------
# protocol.py
# ---------------------------------------------------------------------------

def test_protocol_encode_decode_roundtrip():
    from plugins.google_meet.node import protocol

    msg = protocol.make_request("ping", "tok", {"x": 1}, req_id="abc")
    raw = protocol.encode(msg)
    out = protocol.decode(raw)
    assert out == msg
    assert out["type"] == "ping"
    assert out["id"] == "abc"
    assert out["token"] == "tok"
    assert out["payload"] == {"x": 1}


def test_protocol_make_request_autogenerates_id():
    from plugins.google_meet.node import protocol

    a = protocol.make_request("ping", "tok", {})
    b = protocol.make_request("ping", "tok", {})
    assert a["id"] != b["id"]
    assert len(a["id"]) >= 16  # uuid4 hex


def test_protocol_make_request_rejects_bad_input():
    from plugins.google_meet.node import protocol

    with pytest.raises(ValueError):
        protocol.make_request("", "tok", {})
    with pytest.raises(ValueError):
        protocol.make_request("unknown_type", "tok", {})
    with pytest.raises(ValueError):
        protocol.make_request("ping", "tok", "not a dict")  # type: ignore[arg-type]


def test_protocol_decode_raises_on_malformed():
    from plugins.google_meet.node import protocol

    with pytest.raises(ValueError):
        protocol.decode("not json at all")
    with pytest.raises(ValueError):
        protocol.decode("[]")  # list, not object
    with pytest.raises(ValueError):
        protocol.decode(json.dumps({"id": "x"}))  # missing type
    with pytest.raises(ValueError):
        protocol.decode(json.dumps({"type": "ping"}))  # missing id


def test_protocol_validate_request_happy_path():
    from plugins.google_meet.node import protocol

    msg = protocol.make_request("status", "secret", {})
    ok, reason = protocol.validate_request(msg, "secret")
    assert ok is True
    assert reason == ""


def test_protocol_validate_request_rejects_bad_token():
    from plugins.google_meet.node import protocol

    msg = protocol.make_request("status", "wrong", {})
    ok, reason = protocol.validate_request(msg, "right")
    assert ok is False
    assert "token" in reason.lower()


def test_protocol_validate_request_rejects_unknown_type():
    from plugins.google_meet.node import protocol

    raw = {"type": "nope", "id": "1", "token": "t", "payload": {}}
    ok, reason = protocol.validate_request(raw, "t")
    assert ok is False
    assert "unknown" in reason.lower()


def test_protocol_validate_request_rejects_missing_id():
    from plugins.google_meet.node import protocol

    raw = {"type": "ping", "token": "t", "payload": {}}
    ok, reason = protocol.validate_request(raw, "t")
    assert ok is False
    assert "id" in reason.lower()


def test_protocol_validate_request_rejects_non_dict_payload():
    from plugins.google_meet.node import protocol

    raw = {"type": "ping", "id": "1", "token": "t", "payload": "oops"}
    ok, reason = protocol.validate_request(raw, "t")
    assert ok is False


def test_protocol_error_envelope_shape():
    from plugins.google_meet.node import protocol

    err = protocol.make_error("abc", "nope")
    assert err == {"type": "error", "id": "abc", "error": "nope"}


# ---------------------------------------------------------------------------
# registry.py
# ---------------------------------------------------------------------------

def test_registry_add_get_roundtrip_persists(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    p = tmp_path / "nodes.json"
    r = NodeRegistry(path=p)
    r.add("mac", "ws://mac.local:18789", "deadbeef")

    # Second instance sees it.
    r2 = NodeRegistry(path=p)
    entry = r2.get("mac")
    assert entry is not None
    assert entry["name"] == "mac"
    assert entry["url"] == "ws://mac.local:18789"
    assert entry["token"] == "deadbeef"
    assert "added_at" in entry


def test_registry_get_returns_none_when_missing(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    r = NodeRegistry(path=tmp_path / "n.json")
    assert r.get("ghost") is None


def test_registry_remove(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    r = NodeRegistry(path=tmp_path / "n.json")
    r.add("a", "ws://a", "t")
    assert r.remove("a") is True
    assert r.get("a") is None
    assert r.remove("a") is False  # idempotent


def test_registry_list_all_sorted(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    r = NodeRegistry(path=tmp_path / "n.json")
    r.add("zeta", "ws://z", "t1")
    r.add("alpha", "ws://a", "t2")
    names = [n["name"] for n in r.list_all()]
    assert names == ["alpha", "zeta"]


def test_registry_resolve_auto_picks_single(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    r = NodeRegistry(path=tmp_path / "n.json")
    r.add("mac", "ws://mac", "t")
    picked = r.resolve(None)
    assert picked is not None
    assert picked["name"] == "mac"


def test_registry_resolve_ambiguous_returns_none(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    r = NodeRegistry(path=tmp_path / "n.json")
    r.add("a", "ws://a", "t")
    r.add("b", "ws://b", "t")
    assert r.resolve(None) is None


def test_registry_resolve_empty_returns_none(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    r = NodeRegistry(path=tmp_path / "n.json")
    assert r.resolve(None) is None


def test_registry_resolve_by_name(tmp_path):
    from plugins.google_meet.node.registry import NodeRegistry

    r = NodeRegistry(path=tmp_path / "n.json")
    r.add("a", "ws://a", "t")
    r.add("b", "ws://b", "t")
    picked = r.resolve("b")
    assert picked is not None
    assert picked["name"] == "b"
    assert r.resolve("ghost") is None


def test_registry_defaults_to_hermes_home(tmp_path, monkeypatch):
    from plugins.google_meet.node.registry import NodeRegistry

    # _isolate_home already set HERMES_HOME to tmp_path/.hermes; the
    # registry default path must live inside that tree.
    r = NodeRegistry()
    r.add("x", "ws://x", "t")
    expected = Path(tmp_path) / ".hermes" / "workspace" / "meetings" / "nodes.json"
    assert expected.is_file()


# ---------------------------------------------------------------------------
# server.py — token + dispatch
# ---------------------------------------------------------------------------

def test_server_ensure_token_generates_and_persists(tmp_path):
    from plugins.google_meet.node.server import NodeServer

    p = tmp_path / "tok.json"
    s1 = NodeServer(token_path=p)
    t1 = s1.ensure_token()
    assert isinstance(t1, str) and len(t1) == 32

    # Reuse on a fresh instance.
    s2 = NodeServer(token_path=p)
    t2 = s2.ensure_token()
    assert t1 == t2

    data = json.loads(p.read_text(encoding="utf-8"))
    assert data["token"] == t1
    assert "generated_at" in data


def test_server_get_token_is_idempotent(tmp_path):
    from plugins.google_meet.node.server import NodeServer

    s = NodeServer(token_path=tmp_path / "t.json")
    assert s.get_token() == s.get_token()


def _run(coro):
    return asyncio.new_event_loop().run_until_complete(coro) if False else asyncio.run(coro)


def test_server_handle_request_rejects_bad_token(tmp_path):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol

    s = NodeServer(token_path=tmp_path / "t.json")
    s.ensure_token()
    bad = protocol.make_request("ping", "not-the-token", {})
    resp = asyncio.run(s._handle_request(bad))
    assert resp["type"] == "error"
    assert "token" in resp["error"].lower()


def test_server_handle_request_ping(tmp_path):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol

    s = NodeServer(token_path=tmp_path / "t.json", display_name="node-x")
    tok = s.ensure_token()
    req = protocol.make_request("ping", tok, {})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "pong"
    assert resp["id"] == req["id"]
    assert resp["payload"]["display_name"] == "node-x"


def test_server_handle_request_status_dispatches_to_pm(tmp_path, monkeypatch):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol
    from plugins.google_meet import process_manager as pm

    monkeypatch.setattr(pm, "status",
                        lambda: {"ok": True, "alive": True, "meetingId": "abc"})

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("status", tok, {})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "response"
    assert resp["id"] == req["id"]
    assert resp["payload"] == {"ok": True, "alive": True, "meetingId": "abc"}


def test_server_handle_request_start_bot_dispatches(tmp_path, monkeypatch):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol
    from plugins.google_meet import process_manager as pm

    captured = {}

    def fake_start(**kwargs):
        captured.update(kwargs)
        return {"ok": True, "pid": 42, "meeting_id": "abc-defg-hij"}

    monkeypatch.setattr(pm, "start", fake_start)

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("start_bot", tok, {
        "url": "https://meet.google.com/abc-defg-hij",
        "guest_name": "Bot",
        "duration": "30m",
    })
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "response"
    assert resp["payload"]["ok"] is True
    assert captured["url"] == "https://meet.google.com/abc-defg-hij"
    assert captured["guest_name"] == "Bot"
    assert captured["duration"] == "30m"


def test_server_handle_request_start_bot_missing_url(tmp_path):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("start_bot", tok, {"guest_name": "x"})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "error"
    assert "url" in resp["error"]


def test_server_handle_request_stop_dispatches(tmp_path, monkeypatch):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol
    from plugins.google_meet import process_manager as pm

    got = {}

    def fake_stop(*, reason="requested"):
        got["reason"] = reason
        return {"ok": True, "reason": reason}

    monkeypatch.setattr(pm, "stop", fake_stop)

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("stop", tok, {"reason": "user-cancel"})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "response"
    assert got["reason"] == "user-cancel"


def test_server_handle_request_transcript(tmp_path, monkeypatch):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol
    from plugins.google_meet import process_manager as pm

    got = {}

    def fake_transcript(last=None):
        got["last"] = last
        return {"ok": True, "lines": ["a", "b"], "total": 2}

    monkeypatch.setattr(pm, "transcript", fake_transcript)

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("transcript", tok, {"last": 5})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "response"
    assert resp["payload"]["lines"] == ["a", "b"]
    assert got["last"] == 5


def test_server_handle_request_say_enqueues_when_active(tmp_path, monkeypatch):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol
    from plugins.google_meet import process_manager as pm

    out = tmp_path / "meet-out"
    out.mkdir()
    monkeypatch.setattr(pm, "_read_active",
                        lambda: {"pid": 1, "meeting_id": "m", "out_dir": str(out)})

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("say", tok, {"text": "hello"})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "response"
    assert resp["payload"]["ok"] is True
    assert resp["payload"]["enqueued"] is True
    q = (out / "say_queue.jsonl").read_text(encoding="utf-8").strip().splitlines()
    assert len(q) == 1
    assert json.loads(q[0])["text"] == "hello"


def test_server_handle_request_say_without_active_still_ok(tmp_path, monkeypatch):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol
    from plugins.google_meet import process_manager as pm

    monkeypatch.setattr(pm, "_read_active", lambda: None)

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("say", tok, {"text": "hi"})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "response"
    assert resp["payload"]["ok"] is True
    assert resp["payload"]["enqueued"] is False


def test_server_handle_request_wraps_pm_exceptions(tmp_path, monkeypatch):
    from plugins.google_meet.node.server import NodeServer
    from plugins.google_meet.node import protocol
    from plugins.google_meet import process_manager as pm

    def boom():
        raise ValueError("kaboom")

    monkeypatch.setattr(pm, "status", boom)

    s = NodeServer(token_path=tmp_path / "t.json")
    tok = s.ensure_token()
    req = protocol.make_request("status", tok, {})
    resp = asyncio.run(s._handle_request(req))
    assert resp["type"] == "error"
    assert "kaboom" in resp["error"]


# ---------------------------------------------------------------------------
# client.py
# ---------------------------------------------------------------------------

class _FakeWS:
    """Minimal context-manager stand-in for websockets.sync.client.connect."""

    def __init__(self, reply_builder):
        self._reply_builder = reply_builder
        self.sent = []

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def send(self, raw):
        self.sent.append(raw)

    def recv(self, timeout=None):
        return self._reply_builder(self.sent[-1])


def _install_fake_ws(monkeypatch, reply_builder):
    fake_ws_holder = {}

    def _connect(url, **kwargs):
        ws = _FakeWS(reply_builder)
        fake_ws_holder["ws"] = ws
        fake_ws_holder["url"] = url
        fake_ws_holder["kwargs"] = kwargs
        return ws

    # Patch the concrete import site inside client._rpc
    import websockets.sync.client as wsc  # type: ignore
    monkeypatch.setattr(wsc, "connect", _connect)
    return fake_ws_holder


def test_client_rpc_sends_correct_envelope_and_parses_response(monkeypatch):
    from plugins.google_meet.node.client import NodeClient
    from plugins.google_meet.node import protocol

    def reply(raw_out):
        req = protocol.decode(raw_out)
        return protocol.encode(protocol.make_response(req["id"], {"ok": True, "echo": req["type"]}))

    holder = _install_fake_ws(monkeypatch, reply)

    c = NodeClient("ws://remote:1", "tok123")
    out = c._rpc("ping", {"hello": 1})
    assert out == {"ok": True, "echo": "ping"}

    sent = json.loads(holder["ws"].sent[0])
    assert sent["type"] == "ping"
    assert sent["token"] == "tok123"
    assert sent["payload"] == {"hello": 1}
    assert sent["id"]  # non-empty
    assert holder["url"] == "ws://remote:1"


def test_client_rpc_raises_on_error_envelope(monkeypatch):
    from plugins.google_meet.node.client import NodeClient
    from plugins.google_meet.node import protocol

    def reply(raw_out):
        req = protocol.decode(raw_out)
        return protocol.encode(protocol.make_error(req["id"], "nope"))

    _install_fake_ws(monkeypatch, reply)

    c = NodeClient("ws://x", "t")
    with pytest.raises(RuntimeError, match="nope"):
        c._rpc("ping", {})


def test_client_rpc_raises_on_id_mismatch(monkeypatch):
    from plugins.google_meet.node.client import NodeClient
    from plugins.google_meet.node import protocol

    def reply(raw_out):
        return protocol.encode(protocol.make_response("different-id", {"ok": True}))

    _install_fake_ws(monkeypatch, reply)

    c = NodeClient("ws://x", "t")
    with pytest.raises(RuntimeError, match="mismatch"):
        c._rpc("ping", {})


def test_client_convenience_methods_hit_correct_types(monkeypatch):
    from plugins.google_meet.node.client import NodeClient
    from plugins.google_meet.node import protocol

    seen = []

    def reply(raw_out):
        req = protocol.decode(raw_out)
        seen.append((req["type"], req["payload"]))
        return protocol.encode(protocol.make_response(req["id"], {"ok": True}))

    _install_fake_ws(monkeypatch, reply)

    c = NodeClient("ws://x", "t")
    c.start_bot("https://meet.google.com/a-b-c", guest_name="G", duration="10m")
    c.stop()
    c.status()
    c.transcript(last=3)
    c.say("hi")
    c.ping()

    types = [t for t, _ in seen]
    assert types == ["start_bot", "stop", "status", "transcript", "say", "ping"]
    # Check specific payload routing
    assert seen[0][1]["url"] == "https://meet.google.com/a-b-c"
    assert seen[0][1]["guest_name"] == "G"
    assert seen[0][1]["duration"] == "10m"
    assert seen[3][1]["last"] == 3
    assert seen[4][1]["text"] == "hi"


def test_client_init_rejects_bad_args():
    from plugins.google_meet.node.client import NodeClient

    with pytest.raises(ValueError):
        NodeClient("", "t")
    with pytest.raises(ValueError):
        NodeClient("ws://x", "")


# ---------------------------------------------------------------------------
# cli.py
# ---------------------------------------------------------------------------

def _build_parser():
    from plugins.google_meet.node.cli import register_cli

    parser = argparse.ArgumentParser(prog="meet-node-test")
    register_cli(parser)
    return parser


def test_cli_approve_list_remove(capsys):
    from plugins.google_meet.node.registry import NodeRegistry

    p = _build_parser()

    args = p.parse_args(["approve", "mac", "ws://mac:1", "tok"])
    rc = args.func(args)
    assert rc == 0
    assert NodeRegistry().get("mac") is not None

    args = p.parse_args(["list"])
    rc = args.func(args)
    assert rc == 0
    out = capsys.readouterr().out
    assert "mac" in out
    assert "ws://mac:1" in out

    args = p.parse_args(["remove", "mac"])
    rc = args.func(args)
    assert rc == 0
    assert NodeRegistry().get("mac") is None


def test_cli_list_empty(capsys):
    p = _build_parser()
    args = p.parse_args(["list"])
    rc = args.func(args)
    assert rc == 0
    assert "no nodes" in capsys.readouterr().out


def test_cli_remove_missing_returns_nonzero():
    p = _build_parser()
    args = p.parse_args(["remove", "ghost"])
    rc = args.func(args)
    assert rc == 1


def test_cli_status_pings_via_node_client(capsys, monkeypatch):
    from plugins.google_meet.node.registry import NodeRegistry
    from plugins.google_meet.node import cli as node_cli

    NodeRegistry().add("mac", "ws://mac:1", "tok")

    class _FakeClient:
        def __init__(self, url, token):
            assert url == "ws://mac:1"
            assert token == "tok"

        def ping(self):
            return {"type": "pong", "display_name": "hermes-meet-node"}

    monkeypatch.setattr(node_cli, "NodeClient", _FakeClient)

    p = _build_parser()
    args = p.parse_args(["status", "mac"])
    rc = args.func(args)
    assert rc == 0
    out = capsys.readouterr().out.strip()
    data = json.loads(out)
    assert data["ok"] is True
    assert data["node"] == "mac"


def test_cli_status_unknown_node_fails(capsys):
    p = _build_parser()
    args = p.parse_args(["status", "ghost"])
    rc = args.func(args)
    assert rc == 1


def test_cli_status_reports_client_error(capsys, monkeypatch):
    from plugins.google_meet.node.registry import NodeRegistry
    from plugins.google_meet.node import cli as node_cli

    NodeRegistry().add("mac", "ws://mac:1", "tok")

    class _FakeClient:
        def __init__(self, url, token):
            pass

        def ping(self):
            raise RuntimeError("connection refused")

    monkeypatch.setattr(node_cli, "NodeClient", _FakeClient)

    p = _build_parser()
    args = p.parse_args(["status", "mac"])
    rc = args.func(args)
    assert rc == 1
    data = json.loads(capsys.readouterr().out.strip())
    assert data["ok"] is False
    assert "connection refused" in data["error"]