resilience

Author: chrishagglund-ship-it

tests/integration/client/orkes/test_orkes_clients.py

@@ -26,6 +26,7 @@
 from conductor.client.workflow.conductor_workflow import ConductorWorkflow
 from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
 from conductor.client.workflow.task.simple_task import SimpleTask
+from tests.integration.retry_helpers import retry_scenario
 
 SUFFIX = str(uuid())
 WORKFLOW_NAME = 'IntegrationTestOrkesClientsWf_' + SUFFIX
@@ -66,7 +67,7 @@ def __init__(self, configuration: Configuration):
         self.authorization_client = orkes_clients.get_authorization_client()
         self.workflow_id = None
 
-    def run(self) -> None:
+    def run(self, deadline=None) -> None:
         workflow = ConductorWorkflow(
             executor=self.workflow_executor,
             name=WORKFLOW_NAME,
@@ -77,13 +78,24 @@ def run(self) -> None:
         workflow >> SimpleTask("simple_task", "simple_task_ref")
         workflowDef = workflow.to_workflow_def()
 
-        self.test_workflow_lifecycle(workflowDef, workflow)
-        self.test_task_lifecycle()
-        self.test_secret_lifecycle()
-        self.test_scheduler_lifecycle(workflowDef)
-        self.test_application_lifecycle()
-        self.__test_unit_test_workflow()
-        self.test_user_group_permissions_lifecycle(workflowDef)
+        # Each lifecycle is a scenario: on a transient (status 0) blip against
+        # the shared dev server it retries from the top until the shared deadline
+        # passes (see retry_helpers.retry_scenario); real failures raise at once.
+        retry_scenario('test_workflow_lifecycle', self.test_workflow_lifecycle,
+                       workflowDef, workflow, deadline=deadline)
+        retry_scenario('test_task_lifecycle', self.test_task_lifecycle,
+                       deadline=deadline)
+        retry_scenario('test_secret_lifecycle', self.test_secret_lifecycle,
+                       deadline=deadline)
+        retry_scenario('test_scheduler_lifecycle', self.test_scheduler_lifecycle,
+                       workflowDef, deadline=deadline)
+        retry_scenario('test_application_lifecycle', self.test_application_lifecycle,
+                       deadline=deadline)
+        retry_scenario('__test_unit_test_workflow', self.__test_unit_test_workflow,
+                       deadline=deadline)
+        retry_scenario('test_user_group_permissions_lifecycle',
+                       self.test_user_group_permissions_lifecycle, workflowDef,
+                       deadline=deadline)
 
     def test_workflow_lifecycle(self, workflowDef, workflow):
         self.__test_register_workflow_definition(workflowDef)
@@ -382,6 +394,22 @@ def __test_unit_test_workflow(self):
         execution = self.workflow_client.test_workflow(testRequest)
         assert execution != None
 
+        print(
+            f"[test_workflow] workflow_id={getattr(execution, 'workflow_id', None)} "
+            f"status={execution.status} "
+            f"task_count={len(execution.tasks or [])}"
+        )
+        for t in (execution.tasks or []):
+            print(
+                f"[test_workflow]   task ref={t.reference_task_name} "
+                f"type={t.task_type} status={t.status} "
+                f"retried={getattr(t, 'retry_count', None)}"
+            )
+        if execution.status != "COMPLETED":
+            print(
+                f"[test_workflow] non-terminal output={getattr(execution, 'output', None)}"
+            )
+
         # Ensure workflow is completed successfully
         assert execution.status == "COMPLETED"
 
@@ -569,9 +597,9 @@ def __test_task_execution_lifecycle(self):
         self.task_client.add_task_log(polledTask.task_id, "Polled task...")
 
         taskExecLogs = self.task_client.get_task_logs(polledTask.task_id)
-        taskExecLogs[0].log == "Polled task..."
+        assert taskExecLogs[0].log == "Polled task..."
 
-        # First task of second workflow is in the queue
+        # First task of second workflow is still in the queue
         assert self.task_client.get_queue_size_for_task(TASK_TYPE) == 1
 
         taskResult = TaskResult(

tests/integration/metadata/test_workflow_definition.py

@@ -17,14 +17,17 @@
 from conductor.client.workflow.task.sub_workflow_task import SubWorkflowTask, InlineSubWorkflowTask
 from conductor.client.workflow.task.switch_task import SwitchTask
 from conductor.client.workflow.task.terminate_task import TerminateTask, WorkflowStatus
+from tests.integration.retry_helpers import retry_scenario
 
 WORKFLOW_NAME = 'python_test_workflow'
 TASK_NAME = 'python_test_simple_task'
 WORKFLOW_OWNER_EMAIL = "test@test"
 
 
-def run_workflow_definition_tests(workflow_executor: WorkflowExecutor) -> None:
-    test_kitchensink_workflow_registration(workflow_executor)
+def run_workflow_definition_tests(workflow_executor: WorkflowExecutor, deadline=None) -> None:
+    retry_scenario('test_kitchensink_workflow_registration',
+                   test_kitchensink_workflow_registration, workflow_executor,
+                   deadline=deadline)
 
 
 def generate_tasks_defs() -> List[TaskDef]:

tests/integration/test_async_lease_extension.py

@@ -414,6 +414,15 @@ def test_02_async_without_heartbeat_times_out(self):
                       f"Task should be TIMED_OUT/FAILED, got {task.status}")
         print("\n  PASS: Async task timed out as expected without heartbeat")
 
+    # NOTE: excluded from the long-* CI buckets (reusing the
+    # server_timeout_unreliable marker). This "overhead" test compares
+    # server-side task durations (end_time - start_time) on the shared sdkdev
+    # server, so it mostly measures queue/scheduling latency rather than the
+    # SDK's LeaseManager.track() bookkeeping (a dict insert). On a loaded server
+    # a fast task can stay non-terminal past the wait budget, so it flakes for
+    # reasons unrelated to the SDK. A deterministic in-process micro-benchmark
+    # would be the right home for the no-overhead claim if we want to keep it.
+    @pytest.mark.server_timeout_unreliable
     def test_03_no_performance_overhead(self):
         """Heartbeat tracking adds no meaningful overhead to fast async tasks."""
         print("\n" + "=" * 80)

tests/integration/test_workflow_client_intg.py

@@ -8,11 +8,11 @@
 from tests.integration.client.orkes.test_orkes_clients import TestOrkesClients
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
-from conductor.client.http.rest import ApiException
 from conductor.client.orkes.orkes_workflow_client import OrkesWorkflowClient
 from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
 from tests.integration.metadata.test_workflow_definition import run_workflow_definition_tests
 from tests.integration.workflow.test_workflow_execution import run_workflow_execution_tests
+from tests.integration.retry_helpers import DEFAULT_OVERALL_DEADLINE_SECONDS
 
 WORKFLOW_NAME = 'ut_wf'
 WORKFLOW_UUID = 'ut_wf_uuid'
@@ -33,52 +33,6 @@ def get_configuration():
     return configuration
 
 
-def _first_transient_api_exception(exc):
-    """Walk the exception chain (cause/context) and return the first transient
-    ApiException (status 0 / flagged transient), if any.
-
-    Inner test helpers sometimes catch an ApiException and re-raise it as a bare
-    ``Exception`` (losing the type), so we can't rely on the outermost exception
-    type alone. Implicit chaining still records the original on ``__context__``
-    (or ``__cause__`` when ``raise ... from`` is used), so we follow that chain.
-    """
-    seen = set()
-    cur = exc
-    while cur is not None and id(cur) not in seen:
-        seen.add(id(cur))
-        if isinstance(cur, ApiException) and (
-                getattr(cur, 'transient', False) or cur.status in (0, None)):
-            return cur
-        cur = cur.__cause__ or cur.__context__
-    return None
-
-
-def _run_tolerating_transient(label, func, *args, retries=3, **kwargs):
-    """Run a sub-suite, retrying only on a transient (status 0) transport blip
-    against the shared dev server.
-
-    A `(0)` ApiException is a raw connection/protocol hiccup (read timeout,
-    stale keep-alive, HTTP/2 GOAWAY race, client closed by a fork cleanup, etc.)
-    — not a real assertion or server failure — and was observed flaking the
-    `test-all` bucket. Retrying absorbs that network noise while still surfacing
-    genuine failures immediately (any non-transient error, or a transient one on
-    the final attempt, re-raises). We inspect the whole exception chain because
-    sub-suites may wrap the ApiException in a generic Exception.
-    """
-    for attempt in range(retries):
-        try:
-            return func(*args, **kwargs)
-        except Exception as e:
-            transient = _first_transient_api_exception(e)
-            if transient is not None and attempt < retries - 1:
-                logger.warning(
-                    'transient (%s) API error in %s (attempt %d/%d): %s; retrying',
-                    transient.status, label, attempt + 1, retries, transient)
-                time.sleep(2 ** attempt)
-                continue
-            raise
-
-
 class TestOrkesWorkflowClientIntg(unittest.TestCase):
 
     @classmethod
@@ -96,14 +50,14 @@ def test_all(self):
         configuration = self.config
         workflow_executor = WorkflowExecutor(configuration)
 
-        # test_async.test_async_method(api_client)
-        _run_tolerating_transient(
-            'workflow_definition_tests',
-            run_workflow_definition_tests, workflow_executor)
-        _run_tolerating_transient(
-            'workflow_execution_tests',
-            run_workflow_execution_tests, configuration, workflow_executor)
-        _run_tolerating_transient(
-            'orkes_clients',
-            TestOrkesClients(configuration=configuration).run)
+        # One shared wall-clock budget for the whole aggregate suite. Each
+        # sub-suite runs its scenarios once; a scenario that hits a transient
+        # (status 0) transport blip against the shared dev server retries until
+        # this deadline passes, with capped exponential backoff (see
+        # tests/integration/retry_helpers.py). Real failures raise immediately.
+        deadline = time.monotonic() + DEFAULT_OVERALL_DEADLINE_SECONDS
+
+        run_workflow_definition_tests(workflow_executor, deadline=deadline)
+        run_workflow_execution_tests(configuration, workflow_executor, deadline=deadline)
+        TestOrkesClients(configuration=configuration).run(deadline=deadline)
         logger.info('END: integration tests')

tests/integration/workflow/test_workflow_execution.py

@@ -12,6 +12,7 @@
 from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
 from conductor.client.workflow.task.simple_task import SimpleTask
 from tests.integration.resources.worker.python.python_worker import *
+from tests.integration.retry_helpers import retry_scenario
 
 WORKFLOW_NAME = "sdk_python_integration_test_workflow"
 WORKFLOW_DESCRIPTION = "Python SDK Integration Test"
@@ -36,7 +37,8 @@
 )
 
 
