diff --git a/src/labone/core/session.py b/src/labone/core/session.py index 4e31837..f2abb69 100644 --- a/src/labone/core/session.py +++ b/src/labone/core/session.py @@ -9,13 +9,13 @@ from typing_extensions import Literal, NotRequired, TypeAlias, TypedDict from labone.core import errors, result -from labone.core import value as annotated_value from labone.core.connection_layer import ( KernelInfo, ServerInfo, create_session_client_stream, ) from labone.core.resources import session_protocol_capnp # type: ignore[attr-defined] +from labone.core.value import AnnotatedValue NodeType: TypeAlias = Literal[ "Integer (64 bit)", @@ -407,24 +407,24 @@ async def list_nodes_info( response = await _send_and_wait_request(request) return json.loads(response.nodeProps) - async def set_value( - self, - value: annotated_value.AnnotatedValue, - ) -> annotated_value.AnnotatedValue: + async def set_value(self, value: AnnotatedValue) -> AnnotatedValue: """Set the value of a node. - TODO: Tests - Args: value: Annotated value of the node. - TODO: To accept list of `AnnotatedValue`s + + Returns: + Acknowledged value from the device. Raises: + TypeError: If the node path is of wrong type. + LabOneCoreError: If the node value type is not supported. LabOneConnectionError: If there is a problem in the connection. """ + # T O D O: To accept list of `AnnotatedValue`s + capnp_value = value.to_capnp() request = self._session.setValue_request() - request.path = value.path - request.value = value.value + request.path = capnp_value.metadata.path + request.value = capnp_value.value response = await _send_and_wait_request(request) - res = result.unwrap(response.result) - return annotated_value.AnnotatedValue.from_capnp(res) + return AnnotatedValue.from_capnp(result.unwrap(response.result)) diff --git a/tests/core/test_session.py b/tests/core/test_session.py index 2c14896..17f76fc 100644 --- a/tests/core/test_session.py +++ b/tests/core/test_session.py @@ -1,7 +1,9 @@ """Tests for `labone.core.session.Session` functionality that requires a server.""" +import asyncio import json import random +from dataclasses import dataclass, field from typing import Any from unittest.mock import MagicMock @@ -17,6 +19,7 @@ _request_field_type_description, _send_and_wait_request, ) +from labone.core.value import AnnotatedValue from . import utils from .resources import testfile_capnp @@ -34,6 +37,9 @@ async def listNodes(self, _context, **_): # noqa: N802 async def listNodesJson(self, _context, **_): # noqa: N802 return self._mock.listNodesJson(_context.params, _context.results) + async def setValue(self, _context, **_): # noqa: N802 + return self._mock.setValue(_context.params, _context.results) + @pytest.fixture() async def session_server() -> tuple[Session, MagicMock]: @@ -250,9 +256,79 @@ async def test_a_wait_misc_error(self, error): @utils.ensure_event_loop async def test_capnp_request_field_type_description(): class TestInterface(testfile_capnp.TestInterface.Server): - ... + pass client = testfile_capnp.TestInterface._new_client(TestInterface()) request = client.testMethod_request() assert _request_field_type_description(request, "testUint32Field") == "uint32" assert _request_field_type_description(request, "testTextField") == "text" + + +def session_proto_value_to_python(builder): + """`labone.core.resources.session_protocol_capnp:Value` to a Python value.""" + return getattr(builder, builder.which()) + + +@dataclass +class ServerRecords: + params: list[Any] = field(default_factory=list) + + +class TestSetValue: + """Integration tests for Session node set values functionality.""" + + @pytest.fixture() + async def session_recorder(self, session_server) -> tuple[Session, ServerRecords]: + session, server = await session_server + recorder = ServerRecords() + + def mock_method(params, _): + param_builder = params.as_builder() + recorder.params.append(param_builder) + + server.setValue.side_effect = mock_method + return session, recorder + + @utils.ensure_event_loop + async def test_server_receives_correct_values_single(self, session_recorder): + session, recorder = await session_recorder + value = AnnotatedValue(value=12, path="/foo/bar") + await session.set_value(value) + assert len(recorder.params) == 1 + assert recorder.params[0].path == "/foo/bar" + assert session_proto_value_to_python(recorder.params[0].value) == 12 + + @utils.ensure_event_loop + async def test_server_receives_correct_values_gather(self, session_recorder): + session, recorder = await session_recorder + + await asyncio.gather( + session.set_value( + AnnotatedValue(value=12, path="/foo/bar"), + ), + session.set_value( + AnnotatedValue(value="text", path="/bar/foo"), + ), + session.set_value( + AnnotatedValue(value=False, path="/bar/foobar"), + ), + ) + assert len(recorder.params) == 3 + assert recorder.params[0].path == "/foo/bar" + assert session_proto_value_to_python(recorder.params[0].value) == 12 + assert recorder.params[1].path == "/bar/foo" + assert session_proto_value_to_python(recorder.params[1].value) == "text" + assert recorder.params[2].path == "/bar/foobar" + assert session_proto_value_to_python(recorder.params[2].value) == 0 + + @utils.ensure_event_loop + async def test_server_response(self, session_server): + session, server = await session_server + value = AnnotatedValue(value=123, path="/bar/foobar", timestamp=0) + + def mock_method(_, results): + results.result.ok = value.to_capnp() + + server.setValue.side_effect = mock_method + response = await session.set_value(value) + assert response == value diff --git a/tests/core/test_value.py b/tests/core/test_value.py index 03d2f47..cc9b7d2 100644 --- a/tests/core/test_value.py +++ b/tests/core/test_value.py @@ -9,115 +9,128 @@ from hypothesis import given from hypothesis import strategies as st from hypothesis.extra.numpy import arrays -from labone.core import value from labone.core.resources import session_protocol_capnp +from labone.core.shf_vector_data import VectorValueType +from labone.core.value import AnnotatedValue, _VectorElementType -class TestAnnotatedValueFromPyTypes: +class TestAnnotatedValueValue: @given(st.integers(min_value=-np.int64(), max_value=np.int64())) def test_value_from_python_types_int64(self, inp): - assert value._value_from_python_types(inp).int64 == inp + value = AnnotatedValue(value=inp, path="").to_capnp() + assert value.value.int64 == inp @pytest.mark.parametrize(("inp", "out"), [(False, 0), (True, 1)]) def test_value_from_python_types_bool_to_int64(self, inp, out): - assert value._value_from_python_types(inp).int64 == out + value = AnnotatedValue(value=inp, path="").to_capnp() + assert value.value.int64 is out @given(st.floats(allow_nan=False)) def test_value_from_python_types_double(self, inp): - assert value._value_from_python_types(inp).double == inp + value = AnnotatedValue(value=inp, path="").to_capnp() + assert value.value.double == inp def test_value_from_python_types_np_nan(self): - rval = value._value_from_python_types(np.nan).double - assert np.nan_to_num(rval) == 0.0 + value1 = AnnotatedValue(value=np.nan, path="").to_capnp() + assert np.isnan(value1.value.double) inp = complex(real=0.0, imag=np.nan.imag) - out = value._value_from_python_types(inp).complex - assert inp.real == np.nan_to_num(out.real) - assert inp.imag == np.nan.imag + value2 = AnnotatedValue(value=inp, path="").to_capnp() + assert value2.value.complex.real == inp.real + assert value2.value.complex.imag == inp.imag @given(st.complex_numbers(allow_nan=False)) def test_value_from_python_types_complex(self, inp): - obj = value._value_from_python_types(inp).complex + value = AnnotatedValue(value=inp, path="").to_capnp() expected = session_protocol_capnp.Complex( real=inp.real, imag=inp.imag, ) - assert obj.real == expected.real - assert obj.imag == expected.imag + assert value.value.complex.real == expected.real + assert value.value.complex.imag == expected.imag @given(st.text()) def test_value_from_python_types_string(self, inp): - rval = value._value_from_python_types(inp) - assert rval.string == inp + value = AnnotatedValue(value=inp, path="").to_capnp() + assert value.value.string == inp @given(st.binary()) - def test_value_from_python_types_vector_data_bytes(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.BYTE_ARRAY.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.UINT8.value - assert rval.data == vec + def test_value_from_python_types_vector_data_bytes(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.BYTE_ARRAY.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.UINT8.value + assert vec_data.data == inp @given(arrays(dtype=np.uint8, shape=(1, 2))) - def test_value_from_python_types_vector_data_uint8(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.UINT8.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_uint8(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.UINT8.value + assert vec_data.data == inp.tobytes() @given(arrays(dtype=np.uint16, shape=(1, 2))) - def test_value_from_python_types_vector_data_uint16(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.UINT16.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_uint16(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.UINT16.value + assert vec_data.data == inp.tobytes() @given(arrays(dtype=np.uint32, shape=(1, 2))) - def test_value_from_python_types_vector_data_uint32(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.UINT32.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_uint32(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.UINT32.value + assert vec_data.data == inp.tobytes() @given(arrays(dtype=(np.uint64, int), shape=(1, 2))) - def test_value_from_python_types_vector_data_uint64(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.UINT64.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_uint64(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.UINT64.value + assert vec_data.data == inp.tobytes() @given(arrays(dtype=(float, np.double), shape=(1, 2))) - def test_value_from_python_types_vector_data_double(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.DOUBLE.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_double(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.DOUBLE.value + assert vec_data.data == inp.tobytes() @given(arrays(dtype=(np.single), shape=(1, 2))) - def test_value_from_python_types_vector_data_float(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.FLOAT.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_float(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.FLOAT.value + assert vec_data.data == inp.tobytes() @given(arrays(dtype=(np.csingle), shape=(1, 2))) - def test_value_from_python_types_vector_data_complex_float(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.COMPLEX_FLOAT.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_complex_float(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.COMPLEX_FLOAT.value + assert vec_data.data == inp.tobytes() @given(arrays(dtype=(np.cdouble), shape=(1, 2))) - def test_value_from_python_types_vector_data_complex_double(self, vec): - rval = value._value_from_python_types(vec).vectorData - assert rval.valueType == value.VectorValueType.VECTOR_DATA.value - assert rval.extraHeaderInfo == 0 - assert rval.vectorElementType == value.VectorElementType.COMPLEX_DOUBLE.value - assert rval.data == vec.tobytes() + def test_value_from_python_types_vector_data_complex_double(self, inp): + value = AnnotatedValue(value=inp, path="").to_capnp() + vec_data = value.value.vectorData + assert vec_data.valueType == VectorValueType.VECTOR_DATA.value + assert vec_data.extraHeaderInfo == 0 + assert vec_data.vectorElementType == _VectorElementType.COMPLEX_DOUBLE.value + assert vec_data.data == inp.tobytes()