From 34bd4e50f4f0f72998cab6fb0acb4c97dbec809c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Antonio=20V=C3=A1zquez=20Blanco?= Date: Fri, 16 Oct 2026 11:52:13 +0200 Subject: [PATCH] Add a sync client --- README.md | 4 + docs/sync.md | 56 +++++++++ nanokvm/sync_client.py | 199 +++++++++++++++++++++++++++++++ tests/test_sync_client.py | 240 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 499 insertions(+) create mode 100644 docs/sync.md create mode 100644 nanokvm/sync_client.py create mode 100644 tests/test_sync_client.py diff --git a/README.md b/README.md index fc0f05a..ca445bf 100644 --- a/README.md +++ b/README.md @@ -35,11 +35,15 @@ asyncio.run(main()) The client detects the hardware and firmware, and raises `NanoKVMNotSupportedError` instead of calling an endpoint the device lacks. +Scripts that do not use asyncio can use the same API as blocking calls through +`nanokvm.sync_client.NanoKVMSyncClient`. + ## Documentation - [Connecting and authentication](docs/connecting.md): HTTPS and self-signed certificates, sessions, passwords and user accounts - [API reference](docs/api.md): every method and the hardware and firmware it needs +- [Blocking client](docs/sync.md): the API without asyncio - [Errors](docs/errors.md) - [SSH](docs/ssh.md) diff --git a/docs/sync.md b/docs/sync.md new file mode 100644 index 0000000..79bd083 --- /dev/null +++ b/docs/sync.md @@ -0,0 +1,56 @@ +# Blocking client + +`NanoKVMSyncClient` offers the API of `NanoKVMClient` as blocking calls, for +scripts that do not use asyncio: + +```python +from pathlib import Path + +from nanokvm.models import GpioType +from nanokvm.sync_client import NanoKVMSyncClient + +with NanoKVMSyncClient("https://kvm.local/api/") as kvm: + kvm.authenticate("username", "password") + + kvm.push_button(GpioType.POWER, duration_ms=6000) + print(kvm.get_gpio().pwr) + + for frame in kvm.mjpeg_stream_bytes(): + Path("screen.jpg").write_bytes(frame) + break +``` + +- Every coroutine method of `NanoKVMClient` is a blocking method with the same + arguments, results and exceptions. Properties such as `token` and + `hw_version` are available too, and setting `url` points the client at + another device. +- Methods that stream, such as `mjpeg_stream()` and `watch_image_download()`, + return iterators. Leaving the loop early closes the stream. +- The constructor takes the arguments of `NanoKVMClient`, except `session`. +- The client runs on its own event loop in a background thread, so it also + works where an event loop is already running, such as in Jupyter. Use it in a + `with` block or call `close()`. +- Ctrl-C cancels the operation in progress. +- Progress callbacks run on the client's thread and cannot call the client. + +Type checkers see the blocking methods as untyped. `run()` runs a coroutine of +the wrapped `client` on its event loop and keeps the result type: + +```python +gpio = kvm.run(kvm.client.get_gpio()) +``` + +`run()` is also the way to the async-only `open_ws()`, whose WebSocket belongs +to the client's event loop: + +```python +async def read_message(client): + ws = await client.open_ws("stream/h264/webrtc") + try: + return await ws.receive() + finally: + await ws.close() + + +message = kvm.run(read_message(kvm.client)) +``` diff --git a/nanokvm/sync_client.py b/nanokvm/sync_client.py new file mode 100644 index 0000000..7a644f9 --- /dev/null +++ b/nanokvm/sync_client.py @@ -0,0 +1,199 @@ +"""Blocking client for NanoKVM devices.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncGenerator, Coroutine, Iterator +import concurrent.futures +import functools +import inspect +import threading +from types import TracebackType +from typing import Any, TypeVar + +from .client import NanoKVMClient + +__all__ = ["NanoKVMSyncClient"] + +T = TypeVar("T") + +# Their results stay bound to the client's event loop; use ``run()`` instead. +_ASYNC_ONLY = frozenset({"open_ws"}) + +# Wait in short slices: before Python 3.14, an untimed wait on Windows ignores +# Ctrl-C until the operation finishes. +_WAIT_INTERVAL = 0.1 + + +async def _anext(agen: AsyncGenerator[T, None]) -> T: + return await anext(agen) + + +async def _aclose(agen: AsyncGenerator[Any, None]) -> None: + await agen.aclose() + + +class NanoKVMSyncClient: + """Blocking client for the NanoKVM, for code that does not use asyncio. + + Every public coroutine method of ``NanoKVMClient`` is available as a + blocking method with the same arguments, results and exceptions. Methods + that stream return iterators. The async client runs on a private event + loop in a background thread. + """ + + def __init__(self, url: str, **kwargs: Any) -> None: + """Start the client. Arguments are those of ``NanoKVMClient``. + + ``session`` is not supported: the client owns its HTTP session. + """ + if "session" in kwargs: + raise TypeError("NanoKVMSyncClient does not accept a session") + + self._closed = False + self._loop = asyncio.new_event_loop() + self._thread = threading.Thread( + target=self._loop.run_forever, name="nanokvm-sync-client", daemon=True + ) + self._thread.start() + try: + self._client = self.run(self._start(url, kwargs)) + except BaseException: + self._stop(None) + raise + + @property + def client(self) -> NanoKVMClient: + """The wrapped async client. Await its coroutines through ``run()``.""" + return self._client + + def run(self, coro: Coroutine[Any, Any, T]) -> T: + """Run a coroutine on the client's event loop and return its result. + + Use it for typed results, such as ``run(client.get_gpio())``, or for + the async-only API. An interrupt, such as Ctrl-C, cancels it. + """ + if self._closed: + coro.close() + raise RuntimeError("NanoKVMSyncClient is closed") + if threading.current_thread() is self._thread: + coro.close() + raise RuntimeError( + "NanoKVMSyncClient cannot be called from its own event loop, " + "such as from a progress callback" + ) + + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + try: + while not future.done(): + concurrent.futures.wait((future,), timeout=_WAIT_INTERVAL) + except BaseException: + future.cancel() + raise + return future.result() + + def close(self) -> None: + """Close the client and stop its event loop. Repeated calls do nothing.""" + if not self._closed: + self._stop(self._client) + + def __enter__(self) -> NanoKVMSyncClient: + """Context manager entry.""" + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Context manager exit - close the client.""" + self.close() + + def __getattr__(self, name: str) -> Any: + """Expose the async client's public API as blocking calls.""" + if name.startswith("_"): + raise AttributeError( + f"{type(self).__name__!r} object has no attribute {name!r}" + ) + if name in _ASYNC_ONLY: + raise AttributeError( + f"{name} is only available on NanoKVMClient; " + f"call it through run(client.{name}(...))" + ) + try: + attr = getattr(self._client, name) + except AttributeError: + raise AttributeError( + f"{type(self).__name__!r} object has no attribute {name!r}" + ) from None + + if inspect.isasyncgenfunction(attr): + + @functools.wraps(attr) + def iterate(*args: Any, **kwargs: Any) -> Iterator[Any]: + return self._iterate(attr(*args, **kwargs)) + + return iterate + + if inspect.iscoroutinefunction(attr): + + @functools.wraps(attr) + def call(*args: Any, **kwargs: Any) -> Any: + return self.run(attr(*args, **kwargs)) + + return call + + return attr + + def __setattr__(self, name: str, value: Any) -> None: + """Set the async client's properties, such as ``url``, on the client.""" + if not name.startswith("_") and isinstance( + getattr(NanoKVMClient, name, None), property + ): + setattr(self._client, name, value) + else: + super().__setattr__(name, value) + + def __dir__(self) -> list[str]: + """List the blocking API next to the client's own attributes.""" + public = {name for name in dir(self._client) if not name.startswith("_")} + return sorted(set(super().__dir__()) | (public - _ASYNC_ONLY)) + + def _iterate(self, agen: AsyncGenerator[T, None]) -> Iterator[T]: + """Drive an async generator from the calling thread.""" + while True: + try: + item = self.run(_anext(agen)) + except StopAsyncIteration: + return + try: + yield item + except BaseException: + # The caller stopped early, while the async generator waits at a + # yield. A step that raised or was cancelled already ended it. + if not self._closed: + self.run(_aclose(agen)) + raise + + @staticmethod + async def _start(url: str, kwargs: dict[str, Any]) -> NanoKVMClient: + client = NanoKVMClient(url, **kwargs) + await client.__aenter__() + return client + + async def _shutdown(self, client: NanoKVMClient | None) -> None: + # Close open streams before the HTTP session they read from. + await self._loop.shutdown_asyncgens() + if client is not None: + await client.__aexit__(None, None, None) + await self._loop.shutdown_default_executor() + + def _stop(self, client: NanoKVMClient | None) -> None: + try: + self.run(self._shutdown(client)) + finally: + self._closed = True + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join() + self._loop.close() diff --git a/tests/test_sync_client.py b/tests/test_sync_client.py new file mode 100644 index 0000000..92b9efa --- /dev/null +++ b/tests/test_sync_client.py @@ -0,0 +1,240 @@ +"""Tests for the blocking client.""" + +import asyncio +from collections.abc import AsyncGenerator, Iterator +from pathlib import Path +import threading +from typing import Any +from unittest.mock import AsyncMock, patch + +from aioresponses import aioresponses +import pytest +import yarl + +from nanokvm.client import NanoKVMClient, NanoKVMNotSupportedError +from nanokvm.models import GetGpioRsp, GpioType, HWVersion +from nanokvm.sync_client import NanoKVMSyncClient + +from .common import BASE_URL, mark_detected + +_GPIO_URL = f"{BASE_URL}vm/gpio" +_THREAD_NAME = "nanokvm-sync-client" + + +def _ok(data: object = None) -> dict[str, object]: + return {"code": 0, "msg": "success", "data": data} + + +def _client_threads() -> list[threading.Thread]: + return [t for t in threading.enumerate() if t.name == _THREAD_NAME] + + +@pytest.fixture +def kvm() -> Iterator[NanoKVMSyncClient]: + """A blocking client that already holds a session token.""" + with NanoKVMSyncClient(BASE_URL, token="test-token") as kvm: + yield kvm + + +def test_blocking_calls_authenticate_and_return_results( + mocked: aioresponses, +) -> None: + """Coroutine methods block and return what the async client returns.""" + mocked.post(f"{BASE_URL}auth/login", payload=_ok({"token": "new-token"})) + mocked.get(f"{BASE_URL}vm/hardware", payload=_ok({"version": "PCIE"})) + mocked.get(_GPIO_URL, payload=_ok({"pwr": True, "hdd": False})) + mocked.post(_GPIO_URL, payload=_ok()) + + with NanoKVMSyncClient(BASE_URL, use_password_obfuscation=False) as kvm: + kvm.authenticate("synthetic-user", "synthetic-password") + assert kvm.token == "new-token" + assert kvm.hw_version is HWVersion.PCIE + + assert kvm.get_gpio() == GetGpioRsp(pwr=True, hdd=False) + kvm.push_button(GpioType.POWER, duration_ms=6000) + + call = mocked.requests[("POST", yarl.URL(_GPIO_URL))][0] + assert call.kwargs["json"] == {"type": "power", "duration": 6000} + + +def test_run_returns_typed_results( + kvm: NanoKVMSyncClient, mocked: aioresponses +) -> None: + """run() awaits a coroutine of the wrapped client on its event loop.""" + mocked.get(_GPIO_URL, payload=_ok({"pwr": False, "hdd": False})) + + assert kvm.run(kvm.client.get_gpio()) == GetGpioRsp(pwr=False, hdd=False) + + +def test_run_reaches_the_async_only_api(kvm: NanoKVMSyncClient) -> None: + """open_ws() works through run(), on the client's own event loop.""" + mock_ws = AsyncMock() + mock_ws.receive.return_value = "offer" + + async def read_message(client: NanoKVMClient) -> Any: + ws = await client.open_ws("stream/h264/webrtc") + try: + return await ws.receive() + finally: + await ws.close() + + with patch( + "aiohttp.ClientSession.ws_connect", new_callable=AsyncMock, return_value=mock_ws + ) as ws_connect: + assert kvm.run(read_message(kvm.client)) == "offer" + + assert str(ws_connect.call_args.args[0]).endswith("/api/stream/h264/webrtc") + mock_ws.close.assert_awaited_once() + + +def test_errors_propagate_unchanged( + kvm: NanoKVMSyncClient, mocked: aioresponses +) -> None: + """Client and transport exceptions reach the caller as they are.""" + mark_detected(kvm.client, HWVersion.PCIE) + with pytest.raises(NanoKVMNotSupportedError, match="get_led_strip"): + kvm.get_led_strip() + + # A timeout raised by the call is not mistaken for the wait between polls. + mocked.get(_GPIO_URL, exception=TimeoutError()) + with pytest.raises(TimeoutError): + kvm.get_gpio() + + +def test_streams_become_iterators( + kvm: NanoKVMSyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Async generators become iterators that close their stream when left.""" + closed = threading.Event() + + async def frames(self: NanoKVMClient) -> AsyncGenerator[bytes, None]: + closed.clear() + try: + for frame in (b"one", b"two", b"three"): + yield frame + finally: + closed.set() + + monkeypatch.setattr(NanoKVMClient, "mjpeg_stream_bytes", frames) + + assert list(kvm.mjpeg_stream_bytes()) == [b"one", b"two", b"three"] + assert closed.is_set() + + for _ in kvm.mjpeg_stream_bytes(): + break + assert closed.is_set() + + frames_iter = kvm.mjpeg_stream_bytes() + assert next(frames_iter) == b"one" + frames_iter.close() + assert closed.is_set() + + +def test_interrupt_cancels_the_call_in_progress( + kvm: NanoKVMSyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Ctrl-C while waiting cancels the operation on the event loop.""" + started = threading.Event() + cancelled = threading.Event() + + async def get_gpio(self: NanoKVMClient) -> None: + started.set() + try: + await asyncio.sleep(60) + except asyncio.CancelledError: + cancelled.set() + raise + + def interrupted_wait(*args: Any, **kwargs: Any) -> None: + assert started.wait(5) + raise KeyboardInterrupt + + monkeypatch.setattr(NanoKVMClient, "get_gpio", get_gpio) + with ( + patch("concurrent.futures.wait", side_effect=interrupted_wait), + pytest.raises(KeyboardInterrupt), + ): + kvm.get_gpio() + + assert cancelled.wait(5) + + +def test_only_the_public_blocking_api_is_exposed(kvm: NanoKVMSyncClient) -> None: + """Private and async-only attributes are not proxied.""" + names = dir(kvm) + assert {"authenticate", "push_button", "mjpeg_stream", "run"} <= set(names) + assert "open_ws" not in names + + with pytest.raises(AttributeError, match="_session"): + kvm._session # noqa: B018 + with pytest.raises(AttributeError, match=r"run\(client\.open_ws"): + kvm.open_ws # noqa: B018 + with pytest.raises(AttributeError, match="'NanoKVMSyncClient'.*'no_such'"): + kvm.no_such # noqa: B018 + + +def test_setting_properties_reaches_the_client( + kvm: NanoKVMSyncClient, mocked: aioresponses +) -> None: + """Assigned properties change the wrapped client, not the wrapper.""" + new_url = "http://new-kvm.local/api/" + mocked.get(f"{new_url}vm/gpio", payload=_ok({"pwr": True, "hdd": False})) + + kvm.url = yarl.URL(new_url) + + assert kvm.client.url == yarl.URL(new_url) + assert kvm.get_gpio() == GetGpioRsp(pwr=True, hdd=False) + + with pytest.raises(AttributeError, match="token"): + kvm.token = "other-token" + assert kvm.token == "test-token" + + +def test_calls_from_the_event_loop_thread_are_rejected( + kvm: NanoKVMSyncClient, +) -> None: + """A callback running on the client's loop cannot block on the client.""" + + async def reenter() -> None: + kvm.get_gpio() + + with pytest.raises(RuntimeError, match="own event loop"): + kvm.run(reenter()) + + +def test_close_releases_the_client_and_its_thread() -> None: + """close() closes the HTTP session, stops the thread and can be repeated.""" + kvm = NanoKVMSyncClient(BASE_URL) + thread = kvm._thread + + kvm.close() + kvm.close() + + assert not thread.is_alive() + assert kvm.client._session._http_session is None + with pytest.raises(RuntimeError, match="closed"): + kvm.get_gpio() + + +def test_failed_start_stops_the_thread(tmp_path: Path) -> None: + """A client that cannot start leaves no event loop thread behind.""" + with pytest.raises(FileNotFoundError): + NanoKVMSyncClient(BASE_URL, ssl_ca_cert=str(tmp_path / "missing.pem")) + + assert not _client_threads() + + +def test_session_is_rejected() -> None: + """An external aiohttp session would belong to another event loop.""" + with pytest.raises(TypeError, match="session"): + NanoKVMSyncClient(BASE_URL, session=object()) + + assert not _client_threads() + + +async def test_usable_while_an_event_loop_is_running(mocked: aioresponses) -> None: + """The private loop lets blocking code run where a loop already runs.""" + mocked.get(_GPIO_URL, payload=_ok({"pwr": True, "hdd": True})) + + with NanoKVMSyncClient(BASE_URL, token="test-token") as kvm: + assert kvm.get_gpio().pwr is True