airbyte-ci: do not eagerly fetch connector secrets (#38615)

Author: alafanechere

airbyte-ci/connectors/pipelines/README.md

@@ -305,7 +305,7 @@ flowchart TD
 | `--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.                                                                                              |
 | `--<step-id>.<extra-parameter>=<extra-parameter-value>` | True     |               | You can pass extra parameters for specific test steps. More details in the extra parameters section below                                                                                                |
-| `--ci-requirements`                                     | False    |               |                                                                                                                                                                                                          | Output the CI requirements as a JSON payload. It is used to determine the CI runner to use. 
+| `--ci-requirements`                                     | False    |               |                                                                                                                                                                                                          | Output the CI requirements as a JSON payload. It is used to determine the CI runner to use.
 
 Note:
 
@@ -748,6 +748,7 @@ E.G.: running Poe tasks on the modified internal packages of the current branch:
 
 | Version | PR                                                         | Description                                                                                                                  |
 | ------- | ---------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------- |
+| 4.15.1  | [#38615](https://github.com/airbytehq/airbyte/pull/38615)      | Do not eagerly fetch connector secrets.                                                                                      |
 | 4.15.0  | [#38322](https://github.com/airbytehq/airbyte/pull/38322)  | Introduce a SecretStore abstraction to fetch connector secrets from metadata files.                                          |
 | 4.14.1  | [#38582](https://github.com/airbytehq/airbyte/pull/38582)  | Fixed bugs in `up_to_date` flags, `pull_request` version change logic.                                                       |
 | 4.14.0  | [#38281](https://github.com/airbytehq/airbyte/pull/38281)  | Conditionally run test suites according to `connectorTestSuitesOptions` in metadata files.                                   |

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

@@ -26,12 +26,13 @@
 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
-from pipelines.models.secrets import LocalDirectorySecretStore, Secret, SecretStore
+from pipelines.models.secrets import LocalDirectorySecretStore, Secret, SecretNotFoundError, SecretStore
+from pydash import find  # type: ignore
 
 if TYPE_CHECKING:
+    from logging import Logger
     from pathlib import Path as NativePath
     from typing import Dict, FrozenSet, List, Optional, Sequence
-
 # These test suite names are declared in metadata.yaml files
 TEST_SUITE_NAME_TO_STEP_ID = {
     "unitTests": CONNECTOR_TEST_STEP_ID.UNIT,
@@ -127,7 +128,6 @@ def __init__(
         self.s3_build_cache_secret_key = s3_build_cache_secret_key
         self.concurrent_cat = concurrent_cat
         self.targeted_platforms = targeted_platforms
-
         super().__init__(
             pipeline_name=pipeline_name,
             is_local=is_local,
@@ -151,6 +151,7 @@ def __init__(
             enable_report_auto_open=enable_report_auto_open,
             secret_stores=secret_stores,
         )
+        self.step_id_to_secrets_mapping = self._get_step_id_to_secret_mapping()
 
     @property
     def modified_files(self) -> FrozenSet[NativePath]:
@@ -232,6 +233,124 @@ async def get_connector_dir(self, exclude: Optional[List[str]] = None, include:
         vanilla_connector_dir = self.get_repo_dir(str(self.connector.code_directory), exclude=exclude, include=include)
         return await vanilla_connector_dir.with_timestamps(1)
 
+    @staticmethod
+    def _handle_missing_secret_store(
+        secret_info: Dict[str, str | Dict[str, str]], raise_on_missing: bool, logger: Optional[Logger] = None
+    ) -> None:
+        assert isinstance(secret_info["secretStore"], dict), "The secretStore field must be a dict"
+        message = f"Secret {secret_info['name']} can't be retrieved as {secret_info['secretStore']['alias']} is not available"
+        if raise_on_missing:
+            raise SecretNotFoundError(message)
+        if logger is not None:
+            logger.warn(message)
+
+    @staticmethod
+    def _process_secret(
+        secret_info: Dict[str, str | Dict[str, str]],
+        secret_stores: Dict[str, SecretStore],
+        raise_on_missing: bool,
+        logger: Optional[Logger] = None,
+    ) -> Optional[Secret]:
+        assert isinstance(secret_info["secretStore"], dict), "The secretStore field must be a dict"
+        secret_store_alias = secret_info["secretStore"]["alias"]
+        if secret_store_alias not in secret_stores:
+            ConnectorContext._handle_missing_secret_store(secret_info, raise_on_missing, logger)
+            return None
+        else:
+            # All these asserts and casting are there to make MyPy happy
+            # The dict structure being nested MyPy can't figure if the values are str or dict
+            assert isinstance(secret_info["name"], str), "The secret name field must be a string"
+            if file_name := secret_info.get("fileName"):
+                assert isinstance(secret_info["fileName"], str), "The secret fileName must be a string"
+                file_name = str(secret_info["fileName"])
+            else:
+                file_name = None
+            return Secret(secret_info["name"], secret_stores[secret_store_alias], file_name=file_name)
+
+    @staticmethod
+    def get_secrets_from_connector_test_suites_option(
+        connector_test_suites_options: List[Dict[str, str | Dict[str, List[Dict[str, str | Dict[str, str]]]]]],
+        suite_name: str,
+        secret_stores: Dict[str, SecretStore],
+        raise_on_missing_secret_store: bool = True,
+        logger: Logger | None = None,
+    ) -> List[Secret]:
+        """Get secrets declared in metadata connectorTestSuitesOptions for a test suite name.
+        It will use the secret store alias declared in connectorTestSuitesOptions.
+        If the secret store is not available a warning or and error could be raised according to the raise_on_missing_secret_store parameter value.
+        We usually want to raise an error when running in CI context and log a warning when running locally, as locally we can fallback on local secrets.
+
+        Args:
+            connector_test_suites_options (List[Dict[str, str  |  Dict]]): The connector under test test suite options
+            suite_name (str): The test suite name
+            secret_stores (Dict[str, SecretStore]): The available secrets stores
+            raise_on_missing_secret_store (bool, optional): Raise an error if the secret store declared in the connectorTestSuitesOptions is not available. Defaults to True.
+            logger (Logger | None, optional): Logger to log a warning if the secret store declared in the connectorTestSuitesOptions is not available. Defaults to None.
+
+        Raises:
+            SecretNotFoundError: Raised  if the secret store declared in the connectorTestSuitesOptions is not available and raise_on_missing_secret_store is truthy.
+
+        Returns:
+            List[Secret]: List of secrets declared in the connectorTestSuitesOptions for a test suite name.
+        """
+        secrets: List[Secret] = []
+        enabled_test_suite = find(connector_test_suites_options, lambda x: x["suite"] == suite_name)
+
+        if enabled_test_suite and "testSecrets" in enabled_test_suite:
+            for secret_info in enabled_test_suite["testSecrets"]:
+                if secret := ConnectorContext._process_secret(secret_info, secret_stores, raise_on_missing_secret_store, logger):
+                    secrets.append(secret)
+
+        return secrets
+
+    def get_connector_secrets_for_test_suite(
+        self, test_suite_name: str, connector_test_suites_options: List, local_secrets: List[Secret]
+    ) -> List[Secret]:
+        """Get secrets to use for a test suite.
+        Always merge secrets declared in metadata's connectorTestSuiteOptions with secrets declared locally.
+
+        Args:
+            test_suite_name (str): Name of the test suite to get secrets for
+            context (ConnectorContext): The current connector context
+            connector_test_suites_options (Dict): The current connector test suite options (from metadata)
+            local_secrets (List[Secret]): The local connector secrets.
+
+        Returns:
+            List[Secret]: Secrets to use to run the passed test suite name.
+        """
+        return (
+            self.get_secrets_from_connector_test_suites_option(
+                connector_test_suites_options,
+                test_suite_name,
+                self.secret_stores,
+                raise_on_missing_secret_store=self.is_ci,
+                logger=self.logger,
+            )
+            + local_secrets
+        )
+
+    def _get_step_id_to_secret_mapping(self) -> Dict[CONNECTOR_TEST_STEP_ID, List[Secret]]:
+        step_id_to_secrets: Dict[CONNECTOR_TEST_STEP_ID, List[Secret]] = {
+            CONNECTOR_TEST_STEP_ID.UNIT: [],
+            CONNECTOR_TEST_STEP_ID.INTEGRATION: [],
+            CONNECTOR_TEST_STEP_ID.ACCEPTANCE: [],
+        }
+        local_secrets = self.local_secret_store.get_all_secrets() if self.local_secret_store else []
+        connector_test_suites_options = self.metadata.get("connectorTestSuitesOptions", [])
+
+        keep_steps = set(self.run_step_options.keep_steps or [])
+        skip_steps = set(self.run_step_options.skip_steps or [])
+
+        for test_suite_name, step_id in TEST_SUITE_NAME_TO_STEP_ID.items():
+            if step_id in keep_steps or (not keep_steps and step_id not in skip_steps):
+                step_id_to_secrets[step_id] = self.get_connector_secrets_for_test_suite(
+                    test_suite_name, connector_test_suites_options, local_secrets
+                )
+        return step_id_to_secrets
+
+    def get_secrets_for_step_id(self, step_id: CONNECTOR_TEST_STEP_ID) -> List[Secret]:
+        return self.step_id_to_secrets_mapping.get(step_id, [])
+
     async def __aexit__(
         self, exception_type: Optional[type[BaseException]], exception_value: Optional[BaseException], traceback: Optional[TracebackType]
     ) -> bool:

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

@@ -6,7 +6,6 @@
 from copy import deepcopy
 from typing import TYPE_CHECKING
 
-import yaml  # type: ignore
 from base_images import version_registry  # type: ignore
 from connector_ops.utils import ConnectorLanguage  # type: ignore
 from dagger import Directory

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

@@ -12,7 +12,6 @@
 import toml
 from connector_ops.utils import ConnectorLanguage  # type: ignore
 from jinja2 import Environment, PackageLoader, select_autoescape
-from pipelines.airbyte_ci.connectors.build_image.steps.python_connectors import BuildConnectorImages
 from pipelines.airbyte_ci.connectors.bump_version.pipeline import AddChangelogEntry, SetConnectorVersion, get_bumped_version
 from pipelines.airbyte_ci.connectors.consts import CONNECTOR_TEST_STEP_ID
 from pipelines.airbyte_ci.connectors.context import ConnectorContext, PipelineContext

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

@@ -5,14 +5,13 @@
 """This module groups steps made to run tests agnostic to a connector language."""
 
 import datetime
-import logging
 import os
 import time
 from abc import ABC, abstractmethod
 from functools import cached_property
 from pathlib import Path
 from textwrap import dedent
-from typing import ClassVar, Dict, List, Optional
+from typing import ClassVar, List, Optional
 
 import requests  # type: ignore
 import semver
@@ -26,106 +25,8 @@
 from pipelines.dagger.actions import secrets
 from pipelines.dagger.actions.python.poetry import with_poetry
 from pipelines.helpers.utils import METADATA_FILE_NAME, get_exec_result
-from pipelines.models.secrets import Secret, SecretNotFoundError, SecretStore
+from pipelines.models.secrets import Secret
 from pipelines.models.steps import STEP_PARAMS, MountPath, Step, StepResult, StepStatus
-from pydash import find  # type: ignore
-
-
-def _handle_missing_secret_store(
-    secret_info: Dict[str, str | Dict[str, str]], raise_on_missing: bool, logger: Optional[logging.Logger] = None
-) -> None:
-    assert isinstance(secret_info["secretStore"], dict), "The secretStore field must be a dict"
-    message = f"Secret {secret_info['name']} can't be retrieved as {secret_info['secretStore']['alias']} is not available"
-    if raise_on_missing:
-        raise SecretNotFoundError(message)
-    if logger is not None:
-        logger.warn(message)
-
-
-def _process_secret(
-    secret_info: Dict[str, str | Dict[str, str]],
-    secret_stores: Dict[str, SecretStore],
-    raise_on_missing: bool,
-    logger: Optional[logging.Logger] = None,
-) -> Optional[Secret]:
-    assert isinstance(secret_info["secretStore"], dict), "The secretStore field must be a dict"
-    secret_store_alias = secret_info["secretStore"]["alias"]
-    if secret_store_alias not in secret_stores:
-        _handle_missing_secret_store(secret_info, raise_on_missing, logger)
-        return None
-    else:
-        # All these asserts and casting are there to make MyPy happy
-        # The dict structure being nested MyPy can't figure if the values are str or dict
-        assert isinstance(secret_info["name"], str), "The secret name field must be a string"
-        if file_name := secret_info.get("fileName"):
-            assert isinstance(secret_info["fileName"], str), "The secret fileName must be a string"
-            file_name = str(secret_info["fileName"])
-        else:
-            file_name = None
-        return Secret(secret_info["name"], secret_stores[secret_store_alias], file_name=file_name)
-
-
-def get_secrets_from_connector_test_suites_option(
-    connector_test_suites_options: List[Dict[str, str | Dict[str, List[Dict[str, str | Dict[str, str]]]]]],
-    suite_name: str,
-    secret_stores: Dict[str, SecretStore],
-    raise_on_missing_secret_store: bool = True,
-    logger: logging.Logger | None = None,
-) -> List[Secret]:
-    """Get secrets declared in metadata connectorTestSuitesOptions for a test suite name.
-    It will use the secret store alias declared in connectorTestSuitesOptions.
-    If the secret store is not available a warning or and error could be raised according to the raise_on_missing_secret_store parameter value.
-    We usually want to raise an error when running in CI context and log a warning when running locally, as locally we can fallback on local secrets.
-
-    Args:
-        connector_test_suites_options (List[Dict[str, str  |  Dict]]): The connector under test test suite options
-        suite_name (str): The test suite name
-        secret_stores (Dict[str, SecretStore]): The available secrets stores
-        raise_on_missing_secret_store (bool, optional): Raise an error if the secret store declared in the connectorTestSuitesOptions is not available. Defaults to True.
-        logger (logging.Logger | None, optional): Logger to log a warning if the secret store declared in the connectorTestSuitesOptions is not available. Defaults to None.
-
-    Raises:
-        SecretNotFoundError: Raised  if the secret store declared in the connectorTestSuitesOptions is not available and raise_on_missing_secret_store is truthy.
-
-    Returns:
-        List[Secret]: List of secrets declared in the connectorTestSuitesOptions for a test suite name.
-    """
-    secrets: List[Secret] = []
-    enabled_test_suite = find(connector_test_suites_options, lambda x: x["suite"] == suite_name)
-
-    if enabled_test_suite and "testSecrets" in enabled_test_suite:
-        for secret_info in enabled_test_suite["testSecrets"]:
-            if secret := _process_secret(secret_info, secret_stores, raise_on_missing_secret_store, logger):
-                secrets.append(secret)
-
-    return secrets
-
-
-def get_connector_secrets_for_test_suite(
-    test_suite_name: str, context: ConnectorContext, connector_test_suites_options: List, local_secrets: List[Secret]
-) -> List[Secret]:
-    """Get secrets to use for a test suite.
-    Always merge secrets declared in metadata's connectorTestSuiteOptions with secrets declared locally.
-
-    Args:
-        test_suite_name (str): Name of the test suite to get secrets for
-        context (ConnectorContext): The current connector context
-        connector_test_suites_options (Dict): The current connector test suite options (from metadata)
-        local_secrets (List[Secret]): The local connector secrets.
-
-    Returns:
-        List[Secret]: Secrets to use to run the passed test suite name.
-    """
-    return (
-        get_secrets_from_connector_test_suites_option(
-            connector_test_suites_options,
-            test_suite_name,
-            context.secret_stores,
-            raise_on_missing_secret_store=context.is_ci,
-            logger=context.logger,
-        )
-        + local_secrets
-    )
 
 
 class VersionCheck(Step, ABC):