feat(tests): Tests electrical component connections

Adds tests for list_microgrid_electrical_component_connections, covering success, empty results, and error cases to improve reliability.

Signed-off-by: eduardiazf <[email protected]>

Author: eduardiazf

src/frequenz/client/assets/__init__.py

@@ -5,6 +5,7 @@
 
 from ._client import AssetsApiClient
 from ._delivery_area import DeliveryArea, EnergyMarketCodeType
+from ._lifetime import Lifetime
 from ._location import Location
 from ._microgrid import Microgrid, MicrogridStatus
 
@@ -15,4 +16,5 @@
     "Microgrid",
     "MicrogridStatus",
     "Location",
+    "Lifetime",
 ]

src/frequenz/client/assets/_lifetime.py

@@ -13,15 +13,15 @@ class Lifetime:
     """An active operational period of a microgrid asset.
 
     Warning:
-        The [`end`][frequenz.client.microgrid.Lifetime.end] timestamp indicates that the
+        The [`end`][frequenz.client.assets.Lifetime.end] timestamp indicates that the
         asset has been permanently removed from the system.
     """
 
     start: datetime | None = None
     """The moment when the asset became operationally active.
 
     If `None`, the asset is considered to be active in any past moment previous to the
-    [`end`][frequenz.client.microgrid.Lifetime.end].
+    [`end`][frequenz.client.assets.Lifetime.end].
     """
 
     end: datetime | None = None

src/frequenz/client/assets/electrical_component/__init__.py

@@ -14,6 +14,7 @@
 from ._capacitor_bank import CapacitorBank
 from ._category import ElectricalComponentCategory
 from ._chp import Chp
+from ._connection import ComponentConnection
 from ._converter import Converter
 from ._crypto_miner import CryptoMiner
 from ._electrical_component import ElectricalComponent
@@ -87,4 +88,5 @@
     "StaticTransferSwitch",
     "UninterruptiblePowerSupply",
     "WindTurbine",
+    "ComponentConnection",
 ]

tests/client_test_cases/list_microgrid_electrical_component_connections/empty_case.py

@@ -0,0 +1,31 @@
+# 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.assets.v1 import assets_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(
+        assets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
+            microgrid_id=1234, source_component_ids=[], destination_component_ids=[]
+        ),
+        timeout=60.0,
+    )
+
+
+client_args = (1234,)
+grpc_response = assets_pb2.ListMicrogridElectricalComponentConnectionsResponse(
+    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_microgrid_electrical_component_connections/error_case.py

@@ -0,0 +1,32 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test list_electrical_components with error."""
+
+from typing import Any
+
+from frequenz.api.assets.v1 import assets_pb2
+from grpc import StatusCode
+
+from frequenz.client.assets.exceptions import PermissionDenied
+from tests.util import make_grpc_error
+
+
+def assert_stub_method_call(stub_method: Any) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(
+        assets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
+            microgrid_id=1234
+        ),
+        timeout=60.0,
+    )
+
+
+client_args = (1234,)
+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_microgrid_electrical_component_connections/success_case.py

@@ -0,0 +1,61 @@
+# 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.assets.v1 import assets_pb2
+from frequenz.api.common.v1alpha8.microgrid import lifetime_pb2
+from frequenz.api.common.v1alpha8.microgrid.electrical_components import (
+    electrical_components_pb2,
+)
+from frequenz.client.base.conversion import to_timestamp
+from frequenz.client.common.microgrid.components import ComponentId
+
+from frequenz.client.assets import Lifetime
+from frequenz.client.assets.electrical_component import ComponentConnection
+
+
+def assert_stub_method_call(stub_method: Any) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(
+        assets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
+            microgrid_id=1234, source_component_ids=[], destination_component_ids=[]
+        ),
+        timeout=60.0,
+    )
+
+
+client_args = (1234,)
+lifetime_start = datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+grpc_response = assets_pb2.ListMicrogridElectricalComponentConnectionsResponse(
+    connections=[
+        electrical_components_pb2.ElectricalComponentConnection(
+            source_electrical_component_id=1, destination_electrical_component_id=2
+        ),
+        electrical_components_pb2.ElectricalComponentConnection(
+            source_electrical_component_id=2,
+            destination_electrical_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

@@ -50,3 +50,20 @@ async def test_list_microgrid_electrical_components(
 ) -> None:
     """Test list_microgrid_electrical_components method."""
     await spec.test_unary_unary_call(client, "ListMicrogridElectricalComponents")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "spec",
+    get_test_specs(
+        "list_microgrid_electrical_component_connections", tests_dir=TESTS_DIR
+    ),
+    ids=str,
+)
+async def test_list_connections(
+    client: AssetsApiClient, spec: ApiClientTestCaseSpec
+) -> None:
+    """Test list_connections method."""
+    await spec.test_unary_unary_call(
+        client, "ListMicrogridElectricalComponentConnections"
+    )