Add tests for the new options

This commit also restructure the tests a bit because using all
combination for all options explodes very quickly, so instead we just
have a few meaningful combinations.

We also do a small refactoring of the code to make it easier to mock.

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

Author: llucax

src/frequenz/client/base/channel.py

@@ -236,14 +236,18 @@ def _get_contents(
                 case bytes() as default_bytes:
                     return default_bytes
                 case pathlib.Path() as file_path:
-                    pass
+                    return _read_bytes(name, file_path)
                 case unexpected:
                     assert_never(unexpected)
         case pathlib.Path() as file_path:
-            pass
+            return _read_bytes(name, file_path)
         case unexpected:
             assert_never(unexpected)
+
+
+def _read_bytes(name: str, source: pathlib.Path) -> bytes:
+    """Read the contents of a file as bytes."""
     try:
-        return file_path.read_bytes()
+        return source.read_bytes()
     except OSError as exc:
-        raise ValueError(f"Failed to read {name} from '{file_path}': {exc}") from exc
+        raise ValueError(f"Failed to read {name} from '{source}': {exc}") from exc

tests/test_channel.py

@@ -3,56 +3,168 @@
 
 """Test cases for the channel module."""
 
+import dataclasses
+import pathlib
 from unittest import mock
 
 import pytest
 from grpc import ssl_channel_credentials
 from grpc.aio import Channel
 
-from frequenz.client.base.channel import ChannelOptions, SslOptions, parse_grpc_uri
-
-VALID_URLS = [
-    ("grpc://localhost", "localhost", 9090, True),
-    ("grpc://localhost:1234", "localhost", 1234, True),
-    ("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),
-]
+from frequenz.client.base.channel import (
+    ChannelOptions,
+    SslOptions,
+    _to_bool,
+    parse_grpc_uri,
+)
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class _ValidUrlTestCase:
+    title: str
+    uri: str
+    expected_host: str
+    expected_options: ChannelOptions
+    defaults: ChannelOptions = ChannelOptions()
 
 
-@pytest.mark.parametrize("uri, host, port, ssl", VALID_URLS)
-@pytest.mark.parametrize(
-    "default_port", [None, 9090, 1234], ids=lambda x: f"default_port={x}"
-)
 @pytest.mark.parametrize(
-    "default_ssl", [None, True, False], ids=lambda x: f"default_ssl={x}"
+    "case",
+    [
+        _ValidUrlTestCase(
+            title="default",
+            uri="grpc://localhost",
+            expected_host="localhost",
+            expected_options=ChannelOptions(
+                port=9090,
+                ssl=SslOptions(
+                    enabled=True,
+                    root_certificates=None,
+                    private_key=None,
+                    certificate_chain=None,
+                ),
+            ),
+        ),
+        _ValidUrlTestCase(
+            title="default no SSL defaults",
+            uri="grpc://localhost",
+            defaults=ChannelOptions(port=2355, ssl=SslOptions(enabled=False)),
+            expected_host="localhost",
+            expected_options=ChannelOptions(
+                port=2355,
+                ssl=SslOptions(
+                    enabled=False,
+                    root_certificates=None,
+                    private_key=None,
+                    certificate_chain=None,
+                ),
+            ),
+        ),
+        _ValidUrlTestCase(
+            title="default with SSL defaults",
+            uri="grpc://localhost",
+            defaults=ChannelOptions(
+                port=2355,
+                ssl=SslOptions(
+                    enabled=True,
+                    root_certificates=None,
+                    private_key=b"key",
+                    certificate_chain=pathlib.Path("/chain"),
+                ),
+            ),
+            expected_host="localhost",
+            expected_options=ChannelOptions(
+                port=2355,
+                ssl=SslOptions(
+                    enabled=True,
+                    root_certificates=None,
+                    private_key=b"key",
+                    certificate_chain=pathlib.Path("/chain"),
+                ),
+            ),
+        ),
+        _ValidUrlTestCase(
+            title="complete no defaults",
+            uri="grpc://localhost:1234?ssl=1&ssl_root_certificates_path=/root_cert"
+            "&ssl_private_key_path=/key&ssl_certificate_chain_path=/chain",
+            expected_host="localhost",
+            expected_options=ChannelOptions(
+                port=1234,
+                ssl=SslOptions(
+                    enabled=True,
+                    root_certificates=pathlib.Path("/root_cert"),
+                    private_key=pathlib.Path("/key"),
+                    certificate_chain=pathlib.Path("/chain"),
+                ),
+            ),
+        ),
+        _ValidUrlTestCase(
+            title="complete no defaults",
+            uri="grpc://localhost:1234?ssl=1&ssl_root_certificates_path=/root_cert"
+            "&ssl_private_key_path=/key&ssl_certificate_chain_path=/chain",
+            defaults=ChannelOptions(
+                port=4444,
+                ssl=SslOptions(
+                    enabled=True,
+                    root_certificates=pathlib.Path("/root_cert_def"),
+                    private_key=b"key_def",
+                    certificate_chain=b"chain_def",
+                ),
+            ),
+            expected_host="localhost",
+            expected_options=ChannelOptions(
+                port=1234,
+                ssl=SslOptions(
+                    enabled=True,
+                    root_certificates=pathlib.Path("/root_cert"),
+                    private_key=pathlib.Path("/key"),
+                    certificate_chain=pathlib.Path("/chain"),
+                ),
+            ),
+        ),
+    ],
+    ids=lambda case: case.title,
 )
-def test_parse_uri_ok(  # pylint: disable=too-many-arguments,too-many-locals
-    uri: str,
-    host: str,
-    port: int,
-    ssl: bool,
-    default_port: int | None,
-    default_ssl: bool | None,
+def test_parse_uri_ok(  # pylint: disable=too-many-locals
+    case: _ValidUrlTestCase,
 ) -> None:
     """Test successful parsing of gRPC URIs using grpcio."""
+    uri = case.uri
+    defaults = case.defaults
+
+    expected_host = case.expected_host
+    expected_options = case.expected_options
     expected_channel = mock.MagicMock(name="mock_channel", spec=Channel)
     expected_credentials = mock.MagicMock(
         name="mock_credentials", spec=ssl_channel_credentials
     )
-    expected_port = port if f":{port}" in uri or default_port is None else default_port
-    expected_ssl = ssl if "ssl" in uri or default_ssl is None else default_ssl
-
-    defaults = ChannelOptions(port=expected_port, ssl=SslOptions(enabled=expected_ssl))
+    expected_port = (
+        expected_options.port
+        if f":{expected_options.port}" in uri or defaults.port is None
+        else defaults.port
+    )
+    expected_ssl = (
+        expected_options.ssl.enabled
+        if "ssl=" in uri or defaults.ssl.enabled is None
+        else defaults.ssl.enabled
+    )
+    expected_root_certificates = (
+        expected_options.ssl.root_certificates
+        if "ssl_root_certificates_path=" in uri
+        or defaults.ssl.root_certificates is None
+        else defaults.ssl.root_certificates
+    )
+    expected_private_key = (
+        expected_options.ssl.private_key
+        if "ssl_private_key_path=" in uri or defaults.ssl.private_key is None
+        else defaults.ssl.private_key
+    )
+    expected_certificate_chain = (
+        expected_options.ssl.certificate_chain
+        if "ssl_certificate_chain_path=" in uri
+        or defaults.ssl.certificate_chain is None
+        else defaults.ssl.certificate_chain
+    )
 
     with (
         mock.patch(
@@ -67,20 +179,58 @@ def test_parse_uri_ok(  # pylint: disable=too-many-arguments,too-many-locals
             "frequenz.client.base.channel.ssl_channel_credentials",
             return_value=expected_credentials,
         ) as ssl_channel_credentials_mock,
+        mock.patch(
+            "frequenz.client.base.channel._read_bytes",
+            return_value=b"contents",
+        ) as get_contents_mock,
     ):
         channel = parse_grpc_uri(uri, defaults)
 
     assert channel == expected_channel
-    expected_target = f"{host}:{expected_port}"
+    expected_target = f"{expected_host}:{expected_port}"
     if expected_ssl:
-        ssl_channel_credentials_mock.assert_called_once_with(root_certificates=None)
+        if isinstance(expected_root_certificates, pathlib.Path):
+            get_contents_mock.assert_any_call(
+                "root certificates",
+                expected_root_certificates,
+            )
+            expected_root_certificates = b"contents"
+        if isinstance(expected_private_key, pathlib.Path):
+            get_contents_mock.assert_any_call(
+                "private key",
+                expected_private_key,
+            )
+            expected_private_key = b"contents"
+        if isinstance(expected_certificate_chain, pathlib.Path):
+            get_contents_mock.assert_any_call(
+                "certificate chain",
+                expected_certificate_chain,
+            )
+            expected_certificate_chain = b"contents"
+        ssl_channel_credentials_mock.assert_called_once_with(
+            root_certificates=expected_root_certificates,
+            private_key=expected_private_key,
+            certificate_chain=expected_certificate_chain,
+        )
         secure_channel_mock.assert_called_once_with(
             expected_target, expected_credentials
         )
     else:
         insecure_channel_mock.assert_called_once_with(expected_target)
 
 
+@pytest.mark.parametrize("value", ["true", "on", "1", "TrUe", "On", "ON", "TRUE"])
+def test_to_bool_true(value: str) -> None:
+    """Test conversion of valid boolean values to True."""
+    assert _to_bool(value) is True
+
+
+@pytest.mark.parametrize("value", ["false", "off", "0", "FaLsE", "Off", "OFF", "FALSE"])
+def test_to_bool_false(value: str) -> None:
+    """Test conversion of valid boolean values to False."""
+    assert _to_bool(value) is False
+
+
 INVALID_URLS = [
     ("http://localhost", "Invalid scheme 'http' in the URI, expected 'grpc'"),
     ("grpc://", "Host name is missing in URI 'grpc://'"),
@@ -100,6 +250,15 @@ def test_parse_uri_ok(  # pylint: disable=too-many-arguments,too-many-locals
         "grpc://localhost:1234?ssl=1&ffl=true",
         r"Unexpected query parameters \[ffl\]",
     ),
+    (
+        "grpc://localhost:1234?ssl=0&ssl_root_certificates_path=/root&"
+        "ssl_private_key_path=/key&ssl_certificate_chain_path=/chain",
+        r"Option\(s\) ssl_root_certificates_path, ssl_private_key_path, "
+        r"ssl_certificate_chain_path found in URI 'grpc://localhost:1234\?"
+        r"ssl=0\&ssl_root_certificates_path=/root\&"
+        r"ssl_private_key_path=/key\&ssl_certificate_chain_path=/chain', "
+        r"but SSL is disabled",
+    ),
 ]