Implement ListConnections

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/client/microgrid/_client.py

@@ -28,6 +28,8 @@
 from .component._category import ComponentCategory
 from .component._component import Component
 from .component._component_proto import component_from_proto
+from .component._connection import ComponentConnection
+from .component._connection_proto import component_connection_from_proto
 from .component._types import ComponentTypes
 from .metrics._bounds import Bounds
 from .metrics._metric import Metric
@@ -221,6 +223,66 @@ async def list_components(  # noqa: DOC502 (raises ApiClientError indirectly)
 
         return map(component_from_proto, component_list.components)
 
+    async def list_connections(  # noqa: DOC502 (raises ApiClientError indirectly)
+        self,
+        *,
+        sources: Iterable[ComponentId | Component] = (),
+        destinations: Iterable[ComponentId | Component] = (),
+    ) -> Iterable[ComponentConnection]:
+        """Fetch all the connections present in the local microgrid.
+
+        Electrical components are a part of a microgrid's electrical infrastructure
+        are can be connected to each other to form an electrical circuit, which can
+        then be represented as a graph.
+
+        The direction of a connection is always away from the grid endpoint, i.e.
+        aligned with the direction of positive current according to the passive sign
+        convention: https://en.wikipedia.org/wiki/Passive_sign_convention
+
+        The request may be filtered by `source`/`destination` component(s) of individual
+        connections.  If provided, the `sources` and `destinations` filters have an
+        `AND` relationship between each other, meaning that they are applied serially,
+        but an `OR` relationship with other elements in the same list.
+
+        Example:
+            If `sources = {1, 2, 3}`, and `destinations = {4,
+            5, 6}`, then the result should have all the connections where:
+
+            * Each `source` component ID is either `1`, `2`, OR `3`; **AND**
+            * Each `destination` component ID is either `4`, `5`, OR `6`.
+
+        Args:
+            sources: The component from which the connections originate.
+            destinations: The component at which the connections terminate.
+
+        Returns:
+            Iterator whose elements are all the connections in the local microgrid.
+
+        Raises:
+            ApiClientError: If the are any errors communicating with the Microgrid API,
+                most likely a subclass of
+                [GrpcError][frequenz.client.microgrid.GrpcError].
+        """
+        connection_list = await client.call_stub_method(
+            self,
+            lambda: self.stub.ListConnections(
+                microgrid_pb2.ListConnectionsRequest(
+                    starts=map(_get_component_id, sources),
+                    ends=map(_get_component_id, destinations),
+                ),
+                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
+            ),
+            method_name="ListConnections",
+        )
+
+        return (
+            conn
+            for conn in map(
+                component_connection_from_proto, connection_list.connections
+            )
+            if conn is not None
+        )
+
     async def set_component_power_active(  # noqa: DOC502 (raises ApiClientError indirectly)
         self,
         component: ComponentId | Component,

tests/client_test_cases/list_connections/empty_case.py

@@ -0,0 +1,25 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test list_connections with no connections: result should be empty."""
+
+from typing import Any
+
+from frequenz.api.microgrid.v1 import microgrid_pb2
+
+# No client_args or client_kwargs needed for this call
+
+
+def assert_stub_method_call(stub_method: Any) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(
+        microgrid_pb2.ListConnectionsRequest(starts=[], ends=[]), timeout=60.0
+    )
+
+
+grpc_response = microgrid_pb2.ListConnectionsResponse(connections=[])
+
+
+def assert_client_result(result: Any) -> None:  # noqa: D103
+    """Assert that the client result is an empty list."""
+    assert not list(result)

tests/client_test_cases/list_connections/error_case.py

@@ -0,0 +1,30 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test list_connections with error."""
+
+from typing import Any
+
+from frequenz.api.microgrid.v1 import microgrid_pb2
+from grpc import StatusCode
+
+from frequenz.client.microgrid import PermissionDenied
+from tests.util import make_grpc_error
+
+# No client_args or client_kwargs needed for this call
+
+
+def assert_stub_method_call(stub_method: Any) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(
+        microgrid_pb2.ListConnectionsRequest(starts=[], ends=[]), timeout=60.0
+    )
+
+
+grpc_response = make_grpc_error(StatusCode.PERMISSION_DENIED)
+
+
+def assert_client_exception(exception: Exception) -> None:
+    """Assert that the client exception matches the expected error."""
+    assert isinstance(exception, PermissionDenied)
+    assert exception.grpc_error == grpc_response

tests/client_test_cases/list_connections/mixed_filters_case.py

@@ -0,0 +1,43 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test list_connections with mixed ComponentId and Component objects."""
+
+from datetime import datetime, timezone
+from typing import Any
+
+from frequenz.api.microgrid.v1 import microgrid_pb2
+from frequenz.client.common.microgrid import MicrogridId
+from frequenz.client.common.microgrid.components import ComponentId
+
+from frequenz.client.microgrid.component import (
+    GridConnectionPoint,
+    Meter,
+)
+
+# Mix ComponentId and Component objects
+grid_component = GridConnectionPoint(
+    id=ComponentId(1), microgrid_id=MicrogridId(1), rated_fuse_current=10_000
+)
+meter_component = Meter(id=ComponentId(4), microgrid_id=MicrogridId(1))
+
+client_kwargs = {
+    "sources": [grid_component, ComponentId(2)],
+    "destinations": [ComponentId(3), meter_component],
+}
+
+
+def assert_stub_method_call(stub_method: Any) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(
+        microgrid_pb2.ListConnectionsRequest(starts=[1, 2], ends=[3, 4]), timeout=60.0
+    )
+
+
+lifetime_start = datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+grpc_response = microgrid_pb2.ListConnectionsResponse(connections=[])
+
+
+def assert_client_result(actual_result: Any) -> None:
+    """Assert that the client result matches the expected connections list."""
+    assert list(actual_result) == []

tests/client_test_cases/list_connections/success_case.py

@@ -0,0 +1,57 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test data for successful connection listing."""
+
+from datetime import datetime, timezone
+from typing import Any
+
+from frequenz.api.common.v1.microgrid import lifetime_pb2
+from frequenz.api.common.v1.microgrid.components import components_pb2
+from frequenz.api.microgrid.v1 import microgrid_pb2
+from frequenz.client.base.conversion import to_timestamp
+from frequenz.client.common.microgrid.components import ComponentId
+
+from frequenz.client.microgrid import Lifetime
+from frequenz.client.microgrid.component import ComponentConnection
+
+# No client_args or client_kwargs needed for this call
+
+
+def assert_stub_method_call(stub_method: Any) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(
+        microgrid_pb2.ListConnectionsRequest(starts=[], ends=[]), timeout=60.0
+    )
+
+
+lifetime_start = datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+grpc_response = microgrid_pb2.ListConnectionsResponse(
+    connections=[
+        components_pb2.ComponentConnection(
+            source_component_id=1, destination_component_id=2
+        ),
+        components_pb2.ComponentConnection(
+            source_component_id=2,
+            destination_component_id=3,
+            operational_lifetime=lifetime_pb2.Lifetime(
+                start_timestamp=to_timestamp(lifetime_start)
+            ),
+        ),
+    ]
+)
+
+
+def assert_client_result(actual_result: Any) -> None:
+    """Assert that the client result matches the expected connections list."""
+    assert list(actual_result) == [
+        ComponentConnection(
+            source=ComponentId(1),
+            destination=ComponentId(2),
+        ),
+        ComponentConnection(
+            source=ComponentId(2),
+            destination=ComponentId(3),
+            operational_lifetime=Lifetime(start=lifetime_start),
+        ),
+    ]

tests/test_client.py

@@ -111,6 +111,19 @@ async def test_list_components(
     await spec.test_unary_unary_call(client, "ListComponents")
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "spec",
+    get_test_specs("list_connections", tests_dir=TESTS_DIR),
+    ids=str,
+)
+async def test_list_connections(
+    client: MicrogridApiClient, spec: ApiClientTestCaseSpec
+) -> None:
+    """Test list_connections method."""
+    await spec.test_unary_unary_call(client, "ListConnections")
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "spec",