diff --git a/.github/workflows/integration_tests.yml b/.github/workflows/integration_tests.yml
index 377438c9..6ded3be4 100644
--- a/.github/workflows/integration_tests.yml
+++ b/.github/workflows/integration_tests.yml
@@ -21,11 +21,21 @@ on:
       MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT:
         description: "Base URL for the OGC endpoint in MCP Venue Test (i.e. https://abcdef12345.execute-api.us-west-2.amazonaws.com/test/ogc/api)"
         type: string
-      # TODO: add MCP_VENUE_OPS inputs
+      MCP_VENUE_OPS_AIRFLOW_API_ENDPOINT:
+        description: "Base URL for the Airflow API endpoint in MCP Venue Ops (i.e. https://abcdef12345.execute-api.us-west-2.amazonaws.com/test/sps/api/v1)"
+        type: string
+      MCP_VENUE_OPS_OGC_PROCESSES_API_ENDPOINT:
+        description: "Base URL for the OGC endpoint in MCP Venue Ops (i.e. https://abcdef12345.execute-api.us-west-2.amazonaws.com/test/ogc/api)"
+        type: string
+
 jobs:
 
   Dev-Venue-Airflow-API:
     runs-on: ubuntu-latest
+    env:
+      UNITY_USER: ${{ secrets.MCP_VENUE_DEV_UNITY_USERNAME }}
+      UNITY_PASSWORD: ${{ secrets.MCP_VENUE_DEV_UNITY_PASSWORD }}
+      UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_DEV_UNITY_CLIENTID }}
 
     steps:
       - name: Checkout
@@ -35,32 +45,47 @@ jobs:
         uses: ./.github/actions/setup-action
         continue-on-error: false
 
-      - name: MCP Venue Dev - Integration tests with Airflow API
-        id: mcp_venue_dev_integration_tests_with_airflow_api
+      - name: MCP Venue Dev - Karpenter Integration tests with Airflow API
+        id: mcp_venue_dev_karpenter_integration_tests_with_airflow_api
+        continue-on-error: true
+        run: |
+          pytest -vv -s --gherkin-terminal-reporter \
+          unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_airflow_api.py  \
+          --venue="dev" \
+          --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT }}
+
+      - name: MCP Venue Dev - CWL Integration tests with Airflow API
+        id: mcp_venue_dev_cwl_integration_tests_with_airflow_api
         continue-on-error: true
-        env:
-          UNITY_USER: ${{ secrets.MCP_VENUE_DEV_UNITY_USERNAME }}
-          UNITY_PASSWORD: ${{ secrets.MCP_VENUE_DEV_UNITY_PASSWORD }}
-          UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_DEV_UNITY_CLIENTID }}
         run: |
           pytest -vv -s --gherkin-terminal-reporter \
           unity-test/system/integration/step_defs/test_cwl_workflows_with_airflow_api.py  \
           --venue="dev" \
-          --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT }} \
-          --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT }}
+          --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT }}
 
       - name: Check Tests Results
         if: always()
         run: |
-          tests_status=${{ steps.mcp_venue_dev_integration_tests_with_airflow_api.outcome }}
-          echo "Tests Status: $tests_status"
-          if [ "$tests_status" != "success" ]; then
-            echo "Integration Tests with Airflow API on MCP Venue Dev failed."
+          cwl_tests_status=${{ steps.mcp_venue_dev_cwl_integration_tests_with_airflow_api.outcome }}
+          echo "CWL Tests Status: $tests_status"
+          if [ "$cwl_tests_status" != "success" ]; then
+            echo "CWL Integration Tests with Airflow API on MCP Venue Dev failed."
+          fi
+          karpenter_tests_status=${{ steps.mcp_venue_dev_karpenter_integration_tests_with_airflow_api.outcome }}
+          echo "Karpenter Tests Status: $karpenter_tests_status"
+          if [ "$karpenter_tests_status" != "success" ]; then
+            echo "Karpenter Integration Tests with Airflow API on MCP Venue Dev failed."
+          fi
+          if [ "$karpenter_tests_status" != "success" || "$cwl_tests_status" != "success" ]; then
             exit 1
           fi
 
-  Test-Venue-Airflow-API:
+  Dev-Venue-OGC-API:
     runs-on: ubuntu-latest
+    env:
+      UNITY_USER: ${{ secrets.MCP_VENUE_DEV_UNITY_USERNAME }}
+      UNITY_PASSWORD: ${{ secrets.MCP_VENUE_DEV_UNITY_PASSWORD }}
+      UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_DEV_UNITY_CLIENTID }}
 
     steps:
     - name: Checkout
