Add CWL conformance tests to CI (#205)

This commit adds the full suite of CWL v1.3 conformance tests to
the CI pipeline, in order to prevent regressions with real usage.

Author: GlassOfWhiskey

.github/workflows/ci-tests.yaml

@@ -22,6 +22,74 @@ jobs:
           config-file: .github/codeql/config.yml
           languages: python
       - uses: github/codeql-action/analyze@v4
+  cwl-conformance:
+    name: "CWL conformance tests"
+    strategy:
+      matrix:
+        on: [ "ubuntu-24.04", "macos-15-intel" ]
+    runs-on: ${{ matrix.on }}
+    env:
+      POSTGRES_DB: streamflow
+      POSTGRES_PASSWORD: streamflow
+      POSTGRES_USER: streamflow
+      POSTGRES_HOST: 127.0.0.1
+    steps:
+      - uses: actions/checkout@v6
+      - uses: astral-sh/setup-uv@v7
+        with:
+          version: "0.9.16"
+      - uses: actions/setup-python@v6
+        with:
+          python-version: 3.14
+      - uses: actions/setup-node@v6
+        with:
+          node-version: "24"
+      - name: "Install Docker (MacOS X)"
+        uses: douglascamata/setup-docker-macos-action@v1
+        with:
+          colima-additional-options: '--mount /private/var/folders:w'
+        if: ${{ startsWith(matrix.on, 'macos-') }}
+      - uses: docker/setup-qemu-action@v3
+        if: ${{ startsWith(matrix.on, 'ubuntu-') }}
+      - name: "Start PostgreSQL Docker container"
+        run: |
+          docker run \
+            --rm \
+            --detach \
+            --env POSTGRES_DB=${{ env.POSTGRES_DB }} \
+            --env POSTGRES_PASSWORD=${{ env.POSTGRES_DB }} \
+            --env POSTGRES_USER=${{ env.POSTGRES_DB }} \
+            --publish 5432:5432 \
+            postgres:18.1-alpine
+      - name: "Install Postgresql client"
+        run: brew install libpq@17
+        if: ${{ startsWith(matrix.on, 'macos-') }}
+      - name: "Download CWL conformance test script from StreamFlow"
+        run: |
+          curl -fsSL https://raw.githubusercontent.com/alpha-unito/streamflow/refs/tags/0.2.0.dev14/cwl-conformance-test.sh -o cwl-conformance-test.sh
+          chmod 755 cwl-conformance-test.sh
+      - name: "Test CWL v1.3 conformance"
+        env:
+          VERSION: "v1.3"
+          COMMIT: "2063e9095f421f2a2ce12abaf196b2ba06ca5aae"
+          EXCLUDE: "docker_entrypoint,modify_file_content,iwd-container-entryname1"
+          DOCKER: "docker"
+        run: ./cwl-conformance-test.sh
+      - name: "Upload test results"
+        if: ${{ !cancelled() }}
+        uses: codecov/codecov-action@v5
+        with:
+          flags: ${{ startsWith(matrix.on, 'macos-') && 'macos' || 'ubuntu' }}
+          report_type: test_results
+          token: ${{ secrets.CODECOV_TOKEN }}
+      - name: "Upload coverage report for conformance tests"
+        uses: actions/upload-artifact@v6
+        with:
+          name: ${{ format('{0}-py3.14-cwlv1.3-docker-conformance-tests', matrix.on) }}
+          path: |
+            ./coverage.xml
+          retention-days: 1
+          if-no-files-found: error
   static-checks:
     name: "Static checks"
     runs-on: ubuntu-24.04

.github/workflows/release.yml

@@ -21,15 +21,15 @@ jobs:
           version: "0.9.16"
       - uses: actions/setup-python@v6
         with:
-          python-version: 3.13
+          python-version: 3.14
       - name: "Install plugin"
         run: |
           uv venv
           uv pip install .
       - name: "Get version"
         run: |
           source .venv/bin/activate
-          echo "STREAMFLOW_VERSION=$(python -c "from streamflow.plugins.unito.postgresql.version import VERSION; print(VERSION)")" >> $GITHUB_ENV
+          echo "PLUGIN_VERSION=$(python -c "from streamflow.plugins.unito.postgresql.version import VERSION; print(VERSION)")" >> $GITHUB_ENV
       - name: "Check tag existence"
         uses: mukunku/tag-exists-action@v1.7.0
         id: check-tag
@@ -70,7 +70,7 @@ jobs:
       - name: "Get local version"
         run: |
           source .venv/bin/activate
-          echo "STREAMFLOW_VERSION=$(python -c "from streamflow.plugins.unito.postgresql.version import VERSION; print(VERSION)")" >> $GITHUB_ENV
+          echo "PLUGIN_VERSION=$(python -c "from streamflow.plugins.unito.postgresql.version import VERSION; print(VERSION)")" >> $GITHUB_ENV
       - name: "Get PyPI version"
         run: echo "PYPI_VERSION=$(pip index versions --pre streamflow-postgresql | grep streamflow-postgresql | sed 's/.*(\(.*\))/\1/')" >> $GITHUB_ENV
       - name: "Build Python packages"

pyproject.toml

@@ -59,6 +59,7 @@ lint = [
     "pyupgrade==3.21.2"
 ]
 test = [
+    "cwltest==2.6.20251216093331",
     "pytest==9.0.2",
     "pytest-asyncio==1.3.0",
     "pytest-cov==7.0.0",

streamflow/plugins/unito/postgresql/database.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 from collections.abc import MutableMapping, MutableSequence
 from importlib.resources import files
@@ -54,12 +55,17 @@ async def __aenter__(self):
             )
             async with self._pool.acquire() as conn:
                 async with conn.transaction():
-                    await conn.execute(
+                    schema = (
                         files(__package__)
                         .joinpath("schemas")
                         .joinpath("postgresql.sql")
                         .read_text("utf-8")
                     )
+                    schema_checksum = hashlib.new("sha1", usedforsecurity=False)
+                    schema_checksum.update(schema.encode("utf-8"))
+                    lock_key = int(schema_checksum.hexdigest(), 16) % (2**63 - 1)
+                    await conn.execute(f"SELECT pg_advisory_xact_lock({lock_key})")
+                    await conn.execute(schema)
         return self._pool
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
@@ -148,17 +154,17 @@ async def add_deployment(
                         json.dumps(wraps) if wraps else None,
                     )
 
-    async def add_execution(self, step_id: int, tag: str, cmd: str) -> int:
+    async def add_execution(self, step_id: int, job_token_id: int, cmd: str) -> int:
         async with self.pool as pool:
             async with pool.acquire() as conn:
                 async with conn.transaction():
                     return await conn.fetchval(
-                        "INSERT INTO execution(step, tag, cmd) "
+                        "INSERT INTO execution(step, job_token, cmd) "
                         "VALUES($1, $2, $3) "
                         "RETURNING id",
                         step_id,
-                        tag,
-                        cmd.encode("utf-8"),
+                        job_token_id,
+                        cmd,
                     )
 
     async def add_filter(

streamflow/plugins/unito/postgresql/schemas/postgresql.sql

@@ -69,6 +69,13 @@ CREATE TABLE IF NOT EXISTS token
 );
 
 
+CREATE TABLE IF NOT EXISTS recoverable
+(
+    id    SERIAL PRIMARY KEY,
+    FOREIGN KEY (id) REFERENCES token (id)
+);
+
+
 CREATE TABLE IF NOT EXISTS provenance
 (
     dependee INTEGER,
@@ -106,13 +113,6 @@ CREATE TABLE IF NOT EXISTS target
 );
 
 
-CREATE TABLE IF NOT EXISTS recoverable
-(
-    id    SERIAL PRIMARY KEY,
-    FOREIGN KEY (id) REFERENCES token (id)
-);
-
-
 CREATE TABLE IF NOT EXISTS filter
 (
     id          SERIAL PRIMARY KEY,

tests/cwl-conformance/conftest.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import io
+import json
+import os
+from contextlib import redirect_stdout
+from typing import Any
+
+from cwltest import utils
+
+
+def pytest_cwl_execute_test(
+    config: utils.CWLTestConfig, processfile: str, jobfile: str | None
+) -> tuple[int, dict[str, Any] | None]:
+    from streamflow.core.exception import WorkflowException
+    from streamflow.cwl.runner import main
+
+    this_directory = os.path.abspath(os.path.dirname(__file__))
+    args = [
+        "--streamflow-file",
+        os.path.join(this_directory, "streamflow.yml"),
+        "--outdir",
+        config.outdir,
+        processfile,
+    ]
+    if jobfile is not None:
+        args.append(jobfile)
+
+    try:
+        f = io.StringIO()
+        with redirect_stdout(f):
+            result = main(args)
+            out = f.getvalue()
+            return result, json.loads(out) if out else {}
+    except WorkflowException:
+        return 1, {}

tests/cwl-conformance/streamflow-docker.yml

@@ -0,0 +1,8 @@
+version: v1.0
+database:
+  type: unito.postgresql
+  config:
+    dbname: streamflow
+    username: streamflow
+    password: streamflow
+    hostname: 127.0.0.1