Expect only one username-password pair in ConnectionParameters.from_pydantic_multihost_hosts() (#72)

Author: vrslev

pyproject.toml

@@ -22,14 +22,14 @@ repository = "https://github.com/vrslev/stompman"
 [tool.uv]
 dev-dependencies = [
     "anyio~=4.4.0",
-    "mypy~=1.11.1",
+    "mypy~=1.11.2",
     "pytest-cov~=5.0.0",
     "pytest~=8.3.2",
-    "ruff~=0.5.7",
-    "uvloop~=0.19.0",
-    "hypothesis~=6.111.0",
+    "ruff~=0.6.2",
+    "uvloop~=0.20.0",
+    "hypothesis~=6.111.2",
     "polyfactory~=2.16.2",
-    "faker~=26.3.0",
+    "faker~=28.0.0",
 ]
 
 [build-system]
@@ -67,7 +67,7 @@ ignore = [
     "PLC2801",
     "PLR0913",
 ]
-extend-per-file-ignores = { "tests/*" = ["S101", "SLF001", "ARG"] }
+extend-per-file-ignores = { "tests/*" = ["S101", "SLF001", "ARG", "PLR6301"] }
 
 [tool.pytest.ini_options]
 addopts = "--cov -s -vv"

stompman/config.py

@@ -37,40 +37,56 @@ def unescaped_passcode(self) -> str:
 
     @classmethod
     def from_pydantic_multihost_hosts(cls, hosts: list[MultiHostHostLike]) -> list[Self]:
-        """Create connection parameters from a list of `MultiHostUrl` objects.
+        """Create connection parameters from `pydantic_code.MultiHostUrl.hosts()`.
 
         .. code-block:: python
-        import stompman.
-
-        ArtemisDsn = typing.Annotated[
-            pydantic_core.MultiHostUrl,
-            pydantic.UrlConstraints(
-                host_required=True,
-                allowed_schemes=["tcp"],
-            ),
-        ]
-
-        async with stompman.Client(
-            servers=stompman.ConnectionParameters.from_pydantic_multihost_hosts(
-                ArtemisDsn("tcp://lev:pass@host1:61616,lev:pass@host1:61617,lev:pass@host2:61616").hosts()
-            ),
-        ):
-            ...
+            import stompman
+
+            ArtemisDsn = typing.Annotated[
+                pydantic_core.MultiHostUrl,
+                pydantic.UrlConstraints(
+                    host_required=True,
+                    allowed_schemes=["tcp"],
+                ),
+            ]
+
+            async with stompman.Client(
+                servers=stompman.ConnectionParameters.from_pydantic_multihost_hosts(
+                    ArtemisDsn("tcp://lev:pass@host1:61616,host2:61617,host3:61618").hosts()
+                ),
+            ):
+                ...
         """
-        servers: list[Self] = []
+        all_hosts: list[tuple[str, int]] = []
+        all_credentials: list[tuple[str, str]] = []
+
         for host in hosts:
             if host["host"] is None:
                 msg = "host must be set"
                 raise ValueError(msg)
             if host["port"] is None:
                 msg = "port must be set"
                 raise ValueError(msg)
-            if host["username"] is None:
-                msg = "username must be set"
-                raise ValueError(msg)
-            if host["password"] is None:
-                msg = "password must be set"
-                raise ValueError(msg)
+            all_hosts.append((host["host"], host["port"]))
+
+            username, password = host["username"], host["password"]
+            if username is None:
+                if password is not None:
+                    msg = "password is set, username must be set"
+                    raise ValueError(msg)
+            elif password is None:
+                if username is not None:
+                    msg = "username is set, password must be set"
+                    raise ValueError(msg)
+            else:
+                all_credentials.append((username, password))
+
+        if not all_credentials:
+            msg = "username and password must be set"
+            raise ValueError(msg)
+        if len(all_credentials) != 1:
+            msg = "only one username-password pair must be set"
+            raise ValueError(msg)
 
-            servers.append(cls(host=host["host"], port=host["port"], login=host["username"], passcode=host["password"]))
-        return servers
+        login, passcode = all_credentials[0]
+        return [cls(host=host, port=port, login=login, passcode=passcode) for (host, port) in all_hosts]

tests/conftest.py

@@ -24,7 +24,7 @@ def anyio_backend(request: pytest.FixtureRequest) -> object:
 
 
 @pytest.fixture
-def mock_sleep(monkeypatch: pytest.MonkeyPatch) -> None:  # noqa: PT004
+def mock_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
     original_sleep = asyncio.sleep
     monkeypatch.setattr("asyncio.sleep", lambda _: original_sleep(0))
 

tests/test_config.py

@@ -1,25 +1,92 @@
-from typing import Any
+from typing import TypedDict
 
 import faker
 import pytest
+from polyfactory.factories.typed_dict_factory import TypedDictFactory
 
 import stompman
+from stompman.config import MultiHostHostLike
 
 
-def test_connection_parameters_from_pydantic_multihost_hosts(faker: faker.Faker) -> None:
-    full_host: dict[str, Any] = {
-        "username": faker.pystr(),
-        "password": faker.pystr(),
-        "host": faker.pystr(),
-        "port": faker.pyint(),
-    }
-    assert stompman.ConnectionParameters.from_pydantic_multihost_hosts(
-        [{**full_host, "port": index} for index in range(5)]  # type: ignore[typeddict-item]
-    ) == [
-        stompman.ConnectionParameters(full_host["host"], index, full_host["username"], full_host["password"])
-        for index in range(5)
-    ]
-
-    for key in ("username", "password", "host", "port"):
-        with pytest.raises(ValueError, match=f"{key} must be set"):
-            assert stompman.ConnectionParameters.from_pydantic_multihost_hosts([{**full_host, key: None}, full_host])  # type: ignore[typeddict-item, list-item]
+class StrictMultiHostHostLike(TypedDict):
+    username: str
+    password: str
+    host: str
+    port: int
+
+
+class StrictMultiHostHostLikeFactory(TypedDictFactory[StrictMultiHostHostLike]): ...
+
+
+class MultiHostHostLikeFactory(TypedDictFactory[MultiHostHostLike]): ...
+
+
+class TestConnectionParametersFromPydanticMultiHostHosts:
+    def test_ok(self, faker: faker.Faker) -> None:
+        hosts: list[MultiHostHostLike] = [
+            {"host": "host1", "port": 1, "username": None, "password": None},
+            {"host": "host2", "port": 2, "username": None, "password": None},
+            {"host": "host3", "port": 3, "username": None, "password": None},
+            {"host": "host4", "port": 4, "username": None, "password": None},
+        ]
+        host_with_credentials = faker.pyint(min_value=0, max_value=3)
+        hosts[host_with_credentials]["username"] = "lev"
+        hosts[host_with_credentials]["password"] = "pass"  # noqa: S105
+
+        result = stompman.ConnectionParameters.from_pydantic_multihost_hosts(hosts)
+
+        assert result == [
+            stompman.ConnectionParameters("host1", 1, "lev", "pass"),
+            stompman.ConnectionParameters("host2", 2, "lev", "pass"),
+            stompman.ConnectionParameters("host3", 3, "lev", "pass"),
+            stompman.ConnectionParameters("host4", 4, "lev", "pass"),
+        ]
+
+    def test_no_host_or_port_or_both(self, faker: faker.Faker) -> None:
+        cases: list[MultiHostHostLike] = [
+            {"host": None, "port": faker.pyint(), "username": faker.pystr(), "password": faker.pystr()},
+            {"host": faker.pystr(), "port": None, "username": faker.pystr(), "password": faker.pystr()},
+            {"host": None, "port": None, "username": faker.pystr(), "password": faker.pystr()},
+        ]
+
+        for host in cases:
+            with pytest.raises(ValueError, match="must be set"):
+                stompman.ConnectionParameters.from_pydantic_multihost_hosts([host])
+
+    def test_no_username(self, faker: faker.Faker) -> None:
+        hosts: list[MultiHostHostLike] = [
+            {"host": faker.pystr(), "port": faker.pyint(), "username": None, "password": faker.pystr()},
+            {"host": faker.pystr(), "port": faker.pyint(), "username": faker.pystr(), "password": faker.pystr()},
+        ]
+
+        with pytest.raises(ValueError, match="username must be set"):
+            stompman.ConnectionParameters.from_pydantic_multihost_hosts(hosts)
+
+    def test_no_password(self, faker: faker.Faker) -> None:
+        hosts: list[MultiHostHostLike] = [
+            {"host": faker.pystr(), "port": faker.pyint(), "username": faker.pystr(), "password": None},
+            {"host": faker.pystr(), "port": faker.pyint(), "username": faker.pystr(), "password": faker.pystr()},
+        ]
+
+        with pytest.raises(ValueError, match="password must be set"):
+            stompman.ConnectionParameters.from_pydantic_multihost_hosts(hosts)
+
+    def test_no_credentials(self, faker: faker.Faker) -> None:
+        cases: list[MultiHostHostLike] = [
+            {"host": faker.pystr(), "port": faker.pyint(), "username": None, "password": None},
+            {"host": faker.pystr(), "port": faker.pyint(), "username": None, "password": None},
+        ]
+
+        for host in cases:
+            with pytest.raises(ValueError, match="username and password must be set"):
+                stompman.ConnectionParameters.from_pydantic_multihost_hosts([host])
+
+    def test_multiple_credentials(self, faker: faker.Faker) -> None:
+        hosts: list[MultiHostHostLike] = [
+            {"host": faker.pystr(), "port": faker.pyint(), "username": None, "password": None},
+            {"host": faker.pystr(), "port": faker.pyint(), "username": faker.pystr(), "password": faker.pystr()},
+            {"host": faker.pystr(), "port": faker.pyint(), "username": faker.pystr(), "password": faker.pystr()},
+        ]
+
+        with pytest.raises(ValueError, match="only one username-password pair must be set"):
+            stompman.ConnectionParameters.from_pydantic_multihost_hosts(hosts)