feat(tests): Add tests for listing electrical component connections

    This commit introduces new test cases for the `list_microgrid_electrical_component_connections` method, covering scenarios for successful connections, empty results, and error handling. The tests ensure that the API behaves as expected under various conditions, enhancing the reliability of the electrical component connection functionality.

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