Add setup and teardown hooks to CATs (#40551)

airbyte-integrations/bases/connector-acceptance-test/connector_acceptance_test/config.py

@@ -158,6 +158,14 @@ def no_unsupported_types_when_skip_test(cls, skip_test: bool, values: Dict[str,
         return skip_test
 
 
+class SetupTeardownConfig(BaseConfig):
+    setup_teardown_dockerfile_path: str = Field(
+        None, description="Path to Dockerfile to run before each test for which a config is provided."
+    )
+    setup_command: List[str] = Field(None, description="Command for running the setup/teardown container for setup")
+    teardown_command: List[str] = Field(None, description="Command for running the setup/teardown container for teardown")
+
+
 class BasicReadTestConfig(BaseConfig):
     config_path: str = config_path
     deployment_mode: Optional[str] = deployment_mode
@@ -181,6 +189,10 @@ class BasicReadTestConfig(BaseConfig):
         default_factory=FileTypesConfig,
         description="For file-based connectors, unsupported by source file types can be configured or a test can be skipped at all",
     )
+    setup_teardown_config: Optional[SetupTeardownConfig] = Field(
+        default_factory=SetupTeardownConfig,
+        description="Information required to run a setup & teardown Docker container before each test.",
+    )
 
 
 class FullRefreshConfig(BaseConfig):
@@ -197,6 +209,10 @@ class FullRefreshConfig(BaseConfig):
     ignored_fields: Optional[Mapping[str, List[IgnoredFieldsConfiguration]]] = Field(
         description="For each stream, list of fields path ignoring in sequential reads test"
     )
+    setup_teardown_config: Optional[SetupTeardownConfig] = Field(
+        default_factory=SetupTeardownConfig,
+        description="Information required to run a setup & teardown Docker container before each test.",
+    )
 
 
 class FutureStateCursorFormatStreamConfiguration(BaseConfig):
@@ -233,6 +249,10 @@ class IncrementalConfig(BaseConfig):
     skip_comprehensive_incremental_tests: Optional[bool] = Field(
         description="Determines whether to skip more granular testing for incremental syncs", default=False
     )
+    setup_teardown_config: Optional[SetupTeardownConfig] = Field(
+        default_factory=SetupTeardownConfig,
+        description="Information required to run a setup & teardown Docker container before each test.",
+    )
 
     class Config:
         smart_union = True

airbyte-integrations/bases/connector-acceptance-test/connector_acceptance_test/conftest.py

@@ -19,7 +19,13 @@
 import pytest
 from airbyte_protocol.models import AirbyteRecordMessage, AirbyteStream, ConfiguredAirbyteCatalog, ConnectorSpecification, Type
 from connector_acceptance_test.base import BaseTest
-from connector_acceptance_test.config import Config, EmptyStreamConfiguration, ExpectedRecordsConfig, IgnoredFieldsConfiguration
+from connector_acceptance_test.config import (
+    Config,
+    EmptyStreamConfiguration,
+    ExpectedRecordsConfig,
+    IgnoredFieldsConfiguration,
+    SetupTeardownConfig,
+)
 from connector_acceptance_test.tests import TestBasicRead
 from connector_acceptance_test.utils import (
     SecretDict,
@@ -29,6 +35,7 @@
     filter_output,
     load_config,
     load_yaml_or_json_path,
+    setup_and_teardown_runner,
 )
 
 
@@ -124,6 +131,13 @@ def connector_config_fixture(base_path, connector_config_path) -> SecretDict:
     return SecretDict(json.loads(contents))
 
 
+@pytest.fixture(name="setup_teardown_dockerfile_config")
+def setup_teardown_dockerfile_config_fixture(inputs, base_path, acceptance_test_config) -> Optional[SetupTeardownConfig]:
+    """Fixture with connector's setup/teardown Dockerfile path, if it exists."""
+    if hasattr(inputs, "setup_teardown_config"):
+        return getattr(inputs, "setup_teardown_config")
+
+
 @pytest.fixture(name="invalid_connector_config")
 def invalid_connector_config_fixture(base_path, invalid_connector_config_path) -> MutableMapping[str, Any]:
     """TODO: implement default value - generate from valid config"""
@@ -186,6 +200,33 @@ def docker_runner_fixture(
     )
 
 
+@pytest.fixture(autouse=True)
+async def setup_and_teardown(
+    base_path: Path,
+    connector_config: SecretDict,
+    dagger_client: dagger.Client,
+    setup_teardown_dockerfile_config: Optional[SetupTeardownConfig],
+):
+    setup_teardown_container = None
+    if setup_teardown_dockerfile_config:
+        logging.info("Running setup")
+        setup_teardown_container = await setup_and_teardown_runner.do_setup(
+            dagger_client,
+            base_path / setup_teardown_dockerfile_config.setup_teardown_dockerfile_path,
+            setup_teardown_dockerfile_config.setup_command,
+            connector_config,
+        )
+        print(f">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> {await setup_teardown_container.stdout()}")
+    yield None
+    if setup_teardown_container:
+        logging.info("Running teardown")
+        setup_teardown_container = await setup_and_teardown_runner.do_teardown(
+            setup_teardown_container,
+            setup_teardown_dockerfile_config.teardown_command,
+        )
+        print(f">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> {await setup_teardown_container.stdout()}")
+
+
 @pytest.fixture(name="previous_connector_image_name")
 def previous_connector_image_name_fixture(image_tag, inputs) -> str:
     """Fixture with previous connector image name to use for backward compatibility tests"""

airbyte-integrations/bases/connector-acceptance-test/connector_acceptance_test/utils/setup_and_teardown_runner.py

@@ -0,0 +1,46 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+import json
+import os
+from pathlib import Path
+from typing import List
+
+import dagger
+from connector_acceptance_test.utils import SecretDict
+
+IN_CONTAINER_CONFIG_PATH = Path("/tmp/config.json")
+IN_CONTAINER_OUTPUT_PATH = Path("/tmp/output.txt")
+
+
+async def _build_container(dagger_client: dagger.Client, dockerfile_path: Path) -> dagger.Container:
+    workspace = (
+        dagger_client.container()
+        .with_directory("/tmp", dagger_client.host().directory(os.path.dirname(dockerfile_path)))
+        .with_workdir("/tmp")
+        .with_file("/tmp/Dockerfile.cat_setup_teardown", dagger_client.host().file(str(dockerfile_path.expanduser())))
+        .directory("/tmp")
+    )
+    return await dagger_client.container().build(context=workspace, dockerfile="Dockerfile.cat_setup_teardown")
+
+
+async def _build_setup_container(dagger_client: dagger.Client, dockerfile_path: Path) -> dagger.Container:
+    return await _build_container(dagger_client, dockerfile_path)
+
+
+async def _run_with_config(container: dagger.Container, command: List[str], config: SecretDict) -> dagger.Container:
+    container = container.with_new_file(str(IN_CONTAINER_CONFIG_PATH), contents=json.dumps(dict(config)))
+    return await _run(container, command)
+
+
+async def _run(container: dagger.Container, command: List[str]) -> dagger.Container:
+    return await container.with_exec(command, skip_entrypoint=True)
+
+
+async def do_setup(dagger_client: dagger.Client, dockerfile_path: Path, command: List[str], connector_config: SecretDict):
+    return await _run_with_config(await _build_setup_container(dagger_client, dockerfile_path), command, connector_config)
+
+
+async def do_teardown(container: dagger.Container, command: List[str]):
+    return await _run(container, command)