wip, testing/fixes

Author: chrishagglund-ship-it

.github/workflows/pull_request.yml

@@ -106,6 +106,11 @@ jobs:
   integration-test:
     runs-on: ubuntu-latest
     timeout-minutes: 30
+    strategy:
+      fail-fast: false
+      matrix:
+        bucket: [test-all, long-sync, long-async, core]
+    name: integration-test (${{ matrix.bucket }})
     env:
       CONDUCTOR_SERVER_URL: ${{ vars.SDKDEV_V5_SERVER_URL }}
       CONDUCTOR_AUTH_KEY: ${{ vars.SDKDEV_V5_AUTH_KEY }}
@@ -127,4 +132,4 @@ jobs:
           pip install pytest
 
       - name: Run integration tests
-        run: bash scripts/run_integration_tests.sh
+        run: bash scripts/run_integration_tests.sh --bucket=${{ matrix.bucket }}

pyproject.toml

@@ -192,11 +192,17 @@ line-ending = "auto"
 [tool.pytest.ini_options]
 pythonpath = ["src"]
 markers = [
+<<<<<<< HEAD
     "integration: requires a live Agentspan-compatible server",
     "e2e: requires a live server and AGENTSPAN_SERVER_URL",
     "sse: requires SSE streaming (AGENTSPAN_STREAMING_ENABLED=true)",
     "agent_correctness: marks tests as agent correctness tests",
     "semantic: marks tests that use an LLM judge for semantic assertions",
+=======
+    "slow_sync: long-running sync lease-extension tests (~90s)",
+    "slow_async: long-running async lease-extension tests (~90s)",
+    "slow_test_all: long-running aggregate workflow-client test_all (~83s)",
+>>>>>>> 973c79cf (wip, testing/fixes)
 ]
 
 [tool.coverage.run]

scripts/run_integration_tests.sh

@@ -24,6 +24,19 @@
 #   ./scripts/run_integration_tests.sh
 #   ./scripts/run_integration_tests.sh --with-perf   # also run the perf test
 #
+# Buckets (--bucket=<name>): the slowest tests are split into their own buckets
+# so they can run as separate parallel CI jobs and are skipped by default
+# locally. Each slow bucket is path-scoped so it only imports its own module.
+#   core        (default) everything except the slow buckets below
+#   long-sync   sync lease-extension tests   (test_lease_extension.py, ~90s)
+#   long-async  async lease-extension tests  (test_async_lease_extension.py, ~90s)
+#   test-all    aggregate workflow-client test_all (test_workflow_client_intg.py, ~83s)
+#   all         the full suite (no bucket filtering) — for a complete local run
+#
+#   ./scripts/run_integration_tests.sh                  # fast: skips slow buckets
+#   ./scripts/run_integration_tests.sh --bucket=long-sync
+#   ./scripts/run_integration_tests.sh --bucket=all     # run everything
+#
 # Any other arguments are passed straight through to pytest, which is handy for
 # targeting a subset of tests or getting more detail on failures:
 #
@@ -48,22 +61,66 @@ set -euo pipefail
 REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
 cd "$REPO_ROOT"
 
+# Slow-test files that get their own buckets / CI jobs (see --bucket above).
+LEASE_SYNC=tests/integration/test_lease_extension.py
+LEASE_ASYNC=tests/integration/test_async_lease_extension.py
+TEST_ALL_FILE=tests/integration/test_workflow_client_intg.py
+
 # The perf test is skipped by default; --with-perf opts back in.
 perf_ignore=(--ignore=tests/integration/test_update_task_v2_perf.py)
+bucket="core"
 pytest_args=()
 for arg in "$@"; do
-  if [[ "$arg" == "--with-perf" ]]; then
-    perf_ignore=()
-  else
-    pytest_args+=("$arg")
-  fi
+  case "$arg" in
+    --with-perf) perf_ignore=() ;;
+    --bucket=*)  bucket="${arg#*=}" ;;
+    *)           pytest_args+=("$arg") ;;
+  esac
 done
 
