Tests for Session set_value

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))

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

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()