Reject out-of-range ports at URL-parse time

SQLAlchemy's URL parser rejects a non-integer port, but ``URL.create(
port=0)`` and similar integer-but-illegal values flow through
untouched. Mirror the strict validation already applied to URL query
parameters: require ``1 <= port <= 65535`` and raise ``ArgumentError``
otherwise so typos fail before a pool is built rather than at first
connect.

Author: antoineleclair

src/sqlalchemydqlite/base.py

@@ -216,6 +216,13 @@ def create_connect_args(self, url: URL) -> tuple[list[Any], dict[str, Any]]:
         any pool is built.
         """
         host = url.host or "localhost"
+        if url.port is not None and not (1 <= url.port <= 65535):
+            # SQLAlchemy's URL parser normally rejects non-integer ports
+            # but will happily carry an integer outside the legal TCP
+            # range if the URL was constructed via ``URL.create(port=…)``.
+            # Catch here so typos fail at URL-parse time, matching the
+            # validation style used for known query parameters.
+            raise ArgumentError(f"dqlite URL port {url.port!r} is out of the valid 1..65535 range")
         port = url.port or 9001
         database = url.database or "default"
 

tests/test_dialect_dialect_config.py

@@ -98,6 +98,28 @@ def test_max_total_rows_unparseable_raises(self) -> None:
         with pytest.raises(ArgumentError, match="int"):
             dialect.create_connect_args(url)
 
+    @pytest.mark.parametrize("bad_port", [0, -1, 65536, 70000])
+    def test_invalid_port_raises(self, bad_port: int) -> None:
+        """A ``URL.create(port=…)`` call that smuggles an out-of-range
+        port past SQLAlchemy's own parser must still fail at
+        ``create_connect_args`` time, matching the policy used for URL
+        query parameters."""
+        from sqlalchemy.engine import URL
+
+        dialect = DqliteDialect()
+        url = URL.create("dqlite", host="host", port=bad_port, database="db")
+        with pytest.raises(ArgumentError, match="out of the valid"):
+            dialect.create_connect_args(url)
+
+    @pytest.mark.parametrize("good_port", [1, 9001, 65535])
+    def test_valid_port_accepted(self, good_port: int) -> None:
+        from sqlalchemy.engine import URL
+
+        dialect = DqliteDialect()
+        url = URL.create("dqlite", host="host", port=good_port, database="db")
+        _, kwargs = dialect.create_connect_args(url)
+        assert kwargs["address"] == f"host:{good_port}"
+
     def test_timeout_and_max_total_rows_together(self) -> None:
         dialect = DqliteDialect()
         url = make_url("dqlite://host:19001/db?timeout=3.5&max_total_rows=250")