131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import browser_cli.native_host as native_host
|
|
|
|
|
|
def _raise_system_exit(code: int):
|
|
raise SystemExit(code)
|
|
|
|
|
|
def test_cleanup_removes_socket_and_registry_entry(monkeypatch, tmp_path):
|
|
alias = "work"
|
|
socket_path = tmp_path / "work.sock"
|
|
socket_path.write_text("")
|
|
registry_path = tmp_path / "registry.json"
|
|
registry_path.write_text(json.dumps({alias: str(socket_path), "other": str(tmp_path / "other.sock")}))
|
|
|
|
monkeypatch.setattr(native_host, "REGISTRY_PATH", registry_path)
|
|
monkeypatch.setattr(native_host, "_socket_path_for", lambda alias: str(socket_path))
|
|
monkeypatch.setattr(native_host, "is_windows", lambda: False)
|
|
|
|
native_host._cleanup(alias)
|
|
|
|
assert not socket_path.exists()
|
|
assert json.loads(registry_path.read_text()) == {"other": str(tmp_path / "other.sock")}
|
|
|
|
|
|
def test_stdin_reader_cleans_up_on_eof(monkeypatch):
|
|
cleaned = []
|
|
|
|
monkeypatch.setattr(native_host, "read_native_message", lambda stream: None)
|
|
monkeypatch.setattr(native_host, "_cleanup", cleaned.append)
|
|
monkeypatch.setattr(native_host.os, "_exit", _raise_system_exit)
|
|
monkeypatch.setattr(native_host.sys, "stdin", SimpleNamespace(buffer=object()))
|
|
|
|
with pytest.raises(SystemExit, match="0"):
|
|
native_host.stdin_reader("work")
|
|
|
|
assert cleaned == ["work"]
|
|
|
|
|
|
def test_cleanup_windows_skips_socket_unlink(monkeypatch, tmp_path):
|
|
registry_path = tmp_path / "registry.json"
|
|
registry_path.write_text(json.dumps({"work": r"\\.\pipe\browser-cli-work"}))
|
|
|
|
monkeypatch.setattr(native_host, "REGISTRY_PATH", registry_path)
|
|
monkeypatch.setattr(native_host, "is_windows", lambda: True)
|
|
|
|
native_host._cleanup("work")
|
|
|
|
assert json.loads(registry_path.read_text()) == {}
|
|
|
|
|
|
def test_stdin_reader_cleans_up_on_bye(monkeypatch):
|
|
cleaned = []
|
|
messages = iter([{"type": "bye"}])
|
|
|
|
monkeypatch.setattr(native_host, "read_native_message", lambda stream: next(messages))
|
|
monkeypatch.setattr(native_host, "_cleanup", cleaned.append)
|
|
monkeypatch.setattr(native_host.os, "_exit", _raise_system_exit)
|
|
monkeypatch.setattr(native_host.sys, "stdin", SimpleNamespace(buffer=object()))
|
|
|
|
with pytest.raises(SystemExit, match="0"):
|
|
native_host.stdin_reader("work")
|
|
|
|
assert cleaned == ["work"]
|
|
|
|
|
|
def test_stdin_reader_routes_response_messages(monkeypatch):
|
|
response_queue = native_host.queue.Queue()
|
|
native_host.PENDING["msg-1"] = response_queue
|
|
messages = iter([{"type": "hello"}, {"id": "msg-1", "success": True}, None])
|
|
|
|
monkeypatch.setattr(native_host, "read_native_message", lambda stream: next(messages))
|
|
monkeypatch.setattr(native_host, "_cleanup", lambda alias: None)
|
|
monkeypatch.setattr(native_host.os, "_exit", _raise_system_exit)
|
|
monkeypatch.setattr(native_host.sys, "stdin", SimpleNamespace(buffer=object()))
|
|
|
|
with pytest.raises(SystemExit, match="0"):
|
|
native_host.stdin_reader("work")
|
|
|
|
assert response_queue.get_nowait() == {"id": "msg-1", "success": True}
|
|
native_host.PENDING.clear()
|
|
|
|
|
|
def test_collect_paged_browser_command_accumulates_pages(monkeypatch):
|
|
calls = []
|
|
pages = iter([
|
|
{"success": True, "data": {"__browserCliPage": True, "items": [1, 2], "total": 3, "nextOffset": 2}},
|
|
{"success": True, "data": {"__browserCliPage": True, "items": [3], "total": 3, "nextOffset": None}},
|
|
])
|
|
|
|
def fake_send(cmd):
|
|
calls.append(cmd)
|
|
return next(pages)
|
|
|
|
monkeypatch.setattr(native_host, "PAGE_SIZE", 2)
|
|
monkeypatch.setattr(native_host, "_send_browser_command", fake_send)
|
|
|
|
result = native_host._collect_paged_browser_command({"id": "orig", "command": "tabs.list", "args": {"foo": "bar"}})
|
|
|
|
assert result == {"id": "orig", "success": True, "data": [1, 2, 3], "pageSize": 2, "total": 3}
|
|
assert [call["args"]["__page"] for call in calls] == [
|
|
{"offset": 0, "limit": 2},
|
|
{"offset": 2, "limit": 2},
|
|
]
|
|
assert all(call["args"]["foo"] == "bar" for call in calls)
|
|
assert all(call["id"] != "orig" for call in calls)
|
|
|
|
|
|
def test_collect_paged_browser_command_passes_through_non_paged_response(monkeypatch):
|
|
monkeypatch.setattr(native_host, "_send_browser_command", lambda cmd: {"id": cmd["id"], "success": True, "data": {"value": 1}})
|
|
|
|
result = native_host._collect_paged_browser_command({"id": "orig", "command": "tabs.list", "args": {}})
|
|
|
|
assert result == {"id": "orig", "success": True, "data": {"value": 1}}
|
|
|
|
|
|
def test_handle_browser_command_pages_known_list_commands(monkeypatch):
|
|
seen = []
|
|
|
|
monkeypatch.setattr(native_host, "_collect_paged_browser_command", lambda cmd: seen.append(cmd) or {"success": True, "data": []})
|
|
|
|
result = native_host._handle_browser_command({"id": "orig", "command": "tabs.list", "args": {}})
|
|
|
|
assert result == {"success": True, "data": []}
|
|
assert seen[0]["command"] == "tabs.list"
|