+# The AI/agentic tests always target a dedicated server (see header) and are
+# never part of these buckets.
+ai_ignore=(
+  --ignore=tests/integration/test_ai_task_types.py
+  --ignore=tests/integration/test_ai_examples.py
+  --ignore=tests/integration/test_agentic_workflows.py
+)
+
+# Build the target paths + selection for the chosen bucket. The slow buckets are
+# path-scoped so each job only imports its own module: this keeps
+# scan_for_annotated_workers from starting unrelated workers and lets the four
+# buckets run in parallel against one server without stealing each other's tasks.
+case "$bucket" in
+  core)
+    targets=(tests/integration)
+    select=("${ai_ignore[@]}" ${perf_ignore[@]+"${perf_ignore[@]}"} \
+      --ignore="$LEASE_SYNC" --ignore="$LEASE_ASYNC" --ignore="$TEST_ALL_FILE")
+    ;;
+  all)
+    targets=(tests/integration)
+    select=("${ai_ignore[@]}" ${perf_ignore[@]+"${perf_ignore[@]}"})
+    ;;
+  long-sync)
+    targets=("$LEASE_SYNC")
+    select=(-m slow_sync)
+    ;;
+  long-async)
+    targets=("$LEASE_ASYNC")
+    select=(-m slow_async)
+    ;;
+  test-all)
+    targets=("$TEST_ALL_FILE")
+    select=(-m slow_test_all)
+    ;;
+  *)
+    echo "Unknown --bucket='$bucket' (expected: core, long-sync, long-async, test-all, all)" >&2
+    exit 2
+    ;;
+esac
+
 # Note: the "${arr[@]+"${arr[@]}"}" form is required so empty arrays don't trip
 # "unbound variable" under `set -u` on bash 3.2 (the default macOS bash).
-exec python3 -m pytest tests/integration -v \
-  --ignore=tests/integration/test_ai_task_types.py \
-  --ignore=tests/integration/test_ai_examples.py \
-  --ignore=tests/integration/test_agentic_workflows.py \
-  ${perf_ignore[@]+"${perf_ignore[@]}"} \
+exec python3 -m pytest -v \
+  "${targets[@]}" \
+  ${select[@]+"${select[@]}"} \
   ${pytest_args[@]+"${pytest_args[@]}"}

tests/integration/README.md

