Simplify tests.

Author: anthony-tuininga

tests/conftest.py

@@ -664,6 +664,30 @@ def run_sql_script(self, conn, script_name, **kwargs):
                 prev_obj_type = obj_type
             print("    %s/%s %s" % (line_num, position, text))
 
+    def skip_unless_client_version(self, major_version, minor_version=0):
+        """
+        Skips the test unless the specified client version or higher is being
+        used.
+        """
+        if not self.has_client_version(major_version, minor_version):
+            if minor_version == 0:
+                version = str(major_version)
+            else:
+                version = f"{major_version}.{minor_version}"
+            pytest.skip(f"requires Oracle Client {version} or higher")
+
+    def skip_unless_server_version(self, major_version, minor_version=0):
+        """
+        Skips the test unless the specified server version or higher is being
+        used.
+        """
+        if not self.has_server_version(major_version, minor_version):
+            if minor_version == 0:
+                version = str(major_version)
+            else:
+                version = f"{major_version}.{minor_version}"
+            pytest.skip(f"requires Oracle Database {version} or higher")
+
 
 @pytest.fixture
 def admin_conn(test_env):

tests/ext/test_ext_1400_inband_notify.py

@@ -33,6 +33,7 @@
 
 @pytest.fixture(scope="module", autouse=True)
 def setup_user(test_env):
+    test_env.skip_unless_server_version(19)
     user = test_env.main_user
     with test_env.get_admin_connection() as admin_conn:
         with admin_conn.cursor() as cursor:

tests/ext/test_ext_1500_pool_grow.py

@@ -33,6 +33,7 @@
 
 def test_ext_1500(skip_unless_run_long_tests, test_env):
     "E1500 - test static pool grows back to the min after sessions killed"
+    test_env.skip_unless_client_version(19)
     pool = test_env.get_pool(min=5, max=5, increment=1, ping_interval=0)
     conns = [pool.acquire() for i in range(5)]
     with test_env.get_admin_connection() as admin_conn:

tests/ext/test_ext_1800_inband_notif_async.py

@@ -45,6 +45,7 @@ def module_checks(anyio_backend, skip_unless_thin_mode):
 
 @pytest.fixture(scope="module", autouse=True)
 def setup_user(test_env):
+    test_env.skip_unless_server_version(19)
     user = test_env.main_user
     with test_env.get_admin_connection() as admin_conn:
         with admin_conn.cursor() as cursor:

tests/sql/create_schema.sql

@@ -664,10 +664,6 @@ insert into &main_user..TestJsonCols values (1,
     '[1, 2, 3]', '[4, 5, 6]', utl_raw.cast_to_raw('[7, 8, 9]'))
 /
 
-insert into &main_user..TestJsonCols values (2,
-    'null', empty_clob(), empty_blob())
-/
-
 commit
 /
 

tests/sql/create_schema_21.sql

@@ -44,6 +44,10 @@ create table &main_user..TestOsonCols (
 )
 /
 
+insert into &main_user..TestJsonCols values (2,
+    'null', empty_clob(), empty_blob())
+/
+
 begin
     dbms_aqadm.create_queue_table('&main_user..JSON_QUEUE_TAB', 'JSON');
     dbms_aqadm.create_queue('&main_user..TEST_JSON_QUEUE',

tests/test_1600_dml_returning.py

@@ -29,6 +29,7 @@
 import datetime
 
 import oracledb
+import pytest
 
 
 def test_1600(cursor):
@@ -601,8 +602,10 @@ def test_1625(conn, cursor):
 
 
 # blocked by bug 37741324 in thick mode
-def test_1626(skip_unless_thin_mode, conn, cursor):
+def test_1626(skip_unless_thin_mode, conn, cursor, test_env):
     "1626 - test DML returning with multiple DbObjects returned"
+    if not test_env.has_server_version(21):
+        pytest.skip("awaiting fix for bug 37756810")
     arrays = [
         (1626, 1627, 1628),
         (1629, 1630, 1631),

tests/test_1900_lob_var.py

@@ -458,8 +458,9 @@ def test_1920(conn, cursor, test_env):
     assert lob.read() == supplemental_chars
 
 
-def test_1921(cursor):
+def test_1921(cursor, test_env):
     "1921 - test automatic conversion to CLOB for PL/SQL"
+    test_env.skip_unless_server_version(12, 2)
     var = cursor.var(str, outconverter=lambda v: v[-15:])
     var.setvalue(0, "A" * 50000)
     cursor.execute(
@@ -477,8 +478,9 @@ def test_1921(cursor):
     assert var.getvalue() == "A" * 10 + "B" * 5
 
 
-def test_1922(cursor):
+def test_1922(cursor, test_env):
     "1922 - test automatic conversion to NCLOB for PL/SQL"
+    test_env.skip_unless_server_version(12, 2)
     var = cursor.var(oracledb.DB_TYPE_NCHAR, outconverter=lambda v: v[-12:])
     var.setvalue(0, "N" * 51234)
     cursor.execute(
@@ -496,8 +498,9 @@ def test_1922(cursor):
     assert var.getvalue() == "N" * 5 + "P" * 7
 
 
-def test_1923(cursor):
+def test_1923(cursor, test_env):
     "1923 - test automatic conversion to BLOB for PL/SQL"
+    test_env.skip_unless_server_version(12, 2)
     var = cursor.var(bytes, outconverter=lambda v: v[-14:])
     var.setvalue(0, b"L" * 52345)
     cursor.execute(

tests/test_2300_object_var.py

@@ -870,8 +870,9 @@ def test_2341(test_env):
         conn.gettype(f"{main_user}.UDT_OBJECTARRAY")
 
 
-def test_2342(conn, cursor):
+def test_2342(conn, cursor, test_env):
     "2342 - test nested records"
+    test_env.skip_unless_server_version(21)
     options = [(None, None), (1, None), (None, 2), (1, 2)]
     typ = conn.gettype("PKG_TESTNESTEDRECORDS.UDT_OUTER")
     for option in options:

tests/test_2400_pool.py

@@ -428,8 +428,7 @@ def test_2410(skip_unless_thick_mode, test_env):
 
 def test_2411(skip_unless_thick_mode, test_env):
     "2411 - test PL/SQL session callbacks"
-    if not test_env.has_client_version(12, 2):
-        pytest.skip("PL/SQL session callbacks not supported before 12.2")
+    test_env.skip_unless_client_version(12, 2)
     callback = "pkg_SessionCallback.TheCallback"
     pool = test_env.get_pool(
         min=2,
@@ -673,6 +672,7 @@ def session_callback(cls, conn, requested_tag):
 
 def test_2424(skip_if_drcp, admin_conn, test_env):
     "2424 - drop the pooled connection on receiving dead connection error"
+    test_env.skip_unless_server_version(19)
     pool = test_env.get_pool(min=2, max=2, increment=2)
 
     # acquire connections from the pool and kill all the sessions
@@ -767,6 +767,7 @@ def test_2430(test_env):
 
 def test_2431(test_env):
     "2431 - verify that connection with different cclass is reused"
+    test_env.skip_unless_server_version(19)
     cclass = "cclass2431"
     pool = test_env.get_pool(min=1, max=1)
     # ignore the first acquire which, depending on the speed with which the
@@ -1044,6 +1045,7 @@ def test_2456(test_env):
 
 def test_2457(skip_if_drcp, test_env):
     "2457 - ping pooled connection on receiving dead connection error"
+    test_env.skip_unless_server_version(18)
     admin_conn = test_env.get_admin_connection()
     pool = test_env.get_pool(min=1, max=1, ping_interval=0)