Implement GetMicrogridMetadata

We call the `MicrogridApiClient` method `get_microgrid_info()` to be
more forward-compatible with upcoming changes.

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

Author: llucax

src/frequenz/client/microgrid/_client.py

@@ -13,9 +13,12 @@
 from frequenz.api.microgrid.v1 import microgrid_pb2_grpc
 from frequenz.client.base import channel, client, retry, streaming
 from frequenz.client.common.microgrid.components import ComponentId
+from google.protobuf.empty_pb2 import Empty
 from typing_extensions import override
 
 from ._exception import ClientNotConnected
+from ._microgrid_info import MicrogridInfo
+from ._microgrid_info_proto import microgrid_info_from_proto
 
 DEFAULT_GRPC_CALL_TIMEOUT = 60.0
 """The default timeout for gRPC calls made by this client (in seconds)."""
@@ -122,3 +125,31 @@ async def __aexit__(
                 "Error while disconnecting from the microgrid API", exceptions
             )
         return result
+
+    async def get_microgrid_info(  # noqa: DOC502 (raises ApiClientError indirectly)
+        self,
+    ) -> MicrogridInfo:
+        """Retrieve information about the local microgrid.
+
+        This consists of information about the overall microgrid, for example, the
+        microgrid ID and its location.  It does not include information about the
+        electrical components or sensors in the microgrid.
+
+        Returns:
+            The information about 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].
+        """
+        microgrid = await client.call_stub_method(
+            self,
+            lambda: self.stub.GetMicrogridMetadata(
+                Empty(),
+                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
+            ),
+            method_name="GetMicrogridMetadata",
+        )
+
+        return microgrid_info_from_proto(microgrid.microgrid)

tests/client_test_cases/get_microgrid_info/defaults_case.py

