working test of tests

Author: aaronsteers

airbyte_cdk/test/declarative/__init__.py

@@ -4,15 +4,3 @@
 This module provides fixtures and utilities for testing Airbyte sources and destinations
 in a declarative way.
 """
-
-from airbyte_cdk.test.declarative.test_suites import (
-    ConnectorTestSuiteBase,
-    DestinationTestSuiteBase,
-    SourceTestSuiteBase,
-)
-
-__all__ = [
-    "ConnectorTestSuiteBase",
-    "DestinationTestSuiteBase",
-    "SourceTestSuiteBase",
-]

airbyte_cdk/test/declarative/models/__init__.py

@@ -1,7 +1,7 @@
 from airbyte_cdk.test.declarative.models.scenario import (
-    AcceptanceTestScenario,
+    ConnectorTestScenario,
 )
 
 __all__ = [
-    "AcceptanceTestScenario",
+    "ConnectorTestScenario",
 ]

airbyte_cdk/test/declarative/models/scenario.py

@@ -16,7 +16,7 @@
 from pydantic import BaseModel
 
 
-class AcceptanceTestScenario(BaseModel):
+class ConnectorTestScenario(BaseModel):
     """Acceptance test instance, as a Pydantic model.
 
     This class represents an acceptance test instance, which is a single test case
@@ -32,17 +32,43 @@ class AcceptanceTestFileTypes(BaseModel):
         skip_test: bool
         bypass_reason: str
 
-    config_path: Path
+    config_path: Path | None = None
+    config_dict: dict | None = None
+
+    id: str | None = None
+
     configured_catalog_path: Path | None = None
     timeout_seconds: int | None = None
     expect_records: AcceptanceTestExpectRecords | None = None
     file_types: AcceptanceTestFileTypes | None = None
     status: Literal["succeed", "failed"] | None = None
 
+    def get_config_dict(self) -> dict:
+        """Return the config dictionary.
+
+        If a config dictionary has already been loaded, return it. Otherwise, load
+        Otherwise, load the config file and return the dictionary.
+        """
+        if self.config_dict:
+            return self.config_dict
+
+        if self.config_path:
+            return yaml.safe_load(self.config_path.read_text())
+
+        raise ValueError("No config dictionary or path provided.")
+
     @property
     def expect_exception(self) -> bool:
         return self.status and self.status == "failed"
 
     @property
     def instance_name(self) -> str:
         return self.config_path.stem
+
+    def __str__(self) -> str:
+        if self.id:
+            return f"'{self.id}' Test Scenario"
+        if self.config_path:
+            return f"'{self.config_path.name}' Test Scenario"
+
+        return f"'{hash(self)}' Test Scenario"

airbyte_cdk/test/declarative/test_suites/__init__.py

