Use utility function to raise for deprecated parameters.

Author: tillahoffmann

arangodb/testcontainers/arangodb/__init__.py

@@ -4,6 +4,7 @@
 from os import environ
 from testcontainers.core.config import TIMEOUT
 from testcontainers.core.generic import DbContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_for_logs
 import typing
 
@@ -40,7 +41,6 @@ def __init__(self,
                  arango_root_password: str = "passwd",
                  arango_no_auth: typing.Optional[bool] = None,
                  arango_random_root_password: typing.Optional[bool] = None,
-                 port_to_expose: None = None,
                  **kwargs) -> None:
         """
         Args:
@@ -54,8 +54,7 @@ def __init__(self,
                 the environment variable `ARANGO_NO_AUTH` if `None` or `False` if the environment
                 variable is not available.
         """
-        if port_to_expose:
-            raise ValueError("use `port` instead of `port_to_expose`")
+        raise_for_deprecated_parameter(kwargs, "port_to_expose", "port")
         super().__init__(image=image, **kwargs)
         self.port = port
         self.with_exposed_ports(self.port)

clickhouse/testcontainers/clickhouse/__init__.py

@@ -17,6 +17,7 @@
 from clickhouse_driver.errors import Error
 
 from testcontainers.core.generic import DbContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
 
@@ -39,18 +40,11 @@ class ClickHouseContainer(DbContainer):
             ...     client.execute("select 'working'")
             [('working',)]
     """
-    def __init__(
-            self,
-            image: str = "clickhouse/clickhouse-server:latest",
-            port: int = 9000,
-            username: Optional[str] = None,
-            password: Optional[str] = None,
-            dbname: Optional[str] = None,
-            user: None = None,
-    ) -> None:
-        super().__init__(image=image)
-        if user:
-            raise ValueError("use `username` instead")
+    def __init__(self, image: str = "clickhouse/clickhouse-server:latest", port: int = 9000,
+                 username: Optional[str] = None, password: Optional[str] = None,
+                 dbname: Optional[str] = None, **kwargs) -> None:
+        raise_for_deprecated_parameter(kwargs, "user", "username")
+        super().__init__(image=image, **kwargs)
         self.username = username or os.environ.get("CLICKHOUSE_USER", "test")
         self.password = password or os.environ.get("CLICKHOUSE_PASSWORD", "test")
         self.dbname = dbname or os.environ.get("CLICKHOUSE_DB", "test")

core/testcontainers/core/generic.py

@@ -14,6 +14,7 @@
 
 from .container import DockerContainer
 from .exceptions import ContainerStartException
+from .utils import raise_for_deprecated_parameter
 from .waiting_utils import wait_container_is_ready
 
 ADDITIONAL_TRANSIENT_ERRORS = []
@@ -39,9 +40,9 @@ def get_connection_url(self) -> str:
 
     def _create_connection_url(self, dialect: str, username: str, password: str,
                                host: Optional[str] = None, port: Optional[int] = None,
-                               dbname: Optional[str] = None, db_name: None = None) -> str:
-        if db_name:
-            raise ValueError("use `dbname` instead of `db_name`")
+                               dbname: Optional[str] = None, **kwargs) -> str:
+        if raise_for_deprecated_parameter(kwargs, "db_name", "dbname"):
+            raise ValueError(f"unexpected arguments: {','.join(kwargs)}")
         if self._container is None:
             raise ContainerStartException("container has not been started")
         host = host or self.get_container_host_ip()

core/testcontainers/core/utils.py

@@ -69,3 +69,12 @@ def default_gateway_ip() -> str:
             return ip_address.decode('utf-8').strip().strip('\n')
     except subprocess.SubprocessError:
         return None
+
+
+def raise_for_deprecated_parameter(kwargs: dict, name: str, replacement: str) -> dict:
+    """
+    Raise an error if a dictionary of keyword arguments contains a key and suggest the replacement.
+    """
+    if kwargs.pop(name, None):
+        raise ValueError(f"use `{replacement}` instead of `{name}`")
+    return kwargs

elasticsearch/testcontainers/elasticsearch/__init__.py

@@ -17,6 +17,7 @@
 from urllib.error import URLError
 
 from testcontainers.core.container import DockerContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
 _FALLBACK_VERSION = 8
@@ -73,10 +74,8 @@ class ElasticSearchContainer(DockerContainer):
             '8.3.3'
     """
 
-    def __init__(self, image: str = "elasticsearch", port: int = 9200, port_to_expose: None = None,
-                 **kwargs) -> None:
-        if port_to_expose:
-            raise ValueError("use `port` instead of `port_to_expose`")
+    def __init__(self, image: str = "elasticsearch", port: int = 9200, **kwargs) -> None:
+        raise_for_deprecated_parameter(kwargs, "port_to_expose", "port")
         super(ElasticSearchContainer, self).__init__(image, **kwargs)
         self.port_to_expose = port
         self.with_exposed_ports(self.port_to_expose)

kafka/testcontainers/kafka/__init__.py

@@ -7,6 +7,7 @@
 from kafka.errors import KafkaError, UnrecognizedBrokerVersion, NoBrokersAvailable
 
 from testcontainers.core.container import DockerContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
 