@@ -0,0 +1,42 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test data for successful microgrid info retrieval."""
+
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import AsyncMock
+
+from frequenz.api.common.v1.microgrid import microgrid_pb2 as microgrid_common_pb2
+from frequenz.api.microgrid.v1 import microgrid_pb2
+from frequenz.client.common.microgrid import EnterpriseId, MicrogridId
+from google.protobuf.empty_pb2 import Empty
+
+from frequenz.client.microgrid import MicrogridInfo, MicrogridStatus
+
+# No client_args or client_kwargs needed for this call
+
+
+def assert_stub_method_call(stub_method: AsyncMock) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(Empty(), timeout=60.0)
+
+
+create_timestamp = datetime(2023, 1, 1, tzinfo=timezone.utc)
+grpc_response = microgrid_pb2.GetMicrogridMetadataResponse(
+    microgrid=microgrid_common_pb2.Microgrid()
+)
+
+
+def assert_client_result(result: Any) -> None:
+    """Assert that the client result matches the expected MicrogridInfo."""
+    assert result == MicrogridInfo(
+        id=MicrogridId(0),
+        enterprise_id=EnterpriseId(0),
+        name=None,
+        status=MicrogridStatus.UNSPECIFIED,
+        location=None,
+        delivery_area=None,
+        create_timestamp=datetime(1970, 1, 1, 0, 0, tzinfo=timezone.utc),
+    )
+    assert result.is_active

tests/client_test_cases/get_microgrid_info/error_case.py

@@ -0,0 +1,28 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test data for microgrid info retrieval with error."""
+
+from typing import Any
+
+from google.protobuf.empty_pb2 import Empty
+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(Empty(), 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/get_microgrid_info/full_case.py

@@ -0,0 +1,71 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Test data for successful microgrid info retrieval."""
+
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+from frequenz.api.common.v1 import location_pb2
+from frequenz.api.common.v1.grid import delivery_area_pb2
+from frequenz.api.common.v1.microgrid import microgrid_pb2 as microgrid_common_pb2
+from frequenz.api.microgrid.v1 import microgrid_pb2
+from frequenz.client.base.conversion import to_timestamp
+from frequenz.client.common.microgrid import EnterpriseId, MicrogridId
+from google.protobuf.empty_pb2 import Empty
+
+from frequenz.client.microgrid import (
+    DeliveryArea,
+    EnergyMarketCodeType,
+    Location,
+    MicrogridInfo,
+    MicrogridStatus,
+)
+
+# No client_args or client_kwargs needed for this call
+
+
+def assert_stub_method_call(stub_method: AsyncMock) -> None:
+    """Assert that the gRPC request matches the expected request."""
+    stub_method.assert_called_once_with(Empty(), timeout=60.0)
+
+
+create_timestamp = datetime(2023, 1, 1, tzinfo=timezone.utc)
+grpc_response = microgrid_pb2.GetMicrogridMetadataResponse(
+    microgrid=microgrid_common_pb2.Microgrid(
+        id=1234,
+        enterprise_id=5678,
+        name="Test Microgrid",
+        delivery_area=delivery_area_pb2.DeliveryArea(
+            code="Test Delivery Area",
+            code_type=delivery_area_pb2.EnergyMarketCodeType.ENERGY_MARKET_CODE_TYPE_EUROPE_EIC,
+        ),
+        location=location_pb2.Location(
+            latitude=37.7749, longitude=-122.4194, country_code="DE"
+        ),
+        status=microgrid_common_pb2.MICROGRID_STATUS_INACTIVE,
+        create_timestamp=to_timestamp(create_timestamp),
+    )
+)
+
+
+def assert_client_result(result: Any) -> None:
+    """Assert that the client result matches the expected MicrogridInfo."""
+    assert result == MicrogridInfo(
+        id=MicrogridId(1234),
+        enterprise_id=EnterpriseId(5678),
+        name="Test Microgrid",
+        delivery_area=DeliveryArea(
+            code="Test Delivery Area", code_type=EnergyMarketCodeType.EUROPE_EIC
+        ),
+        location=Location(
+            latitude=pytest.approx(37.7749),  # type: ignore[arg-type]
+            longitude=pytest.approx(-122.4194),  # type: ignore[arg-type]
+            country_code="DE",
+        ),
+        status=MicrogridStatus.INACTIVE,
+        create_timestamp=datetime(2023, 1, 1, tzinfo=timezone.utc),
+    )
+    assert not result.is_active

tests/test_client.py

@@ -3,6 +3,10 @@
 
 """Tests for the MicrogridApiClient class."""
 
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from pathlib import Path
 
 import pytest
 from frequenz.api.microgrid.v1 import microgrid_pb2_grpc
@@ -15,10 +19,26 @@
     MicrogridApiClient,
 )
 
-from .util import patch_client_class
+from .util import ApiClientTestCaseSpec, get_test_specs, patch_client_class
 
 # pylint: disable=protected-access
 
+TESTS_DIR = Path(__file__).parent / "client_test_cases"
+
+
+@pytest.fixture
+async def client() -> AsyncIterator[MicrogridApiClient]:
+    """Fixture that provides a MicrogridApiClient with a mock gRPC stub and channel."""
+    with patch_client_class(MicrogridApiClient, microgrid_pb2_grpc.MicrogridStub):
+        client = MicrogridApiClient(
+            "grpc://localhost:1234",
+            # Retry very fast to avoid long test times, and also not too many
+            # times to avoid test hanging forever.
+            retry_strategy=LinearBackoff(interval=0.0, jitter=0.0, limit=10),
+        )
+        async with client:
+            yield client
+
 
 @patch_client_class(MicrogridApiClient, microgrid_pb2_grpc.MicrogridStub)
 def test_init_defaults() -> None:
@@ -64,3 +84,15 @@ def test_init_with_custom_retry_strategy() -> None:
         "grpc://localhost:1234", retry_strategy=retry_strategy, connect=False
     )
     client._retry_strategy = retry_strategy  # pylint: disable=protected-access
+
+
+@pytest.mark.parametrize(
+    "spec",
+    get_test_specs("get_microgrid_info", tests_dir=TESTS_DIR),
+    ids=str,
+)
+async def test_get_microgrid_info(
+    client: MicrogridApiClient, spec: ApiClientTestCaseSpec
+) -> None:
+    """Test get_microgrid_info method."""
+    await spec.test_unary_unary_call(client, "GetMicrogridMetadata")