airbyte-ci: run live tests against connection listed in metadata (#42574)

Author: clnoll

.github/workflows/connectors_tests.yml

@@ -86,8 +86,7 @@ jobs:
       - name: Check PAT rate limits
         run: |
           ./tools/bin/find_non_rate_limited_PAT \
-            ${{ secrets.GH_PAT_BUILD_RUNNER_OSS }} \
-            ${{ secrets.GH_PAT_BUILD_RUNNER_BACKUP }}
+            ${{ secrets.GH_PAT_MAINTENANCE_OSS }}
       - name: Extract branch name [WORKFLOW DISPATCH]
         shell: bash
         if: github.event_name == 'workflow_dispatch'
@@ -126,6 +125,7 @@ jobs:
           docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
           docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}
+          gcp_integration_tester_credentials: ${{ secrets.GCLOUD_INTEGRATION_TESTER }}
           sentry_dsn: ${{ secrets.SENTRY_AIRBYTE_CI_DSN }}
           git_branch: ${{ github.head_ref }}
           git_revision: ${{ steps.fetch_last_commit_id_pr.outputs.commit_id }}

airbyte-ci/connectors/pipelines/README.md

@@ -773,6 +773,7 @@ E.G.: running Poe tasks on the modified internal packages of the current branch:
 
 | Version | PR                                                         | Description                                                                                                                  |
 | ------- | ---------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------- |
