diff --git a/clients/python/src/zb_mom_ww_mxgateway/galaxy.py b/clients/python/src/zb_mom_ww_mxgateway/galaxy.py index 8acd4ac..2195f8b 100644 --- a/clients/python/src/zb_mom_ww_mxgateway/galaxy.py +++ b/clients/python/src/zb_mom_ww_mxgateway/galaxy.py @@ -140,6 +140,22 @@ class GalaxyRepositoryClient: ) seen_page_tokens.add(page_token) + async def browse_children_raw( + self, request: galaxy_pb.BrowseChildrenRequest + ) -> galaxy_pb.BrowseChildrenReply: + """Issue one BrowseChildren RPC and return the raw reply. + + Lower-level escape hatch for callers that need direct page-token control + or do not want LazyBrowseNode wrapping. Most callers should use + :py:meth:`browse` and :py:meth:`LazyBrowseNode.expand` instead. + """ + + return await self._unary( + "browse children", + self.raw_stub.BrowseChildren, + request, + ) + async def browse( self, options: BrowseChildrenOptions | None = None, diff --git a/clients/python/tests/test_galaxy.py b/clients/python/tests/test_galaxy.py index b0bdd0a..cccf3a1 100644 --- a/clients/python/tests/test_galaxy.py +++ b/clients/python/tests/test_galaxy.py @@ -507,6 +507,35 @@ async def test_browse_with_filter_forwards_to_request() -> None: assert request.historized_only is True +@pytest.mark.asyncio +async def test_browse_children_raw_returns_reply_unwrapped() -> None: + """browse_children_raw forwards the request to the stub and returns the raw reply.""" + stub = FakeGalaxyStub() + expected = _build_browse_reply( + children=[_obj(1, "Plant", is_area=True)], + child_has_children=[True], + cache_sequence=42, + ) + stub.browse_children.replies = [expected] + + async with await GalaxyRepositoryClient.connect( + endpoint="fake", + plaintext=True, + stub=stub, + ) as client: + request = galaxy_pb.BrowseChildrenRequest( + page_size=10, + tag_name_glob="Plant*", + ) + reply = await client.browse_children_raw(request) + + assert reply.cache_sequence == 42 + assert len(reply.children) == 1 + assert reply.children[0].tag_name == "Plant" + assert len(stub.browse_children.requests) == 1 + assert stub.browse_children.requests[0].tag_name_glob == "Plant*" + + class FakeGalaxyStub: def __init__(self) -> None: self.test_connection = FakeUnary([galaxy_pb.TestConnectionReply(ok=False)])