Enhancements to GaussDB CI and Logical Replication Testing

.github/workflows/release.yml → .github/workflows/release-gaussdb-pool.yml

@@ -1,9 +1,10 @@
-name: Build and Release
+---
+name: Build and Release gaussdb_pool
 
 on:
   push:
     tags:
-      - "*"
+      - "pool-v*.*.*"
 
 permissions:
   contents: write
@@ -26,30 +27,18 @@ jobs:
           pip install --upgrade pip
           pip install --upgrade setuptools wheel build
 
-      - name: Build gaussdb
-        working-directory: gaussdb
-        run: python -m build
-
       - name: Build gaussdb_pool
         working-directory: gaussdb_pool
         run: python -m build
 
-      - name: Build isort_gaussdb
-        working-directory: tools/isort-gaussdb
-        run: python -m build
-
       - name: Show dist dirs content
         run: |
-            ls -l gaussdb/dist/
             ls -l gaussdb_pool/dist/
-            ls -l tools/isort-gaussdb/dist/
 
       - name: Collect all artifacts
         run: |
           mkdir -p all_dist
-          cp gaussdb/dist/* all_dist/
           cp gaussdb_pool/dist/* all_dist/
-          cp tools/isort-gaussdb/dist/* all_dist/
           ls -ltr all_dist/
 
       - name: Upload all dist/* to GitHub Release
@@ -64,4 +53,4 @@ jobs:
           twine upload all_dist/*
         env:
           TWINE_USERNAME: __token__
-          TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
+          TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}

.github/workflows/release-gaussdb.yml

@@ -0,0 +1,56 @@
+---
+name: Build and Release gaussdb
+
+on:
+  push:
+    tags:
+      - "v*.*.*"
+
+permissions:
+  contents: write
+
+jobs:
+  build:
+    runs-on: ubuntu-22.04
+
+    steps:
+      - name: Checkout source
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.9"
+
+      - name: Install build tools
+        run: |
+          pip install --upgrade pip
+          pip install --upgrade setuptools wheel build
+
+      - name: Build gaussdb
+        working-directory: gaussdb
+        run: python -m build
+
+      - name: Show dist dirs content
+        run: |
+            ls -l gaussdb/dist/
+
+      - name: Collect all artifacts
+        run: |
+          mkdir -p all_dist
+          cp gaussdb/dist/* all_dist/
+          ls -ltr all_dist/
+
+      - name: Upload all dist/* to GitHub Release
+        uses: softprops/action-gh-release@v1
+        with:
+          files: all_dist/*
+
+      - name: Upload to PyPI
+        if: startsWith(github.ref, 'refs/tags/')
+        run: |
+          pip install --upgrade twine
+          twine upload all_dist/*
+        env:
+          TWINE_USERNAME: __token__
+          TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}

.github/workflows/release-isort-gaussdb.yml

@@ -0,0 +1,56 @@
+---
+name: Build and Release isort-gaussdb
+
+on:
+  push:
+    tags:
+      - "isort-v*.*.*"
+
+permissions:
+  contents: write
+
+jobs:
+  build:
+    runs-on: ubuntu-22.04
+
+    steps:
+      - name: Checkout source
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.9"
+
+      - name: Install build tools
+        run: |
+          pip install --upgrade pip
+          pip install --upgrade setuptools wheel build
+
+      - name: Build isort-gaussdb
+        working-directory: isort-gaussdb
+        run: python -m build
+
+      - name: Show dist dirs content
+        run: |
+            ls -l isort-gaussdb/dist/
+
+      - name: Collect all artifacts
+        run: |
+          mkdir -p all_dist
+          cp isort-gaussdb/dist/* all_dist/
+          ls -ltr all_dist/
+
+      - name: Upload all dist/* to GitHub Release
+        uses: softprops/action-gh-release@v1
+        with:
+          files: all_dist/*
+
+      - name: Upload to PyPI
+        if: startsWith(github.ref, 'refs/tags/')
+        run: |
+          pip install --upgrade twine
+          twine upload all_dist/*
+        env:
+          TWINE_USERNAME: __token__
+          TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}

tests/test_column.py

@@ -108,9 +108,14 @@ def skip_neg_scale(*args):
     ],
 )
 def test_details(conn, type, precision, scale, dsize, isize):
-    cur = conn.cursor()
-    cur.execute(f"select null::{type}")
-    col = cur.description[0]
+    with conn.cursor() as cur:
+        cur.execute(f"select null::{type}")
+        col = cur.description[0]
+    try:
+        conn.rollback()
+    except Exception:
+        pass
+
     assert type == col.type_display
     assert f" {type} " in (repr(col))
     assert col.precision == precision

tests/test_logical_decoding.py

@@ -0,0 +1,148 @@
+import pytest
+
+SLOT_NAME = "slot_test"
+SCHEMA = "my_schema"
+TABLE = "test01"
+
+
+def _slot_exists(conn, slot_name):
+    cur = conn.cursor()
+    cur.execute(
+        "SELECT count(1) FROM pg_replication_slots WHERE slot_name = %s",
+        (slot_name,),
+    )
+    row = cur.fetchone()
+    return bool(row and row[0] > 0)
+
+
+def _cleanup_slot_and_schema(conn):
+    cur = conn.cursor()
+    # Drop slot if exists
+    try:
+        cur.execute(
+            "SELECT count(1) FROM pg_replication_slots WHERE slot_name = %s",
+            (SLOT_NAME,),
+        )
+        if cur.fetchone()[0] > 0:
+            cur.execute("SELECT pg_drop_replication_slot(%s);", (SLOT_NAME,))
+    except Exception:
+        pass
+
+    # Drop schema cascade
+    try:
+        cur.execute(f"DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;")
+    except Exception:
+        pass
+    conn.commit()
+
+
+@pytest.fixture(scope="function")
+def setup_env(conn):
+    """Ensure clean environment for each test."""
+    _cleanup_slot_and_schema(conn)
+    cur = conn.cursor()
+    cur.execute(f"CREATE SCHEMA {SCHEMA};")
+    cur.execute(f"SET search_path TO {SCHEMA};")
+    cur.execute(f"CREATE TABLE {TABLE} (id int, name varchar(255));")
+    cur.execute(f"ALTER TABLE {TABLE} REPLICA IDENTITY FULL;")
+    conn.commit()
+    yield conn
+    _cleanup_slot_and_schema(conn)
+
+
+def _create_slot(cur):
+    cur.execute(
+        "SELECT * FROM pg_create_logical_replication_slot(%s, %s);",
+        (SLOT_NAME, "mppdb_decoding"),
+    )
+
+
+def _read_changes(cur):
+    cur.execute(
+        "SELECT data FROM pg_logical_slot_get_changes(%s, NULL, %s);",
+        (SLOT_NAME, 4096),
+    )
+    rows = cur.fetchall()
+    return [str(r[0]) for r in rows if r and r[0] is not None]
+
+
+def test_create_replication_slot(setup_env):
+    cur = setup_env.cursor()
+    _create_slot(cur)
+    assert _slot_exists(setup_env, SLOT_NAME)
+
+
+def test_insert_produces_changes(setup_env):
+    cur = setup_env.cursor()
+    _create_slot(cur)
+    assert _slot_exists(setup_env, SLOT_NAME)
+
+    # insert
+    cur.execute(f"INSERT INTO {TABLE} VALUES (%s, %s);", (1, "hello world"))
+    setup_env.commit()
+
+    changes = _read_changes(cur)
+    joined = "\n".join(changes).lower()
+
+    assert "insert" in joined, "Insert event not present"
+    assert "hello world" in joined, "Inserted value missing"
+
+
+def test_update_produces_changes(setup_env):
+    cur = setup_env.cursor()
+    _create_slot(cur)
+    assert _slot_exists(setup_env, SLOT_NAME)
+
+    # prepare row
+    cur.execute(f"INSERT INTO {TABLE} VALUES (%s, %s);", (1, "hello world"))
+    setup_env.commit()
+
+    # update
+    cur.execute(
+        f"UPDATE {TABLE} SET name = %s WHERE id = %s;",
+        ("hello gaussdb", 1),
+    )
+    setup_env.commit()
+
+    changes = _read_changes(cur)
+    joined = "\n".join(changes).lower()
+
+    assert "update" in joined, "Update event not present"
+    assert "hello gaussdb" in joined, "Updated value missing"
+
+
+def test_delete_produces_changes(setup_env):
+    cur = setup_env.cursor()
+    _create_slot(cur)
+    assert _slot_exists(setup_env, SLOT_NAME)
+
+    # prepare row
+    cur.execute(f"INSERT INTO {TABLE} VALUES (%s, %s);", (1, "to_delete"))
+    setup_env.commit()
+
+    # delete
+    cur.execute(f"DELETE FROM {TABLE} WHERE id = %s;", (1,))
+    setup_env.commit()
+
+    changes = _read_changes(cur)
+    joined = "\n".join(changes).lower()
+
+    assert "delete" in joined, "Delete event not present"
+    assert "to_delete" in joined, "Deleted value missing"
+
+
+def test_drop_replication_slot(setup_env):
+    cur = setup_env.cursor()
+    _create_slot(cur)
+    assert _slot_exists(setup_env, SLOT_NAME)
+
+    # drop slot
+    cur.execute("SELECT pg_drop_replication_slot(%s);", (SLOT_NAME,))
+    setup_env.commit()
+
+    # verify removed
+    cur.execute(
+        "SELECT count(1) FROM pg_replication_slots WHERE slot_name = %s;",
+        (SLOT_NAME,),
+    )
+    assert cur.fetchone()[0] == 0