+| 4.27.0  | [#42574](https://github.com/airbytehq/airbyte/pull/42574)  | Live tests: run from connectors test pipeline for connectors with sandbox connections                                        |
 | 4.26.1  | [#42905](https://github.com/airbytehq/airbyte/pull/42905)  | Rename the docker cache volume to avoid using the corrupted previous volume.                                                 |
 | 4.26.0  | [#42849](https://github.com/airbytehq/airbyte/pull/42849)  | Send publish failures messages to `#connector-publish-failures`                                                              |
 | 4.25.4  | [#42463](https://github.com/airbytehq/airbyte/pull/42463)  | Add validation before live test runs                                                                                         |

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

@@ -26,9 +26,6 @@
 GITHUB_GLOBAL_CONTEXT_FOR_TESTS = "Connectors CI tests"
 GITHUB_GLOBAL_DESCRIPTION_FOR_TESTS = "Running connectors tests"
 REGRESSION_TEST_MANUAL_APPROVAL_CONTEXT = "Regression Test Results Reviewed and Approved"
-TESTS_SKIPPED_BY_DEFAULT = [
-    CONNECTOR_TEST_STEP_ID.CONNECTOR_LIVE_TESTS,
-]
 
 
 @click.command(
@@ -120,7 +117,6 @@ async def test(
         raise click.UsageError("Cannot use both --only-step and --skip-step at the same time.")
     if not only_steps:
         skip_steps = list(skip_steps)
-        skip_steps += TESTS_SKIPPED_BY_DEFAULT
     if ctx.obj["is_ci"]:
         fail_if_missing_docker_hub_creds(ctx)
 

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

@@ -19,6 +19,7 @@
     "unitTests": CONNECTOR_TEST_STEP_ID.UNIT,
     "integrationTests": CONNECTOR_TEST_STEP_ID.INTEGRATION,
     "acceptanceTests": CONNECTOR_TEST_STEP_ID.ACCEPTANCE,
+    "liveTests": CONNECTOR_TEST_STEP_ID.CONNECTOR_LIVE_TESTS,
 }
 
 

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

@@ -13,7 +13,7 @@
 from functools import cached_property
 from pathlib import Path
 from textwrap import dedent
-from typing import ClassVar, List, Optional, Set
+from typing import Any, ClassVar, Dict, List, Optional, Set
 
 import requests  # type: ignore
 import semver
@@ -445,7 +445,7 @@ async def _run(self, current_acceptance_tests_result: StepResult) -> StepResult:
 
 
 class LiveTestSuite(Enum):
-    ALL = "all"
+    ALL = "live"
     REGRESSION = "regression"
     VALIDATION = "validation"
 
@@ -557,13 +557,10 @@ def __init__(self, context: ConnectorContext) -> None:
         self.connector_image = context.docker_image.split(":")[0]
         options = self.context.run_step_options.step_params.get(CONNECTOR_TEST_STEP_ID.CONNECTOR_LIVE_TESTS, {})
 
-        self.connection_id = self.context.run_step_options.get_item_or_default(options, "connection-id", None)
-        self.pr_url = self.context.run_step_options.get_item_or_default(options, "pr-url", None)
+        self.test_suite = self.context.run_step_options.get_item_or_default(options, "test-suite", LiveTestSuite.REGRESSION.value)
+        self.connection_id = self._get_connection_id(options)
+        self.pr_url = self._get_pr_url(options)
 
-        if not self.connection_id and self.pr_url:
-            raise ValueError("`connection-id` and `pr-url` are required to run live tests.")
-
-        self.test_suite = self.context.run_step_options.get_item_or_default(options, "test-suite", LiveTestSuite.ALL.value)
         self.test_dir = self.test_suite_to_dir[LiveTestSuite(self.test_suite)]
         self.control_version = self.context.run_step_options.get_item_or_default(options, "control-version", None)
         self.target_version = self.context.run_step_options.get_item_or_default(options, "target-version", "dev")
@@ -573,6 +570,26 @@ def __init__(self, context: ConnectorContext) -> None:
         self.connection_subset = self.context.run_step_options.get_item_or_default(options, "connection-subset", "sandboxes")
         self.run_id = os.getenv("GITHUB_RUN_ID") or str(int(time.time()))
 
+    def _get_connection_id(self, options: Dict[str, List[Any]]) -> Optional[str]:
+        if self.context.is_pr:
+            connection_id = self._get_connection_from_test_connections()
+            self.logger.info(
+                f"Context is {self.context.ci_context}; got connection_id={connection_id} from metadata.yaml liveTests testConnections."
+            )
+        else:
+            connection_id = self.context.run_step_options.get_item_or_default(options, "connection-id", None)
+            self.logger.info(f"Context is {self.context.ci_context}; got connection_id={connection_id} from input options.")
+        return connection_id
+
+    def _get_pr_url(self, options: Dict[str, List[Any]]) -> Optional[str]:
+        if self.context.is_pr:
+            pull_request = self.context.pull_request.url if self.context.pull_request else None
+            self.logger.info(f"Context is {self.context.ci_context}; got pull_request={pull_request} from context.")
+        else:
+            pull_request = self.context.run_step_options.get_item_or_default(options, "pr-url", None)
+            self.logger.info(f"Context is {self.context.ci_context}; got pull_request={pull_request} from input options.")
+        return pull_request
+
     def _validate_job_can_run(self) -> None:
         connector_type = self.context.connector.metadata.get("connectorType")
         connector_subtype = self.context.connector.metadata.get("connectorSubtype")
@@ -582,6 +599,30 @@ def _validate_job_can_run(self) -> None:
                 self.connection_subset == "sandboxes"
             ), f"Live tests for database sources may only be run against sandbox connections, got `connection_subset={self.connection_subset}`."
 
+        assert self.connection_id, "`connection-id` is required to run live tests."
+        assert self.pr_url, "`pr_url` is required to run live tests."
+
+        if self.context.is_pr:
+            connection_id_is_valid = False
+            for test_suite in self.context.connector.metadata.get("connectorTestSuitesOptions", []):
+                if test_suite["suite"] == "liveTests":
+                    assert self.connection_id in [
+                        option["id"] for option in test_suite.get("testConnections", [])
+                    ], f"Connection ID {self.connection_id} was not in the list of valid test connections."
+                    connection_id_is_valid = True
+                    break
+            assert connection_id_is_valid, f"Connection ID {self.connection_id} is not a valid sandbox connection ID."
+
+    def _get_connection_from_test_connections(self) -> Optional[str]:
+        for test_suite in self.context.connector.metadata.get("connectorTestSuitesOptions", []):
+            if test_suite["suite"] == "liveTests":
+                for option in test_suite.get("testConnections", []):
+                    connection_id = option["id"]
+                    connection_name = option["name"]
+                    self.logger.info(f"Using connection name={connection_name}; id={connection_id}")
+                    return connection_id
+        return None
+
     async def _run(self, connector_under_test_container: Container) -> StepResult:
         """Run the regression test suite.
 
@@ -594,9 +635,10 @@ async def _run(self, connector_under_test_container: Container) -> StepResult:
         try:
             self._validate_job_can_run()
         except AssertionError as exc:
+            self.logger.info(f"Skipping live tests for {self.context.connector.technical_name} due to validation error {str(exc)}.")
             return StepResult(
                 step=self,
-                status=StepStatus.FAILURE,
+                status=StepStatus.SKIPPED,
                 exc_info=exc,
             )
 
@@ -624,6 +666,7 @@ async def _run(self, connector_under_test_container: Container) -> StepResult:
             stdout=stdout,
             output=container,
             report=regression_test_report,
+            consider_in_overall_status=False if self.context.is_pr else True,
         )
 
     async def _build_test_container(self, target_container_id: str) -> Container:

airbyte-ci/connectors/pipelines/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "pipelines"
-version = "4.26.1"
+version = "4.27.0"
 description = "Packaged maintained by the connector operations team to perform CI for connectors' pipelines"
 authors = ["Airbyte <[email protected]>"]