Airbyte-ci: Add skippable connector test steps

airbyte-ci/connectors/pipelines/README.md

@@ -259,8 +259,8 @@ flowchart TD
 
 | Option              | Multiple | Default value | Description                                                                                                                                                                                              |
 | ------------------- | -------- | ------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `--skip-step/-x`    | True     |               | Skip steps by id e.g. `-x unit -x acceptance`                                                                                                         |
 | `--fail-fast`       | False    | False         | Abort after any tests fail, rather than continuing to run additional tests. Use this setting to confirm a known bug is fixed (or not), or when you only require a pass/fail result.                      |
-| `--fast-tests-only` | True     | False         | Run unit tests only, skipping integration tests or any tests explicitly tagged as slow. Use this for more frequent checks, when it is not feasible to run the entire test suite.                         |
 | `--code-tests-only` | True     | False         | Skip any tests not directly related to code updates. For instance, metadata checks, version bump checks, changelog verification, etc. Use this setting to help focus on code quality during development. |
 | `--concurrent-cat`  | False    | False         | Make CAT tests run concurrently using pytest-xdist. Be careful about source or destination API rate limits.                                                                                              |
 
@@ -502,6 +502,7 @@ This command runs the Python tests for a airbyte-ci poetry package.
 
 | Version | PR                                                         | Description                                                                                               |
 | ------- | ---------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- |
