Rename to run_step

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/context.py

@@ -20,7 +20,7 @@
 from pipelines.helpers.connectors.modifed import ConnectorWithModifiedFiles
 from pipelines.helpers.github import update_commit_status_check
 from pipelines.helpers.slack import send_message_to_webhook
-from pipelines.helpers.steps import RunStepOptions
+from pipelines.helpers.run_steps import RunStepOptions
 from pipelines.helpers.utils import METADATA_FILE_NAME
 from pipelines.models.contexts import PipelineContext
 

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/test/commands.py

@@ -12,7 +12,7 @@
 from pipelines.cli.dagger_pipeline_command import DaggerPipelineCommand
 from pipelines.consts import ContextState
 from pipelines.helpers.github import update_global_commit_status_check_for_tests
-from pipelines.helpers.steps import RunStepOptions
+from pipelines.helpers.run_steps import RunStepOptions
 from pipelines.helpers.utils import fail_if_missing_docker_hub_creds
 
 

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/test/pipeline.py

@@ -12,7 +12,7 @@
 from pipelines.airbyte_ci.connectors.test.steps import java_connectors, python_connectors
 from pipelines.airbyte_ci.connectors.test.steps.common import QaChecks, VersionFollowsSemverCheck, VersionIncrementCheck
 from pipelines.airbyte_ci.metadata.pipeline import MetadataValidation
