Add a GrpcStatusError subclass for unrecognized status codes

When we receive a gRPC status code that we don't recognize, we raise a
`GrpcStatusError` instead of the base class for all gRPC status errors.

This will allow users to more easily differentiate between known and
unknown status codes.

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

Author: llucax

src/frequenz/client/microgrid/__init__.py

@@ -46,6 +46,7 @@
     ResourceExhausted,
     ServiceUnavailable,
     UnknownError,
+    UnrecognizedGrpcStatus,
 )
 from ._metadata import Location, Metadata
 
@@ -87,4 +88,5 @@
     "ResourceExhausted",
     "ServiceUnavailable",
     "UnknownError",
+    "UnrecognizedGrpcStatus",
 ]

src/frequenz/client/microgrid/_exception.py

@@ -58,15 +58,13 @@ def from_grpc_error(
         server_url: str,
         operation: str,
         grpc_error: grpclib.GRPCError,
-        retryable: bool = True,
     ) -> GrpcStatusError:
         """Create an instance of the appropriate subclass from a gRPC error.
 
         Args:
             server_url: The URL of the server that returned the error.
             operation: The operation that caused the error.
             grpc_error: The gRPC error to convert.
-            retryable: Whether retrying the operation might succeed.
 
         Returns:
             An instance of
@@ -104,12 +102,10 @@ def __call__(
             return ctor(
                 server_url=server_url, operation=operation, grpc_error=grpc_error
             )
-        return GrpcStatusError(
+        return UnrecognizedGrpcStatus(
             server_url=server_url,
             operation=operation,
-            description="Got an unrecognized status code",
             grpc_error=grpc_error,
-            retryable=retryable,
         )
 
 
@@ -156,6 +152,28 @@ def __init__(  # pylint: disable=too-many-arguments
         """The original gRPC error."""
 
 
+class UnrecognizedGrpcStatus(GrpcStatusError):
+    """The gRPC server returned an unrecognized status code."""
+
+    def __init__(
+        self, *, server_url: str, operation: str, grpc_error: grpclib.GRPCError
+    ) -> None:
+        """Create a new instance.
+
+        Args:
+            server_url: The URL of the server that returned the error.
+            operation: The operation that caused the error.
+            grpc_error: The gRPC error originating this exception.
+        """
+        super().__init__(
+            server_url=server_url,
+            operation=operation,
+            description="Got an unrecognized status code",
+            grpc_error=grpc_error,
+            retryable=True,  # We don't know so we assume it's retryable
+        )
+
+
 class OperationCancelled(GrpcStatusError):
     """The operation was cancelled."""
 

tests/test_exception.py

@@ -29,6 +29,7 @@
     ResourceExhausted,
     ServiceUnavailable,
     UnknownError,
+    UnrecognizedGrpcStatus,
 )
 
 
@@ -41,6 +42,12 @@ def __call__(
 
 
 ERROR_TUPLES: list[tuple[type[GrpcStatusError], grpclib.Status, str, bool]] = [
+    (
+        UnrecognizedGrpcStatus,
+        mock.MagicMock(name="unknown_status"),
+        "Got an unrecognized status code",
+        True,
+    ),
     (
         OperationCancelled,
         grpclib.Status.CANCELLED,
@@ -200,16 +207,7 @@ def test_client_error() -> None:
 
 
 @pytest.mark.parametrize(
-    "exception_class, grpc_status, expected_description, retryable",
-    ERROR_TUPLES
-    + [
-        (
-            GrpcStatusError,
-            mock.MagicMock(name="unknown_status"),
-            "Got an unrecognized status code",
-            True,
-        )
-    ],
+    "exception_class, grpc_status, expected_description, retryable", ERROR_TUPLES
 )
 def test_from_grpc_error(
     exception_class: type[GrpcStatusError],