-def run_workflow_execution_tests(configuration: Configuration, workflow_executor: WorkflowExecutor):
+def run_workflow_execution_tests(configuration: Configuration, workflow_executor: WorkflowExecutor,
+                                 deadline=None):
     workers = [
         ClassWorker(TASK_NAME),
         ClassWorkerWithDomain(TASK_NAME),
@@ -59,23 +61,35 @@ def run_workflow_execution_tests(configuration: Configuration, workflow_executor
     # (they'd surface as an opaque generic Exception instead). finally cleans up
     # on both success and failure while letting the original error propagate.
     try:
-        scenario_get_workflow_by_correlation_ids(workflow_executor)
+        retry_scenario('scenario_get_workflow_by_correlation_ids',
+                       scenario_get_workflow_by_correlation_ids, workflow_executor,
+                       deadline=deadline)
         logger.debug('finished workflow correlation ids test')
-        scenario_workflow_registration(workflow_executor)
+        retry_scenario('scenario_workflow_registration',
+                       scenario_workflow_registration, workflow_executor,
+                       deadline=deadline)
         logger.debug('finished workflow registration tests')
-        scenario_workflow_execution(
+        retry_scenario(
+            'scenario_workflow_execution', scenario_workflow_execution,
             workflow_quantity=6,
             workflow_name=WORKFLOW_NAME,
             workflow_executor=workflow_executor,
-            workflow_completion_timeout=5.0
+            workflow_completion_timeout=5.0,
+            deadline=deadline,
         )
-        scenario_decorated_workers(workflow_executor)
+        retry_scenario('scenario_decorated_workers',
+                       scenario_decorated_workers, workflow_executor,
+                       deadline=deadline)
         logger.debug('finished decorated workers tests')
-        scenario_execute_workflow_async_features(workflow_executor)
+        retry_scenario('scenario_execute_workflow_async_features',
+                       scenario_execute_workflow_async_features, workflow_executor,
+                       deadline=deadline)
         logger.debug('finished execute_workflow reactive features tests')
-        scenario_execute_workflow_error_handling(workflow_executor)
+        retry_scenario('scenario_execute_workflow_error_handling',
+                       scenario_execute_workflow_error_handling, workflow_executor,
+                       deadline=deadline)
         logger.debug('finished execute_workflow error handling tests')
-        run_signal_tests(configuration, workflow_executor)
+        run_signal_tests(configuration, workflow_executor, deadline=deadline)
         logger.debug('finished signal API tests')
     finally:
         task_handler.stop_processes()
@@ -473,28 +487,44 @@ def _wait_for_workflow_completion(workflow_executor: WorkflowExecutor, workflow_
 
 # ===== SIGNAL TESTS =====
 
-def run_signal_tests(configuration: Configuration, workflow_executor: WorkflowExecutor):
-    """Run all signal API tests using WorkflowExecutor methods"""
+def run_signal_tests(configuration: Configuration, workflow_executor: WorkflowExecutor,
+                     deadline=None):
+    """Run all signal API tests using WorkflowExecutor methods.
+
+    Each scenario is retried at the scenario level on a transient blip (see
+    retry_scenario): a retry starts a fresh workflow and issues a fresh sync
+    signal, so the asserted SignalResponse is always from a signal this attempt
+    actually sent — no double-signalling of a single workflow.
+    """
     logger.info('START: Signal API tests using WorkflowExecutor')
 
     try:
         # Register signal test workflows (same as original test)
-        _register_signal_test_workflows(workflow_executor)
+        retry_scenario('_register_signal_test_workflows',
+                       _register_signal_test_workflows, workflow_executor,
+                       deadline=deadline)
 
         # Test sync signal with different return strategies
-        scenario_signal_target_workflow(workflow_executor)
-        scenario_signal_blocking_workflow(workflow_executor)
-        scenario_signal_blocking_task(workflow_executor)
-        scenario_signal_blocking_task_input(workflow_executor)
+        retry_scenario('scenario_signal_target_workflow',
+                       scenario_signal_target_workflow, workflow_executor, deadline=deadline)
+        retry_scenario('scenario_signal_blocking_workflow',
+                       scenario_signal_blocking_workflow, workflow_executor, deadline=deadline)
+        retry_scenario('scenario_signal_blocking_task',
+                       scenario_signal_blocking_task, workflow_executor, deadline=deadline)
+        retry_scenario('scenario_signal_blocking_task_input',
+                       scenario_signal_blocking_task_input, workflow_executor, deadline=deadline)
 
         # Test default return strategy
-        scenario_signal_default_strategy(workflow_executor)
+        retry_scenario('scenario_signal_default_strategy',
+                       scenario_signal_default_strategy, workflow_executor, deadline=deadline)
 
         # Test async signal
-        scenario_signal_async(workflow_executor)
+        retry_scenario('scenario_signal_async',
+                       scenario_signal_async, workflow_executor, deadline=deadline)
 
         # Test to_dict fix
-        scenario_signal_to_dict_fix(workflow_executor)
+        retry_scenario('scenario_signal_to_dict_fix',
+                       scenario_signal_to_dict_fix, workflow_executor, deadline=deadline)
 
         logger.info('All signal tests completed successfully')