added classes for proof and cromwell apis; first wdl submit test

.github/workflows/api-tests.yml

@@ -11,6 +11,7 @@ on:
 
 jobs:
   api-tests:
+    if: ${{ ! github.event.pull_request.head.repo.fork }}
     runs-on: self-hosted
     steps:
       - uses: actions/checkout@v4
@@ -26,4 +27,4 @@ jobs:
       - name: Run tests
         env:
           PROOF_API_TOKEN_DEV: ${{ secrets.PROOF_API_TOKEN_DEV }}
-        run: uv run pytest tests/proof-api/ --verbose
+        run: uv run pytest tests/cromwell-api/ --verbose

pyproject.toml

@@ -1,10 +1,11 @@
 [project]
-name = "wdl-unit-tests"
+name = "wdlunittests"
 version = "0.1.0"
 description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "httpx>=0.28.1",
     "pytest>=8.3.4",
+    "tenacity>=9.0.0",
 ]

tests/cromwell-api/conftest.py

@@ -0,0 +1,33 @@
+from pathlib import Path
+
+import pytest
+from cromwell import CromwellApi
+from proof import ProofApi
+
+proof_api = ProofApi()
+cromwell_url = proof_api.cromwell_url()
+
+
+@pytest.fixture(scope="session")
+def cromwell_api():
+    return CromwellApi(url=cromwell_url)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def submit_wdls(cromwell_api):
+    """
+    This fixture runs automatically before any tests.
+    Uses "session" scope to run only once for all tests.
+    """
+    root = Path(__file__).parents[2].resolve()
+    pattern = "**/*.wdl"
+    wdl_paths = list(root.glob(pattern))
+
+    print(f"Submitting {len(wdl_paths)} wdls ...")
+
+    out = [cromwell_api.submit_workflow(wdl_path=path) for path in wdl_paths]
+
+    # Yield to let tests run
+    yield out
+
+    # Cleanup is not possible for Cromwell - but would be here

tests/cromwell-api/cromwell.py

@@ -0,0 +1,71 @@
+import httpx
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
+from utils import TOKEN
+
+
+def as_file_object(path=None):
+    if not path:
+        return None
+    return open(path, mode="rb")
+
+
+def my_before_sleep(state):
+    print(
+        f"Retrying in {state.next_action.sleep} seconds, attempt {state.attempt_number}"
+    )
+
+
+class CromwellApi(object):
+    """CromwellApi class"""
+
+    def __init__(self, url):
+        self.base_url = url.rstrip("/")
+        self.token = TOKEN
+        self.headers = {"Authorization": f"Bearer {TOKEN}"}
+
+    def submit_workflow(
+        self,
+        wdl_path,
+        batch=None,
+        params=None,
+    ):
+        files = {
+            "workflowSource": as_file_object(str(wdl_path.absolute())),
+            "workflowInputs": as_file_object(batch),
+            "workflowInputs_2": as_file_object(params),
+        }
+        files = {k: v for k, v in files.items() if v}
+        res = httpx.post(
+            f"{self.base_url}/api/workflows/v1", headers=self.headers, files=files
+        )
+        res.raise_for_status()
+        return res.json()
+
+    @retry(
+        retry=retry_if_exception_type(httpx.HTTPStatusError),
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=4, max=10),
+        before_sleep=my_before_sleep,
+    )
+    def metadata(self, workflow_id):
+        params = {"expandSubWorkflows": False, "excludeKey": "calls"}
+        res = httpx.get(
+            f"{self.base_url}/api/workflows/v1/{workflow_id}/metadata",
+            headers=self.headers,
+            params=params,
+        )
+        res.raise_for_status()
+        return res.json()
+
+    def version(self):
+        res = httpx.get(
+            f"{self.base_url}/engine/v1/version",
+            headers=self.headers,
+        )
+        res.raise_for_status()
+        return res.json()

tests/cromwell-api/proof.py

@@ -0,0 +1,37 @@
+import httpx
+from utils import PROOF_BASE_URL, TOKEN
+
+
+class ProofApi(object):
+    """ProofApi class"""
+
+    def __init__(self):
+        self.base_url = PROOF_BASE_URL
+        self.token = TOKEN
+        self.headers = {"Authorization": f"Bearer {TOKEN}"}
+
+    def status(self, timeout=10):
+        res = httpx.get(
+            f"{self.base_url}/cromwell-server",
+            headers=self.headers,
+            timeout=timeout,
+        )
+        return res.json()
+
+    def is_cromwell_server_up(self, timeout=10):
+        return not self.status()["canJobStart"]
+
+    def start(self):
+        return httpx.post(
+            f"{self.base_url}/cromwell-server",
+            headers=self.headers,
+            json={"slurm_account": None},
+        )
+
+    def start_if_not_up(self):
+        if not self.is_cromwell_server_up():
+            return self.start()
+
+    def cromwell_url(self):
+        self.start_if_not_up()
+        return self.status()["cromwellUrl"]

tests/cromwell-api/test-metadata.py

@@ -0,0 +1,19 @@
+workflow_meta_keys = [
+    "submittedFiles",
+    "calls",
+    "outputs",
+    "status",
+    "id",
+    "inputs",
+    "labels",
+    "submission",
+]
+
+
+def test_metadata(cromwell_api, submit_wdls):
+    """Getting workflow metadata works"""
+    ids = [wf["id"] for wf in submit_wdls]
+    for x in ids:
+        res = cromwell_api.metadata(x)
+        assert isinstance(res, dict)
+        # assert list(res.keys()) == workflow_meta_keys

tests/cromwell-api/test-submit.py

@@ -0,0 +1,8 @@
+from utils import make_path
+
+
+def test_submit_works(cromwell_api):
+    """Submitting a WDL file works"""
+    res = cromwell_api.submit_workflow(wdl_path=make_path("helloHostname"))
+    assert isinstance(res, dict)
+    assert res["status"] == "Submitted"

tests/cromwell-api/test-version.py

@@ -0,0 +1,3 @@
+def test_version(cromwell_api):
+    res = cromwell_api.version()
+    assert isinstance(res, dict)

tests/cromwell-api/utils.py

@@ -0,0 +1,10 @@
+import os
+from pathlib import Path
+
+PROOF_BASE_URL = "https://proof-api-dev.fredhutch.org"
+TOKEN = os.getenv("PROOF_API_TOKEN_DEV")
+
+
+def make_path(file):
+    path = Path(__file__).parents[2].resolve()
+    return path / f"{file}/{file}.wdl"