Refactor pytest CI architecture with parallel testing and coverage aggregation

.github/CONTRIBUTING.md

@@ -75,6 +75,20 @@ Run tests in local
 make test-local
 ```
 
+Run tests for a specific driver
+```bash
+uv run pytest tests --driver=psycopg2
+```
+
+Run tests with a specific database name
+```bash
+uv run pytest tests --driver=psycopg2 --db-name=custom_db
+```
+
+Available drivers:
+- Sync drivers: `pg8000`, `psycopg2`, `psycopg`, `psycopg2cffi`
+- Async drivers: `asyncpg`
+
 Run tests in docker
 ```bash
 make test-docker

.github/workflows/codecov.yml

@@ -1,5 +1,5 @@
 # This workflow will run tests using pytest and upload the coverage report to Codecov
-# Run test with various Python versions
+# Run test with various Python versions and database drivers
 name: Integration Tests
 
 on:
@@ -9,14 +9,18 @@ on:
         branches: [main, develop]
 
 jobs:
-  build:
+  test:
 
     runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version: ["3.9","3.10","3.11","3.12"]
+        driver: ["pg8000", "psycopg2", "psycopg", "psycopg2cffi", "asyncpg"]
 
-    name: Test pgmq-sqlalchemy
+    name: Test pgmq-sqlalchemy (Python ${{ matrix.python-version }}, Driver ${{ matrix.driver }})
+    env:
+      # Create unique database name for this combination
+      DB_NAME_RAW: pgmq_py${{ matrix.python-version }}_${{ matrix.driver }}
     steps:
         - uses: actions/checkout@v4
         - name: Set up Python ${{ matrix.python-version }}
@@ -35,10 +39,56 @@ jobs:
             cp pgmq_postgres.template.env pgmq_postgres.env
             cp pgmq_tests.template.env pgmq_tests.env
             make start-db
-        - name: Run tests and collect coverage
-          run: uv run pytest tests --cov=pgmq_sqlalchemy.queue --cov-report=xml -n auto tests
+        - name: Setup unique database for this test run
+          run: |
+            # Normalize database name (remove dots and hyphens)
+            # Note: We normalize in each step because GitHub Actions doesn't support
+            # computed environment variables that can be reused across steps
+            export DB_NAME=$(echo "$DB_NAME_RAW" | sed 's/\.//g' | sed 's/-/_/g')
+
+            # Create the database
+            docker compose exec -T pgmq_postgres psql -U postgres -c "DROP DATABASE IF EXISTS ${DB_NAME};" || true
+            docker compose exec -T pgmq_postgres psql -U postgres -c "CREATE DATABASE ${DB_NAME};"
+            docker compose exec -T pgmq_postgres psql -U postgres -d ${DB_NAME} -c "CREATE EXTENSION IF NOT EXISTS pgmq CASCADE;"
+        - name: Run tests for specific driver
+          run: |
+            # Normalize database name (remove dots and hyphens)
+            export DB_NAME=$(echo "$DB_NAME_RAW" | sed 's/\.//g' | sed 's/-/_/g')
+
+            # Create unique coverage file name
+            export COVERAGE_FILE=".coverage.py${{ matrix.python-version }}.${{ matrix.driver }}"
+
+            uv run pytest tests --driver=${{ matrix.driver }} --db-name=${DB_NAME} --cov=pgmq_sqlalchemy.queue --cov-report=xml:coverage-py${{ matrix.python-version }}-${{ matrix.driver }}.xml
           continue-on-error: true
-        - name: Upload coverage reports to Codecov with GitHub Action
+        - name: Upload coverage artifact
+          uses: actions/upload-artifact@v4
+          with:
+            name: coverage-py${{ matrix.python-version }}-${{ matrix.driver }}
+            path: coverage-py${{ matrix.python-version }}-${{ matrix.driver }}.xml
+            retention-days: 1
+        - name: Cleanup database
+          if: always()
+          run: |
+            # Normalize database name (remove dots and hyphens)
+            export DB_NAME=$(echo "$DB_NAME_RAW" | sed 's/\.//g' | sed 's/-/_/g')
+            docker compose exec -T pgmq_postgres psql -U postgres -c "DROP DATABASE IF EXISTS ${DB_NAME};" || true
+
+  upload-coverage:
+    needs: test
+    runs-on: ubuntu-latest
+    name: Upload coverage to Codecov
+    steps:
+        - uses: actions/checkout@v4
+        - name: Download all coverage artifacts
+          uses: actions/download-artifact@v4
+          with:
+            path: coverage-reports
+            pattern: coverage-*
+            merge-multiple: true
+        - name: Upload coverage reports to Codecov
           uses: codecov/[email protected]
+          with:
+            directory: ./coverage-reports
+            fail_ci_if_error: false
           env:
             CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}

tests/conftest.py