@@ -70,32 +95,97 @@ jobs:
       uses: ./.github/actions/setup-action
       continue-on-error: false
 
-    - name: MCP Venue Test - Integration tests with Airflow API
-      id: mcp_venue_test_integration_tests_with_airflow_api
+    - name: MCP Venue Dev - Karpenter Integration tests with OGC API
+      id: mcp_venue_dev_karpenter_integration_tests_with_ogc_api
       continue-on-error: true
-      env:
-        UNITY_USER: ${{ secrets.MCP_VENUE_TEST_UNITY_USERNAME }}
-        UNITY_PASSWORD: ${{ secrets.MCP_VENUE_TEST_UNITY_PASSWORD }}
-        UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_TEST_UNITY_CLIENTID }}
       run: |
         pytest -vv -s --gherkin-terminal-reporter \
-        unity-test/system/integration/step_defs/test_cwl_workflows_with_airflow_api.py \
-        --venue="test" \
-        --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT }} \
-        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT }}
+        --venue="dev" \
+        unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_ogc_api.py  \
+        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT }}
+
+    - name: MCP Venue Dev - CWL Integration tests with OGC API
+      id: mcp_venue_dev_cwl_integration_tests_with_ogc_api
+      continue-on-error: true
+      run: |
+        pytest -vv -s --gherkin-terminal-reporter \
+        unity-test/system/integration/step_defs/test_cwl_workflows_with_ogc_api.py \
+        --venue="dev" \
+        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT }}
 
     - name: Check Tests Results
       if: always()
       run: |
-        tests_status=${{ steps.mcp_venue_test_integration_tests_with_airflow_api.outcome }}
-        echo "Tests Status: $tests_status"
-        if [ "$tests_status" != "success" ]; then
-          echo "Integration Tests with Airflow API on MCP Venue Test failed."
-          exit 1
-        fi
+          cwl_tests_status=${{ steps.mcp_venue_dev_cwl_integration_tests_with_ogc_api.outcome }}
+          echo "CWL Tests Status: $tests_status"
+          if [ "$cwl_tests_status" != "success" ]; then
+            echo "CWL Integration Tests with OGC API on MCP Venue Dev failed."
+          fi
+          karpenter_tests_status=${{ steps.mcp_venue_dev_karpenter_integration_tests_with_ogc_api.outcome }}
+          echo "Karpenter Tests Status: $karpenter_tests_status"
+          if [ "$karpenter_tests_status" != "success" ]; then
+            echo "Karpenter Integration Tests with OGC API on MCP Venue Dev failed."
+          fi
+          if [ "$karpenter_tests_status" != "success" || "$cwl_tests_status" != "success" ]; then
+            exit 1
+          fi
 
-  Dev-Venue-OGC-API:
+  Test-Venue-Airflow-API:
     runs-on: ubuntu-latest
+    env:
+      UNITY_USER: ${{ secrets.MCP_VENUE_TEST_UNITY_USERNAME }}
+      UNITY_PASSWORD: ${{ secrets.MCP_VENUE_TEST_UNITY_PASSWORD }}
+      UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_TEST_UNITY_CLIENTID }}
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup
+        uses: ./.github/actions/setup-action
+        continue-on-error: false
+
+      - name: MCP Venue Test - Karpenter Integration tests with Airflow API
+        id: mcp_venue_test_karpenter_integration_tests_with_airflow_api
+        continue-on-error: true
+        run: |
+          pytest -vv -s --gherkin-terminal-reporter \
+          unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_airflow_api.py  \
+          --venue="test" \
+          --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT }}
+
+      - name: MCP Venue Test - CWL Integration tests with Airflow API
+        id: mcp_venue_test_cwl_integration_tests_with_airflow_api
+        continue-on-error: true
+        run: |
+          pytest -vv -s --gherkin-terminal-reporter \
+          unity-test/system/integration/step_defs/test_cwl_workflows_with_airflow_api.py  \
+          --venue="test" \
+          --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT }}
+
+      - name: Check Tests Results
+        if: always()
+        run: |
+          cwl_tests_status=${{ steps.mcp_venue_test_cwl_integration_tests_with_airflow_api.outcome }}
+          echo "CWL Tests Status: $tests_status"
+          if [ "$cwl_tests_status" != "success" ]; then
+            echo "CWL Integration Tests with Airflow API on MCP Venue Test failed."
+          fi
+          karpenter_tests_status=${{ steps.mcp_venue_test_karpenter_integration_tests_with_airflow_api.outcome }}
+          echo "Karpenter Tests Status: $karpenter_tests_status"
+          if [ "$karpenter_tests_status" != "success" ]; then
+            echo "Karpenter Integration Tests with Airflow API on MCP Venue Test failed."
+          fi
+          if [ "$karpenter_tests_status" != "success" || "$cwl_tests_status" != "success" ]; then
+            exit 1
+          fi
+
+  Test-Venue-OGC-API:
+    runs-on: ubuntu-latest
+    env:
+      UNITY_USER: ${{ secrets.MCP_VENUE_TEST_UNITY_USERNAME }}
+      UNITY_PASSWORD: ${{ secrets.MCP_VENUE_TEST_UNITY_PASSWORD }}
+      UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_TEST_UNITY_CLIENTID }}
 
     steps:
     - name: Checkout