@@ -25,10 +26,9 @@ class KafkaContainer(DockerContainer):
     """
     TC_START_SCRIPT = '/tc-start.sh'
 
-    def __init__(self, image: str = "confluentinc/cp-kafka:5.4.3", port: int = 9093,
-                 port_to_expose: None = None, **kwargs) -> None:
-        if port_to_expose:
-            raise ValueError("use `port` instead of `port_to_expose`")
+    def __init__(self, image: str = "confluentinc/cp-kafka:5.4.3", port: int = 9093, **kwargs) \
+            -> None:
+        raise_for_deprecated_parameter(kwargs, "port_to_expose", "port")
         super(KafkaContainer, self).__init__(image, **kwargs)
         self.port = port
         self.with_exposed_ports(self.port)

minio/testcontainers/minio/__init__.py

@@ -2,6 +2,7 @@
 from requests import ConnectionError, Response, get
 
 from testcontainers.core.container import DockerContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
 
@@ -36,16 +37,15 @@ class MinioContainer(DockerContainer):
 
     def __init__(self, image: str = "minio/minio:RELEASE.2022-12-02T19-19-22Z",
                  port: int = 9000, access_key: str = "minioadmin",
-                 secret_key: str = "minioadmin", port_to_expose: None = None, **kwargs) -> None:
+                 secret_key: str = "minioadmin", **kwargs) -> None:
         """
         Args:
             image: Docker image to use for the MinIO container.
             port: Port to expose on the container.
             access_key: Access key for client connections.
             secret_key: Secret key for client connections.
         """
-        if port_to_expose:
-            raise ValueError("use `port` instead of `port_to_expose`")
+        raise_for_deprecated_parameter(kwargs, "port_to_expose", "port")
         super(MinioContainer, self).__init__(image, **kwargs)
         self.port = port
         self.access_key = access_key

mongodb/testcontainers/mongodb/__init__.py

@@ -13,6 +13,7 @@
 import os
 from pymongo import MongoClient
 from testcontainers.core.generic import DbContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 from typing import Optional
 
@@ -49,9 +50,8 @@ class MongoDbContainer(DbContainer):
     """
     def __init__(self, image: str = "mongo:latest", port: int = 27017,
                  username: Optional[str] = None, password: Optional[str] = None,
-                 dbname: Optional[str] = None, port_to_expose: None = None, **kwargs) -> None:
-        if port_to_expose:
-            raise ValueError("use `port` instead of `port_to_expose`")
+                 dbname: Optional[str] = None, **kwargs) -> None:
+        raise_for_deprecated_parameter(kwargs, "port_to_expose", "port")
         super(MongoDbContainer, self).__init__(image=image, **kwargs)
         self.username = username or os.environ.get("MONGO_INITDB_ROOT_USERNAME", "test")
         self.password = password or os.environ.get("MONGO_INITDB_ROOT_PASSWORD", "test")

mssql/testcontainers/mssql/__init__.py

@@ -1,6 +1,7 @@
 from os import environ
 from typing import Optional
 from testcontainers.core.generic import DbContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 
 
 class SqlServerContainer(DbContainer):
@@ -22,11 +23,9 @@ class SqlServerContainer(DbContainer):
 
     def __init__(self, image: str = "mcr.microsoft.com/mssql/server:2019-latest",
                  username: str = "SA", password: Optional[str] = None, port: int = 1433,
-                 dbname: str = "tempdb", dialect: str = 'mssql+pymssql', user: None = None,
-                 **kwargs) -> None:
+                 dbname: str = "tempdb", dialect: str = 'mssql+pymssql', **kwargs) -> None:
+        raise_for_deprecated_parameter(kwargs, "user", "username")
         super(SqlServerContainer, self).__init__(image, **kwargs)
-        if user:
-            raise ValueError("use `username` instead")
 
         self.port_to_expose = port
         self.with_exposed_ports(self.port_to_expose)

mysql/testcontainers/mysql/__init__.py

@@ -13,6 +13,7 @@
 from os import environ
 from typing import Optional
 from testcontainers.core.generic import DbContainer
+from testcontainers.core.utils import raise_for_deprecated_parameter
 
 
 class MySqlContainer(DbContainer):
@@ -39,18 +40,12 @@ class MySqlContainer(DbContainer):
     """
     def __init__(self, image: str = "mysql:latest", username: Optional[str] = None,
                  root_password: Optional[str] = None, password: Optional[str] = None,
-                 dbname: Optional[str] = None, port: int = 3306, MYSQL_USER: None = None,
-                 MYSQL_ROOT_PASSWORD: None = None, MYSQL_PASSWORD: None = None,
-                 MYSQL_DATABASE: None = None, **kwargs) -> None:
+                 dbname: Optional[str] = None, port: int = 3306, **kwargs) -> None:
+        raise_for_deprecated_parameter(kwargs, "MYSQL_USER", "username")
+        raise_for_deprecated_parameter(kwargs, "MYSQL_ROOT_PASSWORD", "root_password")
+        raise_for_deprecated_parameter(kwargs, "MYSQL_PASSWORD", "password")
+        raise_for_deprecated_parameter(kwargs, "MYSQL_DATABASE", "dbname")
         super(MySqlContainer, self).__init__(image, **kwargs)
-        if MYSQL_USER:
-            raise ValueError("use `username` instead of `MYSQL_USER`")
-        if MYSQL_ROOT_PASSWORD:
-            raise ValueError("use `root_password` instead of `MYSQL_ROOT_PASSWORD`")
-        if MYSQL_PASSWORD:
-            raise ValueError("use `password` instead of `MYSQL_PASSWORD`")
-        if MYSQL_DATABASE:
-            raise ValueError("use `dbname` instead of `MYSQL_DATABASE`")
 
         self.port_to_expose = port
         self.with_exposed_ports(self.port_to_expose)