feat(standard-tests): build sources that require catalog, config and state

airbyte_cdk/test/standard_tests/connector_base.py

@@ -4,9 +4,10 @@
 from __future__ import annotations
 
 import importlib
+import inspect
 import os
 from pathlib import Path
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import pytest
 from boltons.typeutils import classproperty
@@ -28,8 +29,15 @@
 )
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    import logging
+    from collections.abc import Callable, Iterable, Mapping
 
+    from airbyte_cdk.models import (
+        AirbyteMessage,
+        AirbyteStateMessage,
+        ConfiguredAirbyteCatalog,
+    )
+    from airbyte_cdk.sources.source import Source
     from airbyte_cdk.test import entrypoint_wrapper
 
 
@@ -96,12 +104,33 @@ def create_connector(
         cls,
         scenario: ConnectorTestScenario | None,
     ) -> IConnector:
-        """Instantiate the connector class."""
+        """Instantiate the connector class.
+
+        A connector class or factory that requires `catalog`, `config` or `state` (the
+        constructor of a Python source built on `YamlDeclarativeSource`, `FileBasedSource` or
+        `ConcurrentSourceAdapter` usually does) receives the scenario's config, no catalog and no
+        state, as the connector's `run.py` passes them when it launches without a catalog or
+        state. An optional `config` also receives the scenario's config, when the scenario has one.
+        When the class or factory accepts any of these arguments, required or optional, its
+        `read` is done by a source built with the catalog, config and state of the `read` command,
+        as `run.py` builds it: a file-based source takes its stream schemas from the catalog it is
+        built with, and a declarative source evaluates the config in its components when it is
+        built.
+        """
         connector = cls.connector  # type: ignore
         if connector:
             if callable(connector) or isinstance(connector, type):
                 # If the connector is a class or factory function, instantiate it:
-                return cast(IConnector, connector())  # type: ignore [redundant-cast]
+                init_kwargs = _source_init_kwargs(
+                    connector,
+                    scenario=scenario,
+                    connector_root=cls.get_connector_root_dir(),
+                )
+                instance = connector(**init_kwargs)
+                read_args = _source_read_args(connector)
+                if read_args:
+                    _read_with_source_built_for_the_command(instance, connector, read_args)
+                return cast(IConnector, instance)  # type: ignore [redundant-cast]
 
         # Otherwise, we can't instantiate the connector. Fail with a clear error message.
         raise NotImplementedError(
@@ -199,3 +228,129 @@ def test_config_matches_spec(
             connector_name=connector_root.absolute().name,
             scenario_id=scenario.id,
         )
+
+
+_SOURCE_INIT_ARGS = ("catalog", "config", "state")
+"""Constructor arguments the test suite can supply to a connector class or factory."""
+
+
+def _source_init_kwargs(
+    factory: object,
+    *,
+    scenario: ConnectorTestScenario | None,
+    connector_root: Path,
+) -> dict[str, object]:
+    """Return the keyword arguments to build the connector class or factory with.
+
+    Only `catalog`, `config` and `state` can be supplied: `config` is the scenario's config (an
+    empty dict when the scenario has none, or for `spec`), `catalog` and `state` are `None`. Any
+    other required argument raises a `TypeError` naming it, since the suite cannot guess it.
+
+    An optional `config` (accepted by keyword) also receives the scenario's config when the
+    scenario has one, as `run.py` passes the `--config` it launches with: a declarative Python
+    source evaluates its config in its components when it is built.
+    """
+    try:
+        signature = inspect.signature(factory)  # type: ignore [arg-type]
+    except (TypeError, ValueError):
+        # Builtins and some C extensions have no signature; call them without arguments.
+        return {}
+
+    required = [
+        parameter
+        for parameter in signature.parameters.values()
+        if parameter.default is inspect.Parameter.empty
+        and parameter.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
+    ]
+    unsupported = [
+        parameter.name
+        for parameter in required
+        if parameter.name not in _SOURCE_INIT_ARGS
+        or parameter.kind is inspect.Parameter.POSITIONAL_ONLY
+    ]
+    if unsupported:
+        factory_name = getattr(factory, "__name__", repr(factory))
+        raise TypeError(
+            f"The standard tests cannot instantiate `{factory_name}`: it requires the "
+            f"argument(s) {', '.join(f'`{name}`' for name in unsupported)}, and the suite can "
+            f"only supply {', '.join(f'`{name}`' for name in _SOURCE_INIT_ARGS)}. Give those "
+            "arguments defaults, export a factory function under the connector class name, or "
+            "override `create_connector()` in a test suite subclass."
+        )
+
+    kwargs: dict[str, object] = {}
+    for parameter in required:
+        if parameter.name == "config":
+            kwargs["config"] = _scenario_config_or_empty(scenario, connector_root=connector_root)
+        else:
+            kwargs[parameter.name] = None
+
+    optional_config = signature.parameters.get("config")
+    if (
+        "config" not in kwargs
+        and optional_config is not None
+        and optional_config.kind
+        in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
+    ):
+        config = _scenario_config_or_empty(scenario, connector_root=connector_root)
+        if config:
+            kwargs["config"] = config
+    return kwargs
+
+
+def _scenario_config_or_empty(
+    scenario: ConnectorTestScenario | None,
+    *,
+    connector_root: Path,
+) -> dict[str, object]:
+    """Return the scenario's config, or an empty dict when there is none or its file is missing.
+
+    A missing config file is reported by the test that runs the command, not here.
+    """
+    if scenario is None:
+        return {}
+    try:
+        return scenario.get_config_dict(connector_root=connector_root, empty_if_missing=True)
+    except FileNotFoundError:
+        return {}
+
+
+def _source_read_args(factory: object) -> tuple[str, ...]:
+    """Return the names among `catalog`, `config` and `state` that the factory accepts by keyword."""
+    try:
+        signature = inspect.signature(factory)  # type: ignore [arg-type]
+    except (TypeError, ValueError):
+        return ()
+    return tuple(
+        parameter.name
+        for parameter in signature.parameters.values()
+        if parameter.name in _SOURCE_INIT_ARGS
+        and parameter.kind
+        in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
+    )
+
+
+def _read_with_source_built_for_the_command(
+    source: object,
+    factory: Callable[..., object],
+    read_args: tuple[str, ...],
+) -> None:
+    """Make `source.read` read with a source built from the `read` command's arguments.
+
+    The suite builds the source before it knows the catalog and state of a `read`, while the
+    connector's `run.py` builds it with them.
+    """
+    if not callable(getattr(source, "read", None)):
+        return
+
+    def read(
+        logger: logging.Logger,
+        config: Mapping[str, Any],
+        catalog: ConfiguredAirbyteCatalog,
+        state: list[AirbyteStateMessage] | None = None,
+    ) -> Iterable[AirbyteMessage]:
+        command_args: dict[str, object] = {"catalog": catalog, "config": config, "state": state}
+        launched = cast("Source", factory(**{name: command_args[name] for name in read_args}))
+        return launched.read(logger, config, catalog, state)
+
+    setattr(source, "read", read)

cdk-migrations.md

@@ -1,5 +1,13 @@
 # CDK Migration Guide
 
+## Upgrading to the next minor release (unreleased)
+
+### The in-process Standard Tests build sources whose constructor requires `catalog`, `config` and `state`
+
+Previously the in-process Standard Tests (`SourceTestSuiteBase`, which `airbyte-cdk connector test` and the suite from `create_connector_test_suite()` use) called the connector class with no arguments. Python sources built on `YamlDeclarativeSource`, `FileBasedSource` or `ConcurrentSourceAdapter` require `catalog`, `config` and `state`, so every test failed with a `TypeError` before reaching the connector. `create_connector()` now passes the arguments the class or factory requires: `config` receives the scenario's config (an empty dict when the scenario has none, for example for `spec`), and `catalog` and `state` receive `None`, as the connector's `run.py` passes them when it launches without a catalog or state. A class or factory with an optional `config` also receives the scenario's config, when the scenario has one, since `run.py` passes `--config`. When the class or factory accepts `catalog`, `config` or `state`, required or optional, its `read` is done by a source built with the catalog, config and state of the `read` command, as `run.py` builds it: a file-based source takes its stream schemas from the catalog it is built with, and a declarative Python source whose constructor takes an optional `config` evaluates that config in its components when it is built. A class that requires any other argument (a `FileBasedSource` used directly also needs `stream_reader` and `spec_class`) fails with a `TypeError` that names the argument.
+
+No action is needed for connectors whose constructor takes no required arguments. A connector that worked around the old behaviour by overriding `create_connector()` in its own test suite can drop the override once it pins this release, unless the source needs more than `catalog`, `config` and `state`. A source whose constructor requires other arguments keeps overriding `create_connector()`, or exports a factory function under the connector class name.
+
 ## Upgrading to 7.36.0
 
 This release of the CDK adds checks to the Standard Tests (`airbyte-cdk connector test` and `airbyte-cdk image test`) and changes which commands they run for the configs listed in `acceptance-test-config.yml`. There is no change to how connectors run. The release is a minor one, since only the test harness changes, but the connector test suites of connectors that do not satisfy the new expectations start failing. The connector monorepo installs the `airbyte-cdk` CLI unpinned (`uv tool install --upgrade 'airbyte-cdk[dev]'`), so connector CI picks the new behaviour up with this release, without a per-connector CDK bump.

unit_tests/test/test_standard_tests.py

@@ -13,6 +13,7 @@
 from airbyte_cdk.sources.source import Source
 from airbyte_cdk.test.models.scenario import ConnectorTestScenario
 from airbyte_cdk.test.standard_tests._job_runner import IConnector
+from airbyte_cdk.test.standard_tests.connector_base import ConnectorTestSuiteBase
 from airbyte_cdk.test.standard_tests.declarative_sources import md5_checksum
 from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite
 from airbyte_cdk.test.standard_tests.pytest_hooks import _scenario_test_ids
@@ -121,3 +122,167 @@ def test_md5_checksum_matches_injected_text_with_crlf(tmp_path: Path) -> None:
         code_text=components_py_path.read_text(),
         checksums={"md5": md5_checksum(components_py_path)},
     )