@@ -105,32 +195,81 @@ jobs:
       uses: ./.github/actions/setup-action
       continue-on-error: false
 
-    - name: MCP Venue Dev - Integration tests with OGC API
-      id: mcp_venue_dev_integration_tests_with_ogc_api
+    - name: MCP Venue Test - Karpenter Integration tests with OGC API
+      id: mcp_venue_test_karpenter_integration_tests_with_ogc_api
+      continue-on-error: true
+      run: |
+        pytest -vv -s --gherkin-terminal-reporter \
+        --venue="test" \
+        unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_ogc_api.py  \
+        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT }}
+
+    - name: MCP Venue Test - CWL Integration tests with OGC API
+      id: mcp_venue_test_cwl_integration_tests_with_ogc_api
       continue-on-error: true
-      env:
-        UNITY_USER: ${{ secrets.MCP_VENUE_DEV_UNITY_USERNAME }}
-        UNITY_PASSWORD: ${{ secrets.MCP_VENUE_DEV_UNITY_PASSWORD }}
-        UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_DEV_UNITY_CLIENTID }}
       run: |
         pytest -vv -s --gherkin-terminal-reporter \
         unity-test/system/integration/step_defs/test_cwl_workflows_with_ogc_api.py \
-        --venue="dev" \
-        --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_DEV_AIRFLOW_API_ENDPOINT }} \
-        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_DEV_OGC_PROCESSES_API_ENDPOINT }}
+        --venue="test" \
+        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT }}
 
     - name: Check Tests Results
       if: always()
       run: |
-        tests_status=${{ steps.mcp_venue_dev_integration_tests_with_ogc_api.outcome }}
-        echo "Tests Status: $tests_status"
-        if [ "$tests_status" != "success" ]; then
-          echo "Integration Tests with OGC API on MCP Venue Dev failed."
-          exit 1
-        fi
+          cwl_tests_status=${{ steps.mcp_venue_test_cwl_integration_tests_with_ogc_api.outcome }}
+          echo "CWL Tests Status: $tests_status"
+          if [ "$cwl_tests_status" != "success" ]; then
+            echo "CWL Integration Tests with OGC API on MCP Venue Test failed."
+          fi
+          karpenter_tests_status=${{ steps.mcp_venue_test_karpenter_integration_tests_with_ogc_api.outcome }}
+          echo "Karpenter Tests Status: $karpenter_tests_status"
+          if [ "$karpenter_tests_status" != "success" ]; then
+            echo "Karpenter Integration Tests with OGC API on MCP Venue Test failed."
+          fi
+          if [ "$karpenter_tests_status" != "success" || "$cwl_tests_status" != "success" ]; then
+            exit 1
+          fi
 
-  Test-Venue-OGC-API:
+  Ops-Venue-Airflow-API:
+    runs-on: ubuntu-latest
+    env:
+      UNITY_USER: ${{ secrets.MCP_VENUE_OPS_UNITY_USERNAME }}
+      UNITY_PASSWORD: ${{ secrets.MCP_VENUE_OPS_UNITY_PASSWORD }}
+      UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_OPS_UNITY_CLIENTID }}
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup
+        uses: ./.github/actions/setup-action
+        continue-on-error: false
+
+      - name: MCP Venue Ops - Karpenter Integration tests with Airflow API
+        id: mcp_venue_ops_karpenter_integration_tests_with_airflow_api
+        continue-on-error: true
+        run: |
+          pytest -vv -s --gherkin-terminal-reporter \
+          unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_airflow_api.py  \
+          --venue="prod" \
+          --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_OPS_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_OPS_AIRFLOW_API_ENDPOINT }}
+
+      - name: Check Tests Results
+        if: always()
+        run: |
+          karpenter_tests_status=${{ steps.mcp_venue_ops_karpenter_integration_tests_with_airflow_api.outcome }}
+          echo "Karpenter Tests Status: $karpenter_tests_status"
+          if [ "$karpenter_tests_status" != "success" ]; then
+            echo "Karpenter Integration Tests with Airflow API on MCP Ops Test failed."
+            exit 1
+          fi
+
+  Ops-Venue-OGC-API:
     runs-on: ubuntu-latest
