diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py index de567f343..683501497 100644 --- a/airbyte_cdk/test/standard_tests/connector_base.py +++ b/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) diff --git a/cdk-migrations.md b/cdk-migrations.md index ae3781c9b..2f2fc85f9 100644 --- a/cdk-migrations.md +++ b/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. diff --git a/unit_tests/test/test_standard_tests.py b/unit_tests/test/test_standard_tests.py index 323595061..578a44dc7 100644 --- a/unit_tests/test/test_standard_tests.py +++ b/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]