Fix connection secure port management.

NOTES.md

@@ -0,0 +1,15 @@
+# Project notes
+
+## Table of contents
+
+- [How to test](#how-to-test)
+
+### How to test
+
+A way (but not the sole one) to do it:
+
+1. make sure the virtual environment is activated (use `poetry shell` to activate it if necessary);
+2. create a `.env` file (you can copypaste the contents of the [example.env](./example.env) file);
+3. inject the environment variable, on Ubuntu you can do it like `export $(cat .env | xargs)`;
+4. run `docker run --name clickhouse-server -p 8123:8123 -p 9000:9000 --ulimit nofile=262144:262144 -v clickhouse_data:/var/lib/clickhouse -v clickhouse_logs:/var/log/clickhouse-server -e CLICKHOUSE_USER=ch -e CLICKHOUSE_PASSWORD=P@s5W0rD -e CLICKHOUSE_DB=test clickhouse/clickhouse-server` in a terminal (assuming the contents of the [example.env](./example.env) file are copypasted, otherwise adjust the environment variables accordingly);
+5. run `make test` in another terminal.

asynch/connection.py

@@ -15,21 +15,26 @@ def __init__(
         user: str = constants.DEFAULT_USER,
         password: str = constants.DEFAULT_PASSWORD,
         host: str = constants.DEFAULT_HOST,
-        port: int = constants.DEFAULT_PORT,
+        port: Optional[int] = None,
         database: str = constants.DEFAULT_DATABASE,
         cursor_cls=Cursor,
         echo: bool = False,
         stack_track: bool = False,
+        secure: bool = False,
         **kwargs,
     ):
+        self._dsn = dsn
+        if port is None:
+            port = constants.DEFAULT_SECURE_PORT if secure else constants.DEFAULT_PORT
         if dsn:
             config = parse_dsn(dsn)
             self._connection = ProtoConnection(**config, stack_track=stack_track, **kwargs)
-            user = config.get("user", None) or user
-            password = config.get("password", None) or password
-            host = config.get("host", None) or host
-            port = config.get("port", None) or port
-            database = config.get("database", None) or database
+            self._user = config.get("user")
+            self._password = config.get("password")
+            self._host = config.get("host")
+            self._port = config.get("port")
+            self._database = config.get("database")
+            self._secure = config.get("secure")
         else:
             self._connection = ProtoConnection(
                 host=host,
@@ -38,15 +43,15 @@ def __init__(
                 user=user,
                 password=password,
                 stack_track=stack_track,
+                secure=secure,
                 **kwargs,
             )
-        self._dsn = dsn
-        # dsn parts
-        self._user = user
-        self._password = password
-        self._host = host
-        self._port = port
-        self._database = database
+            self._user = user
+            self._password = password
+            self._host = host
+            self._port = port
+            self._database = database
+            self._secure = secure
         # connection additional settings
         self._opened: bool = False
         self._closed: bool = False
@@ -104,23 +109,23 @@ def status(self) -> str:
         raise ConnectionError(f"{self} is in an unknown state")
 
     @property
-    def host(self) -> str:
+    def host(self) -> Optional[str]:
         return self._host
 
     @property
-    def port(self) -> int:
+    def port(self) -> Optional[int]:
         return self._port
 
     @property
-    def user(self) -> str:
+    def user(self) -> Optional[str]:
         return self._user
 
     @property
-    def password(self) -> str:
+    def password(self) -> Optional[str]:
         return self._password
 
     @property
-    def database(self) -> str:
+    def database(self) -> Optional[str]:
         return self._database
 
     @property

asynch/pool.py

@@ -190,7 +190,7 @@ async def _init_connections(self, n: int, *, strict: bool = False) -> None:
             raise ValueError(msg)
         if (self._pool_size + n) > self.maxsize:
             msg = (
-                f"{self} has the {self._pool_size} connections, "
+                f"{self} has {self._pool_size} connections, "
                 f"adding {n} will exceed its maxsize ({self.maxsize})"
             )
             raise AsynchPoolError(msg)
@@ -207,7 +207,7 @@ async def _init_connections(self, n: int, *, strict: bool = False) -> None:
             for i in await asyncio.gather(*tasks, return_exceptions=True)
             if isinstance(i, Exception)
         ):
-            msg = f"failed to create the {n} connection(s) for the {self}"
+            msg = f"failed to create {n} connection(s) for {self}"
             raise AsynchPoolError(msg)
 
     async def _ensure_minsize_connections(self, *, strict: bool = False) -> None:
@@ -243,9 +243,7 @@ async def connection(self) -> AsyncIterator[Connection]:
     async def startup(self) -> "Pool":
         """Initialise the pool.
 
-        When entering the context,
-        the pool get filled with connections
-        up to the pool `minsize` value.
+        The pool is filled with `minsize` connections.
 
         :return: a pool object with `minsize` opened connections
         :rtype: Pool

asynch/proto/connection.py

@@ -77,7 +77,7 @@ def __init__(  # nosec:B107
         user: str = constants.DEFAULT_USER,
         password: str = constants.DEFAULT_PASSWORD,
         host: str = constants.DEFAULT_HOST,
-        port: int = constants.DEFAULT_PORT,
+        port: Optional[int] = None,
         database: str = constants.DEFAULT_DATABASE,
         client_name: str = constants.CLIENT_NAME,
         connect_timeout: int = constants.DBMS_DEFAULT_CONNECT_TIMEOUT_SEC,
@@ -91,21 +91,19 @@ def __init__(  # nosec:B107
         ssl_version=None,
         ca_certs=None,
         ciphers=None,
-        alt_hosts: str = None,
+        alt_hosts: str = "",
         stack_track=False,
         settings_is_important=False,
         **kwargs,
     ):
+        if port is None:
+            port = constants.DEFAULT_SECURE_PORT if secure else constants.DEFAULT_PORT
         self.stack_track = stack_track
-        if secure:
-            default_port = constants.DEFAULT_SECURE_PORT
-        else:
-            default_port = constants.DEFAULT_PORT
-        self.hosts = [(host, port or default_port)]
+        self.hosts = [(host, port)]
         if alt_hosts:
             for host in alt_hosts.split(","):
                 url = urlparse(f"{ClickhouseScheme.clickhouse}://" + host)
-                self.hosts.append((url.hostname, url.port or default_port))
+                self.hosts.append((url.hostname or host, url.port or port))
         self.database = database
         self.host = None
         self.port = None

asynch/proto/utils/dsn.py

@@ -16,8 +16,7 @@
 _TIMEOUTS: set[str] = {"connect_timeout", "send_receive_timeout", "sync_request_timeout"}
 
 
-class DSNError(Exception):
-    pass
+class DSNError(Exception): ...
 
 
 def parse_dsn(dsn: str) -> dict[str, Any]:
@@ -33,9 +32,9 @@ def parse_dsn(dsn: str) -> dict[str, Any]:
 
     :param dsn str: the DSN string
 
-    :raises DSNError: when parsing fails under the strict mode
+    :raises DSNError: DNS parsing failure(s)
 
-    :return: the dictionary of DSN string components
+    :return: DSN configuration dictionary
     :rtype: dict[str, Any]
     """
 

asynch/proto/utils/escape.py

@@ -22,28 +22,20 @@
 def escape_param(item: Any) -> str:
     if item is None:
         return "NULL"
-
     elif isinstance(item, datetime):
         return "'%s'" % item.strftime("%Y-%m-%d %H:%M:%S")
-
     elif isinstance(item, date):
         return "'%s'" % item.strftime("%Y-%m-%d")
-
     elif isinstance(item, string_types):
         return "'%s'" % "".join(escape_chars_map.get(c, c) for c in item)
-
     elif isinstance(item, list):
         return "[%s]" % ", ".join(text_type(escape_param(x)) for x in item)
-
     elif isinstance(item, tuple):
         return "(%s)" % ", ".join(text_type(escape_param(x)) for x in item)
-
     elif isinstance(item, Enum):
         return escape_param(item.value)
-
     elif isinstance(item, UUID):
         return "'%s'" % str(item)
-
     else:
         return str(item)
 

example.env

@@ -1,5 +1,5 @@
 CLICKHOUSE_HOST=localhost
 CLICKHOUSE_PORT=9000
-CLICKHOUSE_USER=default
-CLICKHOUSE_PASSWORD=default
-CLICKHOUSE_DB=default
+CLICKHOUSE_USER=ch
+CLICKHOUSE_PASSWORD=P@s5W0rD
+CLICKHOUSE_DB=test

tests/test_connection.py

@@ -3,6 +3,7 @@
 import pytest
 
 from asynch.connection import Connection
+from tests.conftest import CONNECTION_DSN
 
 HOST = "192.168.15.103"
 PORT = 10000
@@ -134,7 +135,7 @@ def test_connection_status_offline():
 
 @pytest.mark.asyncio
 async def test_connection_status_online():
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
     conn_id = id(conn)
 
     repstr = f"<{conn.__class__.__name__} object at 0x{conn_id:x}"
@@ -158,7 +159,7 @@ async def test_connection_status_online():
 
 @pytest.mark.asyncio
 async def test_async_context_manager_interface():
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
     _test_connectivity_invariant(conn=conn)
 
     async with conn:
@@ -178,7 +179,7 @@ async def test_async_context_manager_interface():
 
 @pytest.mark.asyncio
 async def test_connection_ping():
-    conn = Connection()  # default
+    conn = Connection(dsn=CONNECTION_DSN)
 
     with pytest.raises(ConnectionError):
         await conn.ping()
@@ -210,13 +211,13 @@ async def test_connection_cleanup(get_tcp_connections):
 
     # get the number of total TCP connections to the ClickHouse
     init_tcps = 0
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
     async with conn as cn:
         init_tcps = await get_tcp_connections(cn)
 
     # open-execute-close connections
     for _ in range(100):
-        async with Connection() as cn:
+        async with Connection(dsn=CONNECTION_DSN) as cn:
             async with cn.cursor() as cur:
                 await cur.execute("SELECT 1")
                 ret = await cur.fetchone()
@@ -231,15 +232,15 @@ async def test_connection_cleanup(get_tcp_connections):
 
 @pytest.mark.asyncio
 async def test_connection_close():
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
 
     # it does not break
     await conn.close()
 
     assert not conn.opened
     assert conn.closed
 
-    async with Connection() as conn:
+    async with Connection(dsn=CONNECTION_DSN) as conn:
         assert conn.opened
 
         await conn.close()

tests/test_cursors.py

@@ -58,7 +58,7 @@ async def test_fetchone(conn: Connection):
         ret = await cursor.fetchone()
         assert ret == (1,)
 
-        await cursor.execute("SELECT {val}", args={"val": 2})
+        await cursor.execute("SELECT %(val)s", args={"val": 2})
         ret = await cursor.fetchone()
         assert ret == (2,)
 
@@ -74,7 +74,7 @@ async def test_fetchall(conn: Connection):
         ret = await cursor.fetchall()
         assert ret == [(1,)]
 
-        await cursor.execute("SELECT {val}", args={"val": 2})
+        await cursor.execute("SELECT %(val)s", args={"val": 2})
         ret = await cursor.fetchall()
         assert ret == [(2,)]
 
@@ -86,7 +86,7 @@ async def test_dict_cursor(conn: Connection):
         ret = await cursor.fetchall()
         assert ret == [{"1": 1}]
 
-        await cursor.execute("SELECT {val}", args={"val": 2})
+        await cursor.execute("SELECT %(val)s", args={"val": 2})
         ret = await cursor.fetchall()
         assert ret == [{"2": 2}]
 

tests/test_pool.py

@@ -8,6 +8,7 @@
 from asynch.pool import Pool
 from asynch.proto import constants
 from asynch.proto.models.enums import PoolStatus
+from tests.conftest import CONNECTION_DSN
 
 
 def _get_pool_size(pool: Pool) -> int:
@@ -41,7 +42,7 @@ async def test_pool_repr():
     assert repr(pool) == repstr
 
     min_size, max_size = 2, 3
-    pool = Pool(minsize=min_size, maxsize=max_size)
+    pool = Pool(minsize=min_size, maxsize=max_size, dsn=CONNECTION_DSN)
     async with pool:
         repstr = (
             f"<Pool(minsize={min_size}, maxsize={max_size}) "
@@ -95,10 +96,10 @@ async def _get_pool_connection(pool: Pool):
         async with pool.connection():
             pass
 
-    async with Connection() as conn:
+    async with Connection(dsn=CONNECTION_DSN) as conn:
         init_tcps = await get_tcp_connections(conn)
 
-    async with Pool(minsize=1, maxsize=2) as pool:
+    async with Pool(minsize=1, maxsize=2, dsn=CONNECTION_DSN) as pool:
         async with pool.connection():
             assert pool.free_connections == 0
             assert pool.acquired_connections == 1
@@ -150,7 +151,7 @@ async def _get_pool_connection(pool: Pool):
         assert pool.free_connections == 2
         assert pool.acquired_connections == 0
 
-    async with Connection() as conn:
+    async with Connection(dsn=CONNECTION_DSN) as conn:
         assert init_tcps == await get_tcp_connections(conn)
 
 
@@ -171,20 +172,20 @@ async def _test_pool_connection(pool: Pool, *, selectee: Any = 42):
                 assert ret == (selectee,)
                 return selectee
 
-    async with Connection() as conn:
+    async with Connection(dsn=CONNECTION_DSN) as conn:
         init_tcps = await get_tcp_connections(conn)
 
     min_size, max_size = 10, 21
     selectees = list(range(min_size, max_size + 1))  # exceeding the maxsize
     answers = []
-    async with Pool(minsize=min_size, maxsize=max_size) as pool:
+    async with Pool(minsize=min_size, maxsize=max_size, dsn=CONNECTION_DSN) as pool:
         tasks = [
             asyncio.create_task(_test_pool_connection(pool=pool, selectee=selectee))
             for selectee in selectees
         ]
         answers = await asyncio.gather(*tasks)
 
-    async with Connection() as conn:
+    async with Connection(dsn=CONNECTION_DSN) as conn:
         noc = await get_tcp_connections(conn)
         assert noc == init_tcps
 
@@ -207,7 +208,11 @@ async def _get_answer(pool: Pool, *, raise_exc: bool = True):
                 return ret
 
     min_size, max_size = 1, 1
-    pool = Pool(minsize=min_size, maxsize=max_size)
+    pool = Pool(
+        minsize=min_size,
+        maxsize=max_size,
+        dsn=CONNECTION_DSN,
+    )
     async with pool:
         async with pool.connection() as conn:
             await conn.ping()