+    env:
+      UNITY_USER: ${{ secrets.MCP_VENUE_OPS_UNITY_USERNAME }}
+      UNITY_PASSWORD: ${{ secrets.MCP_VENUE_OPS_UNITY_PASSWORD }}
+      UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_OPS_UNITY_CLIENTID }}
 
     steps:
     - name: Checkout
@@ -140,26 +279,21 @@ jobs:
       uses: ./.github/actions/setup-action
       continue-on-error: false
 
-    - name: MCP Venue Test - Integration tests with OGC API
-      id: mcp_venue_test_integration_tests_with_ogc_api
+    - name: MCP Venue Ops - Karpenter Integration tests with OGC API
+      id: mcp_venue_ops_karpenter_integration_tests_with_ogc_api
       continue-on-error: true
-      env:
-        UNITY_USER: ${{ secrets.MCP_VENUE_TEST_UNITY_USERNAME }}
-        UNITY_PASSWORD: ${{ secrets.MCP_VENUE_TEST_UNITY_PASSWORD }}
-        UNITY_CLIENT_ID: ${{ secrets.MCP_VENUE_TEST_UNITY_CLIENTID }}
       run: |
         pytest -vv -s --gherkin-terminal-reporter \
-        unity-test/system/integration/step_defs/test_cwl_workflows_with_ogc_api.py \
-        --venue="test" \
-        --airflow-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT || vars.MCP_VENUE_TEST_AIRFLOW_API_ENDPOINT }} \
-        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_TEST_OGC_PROCESSES_API_ENDPOINT }}
+        --venue="prod" \
+        unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_ogc_api.py  \
+        --ogc-processes-endpoint=${{ github.event.inputs.MCP_VENUE_OPS_OGC_PROCESSES_API_ENDPOINT || vars.MCP_VENUE_OPS_OGC_PROCESSES_API_ENDPOINT }}
 
     - name: Check Tests Results
       if: always()
       run: |
-        tests_status=${{ steps.mcp_venue_test_integration_tests_with_ogc_api.outcome }}
-        echo "Tests Status: $tests_status"
-        if [ "$tests_status" != "success" ]; then
-          echo "Integration Tests with OGC API on MCP Venue Test failed."
-          exit 1
-        fi
+          karpenter_tests_status=${{ steps.mcp_venue_ops_karpenter_integration_tests_with_ogc_api.outcome }}
+          echo "Karpenter Tests Status: $karpenter_tests_status"
+          if [ "$karpenter_tests_status" != "success" ]; then
+            echo "Karpenter Integration Tests with OGC API on MCP Venue Ops failed."
+            exit 1
+          fi
diff --git a/unity-test/conftest.py b/unity-test/conftest.py
index b01e175e..936a9165 100644
--- a/unity-test/conftest.py
+++ b/unity-test/conftest.py
@@ -28,8 +28,9 @@ def pytest_addoption(parser):
         "--venue",
         action="store",
         default=None,
