Add a new module channel with a function to parse URIs

This new module provides a function to parse URIs to create `grpclib`
client `Channel` instances. For now the URI provides only very basic
options, but it can be extended in the future.

Signed-off-by: Leandro Lucarella <luca-frequenz@llucax.com>

RELEASE_NOTES.md

@@ -11,6 +11,7 @@
 ## New Features
 
 - `GrpcStreamBroadcaster` is now compatible with both `grpcio` and `grpclib` implementations of gRPC. Just install `frequenz-client-base[grpcio]` or `frequenz-client-base[grpclib]` to use the desired implementation and everything should work as expected.
+- A new module `channel` with a function to parse URIs to create `grpclib` client `Channel` instances.
 
 ## Bug Fixes
 

src/frequenz/client/base/channel.py

@@ -0,0 +1,75 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Handling of gRPC channels."""
+
+from urllib.parse import parse_qs, urlparse
+
+from grpclib.client import Channel
+
+
+def _to_bool(value: str) -> bool:
+    value = value.lower()
+    if value in ("true", "on", "1"):
+        return True
+    if value in ("false", "off", "0"):
+        return False
+    raise ValueError(f"Invalid boolean value '{value}'")
+
+
+def parse_grpc_uri(uri: str, /, *, default_port: int = 9090) -> Channel:
+    """Create a grpclib client channel from a URI.
+
+    The URI must have the following format:
+
+    ```
+    grpc://hostname[:port][?ssl=false]
+    ```
+
+    A few things to consider about URI components:
+
+    - If any other components are present in the URI, a [`ValueError`][] is raised.
+    - If the port is omitted, the `default_port` is used.
+    - If a query parameter is passed many times, the last value is used.
+    - The only supported query parameter is `ssl`, which must be a boolean value and
+      defaults to `false`.
+    - Boolean query parameters can be specified with the following values
+      (case-insensitive): `true`, `1`, `on`, `false`, `0`, `off`.
+
+    Args:
+        uri: The gRPC URI specifying the connection parameters.
+        default_port: The default port number to use if the URI does not specify one.
+
+    Returns:
+        A grpclib client channel object.
+
+    Raises:
+        ValueError: If the URI is invalid or contains unexpected components.
+    """
+    parsed_uri = urlparse(uri)
+    if parsed_uri.scheme != "grpc":
+        raise ValueError(
+            f"Invalid scheme '{parsed_uri.scheme}' in the URI, expected 'grpc'", uri
+        )
+    if not parsed_uri.hostname:
+        raise ValueError(f"Host name is missing in URI '{uri}'", uri)
+    for attr in ("path", "fragment", "params", "username", "password"):
+        if getattr(parsed_uri, attr):
+            raise ValueError(
+                f"Unexpected {attr} '{getattr(parsed_uri, attr)}' in the URI '{uri}'",
+                uri,
+            )
+
+    options = {k: v[-1] for k, v in parse_qs(parsed_uri.query).items()}
+    ssl = _to_bool(options.pop("ssl", "false"))
+    if options:
+        raise ValueError(
+            f"Unexpected query parameters {options!r} in the URI '{uri}'",
+            uri,
+        )
+
+    return Channel(
+        host=parsed_uri.hostname,
+        port=parsed_uri.port or default_port,
+        ssl=ssl,
+    )

tests/test_channel.py

@@ -0,0 +1,85 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Test cases for the channel module."""
+
+import unittest.mock
+from dataclasses import dataclass
+
+import pytest
+
+from frequenz.client.base.channel import parse_grpc_uri
+
+
+@dataclass(frozen=True)
+class _FakeChannel:
+    host: str
+    port: int
+    ssl: bool
+
+
+@pytest.mark.parametrize(
+    "uri, host, port, ssl",
+    [
+        ("grpc://localhost", "localhost", 9090, False),
+        ("grpc://localhost:1234", "localhost", 1234, False),
+        ("grpc://localhost:1234?ssl=true", "localhost", 1234, True),
+        ("grpc://localhost:1234?ssl=false", "localhost", 1234, False),
+        ("grpc://localhost:1234?ssl=1", "localhost", 1234, True),
+        ("grpc://localhost:1234?ssl=0", "localhost", 1234, False),
+        ("grpc://localhost:1234?ssl=on", "localhost", 1234, True),
+        ("grpc://localhost:1234?ssl=off", "localhost", 1234, False),
+        ("grpc://localhost:1234?ssl=TRUE", "localhost", 1234, True),
+        ("grpc://localhost:1234?ssl=FALSE", "localhost", 1234, False),
+        ("grpc://localhost:1234?ssl=ON", "localhost", 1234, True),
+        ("grpc://localhost:1234?ssl=OFF", "localhost", 1234, False),
+        ("grpc://localhost:1234?ssl=0&ssl=1", "localhost", 1234, True),
+        ("grpc://localhost:1234?ssl=1&ssl=0", "localhost", 1234, False),
+    ],
+)
+def test_parse_uri_ok(
+    uri: str,
+    host: str,
+    port: int,
+    ssl: bool,
+) -> None:
+    """Test successful parsing of gRPC URIs."""
+    with unittest.mock.patch(
+        "frequenz.client.base.channel.Channel",
+        return_value=_FakeChannel(host, port, ssl),
+    ):
+        channel = parse_grpc_uri(uri)
+
+    assert isinstance(channel, _FakeChannel)
+    assert channel.host == host
+    assert channel.port == port
+    assert channel.ssl == ssl
+
+
+@pytest.mark.parametrize(
+    "uri, error_msg",
+    [
+        ("http://localhost", "Invalid scheme 'http' in the URI, expected 'grpc'"),
+        ("grpc://", "Host name is missing in URI 'grpc://'"),
+        ("grpc://localhost:1234?ssl=invalid", "Invalid boolean value 'invalid'"),
+        ("grpc://localhost:1234?ssl=1&ssl=invalid", "Invalid boolean value 'invalid'"),
+        ("grpc://:1234", "Host name is missing"),
+        ("grpc://host:1234;param", "Port could not be cast to integer value"),
+        ("grpc://host:1234/path", "Unexpected path '/path'"),
+        ("grpc://host:1234#frag", "Unexpected fragment 'frag'"),
+        ("grpc://user@host:1234", "Unexpected username 'user'"),
+        ("grpc://:pass@host:1234?user:pass", "Unexpected password 'pass'"),
+        (
+            "grpc://localhost?ssl=1&ssl=1&ssl=invalid",
+            "Invalid boolean value 'invalid'",
+        ),
+        (
+            "grpc://localhost:1234?ssl=1&ffl=true",
+            "Unexpected query parameters {'ffl': 'true'}",
+        ),
+    ],
+)
+def test_parse_uri_error(uri: str, error_msg: str) -> None:
+    """Test parsing of invalid gRPC URIs."""
+    with pytest.raises(ValueError, match=error_msg):
+        parse_grpc_uri(uri)