Add wal_level check to skip logical decoding tests when not set to logical; skip composite and range type tests on GaussDB

Author: chenyunliang520

tests/conftest.py

@@ -145,6 +145,36 @@ def get_database_type():
         print(f"Failed to get database version: {e}")
         return ""
 
+def get_wal_level():
+    dsn = os.getenv("DSN") or os.getenv("GAUSSDB_TEST_DSN")
+    if not dsn:
+        print("DSN environment variable not set")
+        return ""
+
+    try:
+        conn = pq.PGconn.connect(dsn.encode("utf-8"))
+        if conn.status != pq.ConnStatus.OK:
+            print(f"Connection failed: {conn.error_message.decode()}")
+            conn.finish()
+            return ""
+
+        res = conn.exec_(b"SHOW wal_level;")
+        if res.status != pq.ExecStatus.TUPLES_OK:
+            print(f"Query failed: {conn.error_message.decode()}")
+            res.clear()
+            conn.finish()
+            return ""
+
+        raw_wal_level = res.get_value(0, 0)
+        wal_level = raw_wal_level.decode("utf-8").lower() if raw_wal_level is not None else ""
+
+        res.clear()
+        conn.finish()
+        return wal_level
+    except Exception as e:
+        print(f"Failed to get wal_level: {e}")
+        return ""
+
 
 def pytest_collection_modifyitems(config, items):
     res = get_database_type()

tests/test_logical_decoding.py

@@ -1,10 +1,19 @@
 import pytest
+from .conftest import get_wal_level
 
 SLOT_NAME = "slot_test"
 SCHEMA = "my_schema"
 TABLE = "test01"
 
 
+@pytest.fixture(scope="session", autouse=True)
+def check_wal_level():
+    """Check if wal_level is set to logical, skip tests if not."""
+    wal_level = get_wal_level()
+    if wal_level != "logical":
+        pytest.skip(f"wal_level={wal_level!r}, expected 'logical'")
+
+
 def _slot_exists(conn, slot_name):
     cur = conn.cursor()
     cur.execute(
@@ -37,7 +46,7 @@ def _cleanup_slot_and_schema(conn):
 
 
 @pytest.fixture(scope="function")
-def setup_env(conn):
+def setup_env(conn, check_wal_level):
     """Ensure clean environment for each test."""
     _cleanup_slot_and_schema(conn)
     cur = conn.cursor()

tests/types/test_composite.py

@@ -10,6 +10,7 @@
 from ..utils import eur
 from ..fix_crdb import crdb_skip_message, is_crdb
 from ..test_adapt import StrNoneBinaryDumper, StrNoneDumper
+from ..conftest import get_database_type
 
 pytestmark = pytest.mark.crdb_skip("composite")
 
@@ -162,6 +163,9 @@ def test_load_record_binary(conn, want, rec):
 def testcomp(svcconn):
     if is_crdb(svcconn):
         pytest.skip(crdb_skip_message("composite"))
+    res = get_database_type()
+    if res == "gaussdb":
+        pytest.skip("gaussdb not support.")
     cur = svcconn.cursor()
     cur.execute(
         """

tests/types/test_range.py

@@ -12,6 +12,7 @@
 from ..utils import eur
 from ..fix_crdb import crdb_skip_message, is_crdb
 from ..test_adapt import StrNoneBinaryDumper, StrNoneDumper
+from ..conftest import get_database_type
 
 pytestmark = pytest.mark.crdb_skip("range")
 
@@ -268,27 +269,14 @@ def create_test_range(conn):
     if is_crdb(conn):
         pytest.skip(crdb_skip_message("range"))
 
+    res = get_database_type()
+    if res == "gaussdb":
+        pytest.skip("gaussdb not support.")
+
     conn.execute(
         """
-        ALTER FUNCTION testschema.testrange(
-            double precision,
-            double precision
-        )
-        OWNER TO root;
-
-        ALTER FUNCTION testschema.testrange(
-            double precision,
-            double precision,
-            text
-        )
-        OWNER TO root;
-
-        ALTER FUNCTION public.testrange(text, text, text) OWNER TO root;
-        ALTER FUNCTION public.testrange(text, text) OWNER TO root;
-
         drop schema if exists testschema cascade;
         create schema testschema;
-
         drop type if exists testrange cascade;
         drop type if exists testschema.testrange cascade;
 
@@ -297,7 +285,6 @@ def create_test_range(conn):
         """
     )
 
-
 fetch_cases = [
     ("testrange", "text"),
     ("testschema.testrange", "float8"),