diff --git a/tests/test_account_permissions.py b/tests/test_account_permissions.py index acef537..0fd4da2 100644 --- a/tests/test_account_permissions.py +++ b/tests/test_account_permissions.py @@ -18,15 +18,13 @@ ) from nanokvm.models import GetAccountRsp, HWVersion -from .common import info_payload, mark_detected - -_BASE_URL = "http://kvm.local/api/" +from .common import BASE_URL, info_payload, mark_detected def _websocket_request_info() -> aiohttp.RequestInfo: """Build a typed request descriptor for a synthetic handshake failure.""" return aiohttp.RequestInfo( - yarl.URL("ws://kvm.local/api/ws"), + yarl.URL("ws://localhost:8888/api/ws"), "GET", CIMultiDictProxy(CIMultiDict()), ) @@ -34,9 +32,9 @@ def _websocket_request_info() -> aiohttp.RequestInfo: async def test_get_account_exposes_known_role(mocked: aioresponses) -> None: """The role returned by 2.5.1 is available to callers.""" - async with NanoKVMClient(_BASE_URL, token="synthetic-token") as client: + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: mocked.get( - f"{_BASE_URL}auth/account", + f"{BASE_URL}auth/account", payload={ "code": 0, "msg": "success", @@ -71,8 +69,8 @@ async def test_http_403_is_permission_error_before_body_parsing( body: bytes, mocked: aioresponses ) -> None: """All real 403 bodies map to one structured permission exception.""" - async with NanoKVMClient(_BASE_URL, token="synthetic-token") as client: - mocked.get(f"{_BASE_URL}storage/image", status=403, body=body) + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: + mocked.get(f"{BASE_URL}storage/image", status=403, body=body) with pytest.raises(NanoKVMPermissionError) as exc_info: await client.get_images() @@ -87,9 +85,9 @@ async def test_permission_error_does_not_invalidate_session_or_retry_login( mocked: aioresponses, ) -> None: """A forbidden admin resource leaves the same user session usable.""" - async with NanoKVMClient(_BASE_URL, token="synthetic-token") as client: - mocked.get(f"{_BASE_URL}storage/image", status=403, body=b'"forbidden"') - mocked.get(f"{_BASE_URL}vm/info", payload=info_payload()) + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: + mocked.get(f"{BASE_URL}storage/image", status=403, body=b'"forbidden"') + mocked.get(f"{BASE_URL}vm/info", payload=info_payload()) with pytest.raises(NanoKVMPermissionError): await client.get_images() @@ -100,15 +98,15 @@ async def test_permission_error_does_not_invalidate_session_or_retry_login( assert client.token == "synthetic-token" assert ( "POST", - yarl.URL(f"{_BASE_URL}auth/login"), + yarl.URL(f"{BASE_URL}auth/login"), ) not in mocked.requests async def test_http_401_invalidates_local_session(mocked: aioresponses) -> None: """An authenticated 401 clears the token and local input state.""" - async with NanoKVMClient(_BASE_URL, token="expired-token") as client: + async with NanoKVMClient(BASE_URL, token="expired-token") as client: client._session._mouse_buttons = 1 - mocked.get(f"{_BASE_URL}vm/info", status=401, body=b'"unauthorized"') + mocked.get(f"{BASE_URL}vm/info", status=401, body=b'"unauthorized"') with pytest.raises(NanoKVMNotAuthenticatedError): await client.get_info() @@ -122,9 +120,9 @@ async def test_http_401_invalidates_local_session(mocked: aioresponses) -> None: async def test_http_200_api_error_remains_api_error(mocked: aioresponses) -> None: """An API error envelope at HTTP 200 is distinct from permission failure.""" - async with NanoKVMClient(_BASE_URL, token="synthetic-token") as client: + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: mocked.get( - f"{_BASE_URL}storage/image", + f"{BASE_URL}storage/image", payload={"code": -1, "msg": "read failed", "data": None}, ) @@ -142,8 +140,8 @@ async def test_multipart_403_is_permission_error( script = tmp_path / "synthetic.sh" script.write_text("#!/bin/sh\ntrue\n") - async with NanoKVMClient(_BASE_URL, token="synthetic-token") as client: - mocked.post(f"{_BASE_URL}vm/script/upload", status=403, body=b"") + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: + mocked.post(f"{BASE_URL}vm/script/upload", status=403, body=b"") with pytest.raises(NanoKVMPermissionError) as exc_info: await client.upload_script(script) @@ -167,7 +165,7 @@ async def test_websocket_403_is_permission_error() -> None: new_callable=AsyncMock, side_effect=handshake_error, ): - async with NanoKVMClient(_BASE_URL, token="synthetic-token") as client: + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: mark_detected(client, HWVersion.PCIE, "2.5.1") with pytest.raises(NanoKVMPermissionError) as exc_info: @@ -193,7 +191,7 @@ async def test_websocket_401_clears_session_without_lock_deadlock() -> None: new_callable=AsyncMock, side_effect=handshake_error, ): - async with NanoKVMClient(_BASE_URL, token="expired-token") as client: + async with NanoKVMClient(BASE_URL, token="expired-token") as client: mark_detected(client, HWVersion.PCIE, "2.5.1") client._session._mouse_buttons = 1 @@ -206,8 +204,8 @@ async def test_websocket_401_clears_session_without_lock_deadlock() -> None: async def test_mjpeg_403_is_permission_error(mocked: aioresponses) -> None: """Opening the MJPEG stream classifies forbidden status before parsing.""" - async with NanoKVMClient(_BASE_URL, token="synthetic-token") as client: - mocked.get(f"{_BASE_URL}stream/mjpeg", status=403, body=b"") + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: + mocked.get(f"{BASE_URL}stream/mjpeg", status=403, body=b"") with pytest.raises(NanoKVMPermissionError) as exc_info: async for _frame in client.mjpeg_stream(): @@ -220,8 +218,8 @@ async def test_mjpeg_403_is_permission_error(mocked: aioresponses) -> None: async def test_mjpeg_401_invalidates_session(mocked: aioresponses) -> None: """An expired MJPEG session is classified and cleared like JSON requests.""" - async with NanoKVMClient(_BASE_URL, token="expired-token") as client: - mocked.get(f"{_BASE_URL}stream/mjpeg", status=401, body=b'"unauthorized"') + async with NanoKVMClient(BASE_URL, token="expired-token") as client: + mocked.get(f"{BASE_URL}stream/mjpeg", status=401, body=b'"unauthorized"') with pytest.raises(NanoKVMNotAuthenticatedError): async for _frame in client.mjpeg_stream(): diff --git a/tests/test_auth_concurrency.py b/tests/test_auth_concurrency.py index 52962e8..1b9fb0b 100644 --- a/tests/test_auth_concurrency.py +++ b/tests/test_auth_concurrency.py @@ -1,7 +1,7 @@ """Concurrent authentication must preserve the identity of each operation.""" import asyncio -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager import io from unittest.mock import AsyncMock, patch @@ -21,7 +21,7 @@ def _ok(data: object = None) -> web.Response: @asynccontextmanager -async def _server(app: web.Application) -> AsyncIterator[str]: +async def _server(app: web.Application) -> AsyncGenerator[str, None]: async def hardware(request: web.Request) -> web.Response: return _ok({"version": "PCIE"}) diff --git a/tests/test_client.py b/tests/test_client.py index 2871641..0d2b1fb 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,119 +1,13 @@ -import asyncio -import io -import logging -from pathlib import Path +"""Tests for the client facade and the shared response envelope.""" -import aiohttp -from aiohttp import ClientSession, web +from aiohttp import ClientSession from aioresponses import aioresponses -from PIL import Image import pytest -import yarl - -from nanokvm.client import ( - NanoKVMApiError, - NanoKVMClient, - NanoKVMInvalidResponseError, - NanoKVMNotSupportedError, -) -from nanokvm.compatibility import _parse_version, _version_at_least -from nanokvm.models import ( - ApiResponseCode, - DiskType, - DNSMode, - DownloadStatus, - GetMacRsp, - GetOLEDRsp, - HWVersion, - MouseJigglerMode, - OledType, - ShortcutKey, - StatusImageRsp, - StreamMode, - VirtualDevice, -) - -from .common import info_payload, mark_detected - - -def test_version_parser() -> None: - """Test internal application version parsing and comparison.""" - assert _parse_version("v2.4.1") == (2, 4, 1) - assert _parse_version("dev") is None - assert _version_at_least("1.2.14", "1.2.9") - assert _version_at_least("2.4", "2.4.0") - assert not _version_at_least("2.4.0", "2.4.1") - assert _version_at_least("dev", "2.4.1") - - -@pytest.mark.parametrize( - ("hardware", "expected_prefix"), - [ - (HWVersion.PRO, "/storage/download"), - (HWVersion.PCIE, "/download"), - (HWVersion.UNKNOWN, "/download"), - (None, "/download"), - ], -) -async def test_image_download_prefix_by_hardware_family( - hardware: HWVersion | None, - expected_prefix: str, - client: NanoKVMClient, -) -> None: - """Image download routes use the Pro prefix only for Pro hardware.""" - client._session._hw_version = hardware - - assert client._image_download_prefix() == expected_prefix - - -@pytest.mark.parametrize( - "status", - ["success", "failed", "checksum_failed"], -) -def test_download_status_accepts_terminal_states(status: str) -> None: - """Download status responses include terminal success and failure states.""" - response = StatusImageRsp.model_validate( - {"status": status, "file": "image.iso", "percentage": ""} - ) - - assert response.status is DownloadStatus(status) - - -async def test_get_images_success(client: NanoKVMClient, mocked: aioresponses) -> None: - """Test get_images with a successful response.""" - mocked.get( - "http://localhost:8888/api/storage/image", - payload={ - "code": 0, - "msg": "success", - "data": { - "files": [ - "/data/alpine-standard-3.23.2-x86_64.iso", - "/data/cs10-js.iso", - ] - }, - }, - ) - - response = await client.get_images() - - assert response is not None - assert len(response.files) == 2 - assert "/data/alpine-standard-3.23.2-x86_64.iso" in response.files - assert "/data/cs10-js.iso" in response.files - - -async def test_get_images_empty(client: NanoKVMClient, mocked: aioresponses) -> None: - """Test get_images with an empty list.""" - mocked.get( - "http://localhost:8888/api/storage/image", - payload={"code": 0, "msg": "success", "data": {"files": []}}, - ) - response = await client.get_images() +from nanokvm.client import NanoKVMApiError, NanoKVMClient, NanoKVMInvalidResponseError +from nanokvm.models import GetMacRsp, GetOLEDRsp, HWVersion, OledType - assert response is not None - assert len(response.files) == 0 +from .common import BASE_URL, info_payload, mark_detected async def test_typed_success_response_requires_data( @@ -121,7 +15,7 @@ async def test_typed_success_response_requires_data( ) -> None: """A typed endpoint rejects a successful response without its payload.""" mocked.get( - "http://localhost:8888/api/storage/image", + f"{BASE_URL}storage/image", payload={"code": 0, "msg": "success", "data": None}, ) @@ -134,7 +28,7 @@ async def test_api_error_data_is_not_validated_as_success_model( ) -> None: """An error envelope is classified before its endpoint data is parsed.""" mocked.get( - "http://localhost:8888/api/storage/image", + f"{BASE_URL}storage/image", payload={ "code": -1, "msg": "failed to list images", @@ -148,131 +42,12 @@ async def test_api_error_data_is_not_validated_as_success_model( assert exc_info.value.code == -1 -def test_invalid_response_error_does_not_echo_payload() -> None: - """Malformed response errors do not include arbitrary response values.""" - client = NanoKVMClient("http://localhost:8888/api/") - secret = "SYNTHETIC_SECRET_VALUE" - - with pytest.raises(NanoKVMInvalidResponseError) as exc_info: - client._session.validate_api_response( - {"msg": "invalid", "data": {"token": secret}}, - ) - - assert secret not in str(exc_info.value) - - -async def test_form_response_is_not_logged( - tmp_path: Path, - caplog: pytest.LogCaptureFixture, - mocked: aioresponses, -) -> None: - """Multipart responses are not emitted verbatim to debug logs.""" - script = tmp_path / "test.sh" - script.write_text("#!/bin/sh\ntrue\n") - secret = "SYNTHETIC_FORM_VALUE" - caplog.set_level(logging.DEBUG, logger="nanokvm") - - async with NanoKVMClient( - "http://localhost:8888/api/", token="test-token" - ) as client: - mocked.post( - "http://localhost:8888/api/vm/script/upload", - payload={ - "code": 0, - "msg": "success", - "data": {"file": secret}, - }, - ) - - await client.upload_script(script) - - assert secret not in caplog.text - - -def test_parse_jpeg_returns_fully_loaded_image() -> None: - """JPEG parsing returns an image detached from the source stream.""" - image_buffer = io.BytesIO() - Image.new("RGB", (2, 2), color=(1, 2, 3)).save(image_buffer, format="JPEG") - client = NanoKVMClient("http://localhost:8888/api/") - - image = client._parse_jpeg_from_bytes(image_buffer.getvalue()) - - assert image.getpixel((0, 0)) - assert getattr(image, "fp", None) is None - - -async def test_mjpeg_stream_allows_frames_after_request_timeout() -> None: - """An MJPEG stream remains open beyond the normal request timeout.""" - image_buffer = io.BytesIO() - Image.new("RGB", (2, 2)).save(image_buffer, format="JPEG") - image_data = image_buffer.getvalue() - - async def stream_handler( - request: web.Request, - ) -> web.StreamResponse: - response = web.StreamResponse( - headers={"Content-Type": "multipart/x-mixed-replace; boundary=frame"} - ) - await response.prepare(request) - for _ in range(2): - await response.write( - b"--frame\r\n" - b"Content-Type: image/jpeg\r\n" - + f"Content-Length: {len(image_data)}\r\n\r\n".encode() - + image_data - + b"\r\n" - ) - await asyncio.sleep(0.55) - await response.write(b"--frame--\r\n") - await response.write_eof() - return response - - application = web.Application() - application.router.add_get("/api/stream/mjpeg", stream_handler) - runner = web.AppRunner(application) - await runner.setup() - site = web.TCPSite(runner, "127.0.0.1", 0) - await site.start() - server_sockets = getattr(site._server, "sockets", None) - assert server_sockets - port = server_sockets[0].getsockname()[1] - - frames = [] - try: - async with NanoKVMClient( - f"http://127.0.0.1:{port}/api/", token="test-token", request_timeout=1 - ) as client: - async with asyncio.timeout(3): - async for frame in client.mjpeg_stream(): - frames.append(frame) - finally: - await runner.cleanup() - - assert len(frames) == 2 - - -async def test_get_images_api_error( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test get_images with an API error response.""" - mocked.get( - "http://localhost:8888/api/storage/image", - payload={"code": -1, "msg": "failed to list images", "data": None}, - ) - - with pytest.raises(NanoKVMApiError) as exc_info: - await client.get_images() - - assert exc_info.value.code == ApiResponseCode.FAILURE - assert "failed to list images" in exc_info.value.msg - - async def test_detect_versions_stores_application_and_image( client: NanoKVMClient, mocked: aioresponses ) -> None: """Test detect_versions caches versions from /vm/info.""" mocked.get( - "http://localhost:8888/api/vm/info", + f"{BASE_URL}vm/info", payload=info_payload(application="2.4.1", image="1.4.0"), ) @@ -287,7 +62,7 @@ async def test_api_error_allows_endpoint_specific_codes( ) -> None: """Test endpoint-specific API error codes are surfaced as API errors.""" mocked.post( - "http://localhost:8888/api/storage/image/mount", + f"{BASE_URL}storage/image/mount", payload={"code": -6, "msg": "mount image failed", "data": None}, ) @@ -306,7 +81,7 @@ async def test_none_returning_endpoint_preserves_unknown_api_code( mark_detected(client) mocked.post( - "http://localhost:8888/api/vm/web-title", + f"{BASE_URL}vm/web-title", payload={"code": -4, "msg": "failed to set title", "data": None}, ) @@ -317,236 +92,6 @@ async def test_none_returning_endpoint_preserves_unknown_api_code( assert "failed to set title" in exc_info.value.msg -async def test_get_web_title_returns_default_when_unset( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test the firmware's failed read of an unset title maps to the default.""" - mark_detected(client) - - mocked.get( - "http://localhost:8888/api/vm/web-title", - payload={"code": -1, "msg": "read web title failed", "data": None}, - ) - - assert (await client.get_web_title()).title == "NanoKVM" - - -async def test_mount_image_sends_pro_read_only_flag( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test mount_image sends the Pro readOnly field.""" - mocked.post( - "http://localhost:8888/api/storage/image/mount", - payload={"code": 0, "msg": "success", "data": None}, - ) - - mark_detected(client, HWVersion.PRO, "1.2.1") - await client.mount_image("/data/test.img", read_only=True) - - calls = mocked.requests[ - ("POST", yarl.URL("http://localhost:8888/api/storage/image/mount")) - ] - assert calls[0].kwargs.get("json") == { - "file": "/data/test.img", - "cdrom": False, - "readOnly": True, - } - - -async def test_get_oled_info_raises_for_pro_invalid_file_content( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test Pro OLED invalid file content follows upstream error semantics.""" - client._session._hw_version = HWVersion.PRO - - mocked.get( - "http://localhost:8888/api/vm/oled", - payload={ - "code": -2, - "msg": "invalid file content", - "data": None, - }, - ) - - with pytest.raises(NanoKVMApiError) as exc_info: - await client.get_oled_info() - - assert exc_info.value.code == -2 - assert exc_info.value.msg == "invalid file content" - info_url = yarl.URL("http://localhost:8888/api/vm/info") - assert ("GET", info_url) not in mocked.requests - - -async def test_set_mouse_jiggler_state_noops_when_already_disabled( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test disabling mouse jiggler is idempotent.""" - mark_detected(client) - - mocked.get( - "http://localhost:8888/api/vm/mouse-jiggler", - payload={ - "code": 0, - "msg": "success", - "data": {"enabled": False, "mode": "relative"}, - }, - ) - - await client.set_mouse_jiggler_state( - enabled=False, - mode=MouseJigglerMode.RELATIVE, - ) - - post_url = yarl.URL("http://localhost:8888/api/vm/mouse-jiggler/") - assert ("POST", post_url) not in mocked.requests - - -async def test_update_virtual_device_ignores_success_data( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test update_virtual_device handles non-Pro success payloads.""" - mocked.post( - "http://localhost:8888/api/vm/device/virtual", - payload={"code": 0, "msg": "success", "data": {"on": True}}, - ) - - await client.update_virtual_device(VirtualDevice.DISK) - - calls = mocked.requests[ - ("POST", yarl.URL("http://localhost:8888/api/vm/device/virtual")) - ] - assert calls[0].kwargs.get("json") == {"device": "disk"} - - -async def test_get_virtual_device_status_handles_non_pro_shape( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test virtual device status keeps non-Pro fields intact.""" - mocked.get( - "http://localhost:8888/api/vm/device/virtual", - payload={ - "code": 0, - "msg": "success", - "data": {"network": True, "media": False, "disk": True}, - }, - ) - - response = await client.get_virtual_device_status() - - assert response.network is True - assert response.media is False - assert response.disk is True - assert response.mounted_disk is None - - -async def test_get_virtual_device_status_handles_pro_shape( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test Pro virtual device status exposes Pro fields without fake non-Pro disk.""" - client._session._hw_version = HWVersion.PRO - - mocked.get( - "http://localhost:8888/api/vm/device/virtual", - payload={ - "code": 0, - "msg": "success", - "data": { - "isNetworkEnabled": True, - "isMicEnabled": True, - "mountedDisk": "emmc", - "isSdCardExist": True, - "isEmmcExist": True, - }, - }, - ) - - response = await client.get_virtual_device_status() - - assert response.network is True - assert response.mic is True - assert response.mounted_disk == "emmc" - assert response.is_sd_card_exist is True - assert response.is_emmc_exist is True - assert response.media is None - assert response.disk is None - - -async def test_update_virtual_device_sends_pro_disk_type( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test Pro disk virtual device requests include the disk type.""" - client._session._hw_version = HWVersion.PRO - - mocked.post( - "http://localhost:8888/api/vm/device/virtual", - payload={"code": 0, "msg": "success", "data": None}, - ) - - await client.update_virtual_device( - VirtualDevice.DISK, - disk_type=DiskType.EMMC, - ) - - calls = mocked.requests[ - ("POST", yarl.URL("http://localhost:8888/api/vm/device/virtual")) - ] - assert calls[0].kwargs.get("json") == { - "device": "disk", - "type": "emmc", - } - - -async def test_set_led_strip_partial_preserves_current_config( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test partial LED updates post a complete Pro LED configuration.""" - client._session._hw_version = HWVersion.PRO - - mocked.get( - "http://localhost:8888/api/vm/ledstrip/get", - payload={ - "code": 0, - "msg": "success", - "data": { - "on": True, - "hor": 8, - "ver": 6, - "brightness": 25, - }, - }, - ) - mocked.post( - "http://localhost:8888/api/vm/ledstrip/set", - payload={"code": 0, "msg": "success", "data": None}, - ) - - await client.set_led_strip(brightness=50) - - calls = mocked.requests[ - ("POST", yarl.URL("http://localhost:8888/api/vm/ledstrip/set")) - ] - assert calls[0].kwargs.get("json") == { - "on": True, - "hor": 8, - "ver": 6, - "brightness": 50, - } - - -async def test_get_wol_macs_returns_empty_list_for_missing_file( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test WOL getter normalizes the empty-file device state.""" - mocked.get( - "http://localhost:8888/api/network/wol/mac", - payload={"code": -2, "msg": "open file error", "data": None}, - ) - - response = await client.get_wol_macs() - - assert response.macs == [] - - def test_model_normalizations_for_wol_and_oled() -> None: """Test model-only compatibility normalizers.""" assert GetMacRsp(macs=["", "AA:BB:CC:DD:EE:FF", " "]).macs == [ @@ -558,326 +103,9 @@ def test_model_normalizations_for_wol_and_oled() -> None: ) -async def test_connect_wifi_no_auth_sends_ap_header( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test AP-mode Wi-Fi connection sends setup credentials without auth cookie.""" - mocked.post( - "http://localhost:8888/api/network/wifi", - payload={"code": 0, "msg": "success", "data": None}, - ) - - await client.connect_wifi_no_auth( - "NanoKVM-Setup", - ap_password="setup-secret", - ) - - calls = mocked.requests[ - ("POST", yarl.URL("http://localhost:8888/api/network/wifi")) - ] - assert calls[0].kwargs.get("json") == { - "ssid": "NanoKVM-Setup", - "password": "", - } - assert calls[0].kwargs.get("cookies") == {} - assert calls[0].kwargs.get("headers", {})["X-AP-Key"] == "setup-secret" - - -async def test_get_dns_parses_configuration( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test DNS configuration response parsing.""" - mark_detected(client, application_version="2.4.1") - - mocked.get( - "http://localhost:8888/api/network/dns", - payload={ - "code": 0, - "msg": "success", - "data": { - "mode": "manual", - "servers": ["1.1.1.1", "2606:4700:4700::1111"], - "effective": ["1.1.1.1"], - "dhcp": ["192.168.1.1"], - "info": { - "interface": "eth0", - "type": "Wired", - "address": "192.168.1.20/24", - "subnetMask": "255.255.255.0", - "gateway": "192.168.1.1", - "searchDomains": ["lan"], - }, - }, - }, - ) - - response = await client.get_dns() - - assert response.mode is DNSMode.MANUAL - assert response.servers == ["1.1.1.1", "2606:4700:4700::1111"] - assert response.effective == ["1.1.1.1"] - assert response.dhcp == ["192.168.1.1"] - assert response.info.interface == "eth0" - assert response.info.type == "Wired" - assert response.info.address == "192.168.1.20/24" - assert response.info.subnet_mask == "255.255.255.0" - assert response.info.gateway == "192.168.1.1" - assert response.info.search_domains == ["lan"] - - -async def test_set_dns_manual_sends_servers( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test manual DNS mode sends the configured servers.""" - mark_detected(client, application_version="2.4.1") - - mocked.post( - "http://localhost:8888/api/network/dns", - payload={"code": 0, "msg": "success", "data": None}, - ) - - await client.set_dns( - DNSMode.MANUAL, - ["1.1.1.1", "2606:4700:4700::1111"], - ) - - calls = mocked.requests[("POST", yarl.URL("http://localhost:8888/api/network/dns"))] - assert calls[0].kwargs.get("json") == { - "mode": "manual", - "servers": ["1.1.1.1", "2606:4700:4700::1111"], - } - - -async def test_set_dns_dhcp_string_sends_empty_servers( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test DHCP DNS mode accepts string mode and sends empty servers.""" - mark_detected(client, application_version="2.4.1") - - mocked.post( - "http://localhost:8888/api/network/dns", - payload={"code": 0, "msg": "success", "data": None}, - ) - - await client.set_dns("dhcp") - - calls = mocked.requests[("POST", yarl.URL("http://localhost:8888/api/network/dns"))] - assert calls[0].kwargs.get("json") == { - "mode": "dhcp", - "servers": [], - } - - -async def test_get_dns_pro_is_not_supported( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test DNS management is limited to non-Pro hardware.""" - client._session._hw_version = HWVersion.PRO - - with pytest.raises(NanoKVMNotSupportedError) as exc_info: - await client.get_dns() - - assert "get_dns requires hardware: non-Pro" in str(exc_info.value) - assert not mocked.requests - - -async def test_get_dns_old_application_version_is_not_supported( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test old non-Pro application versions reject DNS before endpoint call.""" - client._session._hw_version = HWVersion.PCIE - - mocked.get( - "http://localhost:8888/api/vm/info", - payload=info_payload(application="2.4.0"), - ) - - with pytest.raises(NanoKVMNotSupportedError) as exc_info: - await client.get_dns() - - assert "get_dns requires non-Pro application version >= 2.4.1" in str( - exc_info.value - ) - dns_url = yarl.URL("http://localhost:8888/api/network/dns") - assert ("GET", dns_url) not in mocked.requests - - -async def test_get_dns_exact_application_version_is_supported( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test DNS is allowed at the first upstream version that introduced it.""" - client._session._hw_version = HWVersion.PCIE - - mocked.get( - "http://localhost:8888/api/vm/info", - payload=info_payload(application="2.4.1"), - ) - mocked.get( - "http://localhost:8888/api/network/dns", - payload={ - "code": 0, - "msg": "success", - "data": { - "mode": "dhcp", - "servers": [], - "effective": [], - "dhcp": [], - "info": {}, - }, - }, - ) - - response = await client.get_dns() - - assert response.mode is DNSMode.DHCP - - -async def test_non_pro_application_version_gate_uses_non_pro_minimum( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test non-Pro devices use the non-Pro minimum application version.""" - mark_detected(client, application_version="2.3.1") - - with pytest.raises(NanoKVMNotSupportedError) as exc_info: - await client.get_shortcuts() - - assert "get_shortcuts requires non-Pro application version >= 2.3.2" in str( - exc_info.value - ) - shortcuts_url = yarl.URL("http://localhost:8888/api/hid/shortcuts") - assert ("GET", shortcuts_url) not in mocked.requests - - -async def test_pro_application_version_gate_uses_pro_minimum( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test Pro devices use the Pro minimum application version.""" - mark_detected( - client, - hw_version=HWVersion.PRO, - application_version="1.2.7", - ) - - with pytest.raises(NanoKVMNotSupportedError) as exc_info: - await client.get_shortcuts() - - assert "get_shortcuts requires Pro application version >= 1.2.8" in str( - exc_info.value - ) - shortcuts_url = yarl.URL("http://localhost:8888/api/hid/shortcuts") - assert ("GET", shortcuts_url) not in mocked.requests - - -async def test_upload_script_uses_multipart_form( - tmp_path: Path, client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test script upload uses the shared multipart helper.""" - script = tmp_path / "test.sh" - script.write_text("#!/bin/sh\ntrue\n") - - mocked.post( - "http://localhost:8888/api/vm/script/upload", - payload={ - "code": 0, - "msg": "success", - "data": {"file": "test.sh"}, - }, - ) - - response = await client.upload_script(script) - - calls = mocked.requests[ - ("POST", yarl.URL("http://localhost:8888/api/vm/script/upload")) - ] - assert isinstance(calls[0].kwargs.get("data"), aiohttp.FormData) - assert response.file == "test.sh" - - -async def test_shortcut_methods_send_expected_payloads( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test custom shortcut client helpers use the expected endpoints.""" - mark_detected(client, application_version="2.3.4") - - mocked.post( - "http://localhost:8888/api/hid/shortcut", - payload={"code": 0, "msg": "success", "data": None}, - ) - mocked.delete( - "http://localhost:8888/api/hid/shortcut", - payload={"code": 0, "msg": "success", "data": None}, - ) - mocked.post( - "http://localhost:8888/api/hid/shortcut/leader-key", - payload={"code": 0, "msg": "success", "data": None}, - ) - - key = ShortcutKey(code="ControlLeft", label="Ctrl") - await client.add_shortcut([key]) - await client.delete_shortcut("shortcut-1") - await client.set_leader_key("ControlLeft") - - shortcut_calls = mocked.requests[ - ("POST", yarl.URL("http://localhost:8888/api/hid/shortcut")) - ] - delete_calls = mocked.requests[ - ("DELETE", yarl.URL("http://localhost:8888/api/hid/shortcut")) - ] - leader_calls = mocked.requests[ - ( - "POST", - yarl.URL("http://localhost:8888/api/hid/shortcut/leader-key"), - ) - ] - assert shortcut_calls[0].kwargs.get("json") == { - "keys": [{"code": "ControlLeft", "label": "Ctrl"}] - } - assert delete_calls[0].kwargs.get("json") == {"id": "shortcut-1"} - assert leader_calls[0].kwargs.get("json") == {"key": "ControlLeft"} - - -async def test_tailscale_login_returns_url( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test Tailscale login returns the upstream login URL payload.""" - mark_detected(client, application_version="2.1.6") - - mocked.post( - "http://localhost:8888/api/extensions/tailscale/login", - payload={ - "code": 0, - "msg": "success", - "data": {"url": "https://login.tailscale.com/a/test"}, - }, - ) - - response = await client.tailscale_login() - - assert response.url == "https://login.tailscale.com/a/test" - - -async def test_set_stream_mode_accepts_existing_string_values( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Test set_stream_mode remains source-compatible with valid strings.""" - client._session._hw_version = HWVersion.PRO - - mocked.post( - "http://localhost:8888/api/stream/mode", - payload={"code": 0, "msg": "success", "data": None}, - ) - - await client.set_stream_mode(StreamMode.H264_DIRECT.value) - - calls = mocked.requests[("POST", yarl.URL("http://localhost:8888/api/stream/mode"))] - assert calls[0].kwargs.get("json") == {"mode": "h264-direct"} - - 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: + async with NanoKVMClient(BASE_URL, token="test-token") as client: # Verify session is created assert client._session._http_session is not None assert not client._session._http_session.closed @@ -889,12 +117,12 @@ async def test_client_context_manager() -> None: async def test_client_context_manager_external_session() -> None: """Test that client properly deals with an external session.""" async with ClientSession() as session: - client3 = NanoKVMClient("http://localhost:8888/api/", session=session) + client3 = NanoKVMClient(BASE_URL, session=session) # All clients connect with the same external session async with ( - NanoKVMClient("http://localhost:8888/api/", session=session) as client1, - NanoKVMClient("http://localhost:8888/api/", session=session) as client2, + NanoKVMClient(BASE_URL, session=session) as client1, + NanoKVMClient(BASE_URL, session=session) as client2, client3, ): # Verify session is created @@ -913,30 +141,6 @@ async def test_client_context_manager_external_session() -> None: async def test_request_outside_context_manager_explains_how_to_start() -> None: """Using the client without 'async with' raises a clear error.""" - client = NanoKVMClient("http://localhost:8888/api/", token="synthetic-token") + client = NanoKVMClient(BASE_URL, token="synthetic-token") with pytest.raises(RuntimeError, match="async with NanoKVMClient"): await client.get_gpio() - - -async def test_mount_image_read_only_requires_pro_1_2_1( - client: NanoKVMClient, mocked: aioresponses -) -> None: - """Older Pro firmware ignores readOnly, so the client refuses it first.""" - mark_detected(client, HWVersion.PRO, "1.2.0") - with pytest.raises(NanoKVMNotSupportedError, match="1.2.1"): - await client.mount_image("/data/test.img", read_only=True) - assert not mocked.requests - - -async def test_mjpeg_stream_bytes_yields_raw_frames(mocked: aioresponses) -> None: - """The byte stream hands out each JPEG part untouched.""" - mocked.get( - "http://localhost:8888/api/stream/mjpeg", - headers={"Content-Type": "multipart/x-mixed-replace; boundary=frame"}, - body=b"--frame\r\nContent-Type: image/jpeg\r\n\r\n\xff\xd8one\r\n" - b"--frame\r\nContent-Type: image/jpeg\r\n\r\n\xff\xd8two\r\n--frame--\r\n", - ) - async with NanoKVMClient("http://localhost:8888/api/", token="test-token") as c: - frames = [frame async for frame in c.mjpeg_stream_bytes()] - - assert frames == [b"\xff\xd8one", b"\xff\xd8two"] diff --git a/tests/test_component_session.py b/tests/test_component_session.py index 675d37f..724b4da 100644 --- a/tests/test_component_session.py +++ b/tests/test_component_session.py @@ -13,10 +13,12 @@ from nanokvm.exceptions import NanoKVMNotAuthenticatedError, NanoKVMPermissionError from nanokvm.models import HWVersion +from .common import BASE_URL + async def test_components_share_permissions_and_external_session() -> None: """A forbidden storage read leaves HID/network usable until a 401 clears them.""" - url = "http://kvm.local/api/" + url = BASE_URL async with aiohttp.ClientSession() as http: session = SessionController( url, diff --git a/tests/test_current_password_auth.py b/tests/test_current_password_auth.py index 260e183..f30a1b5 100644 --- a/tests/test_current_password_auth.py +++ b/tests/test_current_password_auth.py @@ -9,11 +9,10 @@ from nanokvm.client import NanoKVMApiError, NanoKVMClient, NanoKVMError from nanokvm.models import HWVersion -from .common import mark_detected +from .common import BASE_URL, mark_detected -_BASE_URL = "http://kvm.local/api/" -_ACCOUNT_URL = f"{_BASE_URL}auth/account" -_PASSWORD_URL = f"{_BASE_URL}auth/password" +_ACCOUNT_URL = f"{BASE_URL}auth/account" +_PASSWORD_URL = f"{BASE_URL}auth/password" def _account_payload(username: str = "synthetic-user") -> dict[str, object]: @@ -34,7 +33,7 @@ def _prepare_client( application: str | None = "2.5.1", ) -> NanoKVMClient: client = NanoKVMClient( - _BASE_URL, + BASE_URL, token="session-token", use_password_obfuscation=True, ) @@ -215,11 +214,11 @@ async def test_password_change_detects_hardware_and_version_when_needed() -> Non ), ): mocked.get( - f"{_BASE_URL}vm/hardware", + f"{BASE_URL}vm/hardware", payload={"code": 0, "msg": "success", "data": {"version": "PCIE"}}, ) mocked.get( - f"{_BASE_URL}vm/info", + f"{BASE_URL}vm/info", payload={ "code": 0, "msg": "success", diff --git a/tests/test_endpoint_compatibility.py b/tests/test_endpoint_compatibility.py index 3471d40..e344f0a 100644 --- a/tests/test_endpoint_compatibility.py +++ b/tests/test_endpoint_compatibility.py @@ -5,8 +5,10 @@ from aioresponses import aioresponses import pytest +import yarl from nanokvm.client import NanoKVMClient, NanoKVMNotSupportedError +from nanokvm.compatibility import _parse_version, _version_at_least from nanokvm.models import HWVersion from .common import BASE_URL, mark_detected @@ -98,3 +100,49 @@ async def test_logout_ignores_missing_legacy_endpoint( await client.logout() assert client.token is None + + +def test_version_parser() -> None: + """Test internal application version parsing and comparison.""" + assert _parse_version("v2.4.1") == (2, 4, 1) + assert _parse_version("dev") is None + assert _version_at_least("1.2.14", "1.2.9") + assert _version_at_least("2.4", "2.4.0") + assert not _version_at_least("2.4.0", "2.4.1") + assert _version_at_least("dev", "2.4.1") + + +async def test_non_pro_application_version_gate_uses_non_pro_minimum( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test non-Pro devices use the non-Pro minimum application version.""" + mark_detected(client, application_version="2.3.1") + + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_shortcuts() + + assert "get_shortcuts requires non-Pro application version >= 2.3.2" in str( + exc_info.value + ) + shortcuts_url = yarl.URL(f"{BASE_URL}hid/shortcuts") + assert ("GET", shortcuts_url) not in mocked.requests + + +async def test_pro_application_version_gate_uses_pro_minimum( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test Pro devices use the Pro minimum application version.""" + mark_detected( + client, + hw_version=HWVersion.PRO, + application_version="1.2.7", + ) + + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_shortcuts() + + assert "get_shortcuts requires Pro application version >= 1.2.8" in str( + exc_info.value + ) + shortcuts_url = yarl.URL(f"{BASE_URL}hid/shortcuts") + assert ("GET", shortcuts_url) not in mocked.requests diff --git a/tests/test_hardware_family.py b/tests/test_hardware_family.py index e5aa49c..a589001 100644 --- a/tests/test_hardware_family.py +++ b/tests/test_hardware_family.py @@ -14,6 +14,8 @@ ) from nanokvm.models import HWFamily, HWVersion +from .common import BASE_URL + @pytest.mark.parametrize( ("version", "family"), @@ -39,7 +41,7 @@ async def test_hardware_decorator_accepts_each_non_pro_version() -> None: async def operation(client: NanoKVMClient) -> str: return "ok" - client = NanoKVMClient("http://kvm.local/api/") + client = NanoKVMClient(BASE_URL) for version in (HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE): client._session._hw_version = version assert await operation(client) == "ok" @@ -52,7 +54,7 @@ async def test_hardware_decorator_preserves_exact_version_requirements() -> None async def operation(client: NanoKVMClient) -> str: return "ok" - client = NanoKVMClient("http://kvm.local/api/") + client = NanoKVMClient(BASE_URL) client._session._hw_version = HWVersion.PRO assert await operation(client) == "ok" client._session._hw_version = HWVersion.PCIE @@ -64,12 +66,12 @@ async def test_pro_family_requirement_accepts_future_pro_hardware( mocked: aioresponses, ) -> None: """Pro-only operations accept future hardware versions in the Pro family.""" - async with NanoKVMClient("http://kvm.local/api/", token="test-token") as client: + async with NanoKVMClient(BASE_URL, token="test-token") as client: client._session._hw_version = cast( HWVersion, SimpleNamespace(family=HWFamily.PRO, value="Pro Desk PoE") ) mocked.get( - "http://kvm.local/api/vm/hdmi/capture", + f"{BASE_URL}vm/hdmi/capture", payload={"code": 0, "msg": "success", "data": {"enabled": True}}, ) @@ -86,7 +88,7 @@ async def test_hardware_decorator_family_requirement_requires_detection() -> Non async def operation(client: NanoKVMClient) -> str: return "ok" - client = NanoKVMClient("http://kvm.local/api/") + client = NanoKVMClient(BASE_URL) with pytest.raises(NanoKVMError, match="requires hardware detection"): await operation(client) @@ -101,7 +103,7 @@ async def test_hardware_decorator_family_requirement_rejects_other_versions( async def operation(client: NanoKVMClient) -> str: return "ok" - client = NanoKVMClient("http://kvm.local/api/") + client = NanoKVMClient(BASE_URL) client._session._hw_version = version with pytest.raises(NanoKVMNotSupportedError, match="hardware: non-Pro"): await operation(client) @@ -116,7 +118,7 @@ async def test_hardware_decorator_detects_hardware_with_a_token( async def operation(client: NanoKVMClient) -> str: return "ok" - url = "http://kvm.local/api/" + url = BASE_URL async with NanoKVMClient(url, token="synthetic-token") as client: mocked.get( f"{url}vm/hardware", diff --git a/tests/test_hid.py b/tests/test_hid.py new file mode 100644 index 0000000..eb5685f --- /dev/null +++ b/tests/test_hid.py @@ -0,0 +1,72 @@ +"""Tests for HID operations.""" + +from aioresponses import aioresponses +import yarl + +from nanokvm.client import NanoKVMClient +from nanokvm.models import MouseJigglerMode, ShortcutKey + +from .common import BASE_URL, mark_detected + + +async def test_set_mouse_jiggler_state_noops_when_already_disabled( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test disabling mouse jiggler is idempotent.""" + mark_detected(client) + + mocked.get( + f"{BASE_URL}vm/mouse-jiggler", + payload={ + "code": 0, + "msg": "success", + "data": {"enabled": False, "mode": "relative"}, + }, + ) + + await client.set_mouse_jiggler_state( + enabled=False, + mode=MouseJigglerMode.RELATIVE, + ) + + post_url = yarl.URL(f"{BASE_URL}vm/mouse-jiggler/") + assert ("POST", post_url) not in mocked.requests + + +async def test_shortcut_methods_send_expected_payloads( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test custom shortcut client helpers use the expected endpoints.""" + mark_detected(client, application_version="2.3.4") + + mocked.post( + f"{BASE_URL}hid/shortcut", + payload={"code": 0, "msg": "success", "data": None}, + ) + mocked.delete( + f"{BASE_URL}hid/shortcut", + payload={"code": 0, "msg": "success", "data": None}, + ) + mocked.post( + f"{BASE_URL}hid/shortcut/leader-key", + payload={"code": 0, "msg": "success", "data": None}, + ) + + key = ShortcutKey(code="ControlLeft", label="Ctrl") + await client.add_shortcut([key]) + await client.delete_shortcut("shortcut-1") + await client.set_leader_key("ControlLeft") + + shortcut_calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}hid/shortcut"))] + delete_calls = mocked.requests[("DELETE", yarl.URL(f"{BASE_URL}hid/shortcut"))] + leader_calls = mocked.requests[ + ( + "POST", + yarl.URL(f"{BASE_URL}hid/shortcut/leader-key"), + ) + ] + assert shortcut_calls[0].kwargs.get("json") == { + "keys": [{"code": "ControlLeft", "label": "Ctrl"}] + } + assert delete_calls[0].kwargs.get("json") == {"id": "shortcut-1"} + assert leader_calls[0].kwargs.get("json") == {"key": "ControlLeft"} diff --git a/tests/test_image_transfers.py b/tests/test_image_transfers.py index b42d2b3..be470f4 100644 --- a/tests/test_image_transfers.py +++ b/tests/test_image_transfers.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager import io from pathlib import Path @@ -34,7 +34,7 @@ def _status(status: DownloadStatus) -> StatusImageRsp: @asynccontextmanager -async def _device_server() -> AsyncIterator[tuple[str, dict[str, Any]]]: +async def _device_server() -> AsyncGenerator[tuple[str, dict[str, Any]], None]: state: dict[str, Any] = { "calls": [], "accepted_uploads": [], diff --git a/tests/test_mouse.py b/tests/test_mouse.py index 73f8015..a17a650 100644 --- a/tests/test_mouse.py +++ b/tests/test_mouse.py @@ -15,7 +15,7 @@ ) from nanokvm.models import HWVersion, MouseButton -from .common import mark_detected +from .common import BASE_URL, mark_detected @pytest.fixture @@ -27,9 +27,7 @@ async def client_with_mock_ws() -> AsyncGenerator[tuple[NanoKVMClient, AsyncMock 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: + async with NanoKVMClient(BASE_URL, token="test-token") as client: mark_detected(client, HWVersion.PCIE, "2.3.2") yield client, mock_ws @@ -49,9 +47,7 @@ async def legacy_client_with_mock_ws() -> AsyncGenerator[ 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: + async with NanoKVMClient(BASE_URL, token="test-token") as client: mark_detected(client, HWVersion.PCIE, "2.3.1") yield client, mock_ws @@ -71,7 +67,7 @@ def test_current_mouse_report_uses_selected_mode( mode: Literal["absolute", "relative"], wheel: int, expected: bytes ) -> None: """Current button and wheel reports follow the selected mouse mode.""" - client = NanoKVMClient("http://localhost:8888/api/", token="test-token") + client = NanoKVMClient(BASE_URL, token="test-token") client._session._mouse_mode = mode client._session._mouse_buttons = 3 client._session._mouse_abs_position = (0x1234, 0x5678) @@ -284,9 +280,7 @@ async def connect(*_args: Any, **_kwargs: Any) -> AsyncMock: return mock_ws with patch("aiohttp.ClientSession.ws_connect", new=connect): - async with NanoKVMClient( - "http://localhost:8888/api/", token="test-token" - ) as client: + async with NanoKVMClient(BASE_URL, token="test-token") as client: mark_detected(client, HWVersion.PCIE, "2.3.2") first = asyncio.create_task(client.mouse_move_rel(0.1, 0.0)) await connect_started.wait() @@ -308,9 +302,7 @@ async def test_mouse_send_reconnects_after_transport_failure() -> None: connect = AsyncMock(side_effect=[first_ws, second_ws]) with patch("aiohttp.ClientSession.ws_connect", new=connect): - async with NanoKVMClient( - "http://localhost:8888/api/", token="test-token" - ) as client: + async with NanoKVMClient(BASE_URL, token="test-token") as client: mark_detected(client, HWVersion.PCIE, "2.3.2") with pytest.raises(ConnectionResetError): await client.mouse_move_rel(0.1, 0.0) @@ -330,9 +322,7 @@ async def test_legacy_mouse_send_reconnects_after_transport_failure() -> None: connect = AsyncMock(side_effect=[first_ws, second_ws]) with patch("aiohttp.ClientSession.ws_connect", new=connect): - async with NanoKVMClient( - "http://localhost:8888/api/", token="test-token" - ) as client: + async with NanoKVMClient(BASE_URL, token="test-token") as client: mark_detected(client, HWVersion.PCIE, "2.3.1") with pytest.raises(ConnectionResetError): await client.mouse_move_rel(0.1, 0.0) @@ -347,9 +337,7 @@ async def test_mouse_send_helper_invalidates_failed_websocket() -> None: ws = AsyncMock() ws.closed = False - async with NanoKVMClient( - "http://localhost:8888/api/", token="test-token" - ) as client: + async with NanoKVMClient(BASE_URL, token="test-token") as client: client._session._ws = ws async def send(current_ws: Any) -> None: @@ -415,9 +403,7 @@ async def test_context_manager_cleanup() -> None: 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: + async with NanoKVMClient(BASE_URL, token="test-token") as client: mark_detected(client, HWVersion.PCIE, "2.3.2") await client.mouse_move_abs(0.0, 0.0) assert client._session._ws is not None @@ -455,7 +441,7 @@ async def test_failed_press_does_not_press_on_next_movement(failure: str) -> Non from nanokvm.client import NanoKVMPermissionError info = aiohttp.RequestInfo( - yarl.URL("http://kvm.local/api/ws"), "GET", CIMultiDictProxy(CIMultiDict()) + yarl.URL(f"{BASE_URL}ws"), "GET", CIMultiDictProxy(CIMultiDict()) ) error: BaseException expected: type[BaseException] @@ -469,7 +455,7 @@ async def test_failed_press_does_not_press_on_next_movement(failure: str) -> Non error = aiohttp.ClientConnectionError("synthetic failure") expected = aiohttp.ClientConnectionError ws = AsyncMock(closed=False) - async with NanoKVMClient("http://kvm.local/api/", token="reader") as client: + async with NanoKVMClient(BASE_URL, token="reader") as client: mark_detected(client, HWVersion.PCIE, "2.5.1") with patch( "aiohttp.ClientSession.ws_connect", AsyncMock(side_effect=[error, ws]) diff --git a/tests/test_network.py b/tests/test_network.py new file mode 100644 index 0000000..02b65e4 --- /dev/null +++ b/tests/test_network.py @@ -0,0 +1,196 @@ +"""Tests for Wake-on-LAN, Wi-Fi and DNS operations.""" + +from aioresponses import aioresponses +import pytest +import yarl + +from nanokvm.client import NanoKVMClient, NanoKVMNotSupportedError +from nanokvm.models import DNSMode, HWVersion + +from .common import BASE_URL, info_payload, mark_detected + + +async def test_get_wol_macs_returns_empty_list_for_missing_file( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test WOL getter normalizes the empty-file device state.""" + mocked.get( + f"{BASE_URL}network/wol/mac", + payload={"code": -2, "msg": "open file error", "data": None}, + ) + + response = await client.get_wol_macs() + + assert response.macs == [] + + +async def test_connect_wifi_no_auth_sends_ap_header( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test AP-mode Wi-Fi connection sends setup credentials without auth cookie.""" + mocked.post( + f"{BASE_URL}network/wifi", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.connect_wifi_no_auth( + "NanoKVM-Setup", + ap_password="setup-secret", + ) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}network/wifi"))] + assert calls[0].kwargs.get("json") == { + "ssid": "NanoKVM-Setup", + "password": "", + } + assert calls[0].kwargs.get("cookies") == {} + assert calls[0].kwargs.get("headers", {})["X-AP-Key"] == "setup-secret" + + +async def test_get_dns_parses_configuration( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test DNS configuration response parsing.""" + mark_detected(client, application_version="2.4.1") + + mocked.get( + f"{BASE_URL}network/dns", + payload={ + "code": 0, + "msg": "success", + "data": { + "mode": "manual", + "servers": ["1.1.1.1", "2606:4700:4700::1111"], + "effective": ["1.1.1.1"], + "dhcp": ["192.168.1.1"], + "info": { + "interface": "eth0", + "type": "Wired", + "address": "192.168.1.20/24", + "subnetMask": "255.255.255.0", + "gateway": "192.168.1.1", + "searchDomains": ["lan"], + }, + }, + }, + ) + + response = await client.get_dns() + + assert response.mode is DNSMode.MANUAL + assert response.servers == ["1.1.1.1", "2606:4700:4700::1111"] + assert response.effective == ["1.1.1.1"] + assert response.dhcp == ["192.168.1.1"] + assert response.info.interface == "eth0" + assert response.info.type == "Wired" + assert response.info.address == "192.168.1.20/24" + assert response.info.subnet_mask == "255.255.255.0" + assert response.info.gateway == "192.168.1.1" + assert response.info.search_domains == ["lan"] + + +async def test_set_dns_manual_sends_servers( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test manual DNS mode sends the configured servers.""" + mark_detected(client, application_version="2.4.1") + + mocked.post( + f"{BASE_URL}network/dns", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.set_dns( + DNSMode.MANUAL, + ["1.1.1.1", "2606:4700:4700::1111"], + ) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}network/dns"))] + assert calls[0].kwargs.get("json") == { + "mode": "manual", + "servers": ["1.1.1.1", "2606:4700:4700::1111"], + } + + +async def test_set_dns_dhcp_string_sends_empty_servers( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test DHCP DNS mode accepts string mode and sends empty servers.""" + mark_detected(client, application_version="2.4.1") + + mocked.post( + f"{BASE_URL}network/dns", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.set_dns("dhcp") + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}network/dns"))] + assert calls[0].kwargs.get("json") == { + "mode": "dhcp", + "servers": [], + } + + +async def test_get_dns_pro_is_not_supported( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test DNS management is limited to non-Pro hardware.""" + client._session._hw_version = HWVersion.PRO + + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_dns() + + assert "get_dns requires hardware: non-Pro" in str(exc_info.value) + assert not mocked.requests + + +async def test_get_dns_old_application_version_is_not_supported( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test old non-Pro application versions reject DNS before endpoint call.""" + client._session._hw_version = HWVersion.PCIE + + mocked.get( + f"{BASE_URL}vm/info", + payload=info_payload(application="2.4.0"), + ) + + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_dns() + + assert "get_dns requires non-Pro application version >= 2.4.1" in str( + exc_info.value + ) + dns_url = yarl.URL(f"{BASE_URL}network/dns") + assert ("GET", dns_url) not in mocked.requests + + +async def test_get_dns_exact_application_version_is_supported( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test DNS is allowed at the first upstream version that introduced it.""" + client._session._hw_version = HWVersion.PCIE + + mocked.get( + f"{BASE_URL}vm/info", + payload=info_payload(application="2.4.1"), + ) + mocked.get( + f"{BASE_URL}network/dns", + payload={ + "code": 0, + "msg": "success", + "data": { + "mode": "dhcp", + "servers": [], + "effective": [], + "dhcp": [], + "info": {}, + }, + }, + ) + + response = await client.get_dns() + + assert response.mode is DNSMode.DHCP diff --git a/tests/test_response_redaction.py b/tests/test_response_redaction.py index dd8cf8f..2d968ee 100644 --- a/tests/test_response_redaction.py +++ b/tests/test_response_redaction.py @@ -2,6 +2,7 @@ from collections.abc import Awaitable import logging +from pathlib import Path import traceback import aiohttp @@ -15,7 +16,8 @@ NanoKVMInvalidResponseError, ) -_URL = "http://kvm.local/api/" +from .common import BASE_URL + _SECRET = "synthetic-secret-value" @@ -27,22 +29,22 @@ async def test_invalid_response_traceback_and_exception_log_hide_payload( ) -> None: """logger.exception renders causes as well as the top-level error message.""" caplog.set_level(logging.DEBUG) - async with NanoKVMClient(_URL, token="synthetic-session") as client: + async with NanoKVMClient(BASE_URL, token="synthetic-session") as client: if source == "login-token": mocked.post( - f"{_URL}auth/login", + f"{BASE_URL}auth/login", payload={"code": 0, "msg": "ok", "data": {"token": [_SECRET]}}, ) elif source == "envelope": mocked.get( - f"{_URL}auth/account", + f"{BASE_URL}auth/account", payload={"code": [_SECRET], "msg": "ok", "data": None}, ) elif source == "form": - mocked.post(f"{_URL}upload", payload={"code": [_SECRET]}) + mocked.post(f"{BASE_URL}upload", payload={"code": [_SECRET]}) else: mocked.get( - f"{_URL}auth/account", + f"{BASE_URL}auth/account", payload={"code": 0, "msg": "ok", "data": {"username": [_SECRET]}}, ) try: @@ -69,21 +71,21 @@ async def test_invalid_utf8_response_is_rejected_consistently( ) -> None: """All JSON response paths translate invalid UTF-8 into the client error.""" async with NanoKVMClient( - _URL, + BASE_URL, token="synthetic-session", use_password_obfuscation=True, ) as client: operation: Awaitable[object] if transport == "json": - mocked.get(f"{_URL}auth/account", body=b"\xff") + mocked.get(f"{BASE_URL}auth/account", body=b"\xff") operation = client.get_account() elif transport == "form": - mocked.post(f"{_URL}upload", body=b"\xff") + mocked.post(f"{BASE_URL}upload", body=b"\xff") operation = client._session.api_request_json( "POST", "/upload", data=aiohttp.FormData() ) else: - mocked.post(f"{_URL}auth/login", body=b"\xff") + mocked.post(f"{BASE_URL}auth/login", body=b"\xff") operation = client.authenticate("synthetic-user", "synthetic-password") with pytest.raises(NanoKVMInvalidResponseError, match="Invalid JSON response"): @@ -98,9 +100,9 @@ async def test_server_messages_are_not_rendered_in_logs_or_errors( ) -> None: """Firmware text is untrusted and can echo a credential or token.""" caplog.set_level(logging.DEBUG) - async with NanoKVMClient(_URL, use_password_obfuscation=True) as client: + async with NanoKVMClient(BASE_URL, use_password_obfuscation=True) as client: mocked.post( - f"{_URL}auth/login", + f"{BASE_URL}auth/login", payload={ "code": code, "msg": _SECRET, @@ -108,7 +110,7 @@ async def test_server_messages_are_not_rendered_in_logs_or_errors( }, ) mocked.get( - f"{_URL}vm/hardware", + f"{BASE_URL}vm/hardware", payload={"code": 0, "msg": "ok", "data": {"version": "PCIE"}}, ) try: @@ -120,3 +122,42 @@ async def test_server_messages_are_not_rendered_in_logs_or_errors( assert error.code == code assert error.msg == _SECRET assert _SECRET not in caplog.text + + +def test_invalid_response_error_does_not_echo_payload() -> None: + """Malformed response errors do not include arbitrary response values.""" + client = NanoKVMClient(BASE_URL) + secret = "SYNTHETIC_SECRET_VALUE" + + with pytest.raises(NanoKVMInvalidResponseError) as exc_info: + client._session.validate_api_response( + {"msg": "invalid", "data": {"token": secret}}, + ) + + assert secret not in str(exc_info.value) + + +async def test_form_response_is_not_logged( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + mocked: aioresponses, +) -> None: + """Multipart responses are not emitted verbatim to debug logs.""" + script = tmp_path / "test.sh" + script.write_text("#!/bin/sh\ntrue\n") + secret = "SYNTHETIC_FORM_VALUE" + caplog.set_level(logging.DEBUG, logger="nanokvm") + + async with NanoKVMClient(BASE_URL, token="test-token") as client: + mocked.post( + f"{BASE_URL}vm/script/upload", + payload={ + "code": 0, + "msg": "success", + "data": {"file": secret}, + }, + ) + + await client.upload_script(script) + + assert secret not in caplog.text diff --git a/tests/test_services.py b/tests/test_services.py new file mode 100644 index 0000000..7a0623a --- /dev/null +++ b/tests/test_services.py @@ -0,0 +1,27 @@ +"""Tests for device services.""" + +from aioresponses import aioresponses + +from nanokvm.client import NanoKVMClient + +from .common import BASE_URL, mark_detected + + +async def test_tailscale_login_returns_url( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test Tailscale login returns the upstream login URL payload.""" + mark_detected(client, application_version="2.1.6") + + mocked.post( + f"{BASE_URL}extensions/tailscale/login", + payload={ + "code": 0, + "msg": "success", + "data": {"url": "https://login.tailscale.com/a/test"}, + }, + ) + + response = await client.tailscale_login() + + assert response.url == "https://login.tailscale.com/a/test" diff --git a/tests/test_session_cookie_auth.py b/tests/test_session_cookie_auth.py index b471a0b..950035c 100644 --- a/tests/test_session_cookie_auth.py +++ b/tests/test_session_cookie_auth.py @@ -19,12 +19,11 @@ ) from nanokvm.models import HWVersion -from .common import info_payload, mark_detected +from .common import BASE_URL, info_payload, mark_detected -_BASE_URL = "http://kvm.local/api/" -_LOGIN_URL = f"{_BASE_URL}auth/login" -_HARDWARE_URL = f"{_BASE_URL}vm/hardware" -_INFO_URL = f"{_BASE_URL}vm/info" +_LOGIN_URL = f"{BASE_URL}auth/login" +_HARDWARE_URL = f"{BASE_URL}vm/hardware" +_INFO_URL = f"{BASE_URL}vm/info" _HARDWARE_PAYLOAD = { "code": 0, @@ -41,7 +40,7 @@ async def test_cookie_only_login_accepts_null_data_and_uses_fresh_cookie( mocked: aioresponses, ) -> None: """The 2.5.1 login envelope has no data and returns the token as a cookie.""" - async with NanoKVMClient(_BASE_URL) as client: + async with NanoKVMClient(BASE_URL) as client: mocked.post( _LOGIN_URL, payload=_login_payload(), @@ -60,7 +59,7 @@ async def test_cookie_token_has_priority_over_legacy_body_token( mocked: aioresponses, ) -> None: """A fresh Set-Cookie value wins when old firmware returns both tokens.""" - async with NanoKVMClient(_BASE_URL) as client: + async with NanoKVMClient(BASE_URL) as client: mocked.post( _LOGIN_URL, payload=_login_payload({"token": "body-token"}), @@ -78,7 +77,7 @@ async def test_cookie_login_ignores_unused_legacy_body( data: object, mocked: aioresponses ) -> None: """A valid fresh cookie must not depend on the obsolete body token schema.""" - async with NanoKVMClient(_BASE_URL) as client: + async with NanoKVMClient(BASE_URL) as client: mocked.post( _LOGIN_URL, payload=_login_payload(data), @@ -93,7 +92,7 @@ async def test_empty_cookie_falls_back_to_legacy_body_token( mocked: aioresponses, ) -> None: """An empty Set-Cookie does not hide a valid legacy token.""" - async with NanoKVMClient(_BASE_URL) as client: + async with NanoKVMClient(BASE_URL) as client: mocked.post( _LOGIN_URL, payload=_login_payload({"token": "body-token"}), @@ -106,7 +105,7 @@ async def test_empty_cookie_falls_back_to_legacy_body_token( async def test_cookie_does_not_bypass_envelope_validation(mocked: aioresponses) -> None: """Cookie precedence only applies after the API envelope is validated.""" - async with NanoKVMClient(_BASE_URL) as client: + async with NanoKVMClient(BASE_URL) as client: mocked.post( _LOGIN_URL, payload={"data": {"token": "body-token"}}, @@ -119,7 +118,7 @@ async def test_cookie_does_not_bypass_envelope_validation(mocked: aioresponses) async def test_legacy_body_token_login_remains_supported(mocked: aioresponses) -> None: """Older firmware continues to authenticate with a token in response data.""" - async with NanoKVMClient(_BASE_URL) as client: + async with NanoKVMClient(BASE_URL) as client: mocked.post( _LOGIN_URL, payload=_login_payload({"token": "body-token"}), @@ -135,7 +134,7 @@ async def test_cookie_login_validates_api_code_before_accepting_cookie( mocked: aioresponses, ) -> None: """A cookie on an API error response cannot establish a session.""" - async with NanoKVMClient(_BASE_URL, use_password_obfuscation=True) as client: + async with NanoKVMClient(BASE_URL, use_password_obfuscation=True) as client: mocked.post( _LOGIN_URL, payload={ @@ -156,7 +155,7 @@ async def test_cookie_login_works_with_dummy_cookie_jar() -> None: """Cookie extraction does not depend on the session cookie jar.""" session = aiohttp.ClientSession(cookie_jar=aiohttp.DummyCookieJar()) try: - async with NanoKVMClient(_BASE_URL, session=session) as client: + async with NanoKVMClient(BASE_URL, session=session) as client: with aioresponses() as mocked: mocked.post( _LOGIN_URL, @@ -179,12 +178,12 @@ async def test_login_does_not_reuse_an_old_cookie_from_external_session() -> Non cookie_jar = aiohttp.CookieJar() cookie_jar.update_cookies( {"nano-kvm-token": "old-cookie-token"}, - response_url=yarl.URL(_BASE_URL), + response_url=yarl.URL(BASE_URL), ) session = aiohttp.ClientSession(cookie_jar=cookie_jar) try: - async with NanoKVMClient(_BASE_URL, session=session) as client: + async with NanoKVMClient(BASE_URL, session=session) as client: with aioresponses() as mocked: mocked.post(_LOGIN_URL, payload=_login_payload()) @@ -200,7 +199,7 @@ async def test_empty_login_response_is_not_authentication_disabled( mocked: aioresponses, ) -> None: """An empty successful login is invalid even for the explicit disabled token.""" - async with NanoKVMClient(_BASE_URL, token="disabled") as client: + async with NanoKVMClient(BASE_URL, token="disabled") as client: mocked.post(_LOGIN_URL, payload=_login_payload()) with pytest.raises(NanoKVMInvalidResponseError, match="missing token"): @@ -218,7 +217,7 @@ async def test_failed_reauthentication_cannot_keep_previous_identity( old_ws = AsyncMock() old_ws.closed = False async with NanoKVMClient( - _BASE_URL, token="previous-admin", use_password_obfuscation=True + BASE_URL, token="previous-admin", use_password_obfuscation=True ) as client: client._session._ws = old_ws client._session._mouse_buttons = 1 @@ -254,7 +253,7 @@ async def test_reauthentication_clears_transport_even_when_token_is_reissued( """Transport state belongs to an authentication attempt, not a token string.""" old_ws = AsyncMock() old_ws.closed = False - async with NanoKVMClient(_BASE_URL, token="same-token") as client: + async with NanoKVMClient(BASE_URL, token="same-token") as client: client._session._ws = old_ws client._session._mouse_buttons = 1 mocked.post(_LOGIN_URL, payload=_login_payload({"token": "same-token"})) @@ -270,7 +269,7 @@ async def test_authentication_uses_received_token_for_http_requests( mocked: aioresponses, ) -> None: """The token obtained from a cookie is sent explicitly on later requests.""" - async with NanoKVMClient(_BASE_URL) as client: + async with NanoKVMClient(BASE_URL) as client: mocked.post( _LOGIN_URL, payload=_login_payload(), @@ -300,7 +299,7 @@ async def test_authentication_closes_previous_websocket_before_replacing_identit new_callable=AsyncMock, side_effect=[old_ws, new_ws], ): - async with NanoKVMClient(_BASE_URL, token="old-token") as client: + async with NanoKVMClient(BASE_URL, token="old-token") as client: mark_detected(client, HWVersion.PCIE, "2.5.1") await client.mouse_move_rel(0.1, 0.0) diff --git a/tests/test_session_generation.py b/tests/test_session_generation.py index 9cfb495..a8d7441 100644 --- a/tests/test_session_generation.py +++ b/tests/test_session_generation.py @@ -13,9 +13,7 @@ from nanokvm.client import NanoKVMClient, NanoKVMNotAuthenticatedError from nanokvm.models import HWVersion -from .common import mark_detected - -_URL = "http://kvm.local/api/" +from .common import BASE_URL, mark_detected @pytest.mark.parametrize("transport", ["json", "form", "mjpeg"]) @@ -34,7 +32,7 @@ async def delayed_401(*args: object, **kwargs: object) -> CallbackResult: await release.wait() return CallbackResult(status=401, body='"unauthorized"') - async with NanoKVMClient(_URL, token="old-session") as client: + async with NanoKVMClient(BASE_URL, token="old-session") as client: async def request() -> None: if transport == "json": @@ -48,17 +46,17 @@ async def request() -> None: pass if transport == "json": - mocked.get(f"{_URL}auth/account", callback=delayed_401) + mocked.get(f"{BASE_URL}auth/account", callback=delayed_401) elif transport == "form": - mocked.post(f"{_URL}upload", callback=delayed_401) + mocked.post(f"{BASE_URL}upload", callback=delayed_401) else: - mocked.get(f"{_URL}stream/mjpeg", callback=delayed_401) + mocked.get(f"{BASE_URL}stream/mjpeg", callback=delayed_401) mocked.post( - f"{_URL}auth/login", + f"{BASE_URL}auth/login", payload={"code": 0, "msg": "ok", "data": {"token": new_token}}, ) mocked.get( - f"{_URL}vm/hardware", + f"{BASE_URL}vm/hardware", payload={"code": 0, "msg": "ok", "data": {"version": "PCIE"}}, ) async with asyncio.timeout(2): @@ -84,14 +82,14 @@ async def delayed_success(*args: object, **kwargs: object) -> CallbackResult: await release.wait() return CallbackResult(payload={"code": 0, "msg": "ok", "data": None}) - async with NanoKVMClient(_URL, token="old-session") as client: - mocked.post(f"{_URL}upload", callback=delayed_success) + async with NanoKVMClient(BASE_URL, token="old-session") as client: + mocked.post(f"{BASE_URL}upload", callback=delayed_success) mocked.post( - f"{_URL}auth/login", + f"{BASE_URL}auth/login", payload={"code": 0, "msg": "ok", "data": {"token": "new-session"}}, ) mocked.get( - f"{_URL}vm/hardware", + f"{BASE_URL}vm/hardware", payload={"code": 0, "msg": "ok", "data": {"version": "PCIE"}}, ) pending = asyncio.create_task( @@ -114,12 +112,12 @@ async def test_websocket_401_waiting_for_cleanup_cannot_clear_new_login() -> Non release = asyncio.Event() error = aiohttp.WSServerHandshakeError( aiohttp.RequestInfo( - yarl.URL(_URL + "ws"), "GET", CIMultiDictProxy(CIMultiDict()) + yarl.URL(BASE_URL + "ws"), "GET", CIMultiDictProxy(CIMultiDict()) ), (), status=401, ) - async with NanoKVMClient(_URL, token="old-session") as client: + async with NanoKVMClient(BASE_URL, token="old-session") as client: original_clear = client._session.clear_local_session async def delayed_clear(*args: Any, **kwargs: Any) -> int: @@ -138,11 +136,11 @@ async def delayed_clear(*args: Any, **kwargs: Any) -> int: aioresponses() as mocked, ): mocked.post( - f"{_URL}auth/login", + f"{BASE_URL}auth/login", payload={"code": 0, "msg": "ok", "data": {"token": "new-session"}}, ) mocked.get( - f"{_URL}vm/hardware", + f"{BASE_URL}vm/hardware", payload={"code": 0, "msg": "ok", "data": {"version": "PCIE"}}, ) pending = asyncio.create_task(client._session.get_ws()) @@ -168,7 +166,7 @@ async def close_old() -> None: closing.set() await finish_close.wait() - async with NanoKVMClient(_URL, token="old-session") as client: + async with NanoKVMClient(BASE_URL, token="old-session") as client: old_ws = AsyncMock(closed=False) old_ws.close.side_effect = close_old client._session._ws = old_ws @@ -178,7 +176,7 @@ async def close_old() -> None: try: with aioresponses() as mocked: mocked.post( - f"{_URL}auth/login", + f"{BASE_URL}auth/login", payload={ "code": 0, "msg": "ok", @@ -186,7 +184,7 @@ async def close_old() -> None: }, ) mocked.get( - f"{_URL}vm/hardware", + f"{BASE_URL}vm/hardware", payload={"code": 0, "msg": "ok", "data": {"version": "PCIE"}}, ) await client.authenticate("synthetic-user", "synthetic-password") @@ -216,23 +214,23 @@ async def delayed_success(*args: object, **kwargs: object) -> CallbackResult: await release.wait() return CallbackResult(payload={"code": 0, "msg": "ok", "data": None}) - async with NanoKVMClient(_URL, token="old-session") as client: + async with NanoKVMClient(BASE_URL, token="old-session") as client: mark_detected(client, HWVersion.PCIE, "2.5.1") mocked.get( - f"{_URL}auth/account", + f"{BASE_URL}auth/account", payload={ "code": 0, "msg": "ok", "data": {"username": "synthetic-user"}, }, ) - mocked.post(f"{_URL}auth/{operation}", callback=delayed_success) + mocked.post(f"{BASE_URL}auth/{operation}", callback=delayed_success) mocked.post( - f"{_URL}auth/login", + f"{BASE_URL}auth/login", payload={"code": 0, "msg": "ok", "data": {"token": "new-session"}}, ) mocked.get( - f"{_URL}vm/hardware", + f"{BASE_URL}vm/hardware", payload={"code": 0, "msg": "ok", "data": {"version": "PCIE"}}, ) pending = asyncio.create_task( @@ -255,10 +253,10 @@ async def delayed_success(*args: object, **kwargs: object) -> CallbackResult: async def test_logout_forgets_cached_firmware_versions(mocked: aioresponses) -> None: """A later login re-detects versions, so a firmware update is noticed.""" - async with NanoKVMClient(_URL, token="synthetic-token") as client: + async with NanoKVMClient(BASE_URL, token="synthetic-token") as client: mark_detected(client, HWVersion.PCIE, "2.3.0") client._session._image_version = "1.4.0" - mocked.post(f"{_URL}auth/logout", payload={"code": 0, "msg": "ok"}) + mocked.post(f"{BASE_URL}auth/logout", payload={"code": 0, "msg": "ok"}) await client.logout() assert client.application_version is None diff --git a/tests/test_ssl.py b/tests/test_ssl.py index 5356d2f..8adb266 100644 --- a/tests/test_ssl.py +++ b/tests/test_ssl.py @@ -11,17 +11,20 @@ from nanokvm.client import NanoKVMAuthenticationFailure, NanoKVMClient +from .common import BASE_URL + _HARDWARE_PAYLOAD = { "code": 0, "msg": "success", "data": {"version": "Alpha"}, } -_HARDWARE_URL = "https://kvm.local/api/vm/hardware" +_BASE_URL = "https://kvm.local/api/" +_HARDWARE_URL = f"{_BASE_URL}vm/hardware" async def test_default_ssl_verification_enabled() -> None: """Test that SSL verification is enabled by default.""" - client = NanoKVMClient("https://kvm.local/api/") + client = NanoKVMClient(_BASE_URL) ssl_config = client._session.create_ssl_context() assert ssl_config is True @@ -29,7 +32,7 @@ async def test_default_ssl_verification_enabled() -> None: async def test_ssl_verification_disabled() -> None: """Test SSL verification can be disabled.""" - client = NanoKVMClient("https://kvm.local/api/", verify_ssl=False) + client = NanoKVMClient(_BASE_URL, verify_ssl=False) ssl_config = client._session.create_ssl_context() assert ssl_config is False @@ -44,7 +47,7 @@ async def test_custom_ca_certificate(tmp_path: Path) -> None: mock_ctx = MagicMock(spec=ssl.SSLContext) mock_ssl_context.return_value = mock_ctx - client = NanoKVMClient("https://kvm.local/api/", ssl_ca_cert=str(ca_cert_file)) + client = NanoKVMClient(_BASE_URL, ssl_ca_cert=str(ca_cert_file)) ssl_config = client._session.create_ssl_context() @@ -54,9 +57,7 @@ async def test_custom_ca_certificate(tmp_path: Path) -> None: async def test_nonexistent_ca_cert_raises_error() -> None: """Test that non-existent CA cert file raises FileNotFoundError.""" - client = NanoKVMClient( - "https://kvm.local/api/", ssl_ca_cert="/path/that/does/not/exist.pem" - ) + client = NanoKVMClient(_BASE_URL, ssl_ca_cert="/path/that/does/not/exist.pem") with pytest.raises(FileNotFoundError): client._session.create_ssl_context() @@ -64,9 +65,7 @@ async def test_nonexistent_ca_cert_raises_error() -> None: async def test_context_manager_closes_session_when_ssl_setup_fails() -> None: """A failed context entry does not leak the internally-created session.""" - client = NanoKVMClient( - "https://kvm.local/api/", ssl_ca_cert="/path/that/does/not/exist.pem" - ) + client = NanoKVMClient(_BASE_URL, ssl_ca_cert="/path/that/does/not/exist.pem") try: with pytest.raises(FileNotFoundError): @@ -86,7 +85,7 @@ async def test_context_manager_closes_session_when_ssl_setup_fails() -> None: async def test_http_url_works_regardless_of_ssl_config() -> None: """Test that HTTP URL (not HTTPS) works regardless of SSL config.""" - client = NanoKVMClient("http://kvm.local/api/", verify_ssl=False) + client = NanoKVMClient(BASE_URL, verify_ssl=False) async with client: assert client._session._http_session is not None @@ -94,7 +93,7 @@ async def test_http_url_works_regardless_of_ssl_config() -> None: async def test_session_created_with_tcp_connector() -> None: """Test that session is created with TCPConnector.""" - client = NanoKVMClient("https://kvm.local/api/") + client = NanoKVMClient(_BASE_URL) async with client: assert client._session._http_session is not None @@ -103,30 +102,28 @@ async def test_session_created_with_tcp_connector() -> None: async def test_password_obfuscation_auto_by_default() -> None: """Test that password obfuscation defaults to None (auto-detect).""" - client = NanoKVMClient("https://kvm.local/api/") + client = NanoKVMClient(_BASE_URL) assert client._session._use_password_obfuscation is None async def test_password_obfuscation_can_be_disabled() -> None: """Test that password obfuscation can be disabled.""" - client = NanoKVMClient("https://kvm.local/api/", use_password_obfuscation=False) + client = NanoKVMClient(_BASE_URL, use_password_obfuscation=False) assert client._session._use_password_obfuscation is False async def test_authenticate_with_plain_text_password(mocked: aioresponses) -> None: """Test authentication with plain text password (newer NanoKVM).""" - async with NanoKVMClient( - "https://kvm.local/api/", use_password_obfuscation=False - ) as client: + async with NanoKVMClient(_BASE_URL, use_password_obfuscation=False) as client: mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={"code": 0, "msg": "success", "data": {"token": "abc123"}}, ) mocked.get(_HARDWARE_URL, payload=_HARDWARE_PAYLOAD) await client.authenticate("root", "password123") - calls = mocked.requests[("POST", yarl.URL("https://kvm.local/api/auth/login"))] + calls = mocked.requests[("POST", yarl.URL(f"{_BASE_URL}auth/login"))] assert len(calls) == 1 request_json = calls[0].kwargs.get("json") assert request_json["username"] == "root" @@ -142,11 +139,9 @@ async def test_authentication_debug_logs_redact_credentials( token = "token-must-not-be-logged" caplog.set_level(logging.DEBUG, logger="nanokvm") - async with NanoKVMClient( - "https://kvm.local/api/", use_password_obfuscation=False - ) as client: + async with NanoKVMClient(_BASE_URL, use_password_obfuscation=False) as client: mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={"code": 0, "msg": "success", "data": {"token": token}}, ) mocked.get(_HARDWARE_URL, payload=_HARDWARE_PAYLOAD) @@ -159,18 +154,16 @@ async def test_authentication_debug_logs_redact_credentials( async def test_authenticate_with_obfuscated_password(mocked: aioresponses) -> None: """Test authentication with obfuscated password (older NanoKVM).""" - async with NanoKVMClient( - "https://kvm.local/api/", use_password_obfuscation=True - ) as client: + async with NanoKVMClient(_BASE_URL, use_password_obfuscation=True) as client: mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={"code": 0, "msg": "success", "data": {"token": "abc123"}}, ) mocked.get(_HARDWARE_URL, payload=_HARDWARE_PAYLOAD) await client.authenticate("root", "password123") - calls = mocked.requests[("POST", yarl.URL("https://kvm.local/api/auth/login"))] + calls = mocked.requests[("POST", yarl.URL(f"{_BASE_URL}auth/login"))] assert len(calls) == 1 request_json = calls[0].kwargs.get("json") assert request_json["username"] == "root" @@ -180,9 +173,9 @@ async def test_authenticate_with_obfuscated_password(mocked: aioresponses) -> No async def test_auto_detect_obfuscation_succeeds(mocked: aioresponses) -> None: """Test auto-detect succeeds with obfuscated password on first attempt.""" - async with NanoKVMClient("https://kvm.local/api/") as client: + async with NanoKVMClient(_BASE_URL) as client: mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={ "code": 0, "msg": "success", @@ -193,7 +186,7 @@ async def test_auto_detect_obfuscation_succeeds(mocked: aioresponses) -> None: await client.authenticate("root", "password123") - calls = mocked.requests[("POST", yarl.URL("https://kvm.local/api/auth/login"))] + calls = mocked.requests[("POST", yarl.URL(f"{_BASE_URL}auth/login"))] assert len(calls) == 1 request_json = calls[0].kwargs.get("json") assert request_json["password"].startswith("U2FsdGVkX1") @@ -202,10 +195,10 @@ async def test_auto_detect_obfuscation_succeeds(mocked: aioresponses) -> None: async def test_auto_detect_fallback_to_plain_text(mocked: aioresponses) -> None: """Test auto-detect falls back to plain text after obfuscated fails.""" - async with NanoKVMClient("https://kvm.local/api/") as client: + async with NanoKVMClient(_BASE_URL) as client: # First call: obfuscated fails with code -2 mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={ "code": -2, "msg": "invalid username or password", @@ -214,7 +207,7 @@ async def test_auto_detect_fallback_to_plain_text(mocked: aioresponses) -> None: ) # Second call: plain text succeeds mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={ "code": 0, "msg": "success", @@ -225,7 +218,7 @@ async def test_auto_detect_fallback_to_plain_text(mocked: aioresponses) -> None: await client.authenticate("root", "password123") - calls = mocked.requests[("POST", yarl.URL("https://kvm.local/api/auth/login"))] + calls = mocked.requests[("POST", yarl.URL(f"{_BASE_URL}auth/login"))] assert len(calls) == 2 # First attempt: obfuscated assert calls[0].kwargs.get("json")["password"].startswith("U2FsdGVkX1") @@ -236,10 +229,10 @@ async def test_auto_detect_fallback_to_plain_text(mocked: aioresponses) -> None: async def test_auto_detect_both_fail(mocked: aioresponses) -> None: """Test auto-detect raises NanoKVMAuthenticationFailure when both fail.""" - async with NanoKVMClient("https://kvm.local/api/") as client: + async with NanoKVMClient(_BASE_URL) as client: # First call: obfuscated fails mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={ "code": -2, "msg": "invalid username or password", @@ -248,7 +241,7 @@ async def test_auto_detect_both_fail(mocked: aioresponses) -> None: ) # Second call: plain text also fails mocked.post( - "https://kvm.local/api/auth/login", + f"{_BASE_URL}auth/login", payload={ "code": -2, "msg": "invalid username or password", @@ -262,11 +255,9 @@ async def test_auto_detect_both_fail(mocked: aioresponses) -> None: async def test_full_client_lifecycle_with_ssl(mocked: aioresponses) -> None: """Test full client lifecycle with SSL configuration.""" - async with NanoKVMClient( - "https://kvm.local/api/", token="test-token", verify_ssl=True - ) as client: + async with NanoKVMClient(_BASE_URL, token="test-token", verify_ssl=True) as client: mocked.get( - "https://kvm.local/api/vm/info", + f"{_BASE_URL}vm/info", payload={ "code": 0, "msg": "success", diff --git a/tests/test_storage.py b/tests/test_storage.py new file mode 100644 index 0000000..fd5ce8b --- /dev/null +++ b/tests/test_storage.py @@ -0,0 +1,152 @@ +"""Tests for image and script storage operations.""" + +from pathlib import Path + +import aiohttp +from aioresponses import aioresponses +import pytest +import yarl + +from nanokvm.client import NanoKVMApiError, NanoKVMClient, NanoKVMNotSupportedError +from nanokvm.models import ApiResponseCode, DownloadStatus, HWVersion, StatusImageRsp + +from .common import BASE_URL, mark_detected + + +@pytest.mark.parametrize( + ("hardware", "expected_prefix"), + [ + (HWVersion.PRO, "/storage/download"), + (HWVersion.PCIE, "/download"), + (HWVersion.UNKNOWN, "/download"), + (None, "/download"), + ], +) +async def test_image_download_prefix_by_hardware_family( + hardware: HWVersion | None, + expected_prefix: str, + client: NanoKVMClient, +) -> None: + """Image download routes use the Pro prefix only for Pro hardware.""" + client._session._hw_version = hardware + + assert client._image_download_prefix() == expected_prefix + + +@pytest.mark.parametrize( + "status", + ["success", "failed", "checksum_failed"], +) +def test_download_status_accepts_terminal_states(status: str) -> None: + """Download status responses include terminal success and failure states.""" + response = StatusImageRsp.model_validate( + {"status": status, "file": "image.iso", "percentage": ""} + ) + + assert response.status is DownloadStatus(status) + + +async def test_get_images_success(client: NanoKVMClient, mocked: aioresponses) -> None: + """Test get_images with a successful response.""" + mocked.get( + f"{BASE_URL}storage/image", + payload={ + "code": 0, + "msg": "success", + "data": { + "files": [ + "/data/alpine-standard-3.23.2-x86_64.iso", + "/data/cs10-js.iso", + ] + }, + }, + ) + + response = await client.get_images() + + assert response is not None + assert len(response.files) == 2 + assert "/data/alpine-standard-3.23.2-x86_64.iso" in response.files + assert "/data/cs10-js.iso" in response.files + + +async def test_get_images_empty(client: NanoKVMClient, mocked: aioresponses) -> None: + """Test get_images with an empty list.""" + mocked.get( + f"{BASE_URL}storage/image", + payload={"code": 0, "msg": "success", "data": {"files": []}}, + ) + + response = await client.get_images() + + assert response is not None + assert len(response.files) == 0 + + +async def test_get_images_api_error( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test get_images with an API error response.""" + mocked.get( + f"{BASE_URL}storage/image", + payload={"code": -1, "msg": "failed to list images", "data": None}, + ) + + with pytest.raises(NanoKVMApiError) as exc_info: + await client.get_images() + + assert exc_info.value.code == ApiResponseCode.FAILURE + assert "failed to list images" in exc_info.value.msg + + +async def test_mount_image_sends_pro_read_only_flag( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test mount_image sends the Pro readOnly field.""" + mocked.post( + f"{BASE_URL}storage/image/mount", + payload={"code": 0, "msg": "success", "data": None}, + ) + + mark_detected(client, HWVersion.PRO, "1.2.1") + await client.mount_image("/data/test.img", read_only=True) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}storage/image/mount"))] + assert calls[0].kwargs.get("json") == { + "file": "/data/test.img", + "cdrom": False, + "readOnly": True, + } + + +async def test_upload_script_uses_multipart_form( + tmp_path: Path, client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test script upload uses the shared multipart helper.""" + script = tmp_path / "test.sh" + script.write_text("#!/bin/sh\ntrue\n") + + mocked.post( + f"{BASE_URL}vm/script/upload", + payload={ + "code": 0, + "msg": "success", + "data": {"file": "test.sh"}, + }, + ) + + response = await client.upload_script(script) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/script/upload"))] + assert isinstance(calls[0].kwargs.get("data"), aiohttp.FormData) + assert response.file == "test.sh" + + +async def test_mount_image_read_only_requires_pro_1_2_1( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Older Pro firmware ignores readOnly, so the client refuses it first.""" + mark_detected(client, HWVersion.PRO, "1.2.0") + with pytest.raises(NanoKVMNotSupportedError, match="1.2.1"): + await client.mount_image("/data/test.img", read_only=True) + assert not mocked.requests diff --git a/tests/test_stream.py b/tests/test_stream.py new file mode 100644 index 0000000..2807d3c --- /dev/null +++ b/tests/test_stream.py @@ -0,0 +1,88 @@ +"""Tests for MJPEG streaming.""" + +import asyncio +import io + +from aiohttp import web +from aioresponses import aioresponses +from PIL import Image + +from nanokvm.client import NanoKVMClient + +from .common import BASE_URL + + +def test_parse_jpeg_returns_fully_loaded_image() -> None: + """JPEG parsing returns an image detached from the source stream.""" + image_buffer = io.BytesIO() + Image.new("RGB", (2, 2), color=(1, 2, 3)).save(image_buffer, format="JPEG") + client = NanoKVMClient(BASE_URL) + + image = client._parse_jpeg_from_bytes(image_buffer.getvalue()) + + assert image.getpixel((0, 0)) + assert getattr(image, "fp", None) is None + + +async def test_mjpeg_stream_allows_frames_after_request_timeout() -> None: + """An MJPEG stream remains open beyond the normal request timeout.""" + image_buffer = io.BytesIO() + Image.new("RGB", (2, 2)).save(image_buffer, format="JPEG") + image_data = image_buffer.getvalue() + + async def stream_handler( + request: web.Request, + ) -> web.StreamResponse: + response = web.StreamResponse( + headers={"Content-Type": "multipart/x-mixed-replace; boundary=frame"} + ) + await response.prepare(request) + for _ in range(2): + await response.write( + b"--frame\r\n" + b"Content-Type: image/jpeg\r\n" + + f"Content-Length: {len(image_data)}\r\n\r\n".encode() + + image_data + + b"\r\n" + ) + await asyncio.sleep(0.55) + await response.write(b"--frame--\r\n") + await response.write_eof() + return response + + application = web.Application() + application.router.add_get("/api/stream/mjpeg", stream_handler) + runner = web.AppRunner(application) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + server_sockets = getattr(site._server, "sockets", None) + assert server_sockets + port = server_sockets[0].getsockname()[1] + + frames = [] + try: + async with NanoKVMClient( + f"http://127.0.0.1:{port}/api/", token="test-token", request_timeout=1 + ) as client: + async with asyncio.timeout(3): + async for frame in client.mjpeg_stream(): + frames.append(frame) + finally: + await runner.cleanup() + + assert len(frames) == 2 + + +async def test_mjpeg_stream_bytes_yields_raw_frames(mocked: aioresponses) -> None: + """The byte stream hands out each JPEG part untouched.""" + mocked.get( + f"{BASE_URL}stream/mjpeg", + headers={"Content-Type": "multipart/x-mixed-replace; boundary=frame"}, + body=b"--frame\r\nContent-Type: image/jpeg\r\n\r\n\xff\xd8one\r\n" + b"--frame\r\nContent-Type: image/jpeg\r\n\r\n\xff\xd8two\r\n--frame--\r\n", + ) + async with NanoKVMClient(BASE_URL, token="test-token") as c: + frames = [frame async for frame in c.mjpeg_stream_bytes()] + + assert frames == [b"\xff\xd8one", b"\xff\xd8two"] diff --git a/tests/test_system.py b/tests/test_system.py new file mode 100644 index 0000000..35144f8 --- /dev/null +++ b/tests/test_system.py @@ -0,0 +1,174 @@ +"""Tests for system, virtual device and LED operations.""" + +from aioresponses import aioresponses +import pytest +import yarl + +from nanokvm.client import NanoKVMApiError, NanoKVMClient +from nanokvm.models import DiskType, HWVersion, VirtualDevice + +from .common import BASE_URL, mark_detected + + +async def test_get_web_title_returns_default_when_unset( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test the firmware's failed read of an unset title maps to the default.""" + mark_detected(client) + + mocked.get( + f"{BASE_URL}vm/web-title", + payload={"code": -1, "msg": "read web title failed", "data": None}, + ) + + assert (await client.get_web_title()).title == "NanoKVM" + + +async def test_get_oled_info_raises_for_pro_invalid_file_content( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test Pro OLED invalid file content follows upstream error semantics.""" + client._session._hw_version = HWVersion.PRO + + mocked.get( + f"{BASE_URL}vm/oled", + payload={ + "code": -2, + "msg": "invalid file content", + "data": None, + }, + ) + + with pytest.raises(NanoKVMApiError) as exc_info: + await client.get_oled_info() + + assert exc_info.value.code == -2 + assert exc_info.value.msg == "invalid file content" + info_url = yarl.URL(f"{BASE_URL}vm/info") + assert ("GET", info_url) not in mocked.requests + + +async def test_update_virtual_device_ignores_success_data( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test update_virtual_device handles non-Pro success payloads.""" + mocked.post( + f"{BASE_URL}vm/device/virtual", + payload={"code": 0, "msg": "success", "data": {"on": True}}, + ) + + await client.update_virtual_device(VirtualDevice.DISK) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/device/virtual"))] + assert calls[0].kwargs.get("json") == {"device": "disk"} + + +async def test_get_virtual_device_status_handles_non_pro_shape( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test virtual device status keeps non-Pro fields intact.""" + mocked.get( + f"{BASE_URL}vm/device/virtual", + payload={ + "code": 0, + "msg": "success", + "data": {"network": True, "media": False, "disk": True}, + }, + ) + + response = await client.get_virtual_device_status() + + assert response.network is True + assert response.media is False + assert response.disk is True + assert response.mounted_disk is None + + +async def test_get_virtual_device_status_handles_pro_shape( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test Pro virtual device status exposes Pro fields without fake non-Pro disk.""" + client._session._hw_version = HWVersion.PRO + + mocked.get( + f"{BASE_URL}vm/device/virtual", + payload={ + "code": 0, + "msg": "success", + "data": { + "isNetworkEnabled": True, + "isMicEnabled": True, + "mountedDisk": "emmc", + "isSdCardExist": True, + "isEmmcExist": True, + }, + }, + ) + + response = await client.get_virtual_device_status() + + assert response.network is True + assert response.mic is True + assert response.mounted_disk == "emmc" + assert response.is_sd_card_exist is True + assert response.is_emmc_exist is True + assert response.media is None + assert response.disk is None + + +async def test_update_virtual_device_sends_pro_disk_type( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test Pro disk virtual device requests include the disk type.""" + client._session._hw_version = HWVersion.PRO + + mocked.post( + f"{BASE_URL}vm/device/virtual", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.update_virtual_device( + VirtualDevice.DISK, + disk_type=DiskType.EMMC, + ) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/device/virtual"))] + assert calls[0].kwargs.get("json") == { + "device": "disk", + "type": "emmc", + } + + +async def test_set_led_strip_partial_preserves_current_config( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test partial LED updates post a complete Pro LED configuration.""" + client._session._hw_version = HWVersion.PRO + + mocked.get( + f"{BASE_URL}vm/ledstrip/get", + payload={ + "code": 0, + "msg": "success", + "data": { + "on": True, + "hor": 8, + "ver": 6, + "brightness": 25, + }, + }, + ) + mocked.post( + f"{BASE_URL}vm/ledstrip/set", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.set_led_strip(brightness=50) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/ledstrip/set"))] + assert calls[0].kwargs.get("json") == { + "on": True, + "hor": 8, + "ver": 6, + "brightness": 50, + } diff --git a/tests/test_video.py b/tests/test_video.py new file mode 100644 index 0000000..d9f709a --- /dev/null +++ b/tests/test_video.py @@ -0,0 +1,26 @@ +"""Tests for video stream settings.""" + +from aioresponses import aioresponses +import yarl + +from nanokvm.client import NanoKVMClient +from nanokvm.models import HWVersion, StreamMode + +from .common import BASE_URL + + +async def test_set_stream_mode_accepts_existing_string_values( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test set_stream_mode remains source-compatible with valid strings.""" + client._session._hw_version = HWVersion.PRO + + mocked.post( + f"{BASE_URL}stream/mode", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.set_stream_mode(StreamMode.H264_DIRECT.value) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}stream/mode"))] + assert calls[0].kwargs.get("json") == {"mode": "h264-direct"}