+
+
+class _NoArgSource:
+    def __init__(self) -> None:
+        self.created_with: dict[str, Any] = {}
+
+
+class _CatalogConfigStateSource:
+    def __init__(self, catalog: Any, config: Any, state: Any, **kwargs: Any) -> None:
+        self.created_with = {"catalog": catalog, "config": config, "state": state, **kwargs}
+
+
+class _OptionalArgsSource:
+    def __init__(self, catalog: Any = None, config: Any = None, state: Any = None) -> None:
+        self.created_with = {"catalog": catalog, "config": config, "state": state}
+
+
+class _StreamReaderSource:
+    def __init__(self, stream_reader: Any, spec_class: Any, catalog: Any, config: Any, state: Any):
+        self.created_with = {}
+
+
+def _suite_for(connector: Any, root: Path) -> type[ConnectorTestSuiteBase]:
+    return type(
+        "Suite",
+        (ConnectorTestSuiteBase,),
+        {
+            "connector": connector,
+            "get_connector_root_dir": classmethod(lambda cls: root),
+        },
+    )
+
+
+@pytest.mark.parametrize(
+    "connector, scenario_config, expected",
+    [
+        pytest.param(_NoArgSource, {"api_key": "x"}, {}, id="no_args"),
+        pytest.param(
+            _CatalogConfigStateSource,
+            {"api_key": "x"},
+            {"catalog": None, "config": {"api_key": "x"}, "state": None},
+            id="required_catalog_config_state_get_scenario_config",
+        ),
+        pytest.param(
+            _OptionalArgsSource,
+            {"api_key": "x"},
+            {"catalog": None, "config": {"api_key": "x"}, "state": None},
+            id="optional_config_gets_scenario_config_catalog_and_state_keep_defaults",
+        ),
+    ],
+)
+def test_create_connector_supplies_required_source_args(
+    tmp_path: Path,
+    connector: Any,
+    scenario_config: dict[str, Any],
+    expected: dict[str, Any],
+) -> None:
+    suite = _suite_for(connector, tmp_path)
+    created = suite.create_connector(ConnectorTestScenario(config_dict=scenario_config))
+    assert created.created_with == expected  # type: ignore[attr-defined]
+
+
+def test_create_connector_without_scenario_passes_empty_config(tmp_path: Path) -> None:
+    suite = _suite_for(_CatalogConfigStateSource, tmp_path)
+    created = suite.create_connector(None)
+    assert created.created_with == {"catalog": None, "config": {}, "state": None}  # type: ignore[attr-defined]
+
+
+def test_create_connector_with_missing_config_file_passes_empty_config(tmp_path: Path) -> None:
+    suite = _suite_for(_CatalogConfigStateSource, tmp_path)
+    scenario = ConnectorTestScenario(config_path=Path("secrets/config.json"))
+    created = suite.create_connector(scenario)
+    assert created.created_with["config"] == {}  # type: ignore[attr-defined]
+
+
+def test_create_connector_reads_config_file_relative_to_connector_root(tmp_path: Path) -> None:
+    (tmp_path / "secrets").mkdir()
+    (tmp_path / "secrets" / "config.json").write_text('{"start_date": "2024-01-01"}')
+    suite = _suite_for(_CatalogConfigStateSource, tmp_path)
+    created = suite.create_connector(ConnectorTestScenario(config_path=Path("secrets/config.json")))
+    assert created.created_with["config"] == {"start_date": "2024-01-01"}  # type: ignore[attr-defined]
+
+
+def test_create_connector_names_arguments_it_cannot_supply(tmp_path: Path) -> None:
+    suite = _suite_for(_StreamReaderSource, tmp_path)
+    with pytest.raises(TypeError) as error:
+        suite.create_connector(ConnectorTestScenario(config_dict={}))
+    message = str(error.value)
+    assert "`_StreamReaderSource`" in message
+    assert "`stream_reader`, `spec_class`" in message
+    assert "override `create_connector()`" in message
+
+
+class _CatalogAtInitSource:
+    """Like a file-based source: `read` only knows the streams of the catalog it was built with."""
+
+    def __init__(self, catalog: Any, config: Any, state: Any) -> None:
+        self.created_with = {"catalog": catalog, "config": config, "state": state}
+
+    def read(self, logger: Any, config: Any, catalog: Any, state: Any = None) -> Any:
+        yield {"built_with": self.created_with, "read_with": (config, catalog, state)}
+
+
+def test_create_connector_reads_with_a_source_built_for_the_read_command(tmp_path: Path) -> None:
+    suite = _suite_for(_CatalogAtInitSource, tmp_path)
+    created = suite.create_connector(ConnectorTestScenario(config_dict={"api_key": "x"}))
+    assert created.created_with == {"catalog": None, "config": {"api_key": "x"}, "state": None}  # type: ignore[attr-defined]
+
+    messages = list(created.read(None, {"api_key": "y"}, "the catalog", ["the state"]))  # type: ignore[attr-defined]
+
+    assert messages == [
+        {
+            "built_with": {
+                "catalog": "the catalog",
+                "config": {"api_key": "y"},
+                "state": ["the state"],
+            },
+            "read_with": ({"api_key": "y"}, "the catalog", ["the state"]),
+        }
+    ]
+
+
+class _OptionalConfigSource:
+    """Like a declarative Python source: `config` is optional and evaluated when it is built."""
+
+    def __init__(self, catalog: Any = None, config: Any = None, state: Any = None) -> None:
+        self.created_with = {"catalog": catalog, "config": config, "state": state}
+
+    def read(self, logger: Any, config: Any, catalog: Any, state: Any = None) -> Any:
+        yield {"built_with": self.created_with}
+
+
+def test_create_connector_reads_sources_with_optional_arguments_with_the_command_arguments(
+    tmp_path: Path,
+) -> None:
+    suite = _suite_for(_OptionalConfigSource, tmp_path)
+    created = suite.create_connector(ConnectorTestScenario(config_dict={"page_id": "1"}))
+    # `check` and `discover` use this source, so it is built with the scenario's config.
+    assert created.created_with == {"catalog": None, "config": {"page_id": "1"}, "state": None}  # type: ignore[attr-defined]
+
+    messages = list(created.read(None, {"page_id": "1"}, "the catalog", None))  # type: ignore[attr-defined]
+
+    assert messages == [
+        {"built_with": {"catalog": "the catalog", "config": {"page_id": "1"}, "state": None}}
+    ]
+
+
+def test_create_connector_leaves_an_optional_config_out_when_the_scenario_has_none(
+    tmp_path: Path,
+) -> None:
+    suite = _suite_for(_OptionalConfigSource, tmp_path)
+    created = suite.create_connector(None)
+    assert created.created_with == {"catalog": None, "config": None, "state": None}  # type: ignore[attr-defined]
+
+
+def test_create_connector_keeps_read_of_sources_built_without_arguments(tmp_path: Path) -> None:
+    class _NoArgReadSource:
+        def read(self, logger: Any, config: Any, catalog: Any, state: Any = None) -> Any:
+            yield "own read"
+
+    suite = _suite_for(_NoArgReadSource, tmp_path)
+    created = suite.create_connector(ConnectorTestScenario(config_dict={}))
+    assert "read" not in vars(created)
+    assert list(created.read(None, {}, None)) == ["own read"]  # type: ignore[attr-defined]