@@ -9,6 +9,80 @@
 from pgmq_sqlalchemy import PGMQueue
 from tests.constant import ASYNC_DRIVERS, SYNC_DRIVERS
 
+# Async fixture names for test filtering
+ASYNC_FIXTURE_NAMES = ['pgmq_by_async_dsn', 'pgmq_by_async_engine', 'pgmq_by_async_session_maker']
+
+
+def pytest_addoption(parser):
+    """Add custom command-line options for pytest."""
+    parser.addoption(
+        "--driver",
+        action="store",
+        default=None,
+        help="Specify the database driver to use for testing (e.g., psycopg2, asyncpg, pg8000, etc.)",
+    )
+    parser.addoption(
+        "--db-name",
+        action="store",
+        default=None,
+        help="Specify the database name to use for testing",
+    )
+
+
+def pytest_collection_modifyitems(config, items):
+    """Modify test collection to skip tests not matching the --driver option."""
+    driver_from_cli = config.getoption("--driver")
+
+    if not driver_from_cli:
+        # No driver specified, run all tests
+        return
+
+    # Determine if the specified driver is sync or async
+    is_async_driver = driver_from_cli in ASYNC_DRIVERS
+    is_sync_driver = driver_from_cli in SYNC_DRIVERS
+
+    if not is_async_driver and not is_sync_driver:
+        # Invalid driver
+        return
+
+    # Filter out tests that don't match the specified driver
+    skip_marker = pytest.mark.skip(reason=f"Test uses different driver (--driver={driver_from_cli} specified)")
+
+    for item in items:
+        # Parse the test name to extract driver info
+        # Format is usually: test_name[fixture_name-driver_name]
+        item_id = item.nodeid
+
+        # Check if the test has a specific driver in its ID
+        # Extract driver name from test ID (e.g., test_name[pgmq_by_dsn-psycopg2])
+        if '[' in item_id and ']' in item_id:
+            # Extract the part between brackets
+            bracket_content = item_id[item_id.find('[')+1:item_id.find(']')]
+
+            # Check for async fixtures by name (more precise than string matching)
+            is_async_test = any(async_fixture in bracket_content for async_fixture in ASYNC_FIXTURE_NAMES)
+
+            # Skip async tests if sync driver specified
+            if is_sync_driver and is_async_test:
+                item.add_marker(skip_marker)
+                continue
+
+            # Skip sync tests if async driver specified
+            if is_async_driver and not is_async_test:
+                item.add_marker(skip_marker)
+                continue
+
+            # Check if any known driver is in the bracket content
+            # Sort drivers by length (descending) to match longer names first (e.g., psycopg2cffi before psycopg2)
+            sorted_drivers = sorted(SYNC_DRIVERS + ASYNC_DRIVERS, key=len, reverse=True)
+            for driver in sorted_drivers:
+                if f"-{driver}]" in item_id or f"-{driver}-" in bracket_content:
+                    # This test is for a specific driver
+                    if driver != driver_from_cli:
+                        # Skip if it doesn't match the CLI driver
+                        item.add_marker(skip_marker)
+                    break
+
 
 @pytest.fixture(scope="module")
 def get_sa_host():
@@ -31,7 +105,11 @@ def get_sa_password():
 
 
 @pytest.fixture(scope="module")
-def get_sa_db():
+def get_sa_db(request):
+    """Get database name from CLI argument or environment variable."""
+    db_name_from_cli = request.config.getoption("--db-name")
+    if db_name_from_cli:
+        return db_name_from_cli
     return os.getenv("SQLALCHEMY_DB", "postgres")
 
 
@@ -44,7 +122,15 @@ def get_dsn(
     get_sa_password,
     get_sa_db,
 ):
-    driver = request.param
+    """Get DSN for sync drivers."""
+    driver_from_cli = request.config.getoption("--driver")
+
+    # Use CLI driver if specified, otherwise use parametrized driver
+    if driver_from_cli and driver_from_cli in SYNC_DRIVERS:
+        driver = driver_from_cli
+    else:
+        driver = request.param
+
     return f"postgresql+{driver}://{get_sa_user}:{get_sa_password}@{get_sa_host}:{get_sa_port}/{get_sa_db}"
 
 
@@ -57,7 +143,15 @@ def get_async_dsn(
     get_sa_password,
     get_sa_db,
 ):
-    driver = request.param
+    """Get DSN for async drivers."""
+    driver_from_cli = request.config.getoption("--driver")
+
+    # Use CLI driver if specified, otherwise use parametrized driver
+    if driver_from_cli and driver_from_cli in ASYNC_DRIVERS:
+        driver = driver_from_cli
+    else:
+        driver = request.param
+
     return f"postgresql+{driver}://{get_sa_user}:{get_sa_password}@{get_sa_host}:{get_sa_port}/{get_sa_db}"