-from pipelines.helpers.steps import StepToRun, run_steps
+from pipelines.helpers.run_steps import StepToRun, run_steps
 
 LANGUAGE_MAPPING = {
     "get_test_steps": {

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/test/steps/java_connectors.py

@@ -21,7 +21,7 @@
 from pipelines.consts import LOCAL_BUILD_PLATFORM
 from pipelines.dagger.actions import secrets
 from pipelines.dagger.actions.system import docker
-from pipelines.helpers.steps import StepToRun
+from pipelines.helpers.run_steps import StepToRun
 from pipelines.helpers.utils import export_container_to_tarball
 from pipelines.models.steps import StepResult, StepStatus
 

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/test/steps/python_connectors.py

@@ -17,7 +17,7 @@
 from pipelines.airbyte_ci.connectors.test.steps.common import AcceptanceTests, CheckBaseImageIsUsed
 from pipelines.consts import LOCAL_BUILD_PLATFORM, PYPROJECT_TOML_FILE_PATH
 from pipelines.dagger.actions import secrets
-from pipelines.helpers.steps import StepToRun
+from pipelines.helpers.run_steps import StepToRun
 from pipelines.models.steps import Step, StepResult, StepStatus
 
 

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/metadata/pipeline.py

@@ -12,7 +12,7 @@
 from pipelines.consts import DOCS_DIRECTORY_ROOT_PATH, INTERNAL_TOOL_PATHS
 from pipelines.dagger.actions.python.common import with_pip_packages
 from pipelines.dagger.containers.python import with_python_base
-from pipelines.helpers.steps import StepToRun, run_steps
+from pipelines.helpers.run_steps import StepToRun, run_steps
 from pipelines.helpers.utils import DAGGER_CONFIG, get_secret_host_variable
 from pipelines.models.reports import Report
 from pipelines.models.steps import MountPath, Step, StepResult

airbyte-ci/connectors/pipelines/pipelines/helpers/steps.py → airbyte-ci/connectors/pipelines/pipelines/helpers/run_steps.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 from dataclasses import dataclass, field
+import inspect
 from pipelines import main_logger
 
 from typing import TYPE_CHECKING, Callable, Dict, List, Tuple, Union
@@ -36,7 +37,7 @@ class StepToRun:
     depends_on: List[str] = field(default_factory=list)
 
 async def evaluate_run_args(args: ARGS_TYPE, results: Dict[str, StepResult]) -> Dict:
-    if asyncer.is_coroutine(args):
+    if inspect.iscoroutinefunction(args):
         return await args(results)
     elif callable(args):
         return args(results)

airbyte-ci/connectors/pipelines/pipelines/models/contexts.py

@@ -18,7 +18,7 @@
 from pipelines.helpers.gcs import sanitize_gcs_credentials
 from pipelines.helpers.github import update_commit_status_check
 from pipelines.helpers.slack import send_message_to_webhook
-from pipelines.helpers.steps import RunStepOptions
+from pipelines.helpers.run_steps import RunStepOptions
 from pipelines.helpers.utils import AIRBYTE_REPO_URL
 from pipelines.models.reports import Report
 

airbyte-ci/connectors/pipelines/tests/test_helpers/test_run_steps.py

@@ -0,0 +1,87 @@
+import pytest
+from pipelines.models.contexts import PipelineContext
+from pipelines.models.steps import Step, StepResult, StepStatus
+from pipelines.helpers.run_steps import RunStepOptions, StepToRun, run_steps
+
+test_context = PipelineContext(
+    pipeline_name="test",
+    is_local=True,
+    git_branch="test",
+    git_revision="test"
+)
+
+class TestStep(Step):
+    title = "Test Step"
+    async def _run(self, result_status = StepStatus.SUCCESS) -> StepResult:
+        return StepResult(self, result_status)
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("steps, expected_results, options", [
+    (
+        [
+            StepToRun(id="step1", step=TestStep(test_context)),
+            StepToRun(id="step2", step=TestStep(test_context)),
+            StepToRun(id="step3", step=TestStep(test_context)),
+            StepToRun(id="step4", step=TestStep(test_context)),
+        ],
+        {
+            "step1": StepStatus.SUCCESS,
+            "step2": StepStatus.SUCCESS,
+            "step3": StepStatus.SUCCESS,
+            "step4": StepStatus.SUCCESS
+        },
+        RunStepOptions(fail_fast=True)
+    ),
+    (
+        [
+            StepToRun(id="step1", step=TestStep(test_context)),
+            [
+                StepToRun(id="step2", step=TestStep(test_context)),
+                StepToRun(id="step3", step=TestStep(test_context)),
+            ],
+            StepToRun(id="step4", step=TestStep(test_context)),
+        ],
+        {
+            "step1": StepStatus.SUCCESS,
+            "step2": StepStatus.SUCCESS,
+            "step3": StepStatus.SUCCESS,
+            "step4": StepStatus.SUCCESS
+        },
+        RunStepOptions(fail_fast=True)
+    ),
+    (
+        [
+            StepToRun(id="step1", step=TestStep(test_context)),
+            StepToRun(id="step2", step=TestStep(test_context), args={"result_status": StepStatus.FAILURE}),
+            StepToRun(id="step3", step=TestStep(test_context)),
+            StepToRun(id="step4", step=TestStep(test_context)),
+        ],
+        {
+            "step1": StepStatus.SUCCESS,
+            "step2": StepStatus.FAILURE,
+            "step3": StepStatus.SKIPPED,
+            "step4": StepStatus.SKIPPED
+        },
+        RunStepOptions(fail_fast=True)
+    )
+    (
+        [
+            StepToRun(id="step1", step=TestStep(test_context)),
+            StepToRun(id="step2", step=TestStep(test_context), args={"result_status": StepStatus.FAILURE}),
+            StepToRun(id="step3", step=TestStep(test_context)),
+            StepToRun(id="step4", step=TestStep(test_context)),
+        ],
+        {
+            "step1": StepStatus.SUCCESS,
+            "step2": StepStatus.FAILURE,
+            "step3": StepStatus.SUCCESS,
+            "step4": StepStatus.SUCCESS
+        },
+        RunStepOptions(fail_fast=False)
+    )
+])
+async def test_run_steps_sequential(steps, expected_results, options):
+    results = await run_steps(steps, options=options)
+
+    for step_id, expected_status in expected_results.items():
+        assert results[step_id].status == expected_status