fix: Improved Oracle DB module (testcontainers#363)

Hi,

I took the liberty to improve the Oracle DB module for Testcontainers
Python. The PR has several enhancements:

* Leveraging `oracledb` thin Python driver
    * This makes Oracle DB tests on CI/CD now possible too
* Usage of `gvenzl/oracle-free` image with the latest and greatest
Oracle DB version
* DB version independent readiness check
* Support for various `gvenzl/oracle-free` image features
(`ORACLE_DATABASE`, `APP_USER`, `APP_USER_PASSWORD`, etc)
* Tests for Oracle DB for the various combinations

Ideally, some more documentation on how the Container is supposed to be
used would be handy but I couldn't really find a good example of how
such a ReadMe should be structured.
Any things are gladly appreciated!

---------

Signed-off-by: gvenzl <[email protected]>
Signed-off-by: Gerald Venzl <[email protected]>
Co-authored-by: David Ankin <[email protected]>

Author: gvenzl

index.rst

@@ -35,7 +35,7 @@ testcontainers-python facilitates the use of Docker containers for functional an
     modules/neo4j/README
     modules/nginx/README
     modules/opensearch/README
-    modules/oracle/README
+    modules/oracle-free/README
     modules/postgres/README
     modules/qdrant/README
     modules/rabbitmq/README

modules/oracle/README.rst renamed to modules/oracle-free/README.rst

@@ -0,0 +1,76 @@
+from os import environ
+from secrets import randbits
+from typing import Optional
+
+from testcontainers.core.generic import DbContainer
+
+
+class OracleDbContainer(DbContainer):
+    """
+    Oracle database container.
+
+    Example:
+
+        .. doctest::
+
+            >>> import sys, pytest
+            >>> if sys.platform.startswith('win') or sys.platform == 'darwin':
+            ...     pytest.skip("linux only test")
+
+            >>> import sqlalchemy
+            >>> from testcontainers.oracle import OracleDbContainer
+
+            >>> with OracleDbContainer() as oracle:
+            ...     engine = sqlalchemy.create_engine(oracle.get_connection_url())
+            ...     with engine.begin() as connection:
+            ...         result = connection.execute(sqlalchemy.text("SELECT 1 FROM dual"))
+            ...         result.fetchall()
+            [(1,)]
+    """
+
+    def __init__(
+        self,
+        image: str = "gvenzl/oracle-free:slim",
+        oracle_password: Optional[str] = None,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        port: int = 1521,
+        dbname: Optional[str] = None,
+        **kwargs
+    ) -> None:
+        super().__init__(image=image, **kwargs)
+
+        self.port = port
+        self.with_exposed_ports(self.port)
+
+        self.oracle_password = oracle_password or environ.get("ORACLE_PASSWORD") or hex(randbits(24))
+        self.username = username or environ.get("APP_USER")
+        self.password = password or environ.get("APP_USER_PASSWORD")
+        self.dbname = dbname or environ.get("ORACLE_DATABASE")
+
+    def get_connection_url(self) -> str:
+        return super()._create_connection_url(
+            dialect="oracle+oracledb",
+            username=self.username or "system",
+            password=self.password or self.oracle_password,
+            port=self.port,
+        ) + "/?service_name={}".format(self.dbname or "FREEPDB1")
+        # Default DB is "FREEPDB1"
+
+    def _configure(self) -> None:
+        # if self.oracle_password is not None:
+        #     self.with_env("ORACLE_PASSWORD", self.oracle_password)
+        # # Either ORACLE_PASSWORD or ORACLE_RANDOM_PASSWORD need to be passed on
+        # else:
+        #     self.with_env("ORACLE_RANDOM_PASSWORD", "y")
+        # this module is unusable with a random password
+        self.with_env("ORACLE_PASSWORD", self.oracle_password)
+
+        if self.username is not None:
+            self.with_env("APP_USER", self.username)
+        if self.password is not None:
+            self.with_env("APP_USER_PASSWORD", self.password)
+
+        # FREE and FREEPDB1 are predefined databases, do not pass them on as ORACLE_DATABASE
+        if self.dbname is not None and self.dbname.upper() not in ("FREE", "FREEPDB1"):
+            self.with_env("ORACLE_DATABASE", self.dbname)

modules/oracle-free/testcontainers/oracle/__init__.py

@@ -0,0 +1,81 @@
+import pytest
+import sqlalchemy
+
+from testcontainers.core.utils import is_arm
+from testcontainers.oracle import OracleDbContainer
+
+
+@pytest.mark.skipif(is_arm(), reason="oracle-free container not available for ARM")
+def test_docker_run_oracle_with_system_password():
+    with OracleDbContainer(oracle_password="test") as oracledb:
+        engine = sqlalchemy.create_engine(oracledb.get_connection_url())
+        with engine.begin() as connection:
+            test_val = 1
+            result = connection.execute(sqlalchemy.text("SELECT {} FROM dual".format(test_val)))
+            for row in result:
+                assert row[0] == test_val
+
+
+@pytest.mark.skipif(is_arm(), reason="oracle-free container not available for ARM")
+def test_docker_run_oracle_with_username_password():
+    with OracleDbContainer(username="test", password="test") as oracledb:
+        engine = sqlalchemy.create_engine(oracledb.get_connection_url())
+        with engine.begin() as connection:
+            test_val = 1
+            result = connection.execute(sqlalchemy.text("SELECT {} FROM dual".format(test_val)))
+            for row in result:
+                assert row[0] == test_val
+
+
+@pytest.mark.skipif(is_arm(), reason="oracle-free container not available for ARM")
+def test_docker_run_oracle_with_custom_db_and_system_username_password():
+    with OracleDbContainer(oracle_password="coolpassword", dbname="myTestPDB") as oracledb:
+        engine = sqlalchemy.create_engine(oracledb.get_connection_url())
+        with engine.begin() as connection:
+            test_val = 1
+            result = connection.execute(sqlalchemy.text("SELECT {} FROM dual".format(test_val)))
+            for row in result:
+                assert row[0] == test_val
+
+
+@pytest.mark.skipif(is_arm(), reason="oracle-free container not available for ARM")
+def test_docker_run_oracle_with_custom_db_and_app_username_password():
+    with OracleDbContainer(username="mycooluser", password="123connect", dbname="anotherPDB") as oracledb:
+        engine = sqlalchemy.create_engine(oracledb.get_connection_url())
+        with engine.begin() as connection:
+            test_val = 1
+            result = connection.execute(sqlalchemy.text("SELECT {} FROM dual".format(test_val)))
+            for row in result:
+                assert row[0] == test_val
+
+
+@pytest.mark.skipif(is_arm(), reason="oracle-free container not available for ARM")
+def test_docker_run_oracle_with_default_db_and_app_username_password():
+    with OracleDbContainer(username="mycooluser", password="123connect") as oracledb:
+        engine = sqlalchemy.create_engine(oracledb.get_connection_url())
+        with engine.begin() as connection:
+            test_val = 1
+            result = connection.execute(sqlalchemy.text("SELECT {} FROM dual".format(test_val)))
+            for row in result:
+                assert row[0] == test_val
+
+
+@pytest.mark.skipif(is_arm(), reason="oracle-free container not available for ARM")
+def test_docker_run_oracle_with_cdb_and_system_username():
+    with OracleDbContainer(oracle_password="MyOraclePWD1", dbname="free") as oracledb:
+        engine = sqlalchemy.create_engine(oracledb.get_connection_url())
+        with engine.begin() as connection:
+            test_val = 1
+            result = connection.execute(sqlalchemy.text("SELECT {} FROM dual".format(test_val)))
+            for row in result:
+                assert row[0] == test_val
+
+
+@pytest.mark.skipif(is_arm(), reason="oracle-free container not available for ARM")
+def test_doctest():
+    with OracleDbContainer() as oracle:
+        print(oracle.get_connection_url())
+        engine = sqlalchemy.create_engine(oracle.get_connection_url())
+        with engine.begin() as connection:
+            result = connection.execute(sqlalchemy.text("SELECT 1 FROM dual"))
+            assert result.fetchall() == [(1,)]