fix: url quote passwords

core/testcontainers/core/generic.py

@@ -11,6 +11,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 from typing import Optional
+from urllib.parse import quote
 
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.exceptions import ContainerStartException
@@ -60,7 +61,8 @@ def _create_connection_url(
             raise ContainerStartException("container has not been started")
         host = host or self.get_container_host_ip()
         port = self.get_exposed_port(port)
-        url = f"{dialect}://{username}:{password}@{host}:{port}"
+        quoted_password = quote(password, safe=" +")
+        url = f"{dialect}://{username}:{quoted_password}@{host}:{port}"
         if dbname:
             url = f"{url}/{dbname}"
         return url

modules/mongodb/tests/test_mongodb.py

@@ -26,3 +26,27 @@ def test_docker_run_mongodb(version: str):
 
         cursor = db.restaurants.find({"borough": "Manhattan"})
         assert cursor.next()["restaurant_id"] == doc["restaurant_id"]
+
+
+# This is a feature in the generic DbContainer class
+# but it can't be tested on its own
+# so is tested in various database modules:
+# - mysql / mariadb
+# - postgresql
+# - sqlserver
+# - mongodb
+def test_quoted_password():
+    user = "root"
+    password = "p@$%25+0&%rd :/!=?"
+    quoted_password = "p%40%24%2525+0%26%25rd %3A%2F%21%3D%3F"
+    # driver = "pymongo"
+    kwargs = {
+        "username": user,
+        "password": password,
+    }
+    with MongoDbContainer("mongo:7.0.7", **kwargs) as container:
+        host = container.get_container_host_ip()
+        port = container.get_exposed_port(27017)
+        expected_url = f"mongodb://{user}:{quoted_password}@{host}:{port}"
+        url = container.get_connection_url()
+        assert url == expected_url

modules/mssql/tests/test_mssql.py

@@ -23,3 +23,35 @@ def test_docker_run_azure_sql_edge():
             result = connection.execute(sqlalchemy.text("select @@servicename"))
             for row in result:
                 assert row[0] == "MSSQLSERVER"
+
+
+# This is a feature in the generic DbContainer class
+# but it can't be tested on its own
+# so is tested in various database modules:
+# - mysql / mariadb
+# - postgresql
+# - sqlserver
+# - mongodb
+def test_quoted_password():
+    user = "SA"
+    # spaces seem to cause issues?
+    password = "p@$%25+0&%rd:/!=?"
+    quoted_password = "p%40%24%2525+0%26%25rd%3A%2F%21%3D%3F"
+    driver = "pymssql"
+    port = 1433
+    expected_url = f"mssql+{driver}://{user}:{quoted_password}@localhost:{port}/tempdb"
+    kwargs = {
+        "username": user,
+        "password": password,
+    }
+    with (
+        SqlServerContainer("mcr.microsoft.com/azure-sql-edge:1.0.7", **kwargs)
+        .with_env("ACCEPT_EULA", "Y")
+        .with_env(
+            "MSSQL_SA_PASSWORD", "{" + password + "}"
+        )  # special characters have to be quoted in braces in env vars
+    ) as container:
+        exposed_port = container.get_exposed_port(container.port)
+        expected_url = expected_url.replace(f":{port}", f":{exposed_port}")
+        url = container.get_connection_url()
+        assert url == expected_url

modules/mysql/tests/test_mysql.py

@@ -47,3 +47,31 @@ def test_docker_env_variables():
         url = container.get_connection_url()
         pattern = r"mysql\+pymysql:\/\/demo:test@[\w,.]+:(3306|32785)\/custom_db"
         assert re.match(pattern, url)
+
+
+# This is a feature in the generic DbContainer class
+# but it can't be tested on its own
+# so is tested in various database modules:
+# - mysql / mariadb
+# - postgresql
+# - sqlserver
+# - mongodb
+def test_quoted_password():
+    user = "root"
+    password = "p@$%25+0&%rd :/!=?"
+    quoted_password = "p%40%24%2525+0%26%25rd %3A%2F%21%3D%3F"
+    driver = "pymysql"
+    with MySqlContainer("mariadb:10.6.5", username=user, password=password) as container:
+        host = container.get_container_host_ip()
+        port = container.get_exposed_port(3306)
+        expected_url = f"mysql+{driver}://{user}:{quoted_password}@{host}:{port}/test"
+        url = container.get_connection_url()
+        assert url == expected_url
+
+        with sqlalchemy.create_engine(expected_url).begin() as connection:
+            connection.execute(sqlalchemy.text("select version()"))
+
+        raw_pass_url = f"mysql+{driver}://{user}:{password}@{host}:{port}/test"
+        with pytest.raises(Exception):
+            with sqlalchemy.create_engine(raw_pass_url).begin() as connection:
+                connection.execute(sqlalchemy.text("select version()"))

modules/postgres/tests/test_postgres.py

@@ -42,3 +42,38 @@ def test_docker_run_postgres_with_driver_pg8000():
         engine = sqlalchemy.create_engine(postgres.get_connection_url())
         with engine.begin() as connection:
             connection.execute(sqlalchemy.text("select 1=1"))
+
+
+# This is a feature in the generic DbContainer class
+# but it can't be tested on its own
+# so is tested in various database modules:
+# - mysql / mariadb
+# - postgresql
+# - sqlserver
+# - mongodb
+def test_quoted_password():
+    user = "root"
+    password = "p@$%25+0&%rd :/!=?"
+    quoted_password = "p%40%24%2525+0%26%25rd %3A%2F%21%3D%3F"
+    driver = "psycopg2"
+    kwargs = {
+        "driver": driver,
+        "username": user,
+        "password": password,
+    }
+    with PostgresContainer("postgres:16-alpine", **kwargs) as container:
+        port = container.get_exposed_port(5432)
+        host = container.get_container_host_ip()
+        expected_url = f"postgresql+{driver}://{user}:{quoted_password}@{host}:{port}/test"
+
+        url = container.get_connection_url()
+        assert url == expected_url
+
+        with sqlalchemy.create_engine(expected_url).begin() as connection:
+            connection.execute(sqlalchemy.text("select 1=1"))
+
+        raw_pass_url = f"postgresql+{driver}://{user}:{password}@{host}:{port}/test"
+        with pytest.raises(Exception):
+            # it raises ValueError, but auth (OperationalError) = more interesting
+            with sqlalchemy.create_engine(raw_pass_url).begin() as connection:
+                connection.execute(sqlalchemy.text("select 1=1"))