further measures to ensure no contention between test runs

Author: chrishagglund-ship-it

tests/integration/test_async_lease_extension.py

@@ -25,6 +25,7 @@
 import sys
 import time
 import unittest
+from uuid import uuid4
 
 import pytest
 
@@ -57,15 +58,25 @@
 # Number of fast tasks for performance comparison
 PERF_TASK_COUNT = 5
 
+# Per-run suffix so this suite's task/workflow names don't collide with other
+# runs (or other PRs/developers) on the shared dev server. With fixed names,
+# concurrent runs poll the same queues and steal/strand each other's tasks,
+# producing non-deterministic SCHEDULED/RUNNING failures.
+RUN_ID = uuid4().hex[:8]
+HEARTBEAT_TASK = f'async_lease_heartbeat_task_{RUN_ID}'
+NO_HEARTBEAT_TASK = f'async_lease_no_heartbeat_task_{RUN_ID}'
+FAST_HB_TASK = f'async_lease_fast_with_hb_{RUN_ID}'
+FAST_NO_HB_TASK = f'async_lease_fast_no_hb_{RUN_ID}'
+
 
 # -- Async Workers -----------------------------------------------------------
 
 @worker_task(
-    task_definition_name='async_lease_heartbeat_task',
+    task_definition_name=HEARTBEAT_TASK,
     lease_extend_enabled=True,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_heartbeat_task',
+        name=HEARTBEAT_TASK,
         response_timeout_seconds=RESPONSE_TIMEOUT_SECONDS,
         timeout_seconds=180,
         retry_count=0,
@@ -82,11 +93,11 @@ async def async_lease_heartbeat_task(job_id: str) -> dict:
 
 
 @worker_task(
-    task_definition_name='async_lease_no_heartbeat_task',
+    task_definition_name=NO_HEARTBEAT_TASK,
     lease_extend_enabled=False,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_no_heartbeat_task',
+        name=NO_HEARTBEAT_TASK,
         response_timeout_seconds=RESPONSE_TIMEOUT_SECONDS,
         timeout_seconds=120,
         retry_count=0,
@@ -103,11 +114,11 @@ async def async_lease_no_heartbeat_task(job_id: str) -> dict:
 
 
 @worker_task(
-    task_definition_name='async_lease_fast_with_hb',
+    task_definition_name=FAST_HB_TASK,
     lease_extend_enabled=True,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_fast_with_hb',
+        name=FAST_HB_TASK,
         response_timeout_seconds=60,
         timeout_seconds=120,
         retry_count=0,
@@ -121,11 +132,11 @@ async def async_lease_fast_with_hb(job_id: str) -> dict:
 
 
 @worker_task(
-    task_definition_name='async_lease_fast_no_hb',
+    task_definition_name=FAST_NO_HB_TASK,
     lease_extend_enabled=False,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_fast_no_hb',
+        name=FAST_NO_HB_TASK,
         response_timeout_seconds=60,
         timeout_seconds=120,
         retry_count=0,
@@ -147,10 +158,10 @@ class TestAsyncLeaseExtension(unittest.TestCase):
     # it doesn't spin up every @worker_task registered across the imported test
     # suite (that was ~24 worker processes started/torn down per test).
     WORKER_TASK_NAMES = {
-        'async_lease_heartbeat_task',
-        'async_lease_no_heartbeat_task',
-        'async_lease_fast_with_hb',
-        'async_lease_fast_no_hb',
+        HEARTBEAT_TASK,
+        NO_HEARTBEAT_TASK,
+        FAST_HB_TASK,
+        FAST_NO_HB_TASK,
     }
 
     @classmethod
@@ -224,14 +235,25 @@ def _start_workflow(self, wf_name, job_id):
         logger.info("Started workflow %s: %s", wf_name, wf_id)
         return wf_id
 
+    TERMINAL_STATES = ('COMPLETED', 'FAILED', 'TIMED_OUT', 'TERMINATED')
+
     def _wait_for_workflow(self, wf_id, timeout_seconds=90):
-        """Poll until workflow reaches a terminal state."""
+        """Poll until workflow reaches a terminal state. If it doesn't within
+        the budget, dump server-side diagnostics so the ensuing assertion shows
+        *why* (e.g. a task stuck in SCHEDULED with no poller) rather than only a
+        bare status mismatch.
+        """
         for _ in range(timeout_seconds):
             wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
-            if wf.status in ('COMPLETED', 'FAILED', 'TIMED_OUT', 'TERMINATED'):
+            if wf.status in self.TERMINAL_STATES:
                 return wf
             time.sleep(1)
-        return self.workflow_client.get_workflow(wf_id, include_tasks=True)
+        wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+        if wf.status not in self.TERMINAL_STATES:
+            print(f"  [diag] workflow {wf_id} still {wf.status} "
+                  f"after {timeout_seconds}s")
+            self._dump_workflow_diagnostics(wf)
+        return wf
 
     def _dump_workflow_diagnostics(self, wf):
         """Print task statuses plus server-side poll data / queue size so a
@@ -273,8 +295,8 @@ def test_01_async_with_heartbeat_completes(self):
         print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_name = 'test_async_lease_heartbeat'
-        self._register_workflow(wf_name, 'async_lease_heartbeat_task')
+        wf_name = f'test_async_lease_heartbeat_{RUN_ID}'
+        self._register_workflow(wf_name, HEARTBEAT_TASK)
 
         wf_id = self._start_workflow(wf_name, 'ASYNC-HB-001')
         wf = self._wait_for_workflow(wf_id, timeout_seconds=80)
@@ -288,7 +310,7 @@ def test_01_async_with_heartbeat_completes(self):
                          f"Workflow should COMPLETE with heartbeat, got {wf.status}")
 
         tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
-        task = tasks_by_ref.get('async_lease_heartbeat_task_ref')
+        task = tasks_by_ref.get(f'{HEARTBEAT_TASK}_ref')
         self.assertIsNotNone(task)
         self.assertEqual(task.status, 'COMPLETED')
         self.assertEqual(task.output_data.get('job_id'), 'ASYNC-HB-001')
@@ -302,8 +324,8 @@ def test_02_async_without_heartbeat_times_out(self):
         print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_name = 'test_async_lease_no_heartbeat'
-        self._register_workflow(wf_name, 'async_lease_no_heartbeat_task')
+        wf_name = f'test_async_lease_no_heartbeat_{RUN_ID}'
+        self._register_workflow(wf_name, NO_HEARTBEAT_TASK)
 
         wf_id = self._start_workflow(wf_name, 'ASYNC-NOHB-001')
         wf = self._wait_for_workflow(wf_id, timeout_seconds=80)
@@ -313,14 +335,11 @@ def test_02_async_without_heartbeat_times_out(self):
         for task in (wf.tasks or []):
             print(f"  Task {task.task_def_name}: {task.status}")
 
-        if wf.status not in ('FAILED', 'TIMED_OUT'):
-            self._dump_workflow_diagnostics(wf)
-
         self.assertIn(wf.status, ('FAILED', 'TIMED_OUT'),
                       f"Workflow should FAIL/TIMEOUT without heartbeat, got {wf.status}")
 
         tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
-        task = tasks_by_ref.get('async_lease_no_heartbeat_task_ref')
+        task = tasks_by_ref.get(f'{NO_HEARTBEAT_TASK}_ref')
         self.assertIsNotNone(task)
         self.assertIn(task.status, ('TIMED_OUT', 'FAILED', 'CANCELED'),
                       f"Task should be TIMED_OUT/FAILED, got {task.status}")
@@ -333,10 +352,10 @@ def test_03_no_performance_overhead(self):
         print(f"  Running {PERF_TASK_COUNT} tasks each, sleep={FAST_TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_with_hb = 'test_async_perf_with_hb'
-        wf_no_hb = 'test_async_perf_no_hb'
-        self._register_workflow(wf_with_hb, 'async_lease_fast_with_hb')
-        self._register_workflow(wf_no_hb, 'async_lease_fast_no_hb')
+        wf_with_hb = f'test_async_perf_with_hb_{RUN_ID}'
+        wf_no_hb = f'test_async_perf_no_hb_{RUN_ID}'
+        self._register_workflow(wf_with_hb, FAST_HB_TASK)
+        self._register_workflow(wf_no_hb, FAST_NO_HB_TASK)
 
         # Run tasks WITH heartbeat tracking
         hb_workflow_ids = []

tests/integration/test_comprehensive_e2e.py

@@ -32,6 +32,7 @@
 from dataclasses import dataclass
 from typing import Optional, List, Dict, Union
 from collections import defaultdict
+from uuid import uuid4
 
 from conductor.client.worker.exception import NonRetryableException
 
@@ -70,15 +71,28 @@ def on_task_execution_failure(self, e): self.events['exec_failed'].append(e)
     def on_task_update_failure(self, e): self.events['update_failed'].append(e)
 
 
+# Per-run suffix so this suite's task/workflow names don't collide with other
+# runs (or other PRs/developers) on the shared dev server. With fixed names,
+# concurrent runs poll the same queues and steal/strand each other's tasks,
+# producing non-deterministic SCHEDULED failures.
+RUN_ID = uuid4().hex[:8]
+SYNC_BASIC = f'sync_basic_{RUN_ID}'
+ASYNC_BASIC = f'async_basic_{RUN_ID}'
+COMPLEX_SCHEMA = f'complex_schema_{RUN_ID}'
+TASK_IN_PROGRESS = f'task_in_progress_{RUN_ID}'
+FAILING_TASK = f'failing_task_{RUN_ID}'
+WF_NAME = f'e2e_comprehensive_test_{RUN_ID}'
+
+
 # Test workers covering all scenarios
-@worker_task(task_definition_name='sync_basic', thread_count=5, register_task_def=True)
+@worker_task(task_definition_name=SYNC_BASIC, thread_count=5, register_task_def=True)
 def sync_basic(value: str, count: int) -> dict:
     ctx = get_task_context()
     ctx.add_log(f"Processing {value}")
     return {'value': value, 'count': count, 'worker': 'sync'}
 
 
-@worker_task(task_definition_name='async_basic', thread_count=10, register_task_def=True)
+@worker_task(task_definition_name=ASYNC_BASIC, thread_count=10, register_task_def=True)
 async def async_basic(message: str) -> dict:
     await asyncio.sleep(0.1)
     return {'message': message, 'worker': 'async'}
@@ -92,17 +106,17 @@ class OrderData:
 
 
 @worker_task(
-    task_definition_name='complex_schema',
+    task_definition_name=COMPLEX_SCHEMA,
     register_task_def=True,
     strict_schema=True,
-    task_def=TaskDef(name='complex_schema', retry_count=2, timeout_policy='RETRY')
+    task_def=TaskDef(name=COMPLEX_SCHEMA, retry_count=2, timeout_policy='RETRY')
 )
 def complex_schema(data: OrderData, optional: Optional[str]) -> dict:
     assert data.id is not None
     return {'id': data.id, 'amount': data.amount, 'tag_count': len(data.tags)}
 
 
-@worker_task(task_definition_name='task_in_progress')
+@worker_task(task_definition_name=TASK_IN_PROGRESS)
 def task_in_progress(job_id: str) -> Union[dict, TaskInProgress]:
     ctx = get_task_context()
     polls = ctx.get_poll_count()
@@ -111,7 +125,7 @@ def task_in_progress(job_id: str) -> Union[dict, TaskInProgress]:
     return {'job_id': job_id, 'polls': polls}
 
 
-@worker_task(task_definition_name='failing_task')
+@worker_task(task_definition_name=FAILING_TASK)
 def failing_task(should_fail: bool) -> dict:
     if should_fail:
         raise NonRetryableException("Test failure")
@@ -126,11 +140,11 @@ class TestComprehensiveE2E(unittest.TestCase):
     # SCHEDULED forever (which is exactly how a non-starting task_in_progress
     # worker manifested as a "4 != 5 tasks" failure in CI).
     EXPECTED_WORKERS = (
-        'sync_basic',
-        'async_basic',
-        'complex_schema',
-        'task_in_progress',
-        'failing_task',
+        SYNC_BASIC,
+        ASYNC_BASIC,
+        COMPLEX_SCHEMA,
+        TASK_IN_PROGRESS,
+        FAILING_TASK,
     )
 
     @classmethod
@@ -163,18 +177,18 @@ def test_01_create_workflow(self):
 
         metadata_client = OrkesMetadataClient(self.config)
 
-        workflow = WorkflowDef(name='e2e_comprehensive_test', version=1)
+        workflow = WorkflowDef(name=WF_NAME, version=1)
         tasks = [
-            WorkflowTask(name='sync_basic', task_reference_name='sync_1', 
+            WorkflowTask(name=SYNC_BASIC, task_reference_name='sync_1', 
                         input_parameters={'value': 'test', 'count': 1}),
-            WorkflowTask(name='async_basic', task_reference_name='async_1',
+            WorkflowTask(name=ASYNC_BASIC, task_reference_name='async_1',
                         input_parameters={'message': 'hello'}),
-            WorkflowTask(name='complex_schema', task_reference_name='complex_1',
+            WorkflowTask(name=COMPLEX_SCHEMA, task_reference_name='complex_1',
                         input_parameters={'data': {'id': '123', 'amount': 99.99, 'tags': ['a', 'b']}, 
                                         'optional': None}),
-            WorkflowTask(name='task_in_progress', task_reference_name='tip_1',
+            WorkflowTask(name=TASK_IN_PROGRESS, task_reference_name='tip_1',
                         input_parameters={'job_id': 'JOB1'}),
-            WorkflowTask(name='failing_task', task_reference_name='fail_1',
+            WorkflowTask(name=FAILING_TASK, task_reference_name='fail_1',
                         input_parameters={'should_fail': True}),
         ]
         workflow._tasks = tasks
@@ -260,7 +274,7 @@ def _run_workflow_to_terminal(self, workflow_client, timeout_s=90):
         workflow (which is what produced the flaky "4 != 5 tasks" failure).
         """
         req = StartWorkflowRequest()
-        req.name = 'e2e_comprehensive_test'
+        req.name = WF_NAME
         req.version = 1
         req.input = {}
 
@@ -375,7 +389,7 @@ def test_05_verify_task_definitions(self):
 
         metadata_client = OrkesMetadataClient(self.config)
 
-        tasks_to_check = ['sync_basic', 'async_basic', 'complex_schema']
+        tasks_to_check = [SYNC_BASIC, ASYNC_BASIC, COMPLEX_SCHEMA]
 
         for task_name in tasks_to_check:
             task_def = metadata_client.get_task_def(task_name)
@@ -384,15 +398,15 @@ def test_05_verify_task_definitions(self):
             print(f"✓ Task definition exists: {task_name}")
 
             # Check schemas if they should exist
-            if task_name in ['sync_basic', 'async_basic', 'complex_schema']:
+            if task_name in tasks_to_check:
                 # These have type hints, should have schemas
                 if hasattr(task_def, 'input_schema') and task_def.input_schema:
                     print(f"  ✓ Has input schema")
                 if hasattr(task_def, 'output_schema') and task_def.output_schema:
                     print(f"  ✓ Has output schema")
 
         # Check complex_schema has TaskDef configuration
-        complex_def = metadata_client.get_task_def('complex_schema')
+        complex_def = metadata_client.get_task_def(COMPLEX_SCHEMA)
         self.assertEqual(complex_def.retry_count, 2, "Retry count from task_def should be applied")
         self.assertEqual(complex_def.timeout_policy, 'RETRY', "Timeout policy should be set")