+| 2.11.0  | [#32188](https://github.com/airbytehq/airbyte/pull/32188)  | Add -x option to connector test to allow for skipping steps                                               |
 | 2.10.12 | [#33419](https://github.com/airbytehq/airbyte/pull/33419)  | Make ClickPipelineContext handle dagger logging.                                                          |
 | 2.10.11 | [#33497](https://github.com/airbytehq/airbyte/pull/33497)  | Consider nested .gitignore rules in format.                                                               |
 | 2.10.10 | [#33449](https://github.com/airbytehq/airbyte/pull/33449)  | Add generated metadata models to the default format ignore list.                                          |

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

@@ -0,0 +1,26 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+from enum import Enum
+
+
+class CONNECTOR_TEST_STEP_ID(str, Enum):
+    """
+    An enum for the different step ids of the connector test pipeline.
+    """
+
+    ACCEPTANCE = "acceptance"
+    BUILD_NORMALIZATION = "build_normalization"
+    BUILD_TAR = "build_tar"
+    BUILD = "build"
+    CHECK_BASE_IMAGE = "check_base_image"
+    INTEGRATION = "integration"
+    METADATA_VALIDATION = "metadata_validation"
+    QA_CHECKS = "qa_checks"
+    UNIT = "unit"
+    VERSION_FOLLOW_CHECK = "version_follow_check"
+    VERSION_INC_CHECK = "version_inc_check"
+    TEST_ORCHESTRATOR = "test_orchestrator"
+    DEPLOY_ORCHESTRATOR = "deploy_orchestrator"
+
+    def __str__(self) -> str:
+        return self.value

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

@@ -6,7 +6,7 @@
 
 from datetime import datetime
 from types import TracebackType
-from typing import Iterable, Optional
+from typing import Iterable, List, Optional
 
 import yaml
 from anyio import Path
@@ -18,6 +18,7 @@
 from pipelines.dagger.actions import secrets
 from pipelines.helpers.connectors.modifed import ConnectorWithModifiedFiles
 from pipelines.helpers.github import update_commit_status_check
+from pipelines.helpers.run_steps import RunStepOptions
 from pipelines.helpers.slack import send_message_to_webhook
 from pipelines.helpers.utils import METADATA_FILE_NAME
 from pipelines.models.contexts.pipeline_context import PipelineContext
@@ -50,8 +51,6 @@ def __init__(
         reporting_slack_channel: Optional[str] = None,
         pull_request: PullRequest = None,
         should_save_report: bool = True,
-        fail_fast: bool = False,
-        fast_tests_only: bool = False,
         code_tests_only: bool = False,
         use_local_cdk: bool = False,
         use_host_gradle_dist_tar: bool = False,
@@ -61,6 +60,7 @@ def __init__(
         s3_build_cache_access_key_id: Optional[str] = None,
         s3_build_cache_secret_key: Optional[str] = None,
         concurrent_cat: Optional[bool] = False,
+        run_step_options: RunStepOptions = RunStepOptions(),
         targeted_platforms: Optional[Iterable[Platform]] = BUILD_PLATFORMS,
     ):
         """Initialize a connector context.
@@ -80,8 +80,6 @@ def __init__(
             slack_webhook (Optional[str], optional): The slack webhook to send messages to. Defaults to None.
             reporting_slack_channel (Optional[str], optional): The slack channel to send messages to. Defaults to None.
             pull_request (PullRequest, optional): The pull request object if the pipeline was triggered by a pull request. Defaults to None.
-            fail_fast (bool, optional): Whether to fail fast. Defaults to False.
-            fast_tests_only (bool, optional): Whether to run only fast tests. Defaults to False.
             code_tests_only (bool, optional): Whether to ignore non-code tests like QA and metadata checks. Defaults to False.
             use_host_gradle_dist_tar (bool, optional): Used when developing java connectors with gradle. Defaults to False.
             enable_report_auto_open (bool, optional): Open HTML report in browser window. Defaults to True.
@@ -102,8 +100,6 @@ def __init__(
         self._updated_secrets_dir = None
         self.cdk_version = None
         self.should_save_report = should_save_report
-        self.fail_fast = fail_fast
-        self.fast_tests_only = fast_tests_only
         self.code_tests_only = code_tests_only
         self.use_local_cdk = use_local_cdk
         self.use_host_gradle_dist_tar = use_host_gradle_dist_tar
@@ -113,6 +109,7 @@ def __init__(
         self.s3_build_cache_access_key_id = s3_build_cache_access_key_id
         self.s3_build_cache_secret_key = s3_build_cache_secret_key
         self.concurrent_cat = concurrent_cat
+        self._connector_secrets = None
         self.targeted_platforms = targeted_platforms
 
         super().__init__(
@@ -131,6 +128,7 @@ def __init__(
             ci_gcs_credentials=ci_gcs_credentials,
             ci_git_user=ci_git_user,
             ci_github_access_token=ci_github_access_token,
+            run_step_options=run_step_options,
             enable_report_auto_open=enable_report_auto_open,
         )
 
@@ -210,6 +208,11 @@ def docker_hub_password_secret(self) -> Optional[Secret]:
             return None
         return self.dagger_client.set_secret("docker_hub_password", self.docker_hub_password)
 
+    async def get_connector_secrets(self):
+        if self._connector_secrets is None:
+            self._connector_secrets = await secrets.get_connector_secrets(self)
+        return self._connector_secrets
+
     async def get_connector_dir(self, exclude=None, include=None) -> Directory:
         """Get the connector under test source code directory.
 

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

@@ -3,15 +3,18 @@
 #
 
 import sys
+from typing import List
 
 import asyncclick as click
 from pipelines import main_logger
+from pipelines.airbyte_ci.connectors.consts import CONNECTOR_TEST_STEP_ID
 from pipelines.airbyte_ci.connectors.context import ConnectorContext
 from pipelines.airbyte_ci.connectors.pipeline import run_connectors_pipelines
 from pipelines.airbyte_ci.connectors.test.pipeline import run_connector_test_pipeline
 from pipelines.cli.dagger_pipeline_command import DaggerPipelineCommand
 from pipelines.consts import LOCAL_BUILD_PLATFORM, ContextState
 from pipelines.helpers.github import update_global_commit_status_check_for_tests
+from pipelines.helpers.run_steps import RunStepOptions
 from pipelines.helpers.utils import fail_if_missing_docker_hub_creds
 
 
@@ -30,27 +33,27 @@
     type=bool,
     is_flag=True,
 )
-@click.option(
-    "--fast-tests-only",
-    help="When enabled, slow tests are skipped.",
-    default=False,
-    type=bool,
-    is_flag=True,
-)
 @click.option(
     "--concurrent-cat",
     help="When enabled, the CAT tests will run concurrently. Be careful about rate limits",
     default=False,
     type=bool,
     is_flag=True,
 )
+@click.option(
+    "--skip-step",
+    "-x",
+    multiple=True,
+    type=click.Choice([step_id.value for step_id in CONNECTOR_TEST_STEP_ID]),
+    help="Skip a step by name. Can be used multiple times to skip multiple steps.",
+)
 @click.pass_context
 async def test(
     ctx: click.Context,
     code_tests_only: bool,
     fail_fast: bool,
-    fast_tests_only: bool,
     concurrent_cat: bool,
+    skip_step: List[str],
 ) -> bool:
     """Runs a test pipeline for the selected connectors.
 
@@ -70,6 +73,11 @@ async def test(
         update_global_commit_status_check_for_tests(ctx.obj, "success")
         return True
 
+    run_step_options = RunStepOptions(
+        fail_fast=fail_fast,
+        skip_steps=[CONNECTOR_TEST_STEP_ID(step_id) for step_id in skip_step],
+    )
+
     connectors_tests_contexts = [
         ConnectorContext(
             pipeline_name=f"Testing connector {connector.technical_name}",
@@ -86,19 +94,19 @@ async def test(
             ci_context=ctx.obj.get("ci_context"),
             pull_request=ctx.obj.get("pull_request"),
             ci_gcs_credentials=ctx.obj["ci_gcs_credentials"],
-            fail_fast=fail_fast,
-            fast_tests_only=fast_tests_only,
             code_tests_only=code_tests_only,
             use_local_cdk=ctx.obj.get("use_local_cdk"),
             s3_build_cache_access_key_id=ctx.obj.get("s3_build_cache_access_key_id"),
             s3_build_cache_secret_key=ctx.obj.get("s3_build_cache_secret_key"),
             docker_hub_username=ctx.obj.get("docker_hub_username"),
             docker_hub_password=ctx.obj.get("docker_hub_password"),
             concurrent_cat=concurrent_cat,
+            run_step_options=run_step_options,
             targeted_platforms=[LOCAL_BUILD_PLATFORM],
         )
         for connector in ctx.obj["selected_connectors_with_modified_files"]
     ]
+
     try:
         await run_connectors_pipelines(
             [connector_context for connector_context in connectors_tests_contexts],

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

@@ -3,101 +3,68 @@
 #
 """This module groups factory like functions to dispatch tests steps according to the connector under test language."""
 
-import itertools
 from typing import List
 
 import anyio
-import asyncer
-from connector_ops.utils import METADATA_FILE_NAME, ConnectorLanguage
+from connector_ops.utils import ConnectorLanguage
+from pipelines.airbyte_ci.connectors.consts import CONNECTOR_TEST_STEP_ID
 from pipelines.airbyte_ci.connectors.context import ConnectorContext
 from pipelines.airbyte_ci.connectors.reports import ConnectorReport
 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.models.steps import StepResult
+from pipelines.helpers.run_steps import StepToRun, run_steps
 
 LANGUAGE_MAPPING = {
-    "run_all_tests": {
-        ConnectorLanguage.PYTHON: python_connectors.run_all_tests,
-        ConnectorLanguage.LOW_CODE: python_connectors.run_all_tests,
-        ConnectorLanguage.JAVA: java_connectors.run_all_tests,
-    }
+    "get_test_steps": {
+        ConnectorLanguage.PYTHON: python_connectors.get_test_steps,
+        ConnectorLanguage.LOW_CODE: python_connectors.get_test_steps,
+        ConnectorLanguage.JAVA: java_connectors.get_test_steps,
+    },
 }
 
 
-async def run_metadata_validation(context: ConnectorContext) -> List[StepResult]:
-    """Run the metadata validation on a connector.
-    Args:
-        context (ConnectorContext): The current connector context.
-
-    Returns:
-        List[StepResult]: The results of the metadata validation steps.
-    """
-    return [await MetadataValidation(context).run()]
-
-
-async def run_version_checks(context: ConnectorContext) -> List[StepResult]:
-    """Run the version checks on a connector.
-
-    Args:
-        context (ConnectorContext): The current connector context.
-
-    Returns:
-        List[StepResult]: The results of the version checks steps.
-    """
-    return [await VersionFollowsSemverCheck(context).run(), await VersionIncrementCheck(context).run()]
-
-
-async def run_qa_checks(context: ConnectorContext) -> List[StepResult]:
-    """Run the QA checks on a connector.
+def get_test_steps(context: ConnectorContext) -> List[StepToRun]:
+    """Get all the tests steps according to the connector language.
 
     Args:
         context (ConnectorContext): The current connector context.
 
     Returns:
-        List[StepResult]: The results of the QA checks steps.
+        List[StepResult]: The list of tests steps.
     """
-    return [await QaChecks(context).run()]
-
-
-async def run_all_tests(context: ConnectorContext) -> List[StepResult]:
-    """Run all the tests steps according to the connector language.
-
-    Args:
-        context (ConnectorContext): The current connector context.
-
-    Returns:
-        List[StepResult]: The results of the tests steps.
-    """
-    if _run_all_tests := LANGUAGE_MAPPING["run_all_tests"].get(context.connector.language):
-        return await _run_all_tests(context)
+    if _get_test_steps := LANGUAGE_MAPPING["get_test_steps"].get(context.connector.language):
+        return _get_test_steps(context)
     else:
         context.logger.warning(f"No tests defined for connector language {context.connector.language}!")
         return []
 
 
-async def run_connector_test_pipeline(context: ConnectorContext, semaphore: anyio.Semaphore) -> ConnectorReport:
-    """Run a test pipeline for a single connector.
+async def run_connector_test_pipeline(context: ConnectorContext, semaphore: anyio.Semaphore):
+    """
+    Compute the steps to run for a connector test pipeline.
+    """
 
-    A visual DAG can be found on the README.md file of the pipelines modules.
+    steps_to_run = get_test_steps(context)
 
-    Args:
-        context (ConnectorContext): The initialized connector context.
+    if not context.code_tests_only:
+        steps_to_run += [
+            [
+                StepToRun(id=CONNECTOR_TEST_STEP_ID.METADATA_VALIDATION, step=MetadataValidation(context)),
+                StepToRun(id=CONNECTOR_TEST_STEP_ID.VERSION_FOLLOW_CHECK, step=VersionFollowsSemverCheck(context)),
+                StepToRun(id=CONNECTOR_TEST_STEP_ID.VERSION_INC_CHECK, step=VersionIncrementCheck(context)),
+                StepToRun(id=CONNECTOR_TEST_STEP_ID.QA_CHECKS, step=QaChecks(context)),
+            ]
+        ]
 
-    Returns:
-        ConnectorReport: The test reports holding tests results.
-    """
     async with semaphore:
         async with context:
-            async with asyncer.create_task_group() as task_group:
-                tasks = [task_group.soonify(run_all_tests)(context)]
-                if not context.code_tests_only:
-                    tasks += [
-                        task_group.soonify(run_metadata_validation)(context),
-                        task_group.soonify(run_version_checks)(context),
-                        task_group.soonify(run_qa_checks)(context),
-                    ]
-            results = list(itertools.chain(*(task.value for task in tasks)))
+            result_dict = await run_steps(
+                runnables=steps_to_run,
+                options=context.run_step_options,
+            )
+
+            results = result_dict.values()
             context.report = ConnectorReport(context, steps_results=results, name="TEST RESULTS")
 
         return context.report

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

@@ -287,7 +287,7 @@ async def _build_connector_acceptance_test(self, connector_under_test_container:
             .with_new_file("/tmp/container_id.txt", str(connector_container_id))
             .with_workdir("/test_input")
             .with_mounted_directory("/test_input", test_input)
-            .with_(await secrets.mounted_connector_secrets(self.context, "/test_input/secrets"))
+            .with_(await secrets.mounted_connector_secrets(self.context, self.CONTAINER_SECRETS_DIRECTORY))
         )
         if "_EXPERIMENTAL_DAGGER_RUNNER_HOST" in os.environ:
             self.context.logger.info("Using experimental dagger runner host to run CAT with dagger-in-dagger")