From 88a1d441f0c60696b0064dac36bf35af74c9aa6b Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 14 Oct 2026 15:02:18 +0300 Subject: [PATCH] feat(standard-tests): build sources that require catalog, config and state The in-process Standard Tests called the connector class with no arguments. Python sources built on YamlDeclarativeSource, FileBasedSource or ConcurrentSourceAdapter require catalog, config and state, so every in-process test of those connectors failed with a TypeError before reaching the connector. That blocks running the suite in connector CI (airbyte-internal-issues #17133) for asana, freshcaller, google-ads, google-drive, microsoft-sharepoint, salesforce and sftp-bulk. create_connector() now inspects the signature and passes config from the scenario (an empty dict when there is none) and None for catalog and state, matching what run.py passes when it launches without them. Any other required argument raises a TypeError naming it, since the suite cannot guess it. An optional config also receives the scenario's config when there is one, as run.py passes --config. When the class accepts catalog, config or state, required or optional, its read is done by a source built from the read command's catalog, config and state, as run.py builds it. A file-based source takes its stream schemas from the catalog it is built with, so a source built without one fails every read with a missing json_schema error. A declarative Python source with an optional config (source-facebook-pages) evaluates the config in its components when it is built, so a source built without one sends requests with empty config values. Co-Authored-By: Claude Opus 5.5 --- .../test/standard_tests/connector_base.py | 163 ++++++++++++++++- cdk-migrations.md | 8 + unit_tests/test/test_standard_tests.py | 165 ++++++++++++++++++ 3 files changed, 332 insertions(+), 4 deletions(-) diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py index de567f343e..6835014970 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 ae3781c9b6..2f2fc85f9f 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 3235950611..578a44dc71 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]