diff --git a/docs/api.md b/docs/api.md index a50a64b..25e2bec 100644 --- a/docs/api.md +++ b/docs/api.md @@ -66,6 +66,23 @@ Request and response models live in `nanokvm.models`. `set_tls` restarts the device's web server, which then answers only on the other scheme (`http` or `https`), within about half a minute. +`push_button(button, duration_ms)` holds the power or reset button for +`duration_ms`; `0` presses for the firmware default of 800 ms. The device +answers once the button is released, so the call waits for the press on top of +`request_timeout`. Holding the power button forces a hung target off: + +```python +await client.push_button(GpioType.POWER, 6000) # hold: force off +async with asyncio.timeout(30): + while (await client.get_gpio()).pwr: # power LED + await asyncio.sleep(1) +await client.push_button(GpioType.POWER, 800) # short press: power on +``` + +Check the power LED rather than the video signal: on the PCIe, +`get_hdmi_state().signal` can stay `True` for about ten seconds after the +target is off. + ### Keyboard and mouse | Methods | Requires | diff --git a/nanokvm/components/system.py b/nanokvm/components/system.py index fec126f..ac4e114 100644 --- a/nanokvm/components/system.py +++ b/nanokvm/components/system.py @@ -5,7 +5,7 @@ import contextlib from datetime import UTC, datetime -from aiohttp import ServerDisconnectedError, hdrs +from aiohttp import ClientTimeout, ServerDisconnectedError, hdrs from ..compatibility import require_application_version, require_hardware from ..exceptions import NanoKVMApiError, NanoKVMInvalidResponseError @@ -54,6 +54,8 @@ from .session import Controller _DEFAULT_WEB_TITLE = "NanoKVM" +# The firmware presses for 800 ms when asked for 0 ms. +_DEFAULT_PRESS_MS = 800 class SystemController(Controller): @@ -94,11 +96,19 @@ async def get_gpio(self) -> GetGpioRsp: ) async def push_button(self, button: GpioType, duration_ms: int) -> None: - """Simulate pushing a hardware button.""" - await self._session.api_request_json( + """Simulate pushing a hardware button for ``duration_ms``. + + The device answers once the button is released, so the request waits + for the press on top of the request timeout. ``0`` presses for the + firmware default of 800 ms. + """ + session = self._session + press_seconds = (duration_ms or _DEFAULT_PRESS_MS) / 1000 + await session.api_request_json( hdrs.METH_POST, "/vm/gpio", data=SetGpioReq(type=button, duration=duration_ms), + timeout=ClientTimeout(total=session._request_timeout + press_seconds), ) @require_application_version(non_pro="2.1.6") diff --git a/tests/test_system.py b/tests/test_system.py index 95a2d38..29bac40 100644 --- a/tests/test_system.py +++ b/tests/test_system.py @@ -1,18 +1,44 @@ """Tests for system, virtual device and LED operations.""" +import asyncio +from collections.abc import AsyncGenerator, Awaitable, Callable +from contextlib import asynccontextmanager from datetime import UTC, datetime import aiohttp +from aiohttp import web from aioresponses import aioresponses import pytest import yarl from nanokvm.client import NanoKVMApiError, NanoKVMClient -from nanokvm.models import DiskType, HWVersion, VirtualDevice +from nanokvm.models import DiskType, GpioType, HWVersion, VirtualDevice from .common import BASE_URL, mark_detected +@asynccontextmanager +async def _gpio_server( + handler: Callable[[web.Request], Awaitable[web.Response]], +) -> AsyncGenerator[str, None]: + application = web.Application() + application.router.add_post("/api/vm/gpio", handler) + runner = web.AppRunner(application) + await runner.setup() + try: + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + yield f"http://127.0.0.1:{runner.addresses[0][1]}/api/" + finally: + await runner.cleanup() + + +async def _press_then_answer(request: web.Request) -> web.Response: + """Answer once the press is over, as the firmware does.""" + await asyncio.sleep((await request.json())["duration"] / 1000) + return web.json_response({"code": 0, "msg": "success", "data": None}) + + async def test_get_web_title_returns_default_when_unset( client: NanoKVMClient, mocked: aioresponses ) -> None: @@ -201,3 +227,46 @@ async def test_set_tls_treats_restart_disconnect_as_success( calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/tls"))] assert calls[0].kwargs["json"] == {"enabled": True} + + +async def test_push_button_waits_for_a_press_longer_than_the_request_timeout() -> None: + """A press longer than the request timeout completes.""" + async with ( + _gpio_server(_press_then_answer) as url, + NanoKVMClient(url, token="test-token", request_timeout=1) as client, + ): + await client.push_button(GpioType.POWER, 1500) + + +@pytest.mark.parametrize(("duration_ms", "total"), [(6000, 16.0), (0, 10.8)]) +async def test_push_button_adds_the_press_to_the_request_timeout( + client: NanoKVMClient, mocked: aioresponses, duration_ms: int, total: float +) -> None: + """The request waits for the press; 0 presses for the default 800 ms.""" + mocked.post(f"{BASE_URL}vm/gpio", payload={"code": 0, "msg": "ok", "data": None}) + + await client.push_button(GpioType.POWER, duration_ms) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/gpio"))] + assert calls[0].kwargs["json"] == {"type": "power", "duration": duration_ms} + assert calls[0].kwargs["timeout"].total == pytest.approx(total) + + +async def test_push_button_still_times_out_when_the_device_does_not_answer() -> None: + """The press extends the timeout instead of removing it.""" + release = asyncio.Event() + + async def never_answers(request: web.Request) -> web.Response: + await release.wait() + return web.json_response({"code": 0, "msg": "success", "data": None}) + + async with ( + _gpio_server(never_answers) as url, + NanoKVMClient(url, token="test-token", request_timeout=1) as client, + ): + with pytest.raises(TimeoutError): + async with asyncio.timeout(5) as guard: + await client.push_button(GpioType.RESET, 100) + release.set() + + assert not guard.expired()