test_google_meet_plugin.py
1 """Tests for the google_meet plugin. 2 3 Covers the safety-gated pieces that don't require Playwright: 4 5 * URL regex — only ``https://meet.google.com/`` URLs pass 6 * Meeting-id extraction from Meet URLs 7 * Status / transcript writes round-trip through the file-backed state 8 * Tool handlers return well-formed JSON under all branches 9 * Process manager refuses unsafe URLs and clears stale state cleanly 10 * ``_on_session_end`` hook is defensive (no-ops when no bot active) 11 12 Does NOT spawn a real Chromium — we mock ``subprocess.Popen`` where needed. 13 """ 14 15 from __future__ import annotations 16 17 import json 18 import os 19 import signal 20 from pathlib import Path 21 from unittest.mock import patch 22 23 import pytest 24 25 26 @pytest.fixture(autouse=True) 27 def _isolate_home(tmp_path, monkeypatch): 28 hermes_home = tmp_path / ".hermes" 29 hermes_home.mkdir() 30 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 31 yield hermes_home 32 33 34 # --------------------------------------------------------------------------- 35 # URL safety gate 36 # --------------------------------------------------------------------------- 37 38 def test_is_safe_meet_url_accepts_standard_meet_codes(): 39 from plugins.google_meet.meet_bot import _is_safe_meet_url 40 41 assert _is_safe_meet_url("https://meet.google.com/abc-defg-hij") 42 assert _is_safe_meet_url("https://meet.google.com/abc-defg-hij?pli=1") 43 assert _is_safe_meet_url("https://meet.google.com/new") 44 assert _is_safe_meet_url("https://meet.google.com/lookup/ABC123") 45 46 47 def test_is_safe_meet_url_rejects_non_meet_urls(): 48 from plugins.google_meet.meet_bot import _is_safe_meet_url 49 50 # wrong host 51 assert not _is_safe_meet_url("https://evil.example.com/abc-defg-hij") 52 # wrong scheme 53 assert not _is_safe_meet_url("http://meet.google.com/abc-defg-hij") 54 # malformed code 55 assert not _is_safe_meet_url("https://meet.google.com/not-a-meet-code") 56 # subdomain hijack attempts 57 assert not _is_safe_meet_url("https://meet.google.com.evil.com/abc-defg-hij") 58 assert not _is_safe_meet_url("https://notmeet.google.com/abc-defg-hij") 59 # empty / wrong type 60 assert not _is_safe_meet_url("") 61 assert not _is_safe_meet_url(None) # type: ignore[arg-type] 62 assert not _is_safe_meet_url(123) # type: ignore[arg-type] 63 64 65 def test_meeting_id_extraction(): 66 from plugins.google_meet.meet_bot import _meeting_id_from_url 67 68 assert _meeting_id_from_url("https://meet.google.com/abc-defg-hij") == "abc-defg-hij" 69 assert _meeting_id_from_url("https://meet.google.com/abc-defg-hij?pli=1") == "abc-defg-hij" 70 # fallback for codes we can't parse (e.g. /new before redirect) 71 fallback = _meeting_id_from_url("https://meet.google.com/new") 72 assert fallback.startswith("meet-") 73 74 75 # --------------------------------------------------------------------------- 76 # _BotState — transcript + status file round-trip 77 # --------------------------------------------------------------------------- 78 79 def test_bot_state_dedupes_captions_and_flushes_status(tmp_path): 80 from plugins.google_meet.meet_bot import _BotState 81 82 out = tmp_path / "session" 83 state = _BotState(out_dir=out, meeting_id="abc-defg-hij", 84 url="https://meet.google.com/abc-defg-hij") 85 86 state.record_caption("Alice", "Hey everyone") 87 state.record_caption("Alice", "Hey everyone") # dup — ignored 88 state.record_caption("Bob", "Let's start") 89 90 transcript = (out / "transcript.txt").read_text() 91 assert "Alice: Hey everyone" in transcript 92 assert "Bob: Let's start" in transcript 93 # dedup — Alice line appears exactly once 94 assert transcript.count("Alice: Hey everyone") == 1 95 96 status = json.loads((out / "status.json").read_text()) 97 assert status["meetingId"] == "abc-defg-hij" 98 assert status["transcriptLines"] == 2 99 assert status["transcriptPath"].endswith("transcript.txt") 100 101 102 def test_bot_state_ignores_blank_text(tmp_path): 103 from plugins.google_meet.meet_bot import _BotState 104 105 state = _BotState(out_dir=tmp_path / "s", meeting_id="x-y-z", 106 url="https://meet.google.com/x-y-z") 107 state.record_caption("Alice", "") 108 state.record_caption("Alice", " ") 109 state.record_caption("", "text but no speaker") 110 111 status = json.loads((tmp_path / "s" / "status.json").read_text()) 112 assert status["transcriptLines"] == 1 113 # blank-speaker falls back to "Unknown" 114 assert "Unknown: text but no speaker" in (tmp_path / "s" / "transcript.txt").read_text() 115 116 117 def test_parse_duration(): 118 from plugins.google_meet.meet_bot import _parse_duration 119 120 assert _parse_duration("30m") == 30 * 60 121 assert _parse_duration("2h") == 2 * 3600 122 assert _parse_duration("90s") == 90 123 assert _parse_duration("90") == 90 124 assert _parse_duration("") is None 125 assert _parse_duration("bogus") is None 126 127 128 # --------------------------------------------------------------------------- 129 # process_manager — refuses unsafe URLs, manages active pointer 130 # --------------------------------------------------------------------------- 131 132 def test_start_refuses_unsafe_url(): 133 from plugins.google_meet import process_manager as pm 134 135 res = pm.start("https://evil.example.com/abc-defg-hij") 136 assert res["ok"] is False 137 assert "refusing" in res["error"] 138 139 140 def test_status_reports_no_active_meeting(): 141 from plugins.google_meet import process_manager as pm 142 143 assert pm.status() == {"ok": False, "reason": "no active meeting"} 144 assert pm.transcript() == {"ok": False, "reason": "no active meeting"} 145 assert pm.stop() == {"ok": False, "reason": "no active meeting"} 146 147 148 def test_start_spawns_subprocess_and_writes_active_pointer(tmp_path): 149 """Verify start() wires env vars correctly and records the pid.""" 150 from plugins.google_meet import process_manager as pm 151 152 class _FakeProc: 153 def __init__(self, pid): 154 self.pid = pid 155 156 captured_env = {} 157 captured_argv = [] 158 159 def _fake_popen(argv, **kwargs): 160 captured_argv.extend(argv) 161 captured_env.update(kwargs.get("env") or {}) 162 return _FakeProc(99999) 163 164 with patch.object(pm.subprocess, "Popen", side_effect=_fake_popen): 165 # Also prevent pid liveness probe from stomping on our real pids 166 with patch.object(pm, "_pid_alive", return_value=False): 167 res = pm.start( 168 "https://meet.google.com/abc-defg-hij", 169 guest_name="Test Bot", 170 duration="15m", 171 ) 172 173 assert res["ok"] is True 174 assert res["meeting_id"] == "abc-defg-hij" 175 assert res["pid"] == 99999 176 assert captured_env["HERMES_MEET_URL"] == "https://meet.google.com/abc-defg-hij" 177 assert captured_env["HERMES_MEET_GUEST_NAME"] == "Test Bot" 178 assert captured_env["HERMES_MEET_DURATION"] == "15m" 179 # python -m plugins.google_meet.meet_bot 180 assert any("plugins.google_meet.meet_bot" in a for a in captured_argv) 181 182 # .active.json points at the bot 183 active = pm._read_active() 184 assert active is not None 185 assert active["pid"] == 99999 186 assert active["meeting_id"] == "abc-defg-hij" 187 188 189 def test_transcript_reads_last_n_lines(tmp_path): 190 from plugins.google_meet import process_manager as pm 191 192 meeting_dir = Path(os.environ["HERMES_HOME"]) / "workspace" / "meetings" / "abc-defg-hij" 193 meeting_dir.mkdir(parents=True) 194 (meeting_dir / "transcript.txt").write_text( 195 "[10:00:00] Alice: one\n" 196 "[10:00:01] Bob: two\n" 197 "[10:00:02] Alice: three\n" 198 ) 199 pm._write_active({ 200 "pid": 0, "meeting_id": "abc-defg-hij", 201 "out_dir": str(meeting_dir), 202 "url": "https://meet.google.com/abc-defg-hij", 203 "started_at": 0, 204 }) 205 206 res = pm.transcript(last=2) 207 assert res["ok"] is True 208 assert res["total"] == 3 209 assert len(res["lines"]) == 2 210 assert res["lines"][-1].endswith("Alice: three") 211 212 213 def test_stop_signals_process_and_clears_pointer(tmp_path): 214 from plugins.google_meet import process_manager as pm 215 216 pm._write_active({ 217 "pid": 11111, "meeting_id": "x-y-z", 218 "out_dir": str(tmp_path / "x-y-z"), 219 "url": "https://meet.google.com/x-y-z", 220 "started_at": 0, 221 }) 222 223 alive_seq = iter([True, True, False]) # alive at first, gone after SIGTERM 224 def _alive(pid): 225 try: 226 return next(alive_seq) 227 except StopIteration: 228 return False 229 230 sent = [] 231 def _kill(pid, sig): 232 sent.append((pid, sig)) 233 234 with patch.object(pm, "_pid_alive", side_effect=_alive), \ 235 patch.object(pm.os, "kill", side_effect=_kill), \ 236 patch.object(pm.time, "sleep", lambda _s: None): 237 res = pm.stop() 238 239 assert res["ok"] is True 240 assert (11111, signal.SIGTERM) in sent 241 # .active.json cleared 242 assert pm._read_active() is None 243 244 245 # --------------------------------------------------------------------------- 246 # Tool handlers — JSON shape + safety gates 247 # --------------------------------------------------------------------------- 248 249 def test_meet_join_handler_missing_url_returns_error(): 250 from plugins.google_meet.tools import handle_meet_join 251 252 out = json.loads(handle_meet_join({})) 253 assert out["success"] is False 254 assert "url is required" in out["error"] 255 256 257 def test_meet_join_handler_respects_safety_gate(): 258 from plugins.google_meet.tools import handle_meet_join 259 260 with patch("plugins.google_meet.tools.check_meet_requirements", return_value=True): 261 out = json.loads(handle_meet_join({"url": "https://evil.example.com/foo"})) 262 assert out["success"] is False 263 assert "refusing" in out["error"] 264 265 266 def test_meet_join_handler_returns_error_when_playwright_missing(): 267 from plugins.google_meet.tools import handle_meet_join 268 269 with patch("plugins.google_meet.tools.check_meet_requirements", return_value=False): 270 out = json.loads(handle_meet_join({"url": "https://meet.google.com/abc-defg-hij"})) 271 assert out["success"] is False 272 assert "prerequisites missing" in out["error"] 273 274 275 def test_meet_say_requires_text(): 276 from plugins.google_meet.tools import handle_meet_say 277 278 out = json.loads(handle_meet_say({})) 279 assert out["success"] is False 280 assert "text is required" in out["error"] 281 282 283 def test_meet_say_no_active_meeting(): 284 from plugins.google_meet.tools import handle_meet_say 285 286 out = json.loads(handle_meet_say({"text": "hello everyone"})) 287 assert out["success"] is False 288 # Falls through to pm.enqueue_say which reports no active meeting. 289 assert "no active meeting" in out.get("reason", "") 290 291 292 def test_meet_status_and_transcript_no_active(): 293 from plugins.google_meet.tools import handle_meet_status, handle_meet_transcript 294 295 assert json.loads(handle_meet_status({}))["success"] is False 296 assert json.loads(handle_meet_transcript({}))["success"] is False 297 298 299 def test_meet_leave_no_active(): 300 from plugins.google_meet.tools import handle_meet_leave 301 302 out = json.loads(handle_meet_leave({})) 303 assert out["success"] is False 304 305 306 # --------------------------------------------------------------------------- 307 # _on_session_end — defensive cleanup 308 # --------------------------------------------------------------------------- 309 310 def test_on_session_end_noop_when_nothing_active(): 311 from plugins.google_meet import _on_session_end 312 # Should not raise and should not call stop(). 313 with patch("plugins.google_meet.pm.stop") as stop_mock: 314 _on_session_end() 315 stop_mock.assert_not_called() 316 317 318 def test_on_session_end_stops_live_bot(): 319 from plugins.google_meet import _on_session_end 320 from plugins.google_meet import pm 321 322 with patch.object(pm, "status", return_value={"ok": True, "alive": True}), \ 323 patch.object(pm, "stop") as stop_mock: 324 _on_session_end() 325 stop_mock.assert_called_once() 326 327 328 # --------------------------------------------------------------------------- 329 # Plugin register() — platform gating + tool registration 330 # --------------------------------------------------------------------------- 331 332 def test_register_refuses_on_windows(): 333 import plugins.google_meet as plugin 334 335 calls = {"tools": [], "cli": [], "hooks": []} 336 337 class _Ctx: 338 def register_tool(self, **kw): calls["tools"].append(kw["name"]) 339 def register_cli_command(self, **kw): calls["cli"].append(kw["name"]) 340 def register_hook(self, name, fn): calls["hooks"].append(name) 341 342 with patch.object(plugin.platform, "system", return_value="Windows"): 343 plugin.register(_Ctx()) 344 345 assert calls == {"tools": [], "cli": [], "hooks": []} 346 347 348 def test_register_wires_tools_cli_and_hook_on_linux(): 349 import plugins.google_meet as plugin 350 351 calls = {"tools": [], "cli": [], "hooks": []} 352 353 class _Ctx: 354 def register_tool(self, **kw): calls["tools"].append(kw["name"]) 355 def register_cli_command(self, **kw): calls["cli"].append(kw["name"]) 356 def register_hook(self, name, fn): calls["hooks"].append(name) 357 358 with patch.object(plugin.platform, "system", return_value="Linux"): 359 plugin.register(_Ctx()) 360 361 assert set(calls["tools"]) == { 362 "meet_join", "meet_status", "meet_transcript", "meet_leave", "meet_say", 363 } 364 assert calls["cli"] == ["meet"] 365 assert calls["hooks"] == ["on_session_end"] 366 367 368 # --------------------------------------------------------------------------- 369 # v2: process_manager.enqueue_say + realtime-mode passthrough 370 # --------------------------------------------------------------------------- 371 372 def test_enqueue_say_requires_text(): 373 from plugins.google_meet import process_manager as pm 374 assert pm.enqueue_say("")["ok"] is False 375 assert pm.enqueue_say(" ")["ok"] is False 376 377 378 def test_enqueue_say_no_active_meeting(): 379 from plugins.google_meet import process_manager as pm 380 res = pm.enqueue_say("hi team") 381 assert res["ok"] is False 382 assert "no active meeting" in res["reason"] 383 384 385 def test_enqueue_say_rejects_transcribe_mode(tmp_path): 386 from plugins.google_meet import process_manager as pm 387 388 out_dir = Path(os.environ["HERMES_HOME"]) / "workspace" / "meetings" / "abc-defg-hij" 389 out_dir.mkdir(parents=True) 390 pm._write_active({ 391 "pid": 0, "meeting_id": "abc-defg-hij", 392 "out_dir": str(out_dir), "url": "https://meet.google.com/abc-defg-hij", 393 "started_at": 0, "mode": "transcribe", 394 }) 395 res = pm.enqueue_say("hi team") 396 assert res["ok"] is False 397 assert "transcribe mode" in res["reason"] 398 399 400 def test_enqueue_say_writes_jsonl_in_realtime_mode(): 401 from plugins.google_meet import process_manager as pm 402 403 out_dir = Path(os.environ["HERMES_HOME"]) / "workspace" / "meetings" / "abc-defg-hij" 404 out_dir.mkdir(parents=True) 405 pm._write_active({ 406 "pid": 0, "meeting_id": "abc-defg-hij", 407 "out_dir": str(out_dir), "url": "https://meet.google.com/abc-defg-hij", 408 "started_at": 0, "mode": "realtime", 409 }) 410 res = pm.enqueue_say("hello everyone") 411 assert res["ok"] is True 412 assert "enqueued_id" in res 413 414 queue = out_dir / "say_queue.jsonl" 415 assert queue.is_file() 416 lines = [json.loads(ln) for ln in queue.read_text().splitlines() if ln.strip()] 417 assert len(lines) == 1 418 assert lines[0]["text"] == "hello everyone" 419 420 421 def test_start_passes_mode_into_active_record(): 422 from plugins.google_meet import process_manager as pm 423 424 class _FakeProc: 425 def __init__(self, pid): self.pid = pid 426 427 with patch.object(pm.subprocess, "Popen", return_value=_FakeProc(12345)), \ 428 patch.object(pm, "_pid_alive", return_value=False): 429 res = pm.start( 430 "https://meet.google.com/abc-defg-hij", 431 mode="realtime", 432 ) 433 assert res["ok"] is True 434 assert res["mode"] == "realtime" 435 assert pm._read_active()["mode"] == "realtime" 436 437 438 def test_start_realtime_env_vars_threaded_through(): 439 from plugins.google_meet import process_manager as pm 440 441 class _FakeProc: 442 def __init__(self, pid): self.pid = pid 443 444 captured_env = {} 445 def _fake_popen(argv, **kwargs): 446 captured_env.update(kwargs.get("env") or {}) 447 return _FakeProc(11111) 448 449 with patch.object(pm.subprocess, "Popen", side_effect=_fake_popen), \ 450 patch.object(pm, "_pid_alive", return_value=False): 451 pm.start( 452 "https://meet.google.com/abc-defg-hij", 453 mode="realtime", 454 realtime_model="gpt-realtime", 455 realtime_voice="alloy", 456 realtime_instructions="Be brief.", 457 realtime_api_key="sk-test", 458 ) 459 assert captured_env["HERMES_MEET_MODE"] == "realtime" 460 assert captured_env["HERMES_MEET_REALTIME_MODEL"] == "gpt-realtime" 461 assert captured_env["HERMES_MEET_REALTIME_VOICE"] == "alloy" 462 assert captured_env["HERMES_MEET_REALTIME_INSTRUCTIONS"] == "Be brief." 463 assert captured_env["HERMES_MEET_REALTIME_KEY"] == "sk-test" 464 465 466 def test_meet_join_accepts_realtime_mode(): 467 from plugins.google_meet.tools import handle_meet_join 468 469 with patch("plugins.google_meet.tools.check_meet_requirements", return_value=True), \ 470 patch("plugins.google_meet.tools.pm.start", return_value={"ok": True, "meeting_id": "x-y-z"}) as start_mock: 471 out = json.loads(handle_meet_join({ 472 "url": "https://meet.google.com/abc-defg-hij", 473 "mode": "realtime", 474 })) 475 assert out["success"] is True 476 assert start_mock.call_args.kwargs["mode"] == "realtime" 477 478 479 def test_meet_join_rejects_bad_mode(): 480 from plugins.google_meet.tools import handle_meet_join 481 482 out = json.loads(handle_meet_join({ 483 "url": "https://meet.google.com/abc-defg-hij", 484 "mode": "bogus", 485 })) 486 assert out["success"] is False 487 assert "mode must be" in out["error"] 488 489 490 # --------------------------------------------------------------------------- 491 # v3: NodeClient routing from tool handlers 492 # --------------------------------------------------------------------------- 493 494 def test_meet_join_unknown_node_returns_clear_error(): 495 from plugins.google_meet.tools import handle_meet_join 496 497 out = json.loads(handle_meet_join({ 498 "url": "https://meet.google.com/abc-defg-hij", 499 "node": "my-mac", 500 })) 501 assert out["success"] is False 502 assert "no registered meet node" in out["error"] 503 504 505 def test_meet_join_routes_to_registered_node(): 506 from plugins.google_meet.tools import handle_meet_join 507 from plugins.google_meet.node.registry import NodeRegistry 508 509 reg = NodeRegistry() 510 reg.add("my-mac", "ws://1.2.3.4:18789", "tok") 511 512 with patch("plugins.google_meet.node.client.NodeClient.start_bot", 513 return_value={"ok": True, "meeting_id": "a-b-c"}) as call_mock: 514 out = json.loads(handle_meet_join({ 515 "url": "https://meet.google.com/abc-defg-hij", 516 "node": "my-mac", 517 "mode": "realtime", 518 })) 519 assert out["success"] is True 520 assert out["node"] == "my-mac" 521 assert call_mock.call_args.kwargs["mode"] == "realtime" 522 523 524 def test_meet_say_routes_to_node(): 525 from plugins.google_meet.tools import handle_meet_say 526 from plugins.google_meet.node.registry import NodeRegistry 527 528 reg = NodeRegistry() 529 reg.add("my-mac", "ws://1.2.3.4:18789", "tok") 530 531 with patch("plugins.google_meet.node.client.NodeClient.say", 532 return_value={"ok": True, "enqueued_id": "abc"}) as call_mock: 533 out = json.loads(handle_meet_say({"text": "hello", "node": "my-mac"})) 534 assert out["success"] is True 535 assert out["node"] == "my-mac" 536 call_mock.assert_called_once_with("hello") 537 538 539 def test_meet_join_auto_node_selects_sole_registered(): 540 from plugins.google_meet.tools import handle_meet_join 541 from plugins.google_meet.node.registry import NodeRegistry 542 543 reg = NodeRegistry() 544 reg.add("only-one", "ws://1.2.3.4:18789", "tok") 545 546 with patch("plugins.google_meet.node.client.NodeClient.start_bot", 547 return_value={"ok": True}) as call_mock: 548 out = json.loads(handle_meet_join({ 549 "url": "https://meet.google.com/abc-defg-hij", 550 "node": "auto", 551 })) 552 assert out["success"] is True 553 assert out["node"] == "only-one" 554 assert call_mock.called 555 556 557 def test_meet_join_auto_node_ambiguous_returns_error(): 558 from plugins.google_meet.tools import handle_meet_join 559 from plugins.google_meet.node.registry import NodeRegistry 560 561 reg = NodeRegistry() 562 reg.add("a", "ws://1.2.3.4:18789", "tok") 563 reg.add("b", "ws://5.6.7.8:18789", "tok") 564 565 out = json.loads(handle_meet_join({ 566 "url": "https://meet.google.com/abc-defg-hij", 567 "node": "auto", 568 })) 569 assert out["success"] is False 570 assert "no registered meet node" in out["error"] 571 572 573 def test_cli_register_includes_node_subcommand(): 574 """`hermes meet` argparse tree includes the node subtree.""" 575 import argparse 576 from plugins.google_meet.cli import register_cli 577 578 parser = argparse.ArgumentParser(prog="hermes meet") 579 register_cli(parser) 580 581 # Parse a known-good node invocation to prove the subtree is wired. 582 ns = parser.parse_args(["node", "list"]) 583 assert ns.meet_command == "node" 584 assert ns.node_cmd == "list" 585 586 587 def test_cli_join_accepts_mode_and_node_flags(): 588 import argparse 589 from plugins.google_meet.cli import register_cli 590 591 parser = argparse.ArgumentParser(prog="hermes meet") 592 register_cli(parser) 593 594 ns = parser.parse_args([ 595 "join", "https://meet.google.com/abc-defg-hij", 596 "--mode", "realtime", "--node", "my-mac", 597 ]) 598 assert ns.mode == "realtime" 599 assert ns.node == "my-mac" 600 601 602 def test_cli_say_subcommand_exists(): 603 import argparse 604 from plugins.google_meet.cli import register_cli 605 606 parser = argparse.ArgumentParser(prog="hermes meet") 607 register_cli(parser) 608 609 ns = parser.parse_args(["say", "hello team", "--node", "my-mac"]) 610 assert ns.text == "hello team" 611 assert ns.node == "my-mac" 612 613 614 # --------------------------------------------------------------------------- 615 # v2.1: new _BotState fields + status dict shape 616 # --------------------------------------------------------------------------- 617 618 def test_bot_state_exposes_v2_telemetry_fields(tmp_path): 619 from plugins.google_meet.meet_bot import _BotState 620 621 state = _BotState(out_dir=tmp_path / "s", meeting_id="x-y-z", 622 url="https://meet.google.com/x-y-z") 623 # Defaults for the new fields. 624 status = json.loads((tmp_path / "s" / "status.json").read_text()) 625 for key in ( 626 "realtime", "realtimeReady", "realtimeDevice", 627 "audioBytesOut", "lastAudioOutAt", "lastBargeInAt", 628 "joinAttemptedAt", "leaveReason", 629 ): 630 assert key in status, f"missing v2 telemetry key: {key}" 631 assert status["realtime"] is False 632 assert status["realtimeReady"] is False 633 assert status["audioBytesOut"] == 0 634 635 # Setting them flushes them. 636 state.set(realtime=True, realtime_ready=True, audio_bytes_out=1024, 637 leave_reason="lobby_timeout") 638 status = json.loads((tmp_path / "s" / "status.json").read_text()) 639 assert status["realtime"] is True 640 assert status["realtimeReady"] is True 641 assert status["audioBytesOut"] == 1024 642 assert status["leaveReason"] == "lobby_timeout" 643 644 645 # --------------------------------------------------------------------------- 646 # Admission detection + barge-in helper 647 # --------------------------------------------------------------------------- 648 649 def test_looks_like_human_speaker(): 650 from plugins.google_meet.meet_bot import _looks_like_human_speaker 651 652 # Blank, "unknown", "you", and the bot's own name → not human (no barge-in) 653 for s in ("", " ", "Unknown", "unknown", "You", "you", "Hermes Agent", "hermes agent"): 654 assert not _looks_like_human_speaker(s, "Hermes Agent"), f"{s!r} should NOT be human" 655 # Real names → human (barge-in) 656 for s in ("Alice", "Bob Lee", "@teknium"): 657 assert _looks_like_human_speaker(s, "Hermes Agent"), f"{s!r} SHOULD be human" 658 659 660 def test_detect_admission_returns_false_on_error(): 661 from plugins.google_meet.meet_bot import _detect_admission 662 663 class _FakePage: 664 def evaluate(self, _js): raise RuntimeError("boom") 665 666 assert _detect_admission(_FakePage()) is False 667 668 669 def test_detect_admission_true_when_probe_returns_true(): 670 from plugins.google_meet.meet_bot import _detect_admission 671 672 class _FakePage: 673 def evaluate(self, _js): return True 674 675 assert _detect_admission(_FakePage()) is True 676 677 678 def test_detect_denied_returns_false_on_error(): 679 from plugins.google_meet.meet_bot import _detect_denied 680 681 class _FakePage: 682 def evaluate(self, _js): raise RuntimeError("boom") 683 684 assert _detect_denied(_FakePage()) is False 685 686 687 # --------------------------------------------------------------------------- 688 # Realtime session counters + cancel_response (barge-in) 689 # --------------------------------------------------------------------------- 690 691 def test_realtime_session_cancel_response_when_disconnected(): 692 from plugins.google_meet.realtime.openai_client import RealtimeSession 693 694 sess = RealtimeSession(api_key="sk-test", audio_sink_path=None) 695 # No _ws yet — cancel should no-op and return False. 696 assert sess.cancel_response() is False 697 698 699 def test_realtime_session_cancel_response_sends_cancel_frame(): 700 from plugins.google_meet.realtime.openai_client import RealtimeSession 701 702 sess = RealtimeSession(api_key="sk-test", audio_sink_path=None) 703 sent = [] 704 705 class _FakeWs: 706 def send(self, msg): sent.append(msg) 707 708 sess._ws = _FakeWs() 709 assert sess.cancel_response() is True 710 assert len(sent) == 1 711 import json as _j 712 envelope = _j.loads(sent[0]) 713 assert envelope == {"type": "response.cancel"} 714 715 716 def test_realtime_session_counters_initialized(): 717 from plugins.google_meet.realtime.openai_client import RealtimeSession 718 719 sess = RealtimeSession(api_key="sk-test", audio_sink_path=None) 720 assert sess.audio_bytes_out == 0 721 assert sess.last_audio_out_at is None 722 723 724 # --------------------------------------------------------------------------- 725 # hermes meet install CLI 726 # --------------------------------------------------------------------------- 727 728 def test_cli_install_subcommand_is_registered(): 729 import argparse 730 from plugins.google_meet.cli import register_cli 731 732 parser = argparse.ArgumentParser(prog="hermes meet") 733 register_cli(parser) 734 735 ns = parser.parse_args(["install"]) 736 assert ns.meet_command == "install" 737 assert ns.realtime is False 738 assert ns.yes is False 739 740 741 def test_cli_install_flags_parse(): 742 import argparse 743 from plugins.google_meet.cli import register_cli 744 745 parser = argparse.ArgumentParser(prog="hermes meet") 746 register_cli(parser) 747 748 ns = parser.parse_args(["install", "--realtime", "--yes"]) 749 assert ns.realtime is True 750 assert ns.yes is True 751 752 753 def test_cmd_install_refuses_windows(capsys): 754 from plugins.google_meet.cli import _cmd_install 755 756 with patch("plugins.google_meet.cli.platform" if False else "platform.system", 757 return_value="Windows"): 758 rc = _cmd_install(realtime=False, assume_yes=True) 759 assert rc == 1 760 out = capsys.readouterr().out 761 assert "Windows" in out 762 763 764 def test_cmd_install_runs_pip_and_playwright(capsys): 765 """End-to-end wiring: pip + playwright install invoked, returncodes handled.""" 766 from plugins.google_meet.cli import _cmd_install 767 import subprocess as _sp 768 769 calls = [] 770 class _FakeRes: 771 def __init__(self, rc=0): self.returncode = rc 772 773 def _fake_run(argv, **kwargs): 774 calls.append(list(argv)) 775 return _FakeRes(0) 776 777 with patch("platform.system", return_value="Linux"), \ 778 patch("subprocess.run", side_effect=_fake_run), \ 779 patch("shutil.which", return_value="/usr/bin/paplay"): 780 rc = _cmd_install(realtime=False, assume_yes=True) 781 assert rc == 0 782 # First invocation: pip install 783 pip_cmds = [c for c in calls if len(c) > 2 and c[1:4] == ["-m", "pip", "install"]] 784 assert pip_cmds, f"no pip install run: {calls}" 785 assert "playwright" in pip_cmds[0] 786 assert "websockets" in pip_cmds[0] 787 # Second: playwright install chromium 788 pw_cmds = [c for c in calls if len(c) > 2 and c[1:4] == ["-m", "playwright", "install"]] 789 assert pw_cmds, f"no playwright install run: {calls}" 790 assert "chromium" in pw_cmds[0] 791 792 793 def test_cmd_install_realtime_skips_when_deps_present(capsys): 794 """When paplay + pactl are already on PATH, no sudo call happens.""" 795 from plugins.google_meet.cli import _cmd_install 796 797 calls = [] 798 class _FakeRes: 799 def __init__(self, rc=0): self.returncode = rc 800 801 def _fake_run(argv, **kwargs): 802 calls.append(list(argv)) 803 return _FakeRes(0) 804 805 with patch("platform.system", return_value="Linux"), \ 806 patch("subprocess.run", side_effect=_fake_run), \ 807 patch("shutil.which", return_value="/usr/bin/paplay"): 808 rc = _cmd_install(realtime=True, assume_yes=True) 809 assert rc == 0 810 # No sudo apt-get call — paplay was already on PATH. 811 sudo_calls = [c for c in calls if c and c[0] == "sudo"] 812 assert sudo_calls == [], f"unexpected sudo invocation: {sudo_calls}" 813 out = capsys.readouterr().out 814 assert "already installed" in out