@@ -4,12 +4,21 @@
 Here we have base classes for a robust set of declarative connector test suites.
 """
 
-from airbyte_cdk.test.declarative.test_suites.connector_base import ConnectorTestSuiteBase
+from airbyte_cdk.test.declarative.test_suites.connector_base import (
+    ConnectorTestScenario,
+    generate_tests,
+)
+from airbyte_cdk.test.declarative.test_suites.declarative_sources import (
+    DeclarativeSourceTestSuite,
+)
 from airbyte_cdk.test.declarative.test_suites.destination_base import DestinationTestSuiteBase
 from airbyte_cdk.test.declarative.test_suites.source_base import SourceTestSuiteBase
 
 __all__ = [
+    "ConnectorTestScenario",
     "ConnectorTestSuiteBase",
+    "DeclarativeSourceTestSuite",
     "DestinationTestSuiteBase",
     "SourceTestSuiteBase",
+    "generate_tests",
 ]

airbyte_cdk/test/declarative/test_suites/connector_base.py

@@ -4,12 +4,16 @@
 from __future__ import annotations
 
 import abc
+import functools
+import inspect
+import sys
 from pathlib import Path
-from typing import Any, Literal
+from typing import Any, Callable, Literal
 
 import pytest
 import yaml
 from pydantic import BaseModel
+from typing_extensions import override
 
 from airbyte_cdk import Connector
 from airbyte_cdk.models import (
@@ -18,30 +22,55 @@
 )
 from airbyte_cdk.test import entrypoint_wrapper
 from airbyte_cdk.test.declarative.models import (
-    AcceptanceTestScenario,
+    ConnectorTestScenario,
 )
 from airbyte_cdk.test.declarative.utils.job_runner import run_test_job
 
-ACCEPTANCE_TEST_CONFIG_PATH = Path("acceptance-test-config.yml")
+ACCEPTANCE_TEST_CONFIG = "acceptance-test-config.yml"
 
 
-def _get_acceptance_tests(
-    category: str,
-    accept_test_config_path: Path = ACCEPTANCE_TEST_CONFIG_PATH,
-) -> list[AcceptanceTestScenario]:
-    all_tests_config = yaml.safe_load(accept_test_config_path.read_text())
-    if "acceptance_tests" not in all_tests_config:
-        raise ValueError(f"Acceptance tests config not found in {accept_test_config_path}")
-    if category not in all_tests_config["acceptance_tests"]:
-        return []
-    if "tests" not in all_tests_config["acceptance_tests"][category]:
-        raise ValueError(f"No tests found for category {category}")
+def generate_tests(metafunc):
+    """
+    A helper for pytest_generate_tests hook.
+
+    If a test method (in a class subclassed from our base class)
+    declares an argument 'instance', this function retrieves the
+    'scenarios' attribute from the test class and parametrizes that
+    test with the values from 'scenarios'.
+
+    ## Usage
+
+    ```python
+    from airbyte_cdk.test.declarative.test_suites.connector_base import (
+        generate_tests,
+        ConnectorTestSuiteBase,
+    )
+
+    def pytest_generate_tests(metafunc):
+        generate_tests(metafunc)
 
-    return [
-        AcceptanceTestScenario.model_validate(test)
-        for test in all_tests_config["acceptance_tests"][category]["tests"]
-        if "iam_role" not in test["config_path"]
-    ]
+    class TestMyConnector(ConnectorTestSuiteBase):
+        ...
+
+    ```
+    """
+    # Check if the test function requires an 'instance' argument
+    if "instance" in metafunc.fixturenames:
+        # Retrieve the test class
+        test_class = metafunc.cls
+        if test_class is None:
+            raise ValueError("Expected a class here.")
+        # Get the 'scenarios' attribute from the class
+        scenarios_attr = getattr(test_class, "get_scenarios", None)
+        if scenarios_attr is None:
+            raise ValueError(
+                f"Test class {test_class} does not have a 'scenarios' attribute. "
+                "Please define the 'scenarios' attribute in the test class."
+            )
+
+        scenarios = test_class.get_scenarios()
+        ids = [str(scenario) for scenario in scenarios]
+        metafunc.parametrize("instance", scenarios, ids=ids)
 
 
 class ConnectorTestSuiteBase(abc.ABC):
@@ -57,52 +86,101 @@ class ConnectorTestSuiteBase(abc.ABC):
     connector_class: type[Connector]
     """The connector class to test."""
 
-    # Public Methods - Subclasses may override these
-
-    @abc.abstractmethod
-    def new_connector(self, **kwargs: dict[str, Any]) -> Connector:
-        """Create a new connector instance.
+    working_dir: Path | None = None
+    """The root directory of the connector source code."""
+
+    @override
+    @classmethod
+    def create_connector(cls, scenario: ConnectorTestScenario) -> ConcurrentDeclarativeSource:
+        """Instantiate the connector class."""
+        return ConcurrentDeclarativeSource(
+            config=scenario.get_config_dict(),
+            catalog={},
+            state=None,
+            source_config={},
+        )
 
