87 lines
3.0 KiB
Python
87 lines
3.0 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()
|