@@ -58,6 +58,25 @@ export CONDUCTOR_SERVER_URL="http://localhost:8080/api"
 ./scripts/run_integration_tests.sh --with-perf
 ```
 
+By default this runs the fast `core` bucket and **skips the slowest tests**
+(a few tests deliberately sleep ~50-90s to exercise lease-extension timeouts and
+one aggregate `test_all`). Those live in their own buckets, each of which runs as
+a separate parallel CI job. Select one with `--bucket=<name>`:
+
+| Bucket | What it runs |
+| --- | --- |
+| `core` (default) | everything except the slow buckets below |
+| `long-sync` | sync lease-extension tests (`test_lease_extension.py`, ~90s) |
+| `long-async` | async lease-extension tests (`test_async_lease_extension.py`, ~90s) |
+| `test-all` | aggregate workflow-client `test_all` (`test_workflow_client_intg.py`, ~83s) |
+| `all` | the full suite (no bucket filtering) — for a complete local run |
+
+```bash
+./scripts/run_integration_tests.sh                  # fast: skips the slow buckets
+./scripts/run_integration_tests.sh --bucket=long-sync
+./scripts/run_integration_tests.sh --bucket=all     # run everything
+```
+
 Any extra arguments pass straight through to pytest, which is handy for
 targeting a subset of tests or getting more detail on failures. See additional
 options and examples in the comments at the top of

tests/integration/client/orkes/test_orkes_clients.py

@@ -1,4 +1,5 @@
 import json
+import time
 
 from shortuuid import uuid
 
@@ -38,6 +39,19 @@
 TEST_IP_JSON = 'tests/integration/resources/test_data/loan_workflow_input.json'
 
 
+def _retry_on_404(func, *args, retries=5, **kwargs):
+    # Updating a task by ref name can transiently 404 while the server is still
+    # scheduling the referenced task. Retry with backoff to tolerate that race.
+    for attempt in range(retries):
+        try:
+            return func(*args, **kwargs)
+        except ApiException as e:
+            if e.status == 404 and attempt < retries - 1:
+                time.sleep(1 << attempt)
+                continue
+            raise
+
+
 class TestOrkesClients:
     def __init__(self, configuration: Configuration):
         self.api_client = ApiClient(configuration)
@@ -576,15 +590,17 @@ def __test_task_execution_lifecycle(self):
 
         polledTask = batchPolledTasks[0]
         # Update first task of second workflow
-        self.task_client.update_task_by_ref_name(
+        _retry_on_404(
+            self.task_client.update_task_by_ref_name,
             workflow_uuid_2,
             polledTask.reference_task_name,
             "COMPLETED",
             "task 2 op 2nd wf"
         )
 
         # Update second task of first workflow
-        self.task_client.update_task_by_ref_name(
+        _retry_on_404(
+            self.task_client.update_task_by_ref_name,
             workflow_uuid_2, "simple_task_ref_2", "COMPLETED", "task 2 op 1st wf"
         )
 
@@ -593,7 +609,8 @@ def __test_task_execution_lifecycle(self):
         polledTask = self.task_client.poll_task(TASK_TYPE)
 
         # Update second task of second workflow
-        self.task_client.update_task_sync(
+        _retry_on_404(
+            self.task_client.update_task_sync,
             workflow_uuid, "simple_task_ref_2", "COMPLETED", "task 1 op 2nd wf"
         )
 

tests/integration/metadata/test_workflow_definition.py

@@ -1,6 +1,8 @@
+import time
 from typing import List
 
 from conductor.client.http.models import TaskDef
+from conductor.client.http.rest import ApiException
 from conductor.client.http.models.start_workflow_request import StartWorkflowRequest
 from conductor.client.workflow.conductor_workflow import ConductorWorkflow
 from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
@@ -68,7 +70,23 @@ def test_kitchensink_workflow_registration(workflow_executor: WorkflowExecutor)
     if type(workflow_id) != str or workflow_id == '':
         raise Exception(f'failed to start workflow, name: {WORKFLOW_NAME}')
 
-    workflow_executor.terminate(workflow_id=workflow_id, reason="End test")
+    _terminate_with_retry(workflow_executor, workflow_id, reason="End test")
+
+
+def _terminate_with_retry(workflow_executor: WorkflowExecutor, workflow_id: str,
+                          reason: str, retries: int = 5) -> None:
+    # Terminating a workflow that the server is still evaluating can transiently
+    # return 423 Locked ("Unable to acquire lock"). Retry with backoff so the
+    # test tolerates that race instead of failing.
+    for attempt in range(retries):
+        try:
+            workflow_executor.terminate(workflow_id=workflow_id, reason=reason)
+            return
+        except ApiException as e:
+            if e.status == 423 and attempt < retries - 1:
+                time.sleep(1 << attempt)
+                continue
+            raise
 
 
 def generate_simple_task(id: int) -> SimpleTask:

tests/integration/test_async_lease_extension.py

@@ -26,6 +26,8 @@
 import time
 import unittest
 
+import pytest
+
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
 
 from conductor.client.automator.task_handler import TaskHandler, get_registered_workers
@@ -138,6 +140,7 @@ async def async_lease_fast_no_hb(job_id: str) -> dict:
 
 # -- Test class --------------------------------------------------------------
 
+@pytest.mark.slow_async
 class TestAsyncLeaseExtension(unittest.TestCase):
 
     # Only the workers defined in THIS module. Used to scope the TaskHandler so

tests/integration/test_authorization_client_intg.py

@@ -532,7 +532,9 @@ def test_38_create_role(self):
 
         request = CreateOrUpdateRoleRequest()
         request.name = self.test_role_name
-        request.permissions = ["workflow:read"]
+        # Permission strings must be in the server's ACCESS_RESOURCE form
+        # (e.g. READ_WORKFLOW_DEF); "workflow:read" is rejected as invalid.
+        request.permissions = ["READ_WORKFLOW_DEF"]
 
         result = self.client.create_role(request)
 
@@ -553,7 +555,8 @@ def test_40_update_role(self):
 
         request = CreateOrUpdateRoleRequest()
         request.name = self.test_role_name
-        request.permissions = ["workflow:read", "workflow:execute"]
+        # Valid ACCESS_RESOURCE permission strings (see test_38_create_role).
+        request.permissions = ["READ_WORKFLOW_DEF", "EXECUTE_WORKFLOW_DEF"]
 
         result = self.client.update_role(self.test_role_name, request)
 

tests/integration/test_authorization_complete.py

@@ -292,7 +292,9 @@ def test_user_lifecycle(self, auth_client, test_run_id, cleanup_tracker):
             target_type="WORKFLOW_DEF",
             target_id="test-workflow"
         )
-        assert isinstance(result, bool)
+        # The server returns a per-access permission map
+        # (e.g. {'CREATE': True, 'READ': True, ...}), not a single bool.
+        assert isinstance(result, dict)
 
 
 class TestGroupManagement:

tests/integration/test_comprehensive_e2e.py

@@ -28,7 +28,6 @@
 import os
 import sys
 import time
-import threading
 import unittest
 from dataclasses import dataclass
 from typing import Optional, List, Dict, Union
@@ -144,6 +143,7 @@ def setUpClass(cls):
         )
 
         cls.workers_started = False
+        cls.task_handler = None
 
     def test_01_create_workflow(self):
         """Create test workflow."""
@@ -174,22 +174,23 @@ def test_01_create_workflow(self):
     def test_02_start_workers(self):
         """Start workers and verify they initialize."""
         print("\n" + "="*80 + "\nTEST 2: Start Workers\n" + "="*80)
-        
-        def run_workers():
-            with TaskHandler(
-                configuration=self.config,
-                metrics_settings=self.metrics_settings,
-                scan_for_annotated_workers=True,
-                event_listeners=[self.event_collector]
-            ) as handler:
-                handler.start_processes()
-                time.sleep(90)  # Run for test duration
-                handler.stop_processes()
-        
-        thread = threading.Thread(target=run_workers, daemon=True)
-        thread.start()
+
+        # Start the workers and keep the handler on the class so it stays alive
+        # for the remaining tests and is stopped deterministically in
+        # tearDownClass. (A previous version ran the handler in a daemon thread
+        # with a fixed 90s sleep; if the suite finished sooner, the worker
+        # processes were never stopped and the interpreter hung at exit joining
+        # them.)
+        handler = TaskHandler(
+            configuration=self.config,
+            metrics_settings=self.metrics_settings,
+            scan_for_annotated_workers=True,
+            event_listeners=[self.event_collector]
+        )
+        handler.start_processes()
+        self.__class__.task_handler = handler
         time.sleep(5)  # Wait for startup
-        
+
         print("✓ Workers started")
         self.__class__.workers_started = True
         self.assertTrue(self.workers_started)
@@ -422,6 +423,10 @@ def test_08_summary_assertions(self):
 
     @classmethod
     def tearDownClass(cls):
+        handler = getattr(cls, 'task_handler', None)
+        if handler is not None:
+            handler.stop_processes()
+            cls.task_handler = None
         if os.path.exists(cls.metrics_dir):
             import shutil
             shutil.rmtree(cls.metrics_dir)