-        By default, this returns a new instance of the connector class. Subclasses
-        may override this method to generate a dynamic connector instance.
-        """
-        return self.connector_factory()
+    def run_test_scenario(
+        self,
+        verb: Literal["read", "check", "discover"],
+        test_scenario: ConnectorTestScenario,
+        *,
+        catalog: dict | None = None,
+    ) -> entrypoint_wrapper.EntrypointOutput:
+        """Run a test job from provided CLI args and return the result."""
+        return run_test_job(
+            self.create_connector(test_scenario),
+            verb,
+            test_instance=test_scenario,
+            catalog=catalog,
+        )
 
     # Test Definitions
 
-    @pytest.mark.parametrize(
-        "test_input,expected",
-        [
-            ("3+5", 8),
-            ("2+4", 6),
-            ("6*9", 54),
-        ],
-    )
-    def test_use_plugin_parametrized_test(
-        self,
-        test_input,
-        expected,
-    ):
-        assert eval(test_input) == expected
-
-    @pytest.mark.parametrize(
-        "instance",
-        _get_acceptance_tests("connection"),
-        ids=lambda instance: instance.instance_name,
-    )
     def test_check(
         self,
-        instance: AcceptanceTestScenario,
+        instance: ConnectorTestScenario,
     ) -> None:
         """Run `connection` acceptance tests."""
-        result: entrypoint_wrapper.EntrypointOutput = run_test_job(
-            self.new_connector(),
+        result = self.run_test_scenario(
             "check",
-            test_instance=instance,
+            test_scenario=instance,
         )
         conn_status_messages: list[AirbyteMessage] = [
             msg for msg in result._messages if msg.type == Type.CONNECTION_STATUS
         ]  # noqa: SLF001  # Non-public API
         assert len(conn_status_messages) == 1, (
             "Expected exactly one CONNECTION_STATUS message. Got: \n" + "\n".join(result._messages)
         )
+
+    @classmethod
+    @property
+    def acceptance_test_config_path(self) -> Path:
+        """Get the path to the acceptance test config file.
+
+        Check vwd and parent directories of cwd for the config file, and return the first one found.
+
+        Give up if the config file is not found in any parent directory.
+        """
+        current_dir = Path.cwd()
+        for parent_dir in current_dir.parents:
+            config_path = parent_dir / ACCEPTANCE_TEST_CONFIG
+            if config_path.exists():
+                return config_path
+        raise FileNotFoundError(
+            f"Acceptance test config file not found in any parent directory from : {Path.cwd()}"
+        )
+
+    @classmethod
+    def get_scenarios(
+        cls,
+    ) -> list[ConnectorTestScenario]:
+        """Get acceptance tests for a given category.
+
+        This has to be a separate function because pytest does not allow
+        parametrization of fixtures with arguments from the test class itself.
+        """
+        category = "connection"
+        all_tests_config = yaml.safe_load(cls.acceptance_test_config_path.read_text())
+        if "acceptance_tests" not in all_tests_config:
+            raise ValueError(
+                f"Acceptance tests config not found in {cls.acceptance_test_config_path}."
+                f" Found only: {str(all_tests_config)}."
+            )
+        if category not in all_tests_config["acceptance_tests"]:
+            return []
+        if "tests" not in all_tests_config["acceptance_tests"][category]:
+            raise ValueError(f"No tests found for category {category}")
+
+        tests_scenarios = [
+            ConnectorTestScenario.model_validate(test)
+            for test in all_tests_config["acceptance_tests"][category]["tests"]
+            if "iam_role" not in test["config_path"]
+        ]
+        working_dir = cls.working_dir or Path()
+        for test in tests_scenarios:
+            if test.config_path:
+                test.config_path = working_dir / test.config_path
+            if test.configured_catalog_path:
+                test.configured_catalog_path = working_dir / test.configured_catalog_path
+        return tests_scenarios

airbyte_cdk/test/declarative/test_suites/declarative_sources.py

@@ -0,0 +1,45 @@
+import os
+from hashlib import md5
+from pathlib import Path
+from typing import Any, cast
+
+import yaml
+
+from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
+    ConcurrentDeclarativeSource,
+)
+from airbyte_cdk.test.declarative.models import ConnectorTestScenario
+from airbyte_cdk.test.declarative.test_suites.source_base import (
+    SourceTestSuiteBase,
+)
+
+
+def md5_checksum(file_path: Path) -> str:
+    with open(file_path, "rb") as file:
+        return md5(file.read()).hexdigest()
+
+class DeclarativeSourceTestSuite(SourceTestSuiteBase):
+
+    manifest_path = Path("manifest.yaml")
+    components_py_path: Path | None = None
+
+    def create_connector(self, connector_test: ConnectorTestScenario) -> ConcurrentDeclarativeSource:
+        config = connector_test.get_config_dict()
+        # catalog = connector_test.get_catalog()
+        # state = connector_test.get_state()
+        # source_config = connector_test.get_source_config()
+
+        manifest_dict = yaml.safe_load(self.manifest_path.read_text())
+        if self.components_py_path and self.components_py_path.exists():
+            os.environ["AIRBYTE_ALLOW_CUSTOM_CODE"] = "true"
+            config["__injected_components_py"] = self.components_py_path.read_text()
+            config["__injected_components_py_checksums"] = {
+                "md5": md5_checksum(self.components_py_path),
+            }
+
+        return ConcurrentDeclarativeSource(
+            config=config,
+            catalog=None,
+            state=None,
+            source_config=manifest_dict,
+        )