feat: move Oracle to clientless validation

docs/supported-databases/oracle.rst

@@ -22,6 +22,8 @@ Usage Example
     pytest_plugins = ["pytest_databases.docker.oracle"]
 
     def test(oracle_service: OracleService) -> None:
+        # ``oracledb`` is user-owned application code; pytest-databases only
+        # starts the service and provides connection metadata.
         conn = oracledb.connect(
             user=oracle_service.user,
             password=oracle_service.password,
@@ -34,24 +36,18 @@ Usage Example
             res = cur.fetchone()[0]
             assert res == 1
 
-    def test(oracle_startup_connection: oracledb.Connection) -> None:
-        with oracle_startup_connection.cursor() as cursor:
-            cursor.execute("CREATE or replace view simple_table as SELECT 1 as the_value from dual")
-            cursor.execute("select * from simple_table")
-            result = cursor.fetchall()
-            assert result is not None and result[0][0] == 1
-
 Available Fixtures
 ------------------
 
-* ``oracle_image``: The Docker image to use for Oracle.
-* ``oracle_service``: A fixture that provides an Oracle service.
-* ``oracle_startup_connection``: A fixture that provides an Oracle connection.
+* ``oracle_user``: The application username created in the Oracle container.
+* ``oracle_password``: The application user password.
+* ``oracle_system_password``: The Oracle system password.
+* ``oracle_service``: Alias for the latest supported Oracle service.
 
 The following version-specific fixtures are also available:
 
-* ``oracle_18c_image``, ``oracle_18c_service_name``, ``oracle_18c_service``, ``oracle_18c_connection``: Oracle 18c
-* ``oracle_23ai_image``, ``oracle_23ai_service_name``, ``oracle_23ai_service``, ``oracle_23ai_connection``: Oracle 23ai
+* ``oracle_18c_image``, ``oracle_18c_service_name``, ``oracle_18c_service``: Oracle 18c
+* ``oracle_23ai_image``, ``oracle_23ai_service_name``, ``oracle_23ai_service``: Oracle 23ai
 
 Service API
 -----------

pyproject.toml

@@ -71,7 +71,7 @@ mariadb = []
 mongodb = ["pymongo"]
 mssql = []
 mysql = []
-oracle = ["oracledb"]
+oracle = []
 postgres = ["psycopg>=3"]
 redis = ["redis"]
 spanner = ["google-cloud-spanner"]
@@ -385,7 +385,9 @@ line-length = 120
 
 [tool.pytest.ini_options]
 addopts = "--doctest-glob='*.md' --dist=loadgroup"
-cdist-group-steal = "3:10"
+# Keep the middle shard non-empty when file justification moves a large test file
+# across cdist chunk boundaries. pytest-cdist 0.3.x accepts only one steal entry.
+cdist-group-steal = "2:10"
 cdist-justify-items = "file"
 filterwarnings = [
   "ignore::DeprecationWarning:pkg_resources",

src/pytest_databases/docker/oracle.py

@@ -1,36 +1,62 @@
 from __future__ import annotations
 
 import contextlib
+import shlex
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-import oracledb
 import pytest
 
 from pytest_databases.helpers import get_xdist_worker_num
 from pytest_databases.types import ServiceContainer
 
 if TYPE_CHECKING:
-    from collections.abc import Generator
+    from collections.abc import Generator, Iterator
+
+    from docker.models.containers import Container
 
     from pytest_databases._service import DockerService
 
 
-def oracle_responsive(host: str, port: int, service_name: str, user: str, password: str) -> bool:
-    try:
-        conn = oracledb.connect(
-            host=host,
-            port=port,
-            user=user,
-            service_name=service_name,
-            password=password,
-        )
-        with conn.cursor() as cursor:
-            cursor.execute("SELECT 1 FROM dual")
-            resp = cursor.fetchone()
-        return resp[0] == 1 if resp is not None else False
-    except Exception:  # noqa: BLE001
-        return False
+ORACLE_USER = "app"
+ORACLE_PASSWORD = "super-secret"
+ORACLE_SYSTEM_PASSWORD = "super-secret"
+
+
+def _output_to_bytes(output: bytes | str | Iterator[bytes]) -> bytes:
+    if isinstance(output, bytes):
+        return output
+    if isinstance(output, str):
+        return output.encode()
+    return b"".join(output)
+
+
+def _normalize_sqlplus_statement(sql: str) -> str:
+    statement = sql.strip()
+    if not statement.endswith(";"):
+        statement = f"{statement};"
+    return statement
+
+
+def _exec_sqlplus(
+    container: Container,
+    user: str,
+    password: str,
+    service_name: str,
+    sql: str,
+) -> tuple[int, bytes]:
+    script = "\n".join([
+        "SET HEADING OFF FEEDBACK OFF PAGESIZE 0 VERIFY OFF ECHO OFF",
+        "WHENEVER OSERROR EXIT 9",
+        "WHENEVER SQLERROR EXIT SQL.SQLCODE",
+        _normalize_sqlplus_statement(sql),
+        "EXIT",
+        "",
+    ])
+    connect_string = f"{user}/{password}@//localhost:1521/{service_name}"
+    command = f"printf '%s' {shlex.quote(script)} | sqlplus -L -S {shlex.quote(connect_string)}"
+    result = container.exec_run(["bash", "-lc", command])
+    return result.exit_code if result.exit_code is not None else -1, _output_to_bytes(result.output)
 
 
 @dataclass
@@ -47,26 +73,19 @@ def _provide_oracle_service(
     image: str,
     name: str,
     service_name: str,
+    user: str,
+    password: str,
+    system_password: str,
 ) -> Generator[OracleService, None, None]:
-    user = "app"
-    password = "super-secret"
-    system_password = "super-secret"
-
     def check(_service: ServiceContainer) -> bool:
-        try:
-            conn = oracledb.connect(
-                host=_service.host,
-                port=_service.port,
-                user=user,
-                password=password,
-                service_name=service_name,
-            )
-            with conn.cursor() as cursor:
-                cursor.execute("SELECT 1 FROM dual")
-                resp = cursor.fetchone()
-            return resp[0] == 1 if resp is not None else False
-        except Exception:  # noqa: BLE001
-            return False
+        exit_code, output = _exec_sqlplus(
+            _service.container,
+            user,
+            password,
+            service_name,
+            "SELECT 1 FROM dual",
+        )
+        return exit_code == 0 and output.strip() == b"1"
 
     worker_num = get_xdist_worker_num()
     if worker_num is not None:
@@ -115,27 +134,58 @@ def oracle_18c_service_name() -> str:
     return "xepdb1"
 
 
+@pytest.fixture(autouse=False, scope="session")
+def oracle_user() -> str:
+    return ORACLE_USER
+
+
+@pytest.fixture(autouse=False, scope="session")
+def oracle_password() -> str:
+    return ORACLE_PASSWORD
+
+
+@pytest.fixture(autouse=False, scope="session")
+def oracle_system_password() -> str:
+    return ORACLE_SYSTEM_PASSWORD
+
+
 @pytest.fixture(autouse=False, scope="session")
 def oracle_23ai_service(
-    docker_service: DockerService, oracle_23ai_image: str, oracle_23ai_service_name: str
+    docker_service: DockerService,
+    oracle_23ai_image: str,
+    oracle_23ai_service_name: str,
+    oracle_user: str,
+    oracle_password: str,
+    oracle_system_password: str,
 ) -> Generator[OracleService, None, None]:
     with _provide_oracle_service(
         image=oracle_23ai_image,
         name="oracle23ai",
         service_name=oracle_23ai_service_name,
+        user=oracle_user,
+        password=oracle_password,
+        system_password=oracle_system_password,
         docker_service=docker_service,
     ) as service:
         yield service
 
 
 @pytest.fixture(autouse=False, scope="session")
 def oracle_18c_service(
-    docker_service: DockerService, oracle_18c_image: str, oracle_18c_service_name: str
+    docker_service: DockerService,
+    oracle_18c_image: str,
+    oracle_18c_service_name: str,
+    oracle_user: str,
+    oracle_password: str,
+    oracle_system_password: str,
 ) -> Generator[OracleService, None, None]:
     with _provide_oracle_service(
         image=oracle_18c_image,
         name="oracle18c",
         service_name=oracle_18c_service_name,
+        user=oracle_user,
+        password=oracle_password,
+        system_password=oracle_system_password,
         docker_service=docker_service,
     ) as service:
         yield service
@@ -145,36 +195,3 @@ def oracle_18c_service(
 @pytest.fixture(autouse=False, scope="session")
 def oracle_service(oracle_23ai_service: OracleService) -> OracleService:
     return oracle_23ai_service
-
-
-@pytest.fixture(autouse=False, scope="session")
-def oracle_18c_connection(
-    oracle_18c_service: OracleService,
-) -> Generator[oracledb.Connection, None, None]:
-    with oracledb.connect(
-        host=oracle_18c_service.host,
-        port=oracle_18c_service.port,
-        user=oracle_18c_service.user,
-        service_name=oracle_18c_service.service_name,
-        password=oracle_18c_service.password,
-    ) as db_connection:
-        yield db_connection
-
-
-@pytest.fixture(autouse=False, scope="session")
-def oracle_23ai_connection(
-    oracle_23ai_service: OracleService,
-) -> Generator[oracledb.Connection, None, None]:
-    with oracledb.connect(
-        host=oracle_23ai_service.host,
-        port=oracle_23ai_service.port,
-        user=oracle_23ai_service.user,
-        service_name=oracle_23ai_service.service_name,
-        password=oracle_23ai_service.password,
-    ) as db_connection:
-        yield db_connection
-
-
-@pytest.fixture(autouse=False, scope="session")
-def oracle_startup_connection(oracle_23ai_startup_connection: oracledb.Connection) -> oracledb.Connection:
-    return oracle_23ai_startup_connection

src/pytest_databases/docker/yugabyte.py

@@ -24,6 +24,7 @@
 YUGABYTE_PASSWORD = "yugabyte"
 YUGABYTE_DB = "yugabyte"
 YUGABYTE_DATABASE = "pytest_databases"
+YUGABYTE_TSERVER_FLAGS = "reject_writes_min_disk_space_mb=128"
 
 
 def _output_to_bytes(output: bytes | str | Iterator[bytes]) -> bytes:
@@ -270,7 +271,7 @@ def yugabyte_responsive(_service: ServiceContainer) -> bool:
         container_port=5433,  # YugabyteDB YSQL port
         check=yugabyte_responsive,
         name=container_name,
-        command="bin/yugabyted start --background=false",
+        command=f"bin/yugabyted start --background=false --tserver_flags={YUGABYTE_TSERVER_FLAGS}",
         transient=xdist_yugabyte_isolation_level == "server",
         timeout=120,
         pause=1.0,

tests/test_oracle.py

@@ -3,48 +3,48 @@
 import pytest
 
 
+@pytest.fixture(scope="module")
+def oracle_test_helpers() -> str:
+    return """
+    import shlex
+
+    def run_sqlplus(service, sql):
+        statement = sql.strip()
+        if not statement.endswith(";"):
+            statement = f"{statement};"
+        script = "\\n".join([
+            "SET HEADING OFF FEEDBACK OFF PAGESIZE 0 VERIFY OFF ECHO OFF",
+            "WHENEVER OSERROR EXIT 9",
+            "WHENEVER SQLERROR EXIT SQL.SQLCODE",
+            statement,
+            "EXIT",
+            "",
+        ])
+        connect_string = f"{service.user}/{service.password}@//localhost:1521/{service.service_name}"
+        command = f"printf '%s' {shlex.quote(script)} | sqlplus -L -S {shlex.quote(connect_string)}"
+        result = service.container.exec_run(["bash", "-lc", command])
+        assert result.exit_code == 0, result.output.decode(errors="replace")
+        return [line.strip() for line in result.output.decode().splitlines() if line.strip()]
+    """
+
+
 @pytest.mark.parametrize(
     "service_fixture",
     [
         "oracle_18c_service",
         "oracle_23ai_service",
     ],
 )
-def test_service_fixture(pytester: pytest.Pytester, service_fixture: str) -> None:
+def test_service_fixture(pytester: pytest.Pytester, service_fixture: str, oracle_test_helpers: str) -> None:
     pytester.makepyfile(f"""
-    import oracledb
     pytest_plugins = ["pytest_databases.docker.oracle"]
 
-    def test({service_fixture}):
-        conn = oracledb.connect(
-            user={service_fixture}.user,
-            password={service_fixture}.password,
-            service_name={service_fixture}.service_name,
-            host={service_fixture}.host,
-            port={service_fixture}.port,
-        )
-        with conn.cursor() as cur:
-            cur.execute("SELECT 1 FROM dual")
-            res = cur.fetchone()[0]
-            assert res == 1
-    """)
-
-    result = pytester.runpytest_subprocess("-p", "pytest_databases")
-    result.assert_outcomes(passed=1)
-
-
-@pytest.mark.parametrize("connection_fixture", ["oracle_18c_connection", "oracle_23ai_connection"])
-def test_connection_fixture(pytester: pytest.Pytester, connection_fixture: str) -> None:
-    pytester.makepyfile(f"""
-    import oracledb
-    pytest_plugins = ["pytest_databases.docker.oracle"]
+    {oracle_test_helpers}
 
-    def test({connection_fixture}):
-        with {connection_fixture}.cursor() as cursor:
-            cursor.execute("CREATE or replace view simple_table as SELECT 1 as the_value from dual")
-            cursor.execute("select * from simple_table")
-            result = cursor.fetchall()
-            assert bool(result is not None and result[0][0] == 1)
+    def test({service_fixture}):
+        assert run_sqlplus({service_fixture}, "SELECT 1 FROM dual") == ["1"]
+        run_sqlplus({service_fixture}, "CREATE or replace view simple_table as SELECT 1 as the_value from dual")
+        assert run_sqlplus({service_fixture}, "select * from simple_table") == ["1"]
     """)
 
     result = pytester.runpytest_subprocess("-p", "pytest_databases")