diff --git a/README.md b/README.md index 8a6ef39..8ab7581 100644 --- a/README.md +++ b/README.md @@ -5,25 +5,33 @@ Async Python client for [NanoKVM](https://github.com/sipeed/NanoKVM). ## Usage ```python - -from aiohttp import ClientSession -from nanokvm.models import GpioType from nanokvm.client import NanoKVMClient +from nanokvm.models import GpioType, MouseButton - -async with ClientSession() as session: - client = NanoKVMClient("http://kvm-8b76.local/api/", session) +async with NanoKVMClient("http://kvm-8b76.local/api/") as client: await client.authenticate("username", "password") + # Get device information dev = await client.get_info() hw = await client.get_hardware() gpio = await client.get_gpio() + # List available images + images = await client.get_images() + + # Keyboard input await client.paste_text("Hello\nworld!") + # Mouse control + await client.mouse_click(MouseButton.LEFT, 0.5, 0.5) + await client.mouse_move_abs(0.25, 0.75) + await client.mouse_scroll(0, -3) + + # Stream video async for frame in client.mjpeg_stream(): print(frame) + # Control GPIO await client.push_button(GpioType.POWER, duration_ms=1000) ``` diff --git a/nanokvm/client.py b/nanokvm/client.py index e3084c0..356a957 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -49,6 +49,7 @@ LoginReq, LoginRsp, MountImageReq, + MouseButton, MouseJigglerMode, PasteReq, SetGpioReq, @@ -109,22 +110,45 @@ class NanoKVMClient: def __init__( self, url: str, - session: ClientSession, *, token: str | None = None, request_timeout: int = 10, ) -> None: - """Initialize the NanoKVM client.""" + """ + Initialize the NanoKVM client. + + Args: + url: Base URL of the NanoKVM API (e.g., "http://192.168.1.1/api/") + token: Optional pre-existing authentication token + request_timeout: Request timeout in seconds (default: 10) + """ self.url = yarl.URL(url) - self.session = session + self._session: ClientSession | None = None self._token = token self._request_timeout = request_timeout + self._ws: aiohttp.ClientWebSocketResponse | None = None @property def token(self) -> str | None: """Return the current auth token.""" return self._token + async def __aenter__(self) -> NanoKVMClient: + """Async context manager entry.""" + self._session = ClientSession() + return self + + async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + """Async context manager exit - cleanup resources.""" + # Close WebSocket connection + if self._ws is not None and not self._ws.closed: + await self._ws.close() + self._ws = None + # Close HTTP session + if self._session is not None: + await self._session.close() + self._session = None + @contextlib.asynccontextmanager async def _request( self, @@ -135,13 +159,17 @@ async def _request( **kwargs: Any, ) -> AsyncIterator[ClientResponse]: """Make an API request.""" + assert self._session is not None, ( + "Client session not initialized. " + "Use as context manager: 'async with NanoKVMClient(url) as client:'" + ) cookies = {} if authenticate: if not self._token: raise NanoKVMNotAuthenticatedError("Client is not authenticated") cookies["nano-kvm-token"] = self._token - async with self.session.request( + async with self._session.request( method, self.url / path.lstrip("/"), headers={ @@ -663,3 +691,131 @@ async def set_mouse_jiggler_state( "/vm/mouse-jiggler", data=SetMouseJigglerReq(enabled=enabled, mode=mode), ) + + async def _get_ws(self) -> aiohttp.ClientWebSocketResponse: + """Get or create WebSocket connection for mouse events.""" + if self._ws is None or self._ws.closed: + assert self._session is not None, ( + "Client session not initialized. " + "Use as context manager: 'async with NanoKVMClient(url) as client:'" + ) + + if not self._token: + raise NanoKVMNotAuthenticatedError("Client is not authenticated") + + # WebSocket URL uses ws:// or wss:// scheme + scheme = "ws" if self.url.scheme == "http" else "wss" + ws_url = self.url.with_scheme(scheme) / "ws" + + self._ws = await self._session.ws_connect( + str(ws_url), + headers={"Cookie": f"nano-kvm-token={self._token}"}, + ) + return self._ws + + async def _send_mouse_event( + self, event_type: int, button_state: int, x: float, y: float + ) -> None: + """ + Send a mouse event via WebSocket. + + Args: + event_type: 0=mouse_up, 1=mouse_down, 2=move_abs, 3=move_rel, 4=scroll + button_state: Button state (0=no buttons, 1=left, 2=right, 4=middle) + x: X coordinate (0.0-1.0 for abs/rel/scroll) or 0.0 for button events + y: Y coordinate (0.0-1.0 for abs/rel/scroll) or 0.0 for button events + """ + ws = await self._get_ws() + + # Scale coordinates for absolute/relative movements and scroll + if event_type in (2, 3, 4): # move_abs, move_rel, or scroll + x_val = int(x * 32768) + y_val = int(y * 32768) + else: + x_val = int(x) + y_val = int(y) + + # Message format: [2, event_type, button_state, x_val, y_val] + # where 2 indicates mouse event + message = [2, event_type, button_state, x_val, y_val] + + _LOGGER.debug("Sending mouse event: %s", message) + await ws.send_json(message) + + async def mouse_move_abs(self, x: float, y: float) -> None: + """ + Move mouse to absolute position. + + Args: + x: X coordinate (0.0 to 1.0, left to right) + y: Y coordinate (0.0 to 1.0, top to bottom) + """ + await self._send_mouse_event(2, 0, x, y) + + async def mouse_move_rel(self, dx: float, dy: float) -> None: + """ + Move mouse relative to current position. + + Args: + dx: Horizontal movement (-1.0 to 1.0) + dy: Vertical movement (-1.0 to 1.0) + """ + await self._send_mouse_event(3, 0, dx, dy) + + async def mouse_down(self, button: MouseButton = MouseButton.LEFT) -> None: + """ + Press a mouse button. + + Args: + button: Mouse button to press (MouseButton.LEFT, MouseButton.RIGHT, + MouseButton.MIDDLE) + """ + await self._send_mouse_event(1, int(button), 0.0, 0.0) + + async def mouse_up(self) -> None: + """ + Release a mouse button. + + Note: Mouse up event always uses button_state=0 per the NanoKVM protocol. + """ + await self._send_mouse_event(0, 0, 0.0, 0.0) + + async def mouse_click( + self, + button: MouseButton = MouseButton.LEFT, + x: float | None = None, + y: float | None = None, + ) -> None: + """ + Click a mouse button at current position or specified coordinates. + + Args: + button: Mouse button to click (MouseButton.LEFT, MouseButton.RIGHT, + MouseButton.MIDDLE) + x: Optional X coordinate (0.0 to 1.0) for absolute positioning + before click + y: Optional Y coordinate (0.0 to 1.0) for absolute positioning + before click + """ + # Move to position if coordinates provided + if x is not None and y is not None: + await self.mouse_move_abs(x, y) + # Small delay to ensure position update + await asyncio.sleep(0.05) + + # Send mouse down + await self.mouse_down(button) + # Small delay between down and up + await asyncio.sleep(0.05) + # Send mouse up + await self.mouse_up() + + async def mouse_scroll(self, dx: float, dy: float) -> None: + """ + Scroll the mouse wheel. + + Args: + dx: Horizontal scroll amount (-1.0 to 1.0) + dy: Vertical scroll amount (-1.0 to 1.0) # positive=up, negative=down) + """ + await self._send_mouse_event(4, 0, dx, dy) diff --git a/nanokvm/models.py b/nanokvm/models.py index b910ef8..35f4458 100644 --- a/nanokvm/models.py +++ b/nanokvm/models.py @@ -102,6 +102,14 @@ class MouseJigglerMode(StrEnum): RELATIVE = "relative" +class MouseButton(IntEnum): + """Mouse Button types.""" + + LEFT = 1 + RIGHT = 2 + MIDDLE = 4 + + # Generic Response Wrapper class ApiResponse(BaseModel, Generic[T]): """Generic API response structure.""" diff --git a/tests/test_client.py b/tests/test_client.py index 3fc4ecc..f8d2070 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,24 +1,15 @@ -from aiohttp import ClientSession from aioresponses import aioresponses +import pytest from nanokvm.client import NanoKVMApiError, NanoKVMClient from nanokvm.models import ApiResponseCode -async def test_client() -> None: - """Test the NanoKVMClient.""" - async with ClientSession() as session: - client = NanoKVMClient("http://localhost:8888/api/", session) - assert client is not None - - async def test_get_images_success() -> None: """Test get_images with a successful response.""" - async with ClientSession() as session: - client = NanoKVMClient( - "http://localhost:8888/api/", session, token="test-token" - ) - + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: with aioresponses() as m: m.get( "http://localhost:8888/api/storage/image", @@ -44,11 +35,9 @@ async def test_get_images_success() -> None: async def test_get_images_empty() -> None: """Test get_images with an empty list.""" - async with ClientSession() as session: - client = NanoKVMClient( - "http://localhost:8888/api/", session, token="test-token" - ) - + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: with aioresponses() as m: m.get( "http://localhost:8888/api/storage/image", @@ -63,20 +52,30 @@ async def test_get_images_empty() -> None: async def test_get_images_api_error() -> None: """Test get_images with an API error response.""" - async with ClientSession() as session: - client = NanoKVMClient( - "http://localhost:8888/api/", session, token="test-token" - ) - + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: with aioresponses() as m: m.get( "http://localhost:8888/api/storage/image", payload={"code": -1, "msg": "failed to list images", "data": None}, ) - try: + with pytest.raises(NanoKVMApiError) as exc_info: await client.get_images() - raise AssertionError("Expected NanoKVMApiError to be raised") - except NanoKVMApiError as e: - assert e.code == ApiResponseCode.FAILURE - assert "failed to list images" in e.msg + + assert exc_info.value.code == ApiResponseCode.FAILURE + assert "failed to list images" in exc_info.value.msg + + +async def test_client_context_manager() -> None: + """Test that client properly initializes and cleans up with context manager.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + # Verify session is created + assert client._session is not None + assert not client._session.closed + + # After exiting context, session should be closed + assert client._session is None diff --git a/tests/test_mouse.py b/tests/test_mouse.py new file mode 100644 index 0000000..60304b9 --- /dev/null +++ b/tests/test_mouse.py @@ -0,0 +1,221 @@ +"""Tests for mouse control functionality.""" + +from collections.abc import AsyncGenerator +from typing import Any +from unittest.mock import AsyncMock, patch + +import pytest + +from nanokvm.client import NanoKVMClient +from nanokvm.models import MouseButton + + +@pytest.fixture +async def client_with_mock_ws() -> AsyncGenerator[tuple[NanoKVMClient, AsyncMock], Any]: + """Fixture that provides a client with mocked WebSocket.""" + mock_ws = AsyncMock() + mock_ws.closed = False + + with patch( + "aiohttp.ClientSession.ws_connect", new_callable=AsyncMock, return_value=mock_ws + ): + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + yield client, mock_ws + + +async def test_mouse_move_abs( + client_with_mock_ws: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Test absolute mouse movement.""" + client, mock_ws = client_with_mock_ws + + await client.mouse_move_abs(0.5, 0.5) + + # Verify the WebSocket send was called with correct parameters + mock_ws.send_json.assert_called_once() + message = mock_ws.send_json.call_args[0][0] + # Message format: [2, event_type, button_state, x_val, y_val] + assert message[0] == 2 # mouse event indicator + assert message[1] == 2 # move_abs + assert message[2] == 0 # button state + # 0.5 * 32768 = 16384 + assert message[3] == 16384 + assert message[4] == 16384 + + +async def test_mouse_move_rel( + client_with_mock_ws: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Test relative mouse movement.""" + client, mock_ws = client_with_mock_ws + + await client.mouse_move_rel(0.1, -0.1) + + # Verify the WebSocket send was called with correct parameters + mock_ws.send_json.assert_called_once() + message = mock_ws.send_json.call_args[0][0] + # Message format: [2, event_type, button_state, x_val, y_val] + assert message[0] == 2 # mouse event indicator + assert message[1] == 3 # move_rel + assert message[2] == 0 # button state + # 0.1 * 32768 = 3276.8 -> 3276 + assert message[3] == 3276 + # -0.1 * 32768 = -3276.8 -> -3276 + assert message[4] == -3276 + + +async def test_mouse_down(client_with_mock_ws: tuple[NanoKVMClient, AsyncMock]) -> None: + """Test mouse button down.""" + client, mock_ws = client_with_mock_ws + + await client.mouse_down(MouseButton.LEFT) + + # Verify the WebSocket send was called with correct parameters + mock_ws.send_json.assert_called_once() + message = mock_ws.send_json.call_args[0][0] + # Message format: [2, event_type, button_state, x_val, y_val] + assert message[0] == 2 # mouse event indicator + assert message[1] == 1 # mouse_down + assert message[2] == 1 # left button + assert message[3] == 0 + assert message[4] == 0 + + +async def test_mouse_up(client_with_mock_ws: tuple[NanoKVMClient, AsyncMock]) -> None: + """Test mouse button up.""" + client, mock_ws = client_with_mock_ws + + await client.mouse_up() + + # Verify the WebSocket send was called with correct parameters + mock_ws.send_json.assert_called_once() + message = mock_ws.send_json.call_args[0][0] + # Message format: [2, event_type, button_state, x_val, y_val] + assert message[0] == 2 # mouse event indicator + assert message[1] == 0 # mouse_up + assert message[2] == 0 # button state (always 0 for mouse_up) + assert message[3] == 0 + assert message[4] == 0 + + +async def test_mouse_click_without_position( + client_with_mock_ws: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Test mouse click at current position.""" + client, mock_ws = client_with_mock_ws + + await client.mouse_click(MouseButton.LEFT) + + # Should send two events: down and up + assert mock_ws.send_json.call_count == 2 + + # First call should be mouse_down + first_message = mock_ws.send_json.call_args_list[0][0][0] + assert first_message[0] == 2 # mouse event indicator + assert first_message[1] == 1 # mouse_down + assert first_message[2] == 1 # left button + + # Second call should be mouse_up + second_message = mock_ws.send_json.call_args_list[1][0][0] + assert second_message[0] == 2 # mouse event indicator + assert second_message[1] == 0 # mouse_up + assert second_message[2] == 0 # button state (always 0 for mouse_up) + + +async def test_mouse_click_with_position( + client_with_mock_ws: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Test mouse click at specific position.""" + client, mock_ws = client_with_mock_ws + + await client.mouse_click(MouseButton.MIDDLE, 0.25, 0.75) + + # Should send three events: move_abs, down, and up + assert mock_ws.send_json.call_count == 3 + + # First call should be move_abs + first_message = mock_ws.send_json.call_args_list[0][0][0] + assert first_message[0] == 2 # mouse event indicator + assert first_message[1] == 2 # move_abs + assert first_message[3] == int(0.25 * 32768) + assert first_message[4] == int(0.75 * 32768) + + # Second call should be mouse_down + second_message = mock_ws.send_json.call_args_list[1][0][0] + assert second_message[0] == 2 # mouse event indicator + assert second_message[1] == 1 # mouse_down + assert second_message[2] == 4 # middle button + + # Third call should be mouse_up + third_message = mock_ws.send_json.call_args_list[2][0][0] + assert third_message[0] == 2 # mouse event indicator + assert third_message[1] == 0 # mouse_up + assert third_message[2] == 0 # button state (always 0 for mouse_up) + + +async def test_mouse_scroll( + client_with_mock_ws: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Test mouse scroll.""" + client, mock_ws = client_with_mock_ws + + await client.mouse_scroll(0.1, -0.2) + + # Verify the WebSocket send was called with correct parameters + mock_ws.send_json.assert_called_once() + message = mock_ws.send_json.call_args[0][0] + # Message format: [2, event_type, button_state, x_val, y_val] + assert message[0] == 2 # mouse event indicator + assert message[1] == 4 # scroll + assert message[2] == 0 # button state + # 0.1 * 32768 = 3276.8 -> 3276 + assert message[3] == 3276 + # -0.2 * 32768 = -6553.6 -> -6553 + assert message[4] == -6553 + + +async def test_context_manager_cleanup() -> None: + """Test that context manager properly closes WebSocket and session.""" + mock_ws = AsyncMock() + mock_ws.closed = False + + with patch( + "aiohttp.ClientSession.ws_connect", new_callable=AsyncMock, return_value=mock_ws + ): + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + # Trigger WebSocket creation by making a call + await client.mouse_move_abs(0.0, 0.0) + # Verify WebSocket was created + assert client._ws is not None + assert client._session is not None + + # After exiting context, resources should be cleaned up + mock_ws.close.assert_called_once() + assert client._ws is None + assert client._session is None + + +async def test_button_mapping( + client_with_mock_ws: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Test different button types.""" + client, mock_ws = client_with_mock_ws + + # Test left button + await client.mouse_down(MouseButton.LEFT) + message = mock_ws.send_json.call_args[0][0] + assert message[2] == MouseButton.LEFT + + # Test right button + await client.mouse_down(MouseButton.RIGHT) + message = mock_ws.send_json.call_args[0][0] + assert message[2] == MouseButton.RIGHT + + # Test middle button + await client.mouse_down(MouseButton.MIDDLE) + message = mock_ws.send_json.call_args[0][0] + assert message[2] == MouseButton.MIDDLE