-        choices=("dev", "test", "ops"),
-        help="The venue in which the cluster will be deployed (dev, test, ops).",
+        # Note: unity-py uses "prod" but sps software uses "ops"
+        choices=("dev", "test", "ops", "prod"),
+        help="The venue in which the cluster will be deployed (dev, test, ops, prod).",
     )
     parser.addoption(
         "--developer",
@@ -176,3 +177,15 @@ def cwl_dag_modular_process(ogc_processes):
         if p.id == "cwl_dag_modular":
             return p
     return None
+
+
+@pytest.fixture(scope="session")
+def karpenter_dag_process(ogc_processes):
+    """
+    Selects the Karpenter Test DAG from the list of available OGC processes
+    """
+
+    for p in ogc_processes:
+        if p.id == "karpenter_test":
+            return p
+    return None
diff --git a/unity-test/system/integration/features/karpenter_test_workflow_with_airflow_api.feature b/unity-test/system/integration/features/karpenter_test_workflow_with_airflow_api.feature
new file mode 100644
index 00000000..552727b2
--- /dev/null
+++ b/unity-test/system/integration/features/karpenter_test_workflow_with_airflow_api.feature
@@ -0,0 +1,11 @@
+Feature: Karpenter Test Workflow with Airflow API
+
+  As an SPS user
+  I want to ensure that the system has been successfully deployed to a given venue
+  So that I can execute workflows as DAGs while provisioning nodes as needed
+
+  Scenario: Execute the Karpenter Test Workflow with the Airflow API
+    Given the Airflow API is up and running
+    When I trigger a run for the Karpenter Test DAG using the Airflow API
+    Then I receive a response with status code 200
+    And I see an eventual successful DAG run
diff --git a/unity-test/system/integration/features/karpenter_test_workflow_with_ogc_api.feature b/unity-test/system/integration/features/karpenter_test_workflow_with_ogc_api.feature
new file mode 100644
index 00000000..a16bccd0
--- /dev/null
+++ b/unity-test/system/integration/features/karpenter_test_workflow_with_ogc_api.feature
@@ -0,0 +1,11 @@
+Feature: Karpenter Test Workflow with OGC API
+
+  As an SPS user
+  I want to ensure that the system has been successfully deployed to a given venue
+  So that I can execute workflows as DAGs while provisioning nodes as needed
+
+  Scenario: Execute the Karpenter Test Workflow with the OGC API
+    Given the OGC API is up and running
+    When I trigger a run for the Karpenter Test DAG using the OGC API
+    Then the job starts executing
+    And I see an eventual successful job
diff --git a/unity-test/system/integration/step_defs/test_cwl_workflows_with_ogc_api.py b/unity-test/system/integration/step_defs/test_cwl_workflows_with_ogc_api.py
index 8fe2f426..4a2e88d7 100644
--- a/unity-test/system/integration/step_defs/test_cwl_workflows_with_ogc_api.py
+++ b/unity-test/system/integration/step_defs/test_cwl_workflows_with_ogc_api.py
@@ -185,7 +185,7 @@ def check_process_execution_and_termination(job):
     if job is not None:
         status = job.get_status().status
         while status in [JobStatus.ACCEPTED, JobStatus.RUNNING]:
-            print(f"Job: {job.id} status: {job.get_status().status}")
+            # print(f"Job: {job.id} status: {job.get_status().status}")
             status = job.get_status().status
 
         print(f"Job: {job.id} status: {job.get_status().status}")
diff --git a/unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_airflow_api.py b/unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_airflow_api.py
new file mode 100644
index 00000000..4c0e2111
--- /dev/null
+++ b/unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_airflow_api.py
@@ -0,0 +1,88 @@
+# This test uses the Airflow API
+# to execute the "Karpenter Test" workflow
+# which is a DAG composed of 3 "dummy" tasks,
+# each executing on a different node type.
+# A successful test will confirm
+# that the Airflow API, Keda and Karpenter are working properly,
+# for the given SPS deployment and venue.
+from pathlib import Path
+
+import backoff
+import requests
+from pytest_bdd import given, scenario, then, when
+
+FILE_PATH = Path(__file__)
+FEATURES_DIR = FILE_PATH.parent.parent / "features"
+FEATURE_FILE: Path = FEATURES_DIR / "karpenter_test_workflow_with_airflow_api.feature"
+
+# DAG parameters are venue specific
+DAG_ID = "karpenter_test"
+DAG_PARAMETERS = {"placeholder": 1}
+
+
+@scenario(FEATURE_FILE, "Execute the Karpenter Test Workflow with the Airflow API")
+def test_execute_karpenter_test_workflow_with_the_airflow_api():
+    pass
+
+
+@given("the Airflow API is up and running")
+def api_up_and_running():
+    pass
+
+
+@when("I trigger a run for the Karpenter Test DAG using the Airflow API", target_fixture="response")
+def trigger_dag(airflow_api_url, fetch_token):
+
+    headers = {"Authorization": f"Bearer {fetch_token}", "Content-Type": "application/json"}
+    job_config = {"conf": DAG_PARAMETERS}
+
+    response = requests.post(
+        f"{airflow_api_url}/dags/{DAG_ID}/dagRuns",
+        headers=headers,
+        json=job_config,
+        # nosec
+        verify=False,
+    )
+    return response
+
+
+@then("I receive a response with status code 200")
+def check_status_code(response):
+    assert response.status_code == 200, f"Expected status code 200, but got {response.status_code}"
+
+
+def check_failed(e):
+    if isinstance(e, AssertionError):
+        return "failed" in e.args[0]
+    return False
+
+
+@then("I see an eventual successful DAG run")
+@backoff.on_exception(
+    backoff.constant,
+    (AssertionError, requests.exceptions.HTTPError),
+    max_time=7200,
+    giveup=check_failed,
+    jitter=None,
+    interval=5,
+)
+def poll_dag_run(response, airflow_api_url, fetch_token):
+
+    headers = {"Authorization": f"Bearer {fetch_token}"}
+
+    if response is not None:
+        dag_json = response.json()
+        dag_run_response = requests.get(
+            f"""{airflow_api_url}/dags/{DAG_ID}/dagRuns/{dag_json["dag_run_id"]}""",
+            headers=headers,
+            # nosec
+            verify=False,
+        )
+        assert dag_run_response.status_code == 200, (
+            f"Expected status code 2" f"00, but got {response.status_code}"
+        )
+        json = dag_run_response.json()
+        assert "state" in json, 'Expected "state" element in response'
+        assert json["state"] == "success"
+    else:
+        pass
diff --git a/unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_ogc_api.py b/unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_ogc_api.py
new file mode 100644
index 00000000..200253e3
--- /dev/null
+++ b/unity-test/system/integration/step_defs/test_karpenter_test_workflow_with_ogc_api.py
@@ -0,0 +1,87 @@
+# This test uses the OGC API
+# to execute the "Karpenter Test" workflow
+# which is a DAG composed of 3 "dummy" tasks,
+# each executing on a different node type.
+# A successful test will confirm
+# that the OGC API, Keda and Karpenter are working properly,
+# for the given SPS deployment and venue.
+from pathlib import Path
+
+import backoff
+from pytest_bdd import given, scenario, then, when
+from unity_sds_client.resources.job_status import JobStatus
+from unity_sds_client.unity_exception import UnityException
+from unity_sps_ogc_processes_api_python_client.exceptions import ApiException, ServiceException
+
+FILE_PATH = Path(__file__)
+FEATURES_DIR = FILE_PATH.parent.parent / "features"
+FEATURE_FILE: Path = FEATURES_DIR / "karpenter_test_workflow_with_ogc_api.feature"
+
+# DAG parameters are venue specific
+DAG_ID = "karpenter_test"
+DAG_PARAMETERS = {"placeholder": 1}
+
+
+@scenario(FEATURE_FILE, "Execute the Karpenter Test Workflow with the OGC API")
+def test_execute_karpenter_test_workflow_with_the_ogc_api():
+    pass
+
+
+@given("the OGC API is up and running")
+def api_up_and_running():
+    pass
+
+
+@when("I trigger a run for the Karpenter Test DAG using the OGC API", target_fixture="job")
+def trigger_process(karpenter_dag_process):
+
+    ogc_process = karpenter_dag_process
+    payload = DAG_PARAMETERS
+
+    # print(ogc_process)
+    assert ogc_process is not None
+    # print(payload)
+    assert payload is not None
+
+    # submit job
+    job = ogc_process.execute(payload)
+    assert job is not None
+    return job
+
+
+@then("the job starts executing")
+def check_job_started(job):
+    if job is not None:
+        status = job.get_status().status
+        assert status in [JobStatus.ACCEPTED, JobStatus.RUNNING]
+    else:
+        pass
+
+
+def check_failed(e):
+    if isinstance(e, AssertionError):
+        return "failed" in e.args[0]
+    return False
+
+
+@then("I see an eventual successful job")
+@backoff.on_exception(
+    backoff.constant,
+    (AssertionError, ApiException, ServiceException, UnityException),
+    max_time=7200,
+    giveup=check_failed,
+    jitter=None,
+    interval=5,
+)
+def check_process_execution_and_termination(job):
+
+    if job is not None:
+        status = job.get_status().status
+        while status in [JobStatus.ACCEPTED, JobStatus.RUNNING]:
+            # print(f"Job: {job.id} status: {job.get_status().status}")
+            status = job.get_status().status
+
+        print(f"Job: {job.id} status: {job.get_status().status}")
+        assert job.get_status().status == JobStatus.SUCCESSFUL
+    else:
+        pass