Merge pull request #48 from taskbadger/sk/ci-postgres-for-procrastinate

fix: make procrastinate integration tests runnable in CI

Author: snopoke

.github/workflows/integration_tests.yml

@@ -17,6 +17,19 @@ jobs:
           --health-retries 5
         ports:
           - 6379:6379
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: procrastinate
+        options: >-
+          --health-cmd "pg_isready -U postgres"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
     steps:
       - uses: actions/checkout@v6
       - name: Set up Python
@@ -36,3 +49,4 @@ jobs:
           TASKBADGER_ORG: ${{ vars.TASKBADGER_ORG }}
           TASKBADGER_PROJECT: ${{ vars.TASKBADGER_PROJECT }}
           TASKBADGER_API_KEY: ${{ secrets.TASKBADGER_API_KEY }}
+          PROCRASTINATE_DSN: postgresql://postgres:postgres@localhost:5432/procrastinate

integration_tests/test_procrastinate.py

@@ -13,6 +13,7 @@
 import random
 
 import procrastinate
+import psycopg
 import pytest
 
 import taskbadger
@@ -36,19 +37,36 @@ def _check_log_errors(caplog):
 
 
 @pytest.fixture(scope="session")
-def app():
-    """A Procrastinate app pointed at a real Postgres instance with its schema applied."""
-    conn = procrastinate.SyncPsycopgConnector(conninfo=PROCRASTINATE_DSN)
+def _schema():
+    # apply_schema is NOT idempotent (schema.sql uses bare CREATE TYPE), so
+    # only apply when the schema isn't already present.
+    with psycopg.connect(PROCRASTINATE_DSN) as conn, conn.cursor() as cur:
+        cur.execute("SELECT to_regclass('procrastinate_jobs')")
+        if cur.fetchone()[0] is not None:
+            return
+    schema_conn = procrastinate.SyncPsycopgConnector(conninfo=PROCRASTINATE_DSN)
+    schema_app = procrastinate.App(connector=schema_conn)
+    with schema_app.open():
+        schema_app.schema_manager.apply_schema()
+
+
+@pytest.fixture
+def app(_schema):
+    # Async connector: run_worker raises SyncConnectorConfigurationError on
+    # SyncPsycopgConnector. Async connectors work in sync contexts too.
+    #
+    # Function-scoped because run_worker tears down the sync sub-connector that
+    # PsycopgConnector spawns inside `app.open()`, leaving the next test's
+    # defer() with no usable sync pool.
+    conn = procrastinate.PsycopgConnector(conninfo=PROCRASTINATE_DSN)
     app = procrastinate.App(connector=conn)
     with app.open():
-        # Apply schema (idempotent — Procrastinate's apply_schema is safe to re-run).
-        app.schema_manager.apply_schema()
         yield app
 
 
-def _fetch_job_args(app, job_id):
-    """Read the stored ``args`` JSONB for a Procrastinate job."""
-    with app.connector.pool.connection() as conn:
+def _fetch_job_args(job_id):
+    # Direct sync psycopg connection — the app's pool is async (see fixture).
+    with psycopg.connect(PROCRASTINATE_DSN) as conn:
         with conn.cursor() as cur:
             cur.execute("SELECT args FROM procrastinate_jobs WHERE id = %s", (job_id,))
             row = cur.fetchone()
@@ -75,7 +93,7 @@ def add_manual(a, b):
 
     # The TB task id was stashed in the job kwargs at defer time. Read it back
     # from Procrastinate to verify the final state.
-    args = _fetch_job_args(app, job_id)
+    args = _fetch_job_args(job_id)
     tb_id = args["__taskbadger_task_id__"]
 
     fetched = taskbadger.get_task(tb_id)
@@ -100,7 +118,7 @@ def add_auto(a, b):
         listen_notify=False,
     )
 
-    args = _fetch_job_args(app, job_id)
+    args = _fetch_job_args(job_id)
     tb_id = args["__taskbadger_task_id__"]
 
     fetched = taskbadger.get_task(tb_id)