diff --git a/qtoggleserver/core/api/__init__.py b/qtoggleserver/core/api/__init__.py index 4077359c..d23c1d17 100644 --- a/qtoggleserver/core/api/__init__.py +++ b/qtoggleserver/core/api/__init__.py @@ -95,7 +95,7 @@ def wrapper(request_handler: APIHandler, *args, **kwargs) -> Any: else: raise APIError(403, "forbidden", required_level=ACCESS_LEVEL_MAPPING.get(access_level)) - request = APIRequest(request_handler) + request = request_handler if isinstance(request_handler, APIRequest) else APIRequest(request_handler) return func(request, *args, **kwargs) diff --git a/qtoggleserver/core/api/funcs/ports.py b/qtoggleserver/core/api/funcs/ports.py index b8d19649..8b35e81f 100644 --- a/qtoggleserver/core/api/funcs/ports.py +++ b/qtoggleserver/core/api/funcs/ports.py @@ -28,6 +28,9 @@ from qtoggleserver.utils import json as json_utils +MAX_VALUE_TIMEOUT = 3600 # seconds + + async def add_virtual_port(attrs: GenericJSONDict) -> core_ports.BasePort: id_ = attrs["id"] type_ = attrs["type"] @@ -351,10 +354,23 @@ async def get_port_value(request: core_api.APIRequest, port_id: str) -> Nullable @core_api.api_call(core_api.ACCESS_LEVEL_NORMAL) async def patch_port_value(request: core_api.APIRequest, port_id: str, params: PortValue) -> None: + request_time = time.time() + port = core_ports.get(port_id) if port is None: raise core_api.APIError(404, "no-such-port") + timeout_str = request.query.get("timeout") + timeout = 0 + if timeout_str is not None: + try: + timeout = float(timeout_str) + except ValueError: + raise core_api.APIError(400, "invalid-field", field="timeout") from None + + if timeout < 0 or timeout > MAX_VALUE_TIMEOUT: + raise core_api.APIError(400, "invalid-field", field="timeout") + try: core_api_schema.validate(params, await port.get_value_schema()) except core_api.APIError: @@ -387,6 +403,11 @@ async def patch_port_value(request: core_api.APIRequest, port_id: str, params: P # Transform any unhandled exception into APIError(500) raise core_api.APIError(500, "unexpected-error", message=str(e)) from e + if timeout: + remaining = timeout - (time.time() - request_time) + if not await port.wait_for_read_value(value, timeout=remaining): + raise core_api.APIError(504, "value-timeout") + @core_api.api_call(core_api.ACCESS_LEVEL_NORMAL) async def patch_port_sequence(request: core_api.APIRequest, port_id: str, params: GenericJSONDict) -> None: diff --git a/qtoggleserver/core/ports.py b/qtoggleserver/core/ports.py index b2f09b31..1e5cd7fc 100644 --- a/qtoggleserver/core/ports.py +++ b/qtoggleserver/core/ports.py @@ -249,6 +249,7 @@ def __init__(self, port_id: str) -> None: self._last_read_value: tuple[NullablePortValue, int] | None = None self._read_value_lock = asyncio.Lock() + self._value_match_waiters: list[tuple[NullablePortValue, asyncio.Future]] = [] # Value that's currently being written self._writing_value: NullablePortValue = None @@ -672,6 +673,32 @@ def get_last_read_value(self) -> NullablePortValue: def set_last_read_value(self, value: NullablePortValue) -> None: self._last_read_value = value, int(time.time() * 1000) + for target_value, future in self._value_match_waiters: + if not future.done() and value == target_value: + future.set_result(None) + + async def wait_for_read_value(self, value: NullablePortValue, timeout: float) -> bool: + """Wait until the port's last read value becomes equal to `value`, but no longer than `timeout` seconds. + Returns `True` as soon as a matching value is read (right away if it already matches), or `False` if + `timeout` seconds elapse without a match.""" + + if self.get_last_read_value() == value: + return True + + if timeout <= 0: + return False + + future = asyncio.get_running_loop().create_future() + waiter = (value, future) + self._value_match_waiters.append(waiter) + try: + await asyncio.wait_for(future, timeout=timeout) + return True + except TimeoutError: + return False + finally: + self._value_match_waiters.remove(waiter) + async def read_transformed_value(self) -> NullablePortValue: value = None async with self._read_value_lock: diff --git a/qtoggleserver/frontend/js/api/constants.js b/qtoggleserver/frontend/js/api/constants.js index f686882c..e15df889 100644 --- a/qtoggleserver/frontend/js/api/constants.js +++ b/qtoggleserver/frontend/js/api/constants.js @@ -218,6 +218,11 @@ export const KNOWN_ERRORS = [ code: 'device-timeout', pretty: gettext('Timeout waiting for a response from the device.') }, + { + status: 504, + code: 'value-timeout', + pretty: gettext('Timeout waiting for value to take effect.') + }, /* Other errors */ { diff --git a/qtoggleserver/frontend/js/api/ports.js b/qtoggleserver/frontend/js/api/ports.js index 21ff847e..262c24e3 100644 --- a/qtoggleserver/frontend/js/api/ports.js +++ b/qtoggleserver/frontend/js/api/ports.js @@ -120,11 +120,12 @@ export function getPortValue(id) { * @alias qtoggle.api.ports.patchPortValue * @param {String} id the port identifier * @param {Boolean|Number} value the new port value + * @param {?Number} [confirmTimeout] optional confirmation timeout, in seconds * @param {Number} [expectEventTimeout] optional timeout within which a corresponding event will be expected, in * milliseconds * @returns {Promise} */ -export function patchPortValue(id, value, expectEventTimeout = null) { +export function patchPortValue(id, value, confirmTimeout = null, expectEventTimeout = null) { let port = Cache.getPort(id) let handle = null @@ -142,12 +143,22 @@ export function patchPortValue(id, value, expectEventTimeout = null) { }, expectEventTimeout) } + let query = null + let timeout = APIConstants.LONG_SERVER_TIMEOUT + if (confirmTimeout != null) { + query = {timeout: confirmTimeout} + + /* Make sure the client waits at least as long as the device is expected to, plus some margin */ + timeout = Math.max(timeout, confirmTimeout + APIConstants.DEFAULT_SERVER_TIMEOUT) + } + return BaseAPI.apiCall({ method: 'PATCH', path: `/ports/${id}/value`, data: value, + query: query, expectedHandle: handle, - timeout: APIConstants.LONG_SERVER_TIMEOUT + timeout: timeout }) } diff --git a/qtoggleserver/frontend/js/dashboard/widgets/widget.js b/qtoggleserver/frontend/js/dashboard/widgets/widget.js index 6d4e0707..a97336ea 100644 --- a/qtoggleserver/frontend/js/dashboard/widgets/widget.js +++ b/qtoggleserver/frontend/js/dashboard/widgets/widget.js @@ -2,8 +2,6 @@ import $ from '$qui/lib/jquery.module.js' import Logger from '$qui/lib/logger.module.js' -import ConditionVariable from '$qui/base/condition-variable.js' -import {TimeoutError} from '$qui/base/errors.js' import {gettext} from '$qui/base/i18n.js' import {mix} from '$qui/base/mixwith.js' import StockIcon from '$qui/icons/stock-icon.js' @@ -104,10 +102,6 @@ class Widget extends mix().with(ViewMixin) { this._height = this.constructor.height } - this._valueChangeWaitPortId = null - this._whenValueChange = null - this._valueChangeTimeoutHandle = null - this.logger = Logger.get(this.makeLogName()) } @@ -1129,14 +1123,6 @@ class Widget extends mix().with(ViewMixin) { /* Ports and port values */ handlePortValueChange(portId, value) { - if (this._valueChangeWaitPortId === portId && this._whenValueChange) { - this._whenValueChange.fulfill() - this._whenValueChange = null - this._valueChangeWaitPortId = null - clearTimeout(this._valueChangeTimeoutHandle) - this._valueChangeTimeoutHandle = null - } - this.onPortValueChange(portId, value) } @@ -1168,46 +1154,13 @@ class Widget extends mix().with(ViewMixin) { * @param {String} portId the id of the port whose value will be set * @param {Number|Boolean} value the new port value * @param {Number} [timeout] how long to wait for new port value to take effect (seconds, defaults to - * {@link qtoggle.api.constants.DEFAULT_SERVER_TIMEOUT}) + * {@link qtoggle.api.constants.DEFAULT_SERVER_TIMEOUT}); pass `0` to not wait for confirmation at all * @returns {Promise} */ setPortValue(portId, value, timeout = APIConstants.DEFAULT_SERVER_TIMEOUT) { this.setProgress() - let prevValue = this.getPortValue(portId) - - if (this._whenValueChange) { - this._whenValueChange.fulfill() - this._whenValueChange = null - this._valueChangeWaitPortId = null - clearTimeout(this._valueChangeTimeoutHandle) - this._valueChangeTimeoutHandle = null - } - - this._whenValueChange = new ConditionVariable() - this._valueChangeWaitPortId = portId - - return PortsAPI.patchPortValue(portId, value).then(function () { - - if (value === prevValue || timeout === 0) { - return /* Value was already set or we're not interested in waiting */ - } - - this._valueChangeTimeoutHandle = setTimeout(function () { - /* Cancel waiting after timeout */ - if (this._whenValueChange) { - let msg = gettext('Timeout waiting for value to take effect.') - this._whenValueChange.cancel(new TimeoutError(msg)) - this._whenValueChange = null - this._valueChangeWaitPortId = null - this._valueChangeTimeoutHandle = null - } - - }.bind(this), timeout * 1000) - - return this._whenValueChange - - }.bind(this)).then(function () { + return PortsAPI.patchPortValue(portId, value, timeout).then(function () { this.clearProgress() diff --git a/tests/unit/qtoggleserver/core/api/test_funcs_ports.py b/tests/unit/qtoggleserver/core/api/test_funcs_ports.py index e12dd792..e4bcf9db 100644 --- a/tests/unit/qtoggleserver/core/api/test_funcs_ports.py +++ b/tests/unit/qtoggleserver/core/api/test_funcs_ports.py @@ -1,3 +1,5 @@ +import asyncio + import pytest from qtoggleserver.conf import settings @@ -179,3 +181,130 @@ async def test_read_only_port(self, mock_api_request_maker, mock_num_port1, mock with pytest.raises(core_api.APIError, match="read-only-port") as exc_info: await ports_api_funcs.patch_port_value(request, "nid1", 100) assert exc_info.value.status == 400 + + +class TestPatchPortValueTimeout: + @pytest.fixture(autouse=True) + def mock_slaves(self, mocker) -> None: + mocker.patch("qtoggleserver.slaves.devices.get_all", return_value=[]) + + async def test_non_integer_timeout_raises_invalid_field( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver + ) -> None: + mock_num_port1.set_writable(True) + + request = mock_api_request_maker( + "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "abc"} + ) + with pytest.raises(core_api.APIError, match="invalid-field") as exc_info: + await ports_api_funcs.patch_port_value(request, "nid1", 100) + assert exc_info.value.status == 400 + assert exc_info.value.params["field"] == "timeout" + + async def test_negative_timeout_raises_invalid_field( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver + ) -> None: + mock_num_port1.set_writable(True) + + request = mock_api_request_maker( + "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "-1"} + ) + with pytest.raises(core_api.APIError, match="invalid-field") as exc_info: + await ports_api_funcs.patch_port_value(request, "nid1", 100) + assert exc_info.value.status == 400 + assert exc_info.value.params["field"] == "timeout" + + async def test_too_large_timeout_raises_invalid_field( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver + ) -> None: + mock_num_port1.set_writable(True) + + request = mock_api_request_maker( + "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "3601"} + ) + with pytest.raises(core_api.APIError, match="invalid-field") as exc_info: + await ports_api_funcs.patch_port_value(request, "nid1", 100) + assert exc_info.value.status == 400 + assert exc_info.value.params["field"] == "timeout" + + async def test_no_timeout_does_not_wait_for_confirmation( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver, mocker + ) -> None: + """Without a `timeout` query argument, the request must be responded as soon as the write completes, + without waiting for the read value to catch up.""" + + mock_num_port1.set_writable(True) + spy = mocker.spy(mock_num_port1, "wait_for_read_value") + + request = mock_api_request_maker("PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL) + result = await ports_api_funcs.patch_port_value(request, "nid1", 100) + + assert result is None + spy.assert_not_called() + + async def test_returns_immediately_if_value_already_matches( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver, mocker + ) -> None: + mock_num_port1.set_writable(True) + mock_num_port1.set_last_read_value(100) + spy = mocker.spy(mock_num_port1, "wait_for_read_value") + + request = mock_api_request_maker( + "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "5"} + ) + result = await ports_api_funcs.patch_port_value(request, "nid1", 100) + + assert result is None + spy.assert_called_once() + + async def test_waits_and_succeeds_once_matching_value_is_read( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver + ) -> None: + mock_num_port1.set_writable(True) + mock_num_port1.set_last_read_value(0) + + async def delayed_read() -> None: + await asyncio.sleep(0.01) + mock_num_port1.set_last_read_value(100) + + task = asyncio.create_task(delayed_read()) + + request = mock_api_request_maker( + "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "5"} + ) + try: + result = await ports_api_funcs.patch_port_value(request, "nid1", 100) + finally: + await task + + assert result is None + + async def test_raises_value_timeout_if_value_never_matches( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver + ) -> None: + mock_num_port1.set_writable(True) + mock_num_port1.set_last_read_value(0) + + request = mock_api_request_maker( + "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "0.02"} + ) + with pytest.raises(core_api.APIError, match="value-timeout") as exc_info: + await ports_api_funcs.patch_port_value(request, "nid1", 100) + assert exc_info.value.status == 504 + + async def test_erroneous_write_skips_confirmation( + self, mock_api_request_maker, mock_num_port1, mock_persist_driver, mocker + ) -> None: + """An erroneous write must be responded as usual, without waiting for any confirmation.""" + + mock_num_port1.set_writable(True) + mocker.patch.object(mock_num_port1, "push_write_and_wait", side_effect=core_ports.PortTimeout()) + spy = mocker.spy(mock_num_port1, "wait_for_read_value") + + request = mock_api_request_maker( + "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "5"} + ) + with pytest.raises(core_api.APIError, match="port-timeout") as exc_info: + await ports_api_funcs.patch_port_value(request, "nid1", 100) + assert exc_info.value.status == 504 + spy.assert_not_called() diff --git a/tests/unit/qtoggleserver/core/test_ports.py b/tests/unit/qtoggleserver/core/test_ports.py index a0e7bb30..9e9df34c 100644 --- a/tests/unit/qtoggleserver/core/test_ports.py +++ b/tests/unit/qtoggleserver/core/test_ports.py @@ -180,6 +180,69 @@ async def blocking_transform(value): assert future_b.cancelled() +class TestPortWaitForReadValue: + async def test_returns_true_immediately_if_already_matching(self, mock_num_port1): + mock_num_port1.set_last_read_value(100) + assert await mock_num_port1.wait_for_read_value(100, timeout=1) is True + + async def test_returns_false_immediately_for_nonpositive_timeout_without_match(self, mock_num_port1): + mock_num_port1.set_last_read_value(0) + assert await mock_num_port1.wait_for_read_value(100, timeout=0) is False + + async def test_resolves_when_matching_value_is_later_read(self, mock_num_port1): + mock_num_port1.set_last_read_value(0) + + async def delayed_read(): + await asyncio.sleep(0.01) + mock_num_port1.set_last_read_value(100) + + task = asyncio.create_task(delayed_read()) + try: + assert await mock_num_port1.wait_for_read_value(100, timeout=1) is True + finally: + await task + + async def test_ignores_non_matching_values(self, mock_num_port1): + """A read value that doesn't match the target shouldn't resolve the wait.""" + + mock_num_port1.set_last_read_value(0) + + async def delayed_reads(): + await asyncio.sleep(0.01) + mock_num_port1.set_last_read_value(50) + await asyncio.sleep(0.01) + mock_num_port1.set_last_read_value(100) + + task = asyncio.create_task(delayed_reads()) + try: + assert await mock_num_port1.wait_for_read_value(100, timeout=1) is True + finally: + await task + + async def test_times_out_without_match(self, mock_num_port1): + mock_num_port1.set_last_read_value(0) + assert await mock_num_port1.wait_for_read_value(100, timeout=0.02) is False + + async def test_cleans_up_waiter_after_resolving(self, mock_num_port1): + mock_num_port1.set_last_read_value(0) + + async def delayed_read(): + await asyncio.sleep(0.01) + mock_num_port1.set_last_read_value(100) + + task = asyncio.create_task(delayed_read()) + try: + await mock_num_port1.wait_for_read_value(100, timeout=1) + finally: + await task + assert mock_num_port1._value_match_waiters == [] + + async def test_cleans_up_waiter_after_timeout(self, mock_num_port1): + mock_num_port1.set_last_read_value(0) + await mock_num_port1.wait_for_read_value(100, timeout=0.02) + assert mock_num_port1._value_match_waiters == [] + + class TestPortSetSequence: async def test_pushes_each_value_via_push_write(self, mock_num_port1, mocker): """Should install a sequence that pushes each value to the write queue via push_write.""" diff --git a/tests/unit/qtoggleserver/mock/api.py b/tests/unit/qtoggleserver/mock/api.py index 8cb1ff47..d36aa0ac 100644 --- a/tests/unit/qtoggleserver/mock/api.py +++ b/tests/unit/qtoggleserver/mock/api.py @@ -30,6 +30,7 @@ def __init__( method=method, path=path, query=query, + query_arguments={k: [v.encode()] for k, v in query.items()}, body=body, ), )