from __future__ import annotations
import pytest
from mcp_qemu_lab.qmp import QmpClient, QmpEndpoint
from mcp_qemu_lab.util import LabError
class _DummyWriter:
def write(self, _: str) -> None:
return
def flush(self) -> None:
return
def test_qmp_execute_times_out_on_event_only_stream(monkeypatch: pytest.MonkeyPatch) -> None:
client = QmpClient(QmpEndpoint(host="127.0.0.1", port=65535), timeout=0.05)
client.writer = _DummyWriter()
def _read_event() -> dict[str, str]:
return {"event": "TEST_EVENT"}
monkeypatch.setattr(client, "_read_message", _read_event)
with pytest.raises(LabError) as exc:
client.execute("query-status")
assert exc.value.code == "qmp_timeout"
def test_qmp_execute_times_out_on_socket_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
client = QmpClient(QmpEndpoint(host="127.0.0.1", port=65535), timeout=0.05)
client.writer = _DummyWriter()
def _read_timeout() -> dict[str, str]:
raise TimeoutError("simulated timeout")
monkeypatch.setattr(client, "_read_message", _read_timeout)
with pytest.raises(LabError) as exc:
client.execute("query-status")
assert exc.value.code == "qmp_timeout"
