Module-Test Nodetree

 - Writing test simulating user-interaction. For this purpose,
   a realistic session mock is used
 - Fixing discovered bug in is_child_node

src/labone/nodetree/node.py

@@ -527,7 +527,8 @@ def is_child_node(
         """Checks if a node is a direct child node of this node.
 
         Children of children (etc.) will not be counted as direct children.
-        The node itself is also not counted as its child.
+        The node itself is also not counted as its child. Path redirection will be
+        taken into account.
 
         Args:
             child_node: Potential child node.
@@ -540,11 +541,10 @@ def is_child_node(
             if isinstance(child_node, MetaNode)
             else tuple(child_node)
         )
-        return (
-            bool(path_segments)
-            and self.path_segments == path_segments[:-1]
-            and path_segments[-1] in self._subtree_paths
-        )
+
+        return path_segments in {
+            self._redirect((*self.path_segments, e)) for e in self._subtree_paths
+        }
 
     @property
     def tree_manager(self) -> NodeTreeManager:

tests/nodetree/conftest.py

@@ -1,19 +1,20 @@
 from __future__ import annotations
 
+import fnmatch
 import json
 import typing as t
 from functools import cached_property
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import create_autospec
 
 import pytest
+from labone.core import KernelSession
+from labone.core.subscription import DataQueue
 
 if t.TYPE_CHECKING:
-    from labone.core.subscription import DataQueue
     from labone.nodetree.enum import NodeEnum
 from labone.nodetree.helper import (
     NormalizedPathSegment,
-    join_path,
     paths_to_nested_dict,
 )
 from labone.nodetree.node import (
@@ -25,8 +26,7 @@
     WildcardNode,
 )
 
-if t.TYPE_CHECKING:
-    from src.labone.core.value import AnnotatedValue
+from src.labone.core.value import AnnotatedValue
 from src.labone.nodetree.node import Node
 from tests.nodetree.zi_responses import zi_get_responses
 
@@ -78,41 +78,61 @@ def device_structure(data_dir) -> StructureProvider:
 
 
 @pytest.fixture()
-def session_mock():
+def session_mock(zi_structure, zi_get_responses_prop):
     """Mock a Session connection by redefining multiple methods."""
-    mock = MagicMock()
-
-    async def mock_get(path):
-        return zi_get_responses_prop[path]  # will only work for leaf nodes
-
-    async def mock_get_with_expression(*_, **__):
-        return list(zi_get_responses)  # will give a dummy answer, not the correct one!
-
-    async def mock_list_nodes_info(*_, **__):
-        return zi_structure.nodes_to_info
+    device_state = {}
+    mock = create_autospec(KernelSession)
+    subscription_queues = {}
 
     async def mock_list_nodes(path):
-        if path == "/zi/*/level":
-            return [join_path(("zi", "debug", "level"))]
-        raise NotImplementedError
+        if path[-1] != "*":
+            path = path + "/*"
+        return fnmatch.filter(
+            zi_structure.paths,
+            path,
+        )  # [p for p in zi_structure.paths if fnmatch.fnmatch(p,path)]
+
+    async def mock_get(path):
+        return device_state.get(path, zi_get_responses_prop[path])
 
     async def mock_set(annotated_value, *_, **__):
-        """will NOT change state for later get requests!"""
+        device_state[annotated_value.path] = annotated_value
+        if annotated_value.path in subscription_queues:
+            await subscription_queues[annotated_value.path].put(annotated_value)
         return annotated_value
 
-    # set_with_expression
+    async def mock_get_with_expression(path, **__):
+        paths = await mock_list_nodes(path)
+        return [await mock_get(p) for p in paths]
+
     async def mock_set_with_expression(ann_value, *_, **__):
-        """will NOT change state for later get requests!"""
-        if ann_value.path == "/zi/*/level":
-            return [ann_value]
-        raise NotImplementedError
+        paths = await mock_list_nodes(ann_value.path)
+        return [
+            await mock_set(AnnotatedValue(path=p, value=ann_value.value)) for p in paths
+        ]
+
+    async def mock_list_nodes_info(path="*"):
+        return {
+            p: zi_structure.nodes_to_info[p]
+            for p in zi_structure.paths
+            if fnmatch.fnmatch(p, path)
+        }
+
+    async def subscribe(path, **__):
+        subscription_queues[path] = DataQueue(
+            path=path,
+            register_function=lambda _: None,
+        )
+        return subscription_queues[path]
 
+    mock.list_nodes.side_effect = mock_list_nodes
+    mock.list_nodes_info.side_effect = mock_list_nodes_info
     mock.get.side_effect = mock_get
     mock.get_with_expression.side_effect = mock_get_with_expression
-    mock.list_nodes_info.side_effect = mock_list_nodes_info
     mock.set.side_effect = mock_set
-    mock.list_nodes.side_effect = mock_list_nodes
     mock.set_with_expression.side_effect = mock_set_with_expression
+    mock.subscribe.side_effect = subscribe
+
     return mock
 
 

tests/nodetree/test_node_unit.py

@@ -10,6 +10,7 @@
 from labone.nodetree.helper import WILDCARD, UndefinedStructure, join_path, split_path
 from labone.nodetree.node import (
     LeafNode,
+    MetaNode,
     Node,
     NodeTreeManager,
     PartialNode,
@@ -75,18 +76,31 @@ def test_redirect(self, path_aliases, start_path, expected, zi):
         assert zi._redirect(start_path) == expected
 
     @pytest.mark.parametrize(
-        ("start_path_segments", "subtree_structure", "test_path_segments", "expected"),
+        (
+            "start_path_segments",
+            "subtree_structure",
+            "redirect_dict",
+            "test_path_segments",
+            "expected",
+        ),
         [
-            ((), [], (), False),
-            (("a",), [], ("a",), False),
-            ((), ["a"], ("a",), True),
-            (("a", "b"), [], ("a", "b"), False),
-            (("a",), ["c"], ("a", "c"), True),
-            (("a", "b"), ["c"], ("a", "b", "c"), True),
-            (("a", "b"), ["c"], ("a", "b", "d"), False),
-            (("a", "b", "c", "d"), ["d"], ("a", "b", "c", "d", "d"), True),
-            (("a",), ["c"], ("a", "c", "d"), False),
-            (("a",), ["c"], ("a", "c", "d", "e", "f", "g"), False),
+            ((), [], {}, (), False),
+            (("a",), [], {}, ("a",), False),
+            ((), ["a"], {}, ("a",), True),
+            (("a", "b"), [], {}, ("a", "b"), False),
+            (("a",), ["c"], {}, ("a", "c"), True),
+            (("a", "b"), ["c"], {}, ("a", "b", "c"), True),
+            (("a", "b"), ["c"], {}, ("a", "b", "d"), False),
+            (("a", "b", "c", "d"), ["d"], {}, ("a", "b", "c", "d", "d"), True),
+            (("a",), ["c"], {}, ("a", "c", "d"), False),
+            (("a",), ["c"], {}, ("a", "c", "d", "e", "f", "g"), False),
+            (
+                ("x", "y"),
+                ["c"],
+                {("x", "y", "c"): ("a", "b", "c", "d", "e")},
+                ("a", "b", "c", "d", "e"),
+                True,
+            ),
         ],
     )
     @pytest.mark.parametrize("as_node", [True, False])
@@ -95,13 +109,28 @@ def test_is_child_node(  # noqa: PLR0913
         as_node,
         start_path_segments,
         subtree_structure,
+        redirect_dict,
         test_path_segments,
         expected,
     ):
         node = MockMetaNode(start_path_segments)
-        node._subtree_paths = subtree_structure
-        sub = MockMetaNode(test_path_segments) if as_node else test_path_segments
-        assert node.is_child_node(sub) == expected
+        node._subtree_paths = {s: None for s in subtree_structure}
+
+        def new_redirect(_, path):
+            if path in redirect_dict:
+                return redirect_dict[path]
+            return path
+
+        with patch.object(
+            MetaNode,
+            "_redirect",
+            autospec=True,
+            side_effect=new_redirect,
+        ) as redirect_mock:
+            sub = MockMetaNode(test_path_segments) if as_node else test_path_segments
+            assert node.is_child_node(sub) == expected
+            for s in subtree_structure:
+                redirect_mock.assert_any_call(node, (*start_path_segments, s))
 
     def test_path(self):
         with patch("labone.nodetree.node.join_path", return_value="/") as join_mock:

tests/nodetree/test_nodetree_module.py

@@ -0,0 +1,47 @@
+import asyncio
+
+import pytest
+from labone.core import KernelSession, ServerInfo, ZIKernelInfo
+from labone.nodetree import construct_nodetree
+
+USE_REAL_SESSION = 0
+
+
+@pytest.mark.asyncio()
+async def test_use_case(session_mock):
+    if USE_REAL_SESSION:
+        session = await KernelSession.create(
+            kernel_info=ZIKernelInfo(),
+            server_info=ServerInfo(host="localhost", port=8004),
+        )
+    zi = await construct_nodetree(session=session if USE_REAL_SESSION else session_mock)
+
+    assert (await zi.debug.level(5)).value == 5
+    assert (await zi.debug.level()).value == 5
+
+    result_node = await zi.devices()
+
+    assert result_node.visible.value == ""
+    assert result_node.connected.value == ""
+
+    await zi["*"].groups[0].status(0)
+    result_node = await zi["*/groups"]()
+
+    for sub in result_node:
+        assert sub in result_node
+
+    # set to this before, see above
+    result_node[0][0].status = 0
+
+    # set to this before, see above
+    await zi.debug.level.wait_for_state_change(5, timeout=0.01)
+
+    with pytest.raises(asyncio.TimeoutError):
+        await zi.debug.level.wait_for_state_change(3, timeout=0.01)
+
+    node = zi.debug.level
+
+    await asyncio.gather(
+        node.wait_for_state_change(3, timeout=0.01),
+        node(3),
+    )