From 3b7b58858bb61b7dddb60bc49d5651a15bdcf84a Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 14 Oct 2026 15:14:43 +0300 Subject: [PATCH 1/2] feat(standard-tests): check primary keys, state messages and error traces in read tests The Standard Tests replaced the Connector Acceptance Tests as the only read coverage for most connectors, but three checks of CAT's basic read test were not carried over. This restores them: - Primary keys: for every `basic_read` config, every record of a stream with a `source_defined_primary_key` must have a usable key. No key field may hold an object or an array, and not every key field may be missing or null, as in CAT's `validate_primary_keys_data_type`. Failures name the stream, record number and key path, never the value. Opt out per config with `validate_primary_keys_data_type: false`. - State: every configured stream must be named by a STREAM state message or by a GLOBAL state's `stream_states`; no state may use the LEGACY type; every state must carry `sourceStats.recordCount`. CAT only compared the number of state messages with the number of streams, which one stream's many states could satisfy for another stream that emitted none. Opt out per config with `validate_state_messages: false`. - Error traces: the new `test_fail_read_with_invalid_config` reads with a config the connector's own spec rejects (it holds none of the required top-level properties and no real value, so no request can be sent with real credentials) and requires a well-formed ERROR trace, as CAT's `test_airbyte_trace_message_on_failure` did. `test_fail_read_with_bad_catalog` now checks its error traces the same way. Opt out with `expect_trace_message_on_failure: false`. The new `assert_read_records` arguments default to off, so the Docker read test is unchanged; it can pass them once it collects record messages. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/models/scenario.py | 12 +- .../test/standard_tests/_read_assertions.py | 293 ++++++- .../test/standard_tests/docker_base.py | 19 +- .../test/standard_tests/source_base.py | 104 ++- cdk-migrations.md | 36 + unit_tests/test/read_test_helpers.py | 38 + unit_tests/test/test_read_assertions.py | 2 + .../test/test_read_keys_state_and_traces.py | 740 ++++++++++++++++++ unit_tests/test/test_scenario_sections.py | 15 + 9 files changed, 1243 insertions(+), 16 deletions(-) create mode 100644 unit_tests/test/read_test_helpers.py create mode 100644 unit_tests/test/test_read_keys_state_and_traces.py diff --git a/airbyte_cdk/test/models/scenario.py b/airbyte_cdk/test/models/scenario.py index 217c28549..ea0995b68 100644 --- a/airbyte_cdk/test/models/scenario.py +++ b/airbyte_cdk/test/models/scenario.py @@ -105,6 +105,12 @@ def __hash__(self) -> int: status: Literal["succeed", "failed", "exception"] | None = None # CAT's `basic_read` opt-out from validating records against the stream schemas. validate_schema: bool = True + # CAT's `basic_read` opt-out from checking the primary key of every record. + validate_primary_keys_data_type: bool = True + # CAT's `basic_read` opt-out from checking that every stream emits per-stream state. + validate_state_messages: bool = True + # CAT's `basic_read` opt-out from requiring a well-formed error trace from a failing read. + expect_trace_message_on_failure: bool = True # The connector's top-level `test_strictness_level` in `acceptance-test-config.yml`. Populated # by `get_scenarios()`; at `high`, `validate_schema: false` is rejected, as in CAT. test_strictness_level: Literal["low", "high"] | None = None @@ -189,8 +195,10 @@ def is_basic_read_config(self) -> bool: """Whether this config is listed under `basic_read` (or the scenario was built by hand). Only the records of these configs are checked as in CAT's basic read test: every stream - not declared in `empty_streams` returns records, and every record matches its stream's - schema unless `validate_schema: false` is set. A config read only because it is listed + not declared in `empty_streams` returns records, every record matches its stream's + schema unless `validate_schema: false` is set, every record has a usable primary key + unless `validate_primary_keys_data_type: false` is set, and every stream emits state + unless `validate_state_messages: false` is set. A config read only because it is listed under `full_refresh` never had CAT's basic read checks, and has no `basic_read` entry to declare those exceptions in, so its `read` only has to return some records. """ diff --git a/airbyte_cdk/test/standard_tests/_read_assertions.py b/airbyte_cdk/test/standard_tests/_read_assertions.py index c25ccdeb6..b588ec2ac 100644 --- a/airbyte_cdk/test/standard_tests/_read_assertions.py +++ b/airbyte_cdk/test/standard_tests/_read_assertions.py @@ -7,7 +7,15 @@ - every configured stream returns at least one record (CAT `_validate_empty_streams`); - every record conforms to its stream's JSON schema (CAT `_validate_schema`), with the same validator CAT ran: Draft 7 with strict integers and a lenient `date-time` format check, and - without failing on extra columns. + without failing on extra columns; +- every record of a stream with a `source_defined_primary_key` has a usable primary key: no key + field holds an object or an array, and not every key field is missing or null (CAT + `_validate_primary_keys_data_type`); +- every configured stream emits a `STREAM` (or `GLOBAL`) state message carrying `sourceStats` + (the per-stream form of CAT `_validate_state_messages`). + +They also include the check of CAT's `TestBasicRead.test_airbyte_trace_message_on_failure`: a +read that fails emits a well-formed `TRACE` message of type `ERROR` (`assert_error_trace_messages`). They take the read output and the configured catalog rather than a scenario, so the in-process and the Docker read tests can share them. @@ -17,7 +25,7 @@ import re from collections import Counter -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field from typing import Any @@ -25,12 +33,21 @@ from jsonschema.exceptions import FormatError, UnknownType, ValidationError, best_match from referencing.exceptions import Unresolvable -from airbyte_cdk.models import AirbyteMessage, ConfiguredAirbyteCatalog +from airbyte_cdk.models import ( + AirbyteMessage, + AirbyteStateType, + ConfiguredAirbyteCatalog, + TraceType, + Type, +) from airbyte_cdk.utils.datetime_helpers import ab_datetime_try_parse MAX_SCHEMA_ERRORS_PER_STREAM = 5 """How many distinct schema errors are reported per stream. The total is always reported.""" +MAX_PRIMARY_KEY_ERRORS_PER_STREAM = 5 +"""How many distinct primary key errors are reported per stream. The total is always reported.""" + _MAX_MESSAGE_CHARS = 300 _MAX_CONSTRAINT_CHARS = 80 @@ -339,6 +356,227 @@ def describe_failures(self) -> str | None: return "\n".join(lines) if lines else None +_MISSING = object() +"""The value of a key field that is absent from the record.""" + + +def _key_field_value(data: Any, key_path: Sequence[str]) -> Any: + """Return the value at `key_path` in `data`, or `_MISSING` if the path does not exist.""" + node = data + for part in key_path: + if not isinstance(node, Mapping) or part not in node: + return _MISSING + node = node[part] + return node + + +def _format_primary_key(primary_key: Sequence[Sequence[str]]) -> str: + return ", ".join(f"`{format_record_path(key_path)}`" for key_path in primary_key) + + +class _StreamPrimaryKeyChecker: + """Checks the primary key of each record of one stream, as CAT's + `_validate_primary_keys_data_type` did. + + A key field must not hold an object or an array, and at least one key field must be present + and not null. As in CAT, a composite key may have some of its fields null. Booleans count + as scalars, as in CAT; they are named `boolean`, never `integer`. + """ + + def __init__(self, stream_name: str, primary_key: Sequence[Sequence[str]]) -> None: + self._stream_name = stream_name + self._primary_key = [list(key_path) for key_path in primary_key if key_path] + self.records_checked = 0 + self.invalid_records = 0 + # Distinct problems, keyed by the key field and the kind of problem; first record wins. + self.errors: dict[tuple[Any, ...], str] = {} + + def check(self, data: Any, record_number: int) -> None: + if not self._primary_key: + return + self.records_checked += 1 + problems: dict[tuple[Any, ...], str] = {} + null_fields: list[str] = [] + for key_path in self._primary_key: + value = _key_field_value(data, key_path) + rendered_path = format_record_path(key_path) + if value is _MISSING or value is None: + null_fields.append( + f"`{rendered_path}` is {'missing' if value is _MISSING else 'null'}" + ) + elif isinstance(value, (Mapping, list)): + got = _json_type_name(value) + problems[(tuple(key_path), got)] = ( + f"record #{record_number}: key field `{rendered_path}` is an {got}; a primary " + "key field must hold a scalar (a string, a number or a boolean)" + ) + if len(null_fields) == len(self._primary_key): + problems[("every key field missing or null", *null_fields)] = ( + f"record #{record_number}: every key field is missing or null " + f"({', '.join(null_fields)})" + ) + if problems: + self.invalid_records += 1 + for problem, description in problems.items(): + self.errors.setdefault(problem, description) + + def describe_failures(self) -> str | None: + if not self.invalid_records: + return None + lines = [ + f"- Stream '{self._stream_name}' (primary key {_format_primary_key(self._primary_key)}): " + f"{self.invalid_records} of {self.records_checked} records have an invalid primary key." + ] + shown = list(self.errors.values())[:MAX_PRIMARY_KEY_ERRORS_PER_STREAM] + lines.extend(f" - {error}" for error in shown) + hidden = len(self.errors) - len(shown) + if hidden: + lines.append(f" - ...and {hidden} more distinct primary key error(s).") + return "\n".join(lines) + + +def _find_state_message_failures( + state_messages: Iterable[AirbyteMessage], + configured_catalog: ConfiguredAirbyteCatalog, +) -> list[str]: + """Check the state messages of a full-refresh read, as CAT's `_validate_state_messages` did. + + CAT asserted that the read emitted at least as many state messages as there are configured + streams, that none was of the deprecated `LEGACY` type, and that each carried `sourceStats`. + State is per stream in the current protocol, so the count is checked per stream instead: a + stream that emits several state messages cannot hide one that emits none. + """ + configured = { + (configured_stream.stream.namespace, configured_stream.stream.name) + for configured_stream in configured_catalog.streams + } + covered: set[tuple[str | None, str]] = set() + state_count = 0 + legacy_states: list[int] = [] + states_without_stats: list[int] = [] + for message in state_messages: + state = message.state + if message.type != Type.STATE or state is None: + continue + state_number = state_count + state_count += 1 + if state.type == AirbyteStateType.STREAM and state.stream is not None: + descriptor = state.stream.stream_descriptor + covered.add((descriptor.namespace, descriptor.name)) + elif state.type == AirbyteStateType.GLOBAL and state.global_ is not None: + covered.update( + (stream_state.stream_descriptor.namespace, stream_state.stream_descriptor.name) + for stream_state in state.global_.stream_states or [] + ) + else: + legacy_states.append(state_number) + if state.sourceStats is None or state.sourceStats.recordCount is None: + states_without_stats.append(state_number) + + def _numbers(numbers: list[int]) -> str: + shown = ", ".join(f"#{number}" for number in numbers[:5]) + return shown + (f" and {len(numbers) - 5} more" if len(numbers) > 5 else "") + + failures: list[str] = [] + streams_without_state = sorted( + name if namespace is None else f"{namespace}.{name}" + for namespace, name in configured - covered + ) + if streams_without_state: + failures.append( + f"- {len(streams_without_state)} of {len(configured)} configured streams emitted no " + "`STREAM` or `GLOBAL` state message: " + f"{', '.join(repr(name) for name in streams_without_state)}. A stream must emit a " + "state message once it has been read in full; the CDK emits one when a stream " + "completes." + ) + if legacy_states: + failures.append( + f"- {len(legacy_states)} of {state_count} state messages use the deprecated `LEGACY` " + f"state type (or no type): state message(s) {_numbers(legacy_states)}. Emit " + "per-stream `STREAM` state (or `GLOBAL` state) instead." + ) + if states_without_stats: + failures.append( + f"- {len(states_without_stats)} of {state_count} state messages carry no " + f"`sourceStats.recordCount`: state message(s) {_numbers(states_without_stats)}. The " + "CDK entrypoint sets it on every state message it emits." + ) + return failures + + +def assert_error_trace_messages(messages: Iterable[AirbyteMessage]) -> None: + """Assert that a failed command emitted at least one well-formed `ERROR` trace message. + + Ported from CAT's `TestBasicRead.test_airbyte_trace_message_on_failure`, which only required + one `ERROR` trace. A trace is also well-formed here: it carries the `error` object, a + non-blank `error.message` (what the platform shows the user), an `error.failure_type` (what + the platform uses to tell a configuration problem from a system error) and a positive + `emitted_at`. A `TRACE` line that does not parse as an Airbyte message never counts. + + The failure message names the trace messages by number and the fields they lack. It never + prints their text, which may hold record or config values. + """ + error_traces = [ + message.trace + for message in messages + if message.type == Type.TRACE + and message.trace is not None + and message.trace.type == TraceType.ERROR + ] + if not error_traces: + raise AssertionError( + "The failed read emitted no `TRACE` message of type `ERROR`. A connector must report " + "why it fails as an error trace message: the CDK entrypoint does so for any " + "exception that escapes `read`, so a connector without one bypasses the entrypoint " + "or exits without raising." + ) + + problems: list[str] = [] + for number, trace in enumerate(error_traces): + lacking: list[str] = [] + error = trace.error + if error is None: + lacking.append("the `error` object") + else: + if not isinstance(error.message, str) or not error.message.strip(): + lacking.append("a non-blank `error.message`") + if error.failure_type is None: + lacking.append("an `error.failure_type`") + emitted_at = trace.emitted_at + if ( + not isinstance(emitted_at, (int, float)) + or isinstance(emitted_at, bool) + or emitted_at <= 0 + ): + lacking.append("a positive `emitted_at`") + if lacking: + problems.append(f"- error trace #{number} lacks {', '.join(lacking)}.") + if problems: + raise AssertionError( + f"{len(problems)} of {len(error_traces)} `ERROR` trace messages of the failed read " + "are not well-formed:\n" + "\n".join(problems) + ) + + +INVALID_CONFIG_MARKER = "__airbyte_standard_tests_invalid_config" +"""The only key of the config that `make_invalid_config` returns.""" + + +def make_invalid_config(connection_specification: Mapping[str, Any]) -> dict[str, Any] | None: + """Return a config that the connector's spec rejects, or `None` if none can be built. + + The config holds none of the spec's required top-level properties, and no value from any + real config, so a connector that validates its config fails before it sends a request, and + nothing secret can leak from it. `None` means the spec requires no top-level property, so + an arbitrary config may be valid. + """ + required = connection_specification.get("required") + if not isinstance(required, list) or not required: + return None + return {INVALID_CONFIG_MARKER: True} + + def assert_schema_validation_opt_out_allowed( *, validate_schema: bool, test_strictness_level: str | None ) -> None: @@ -364,6 +602,8 @@ def assert_read_records( configured_catalog: ConfiguredAirbyteCatalog, require_records_per_stream: bool, validate_schema: bool, + validate_primary_keys: bool = False, + state_messages: Iterable[AirbyteMessage] | None = None, ) -> None: """Assert the records a `read` returned against the configured catalog. @@ -371,15 +611,20 @@ def assert_read_records( records are consumed once, so a lazy iterator over a large read is fine. Args: - records: The `RECORD` messages from the read. + records: The `RECORD` messages from the read. Other messages are ignored. configured_catalog: The catalog the read ran with. Streams declared in `empty_streams` are expected to be excluded from it already. require_records_per_stream: Require at least one record from every configured stream. validate_schema: Validate every record against its stream's JSON schema. + validate_primary_keys: Check the primary key of every record of a stream with a + `source_defined_primary_key`: no key field holds an object or an array, and not + every key field is missing or null. + state_messages: The `STATE` messages from the read, to check that every configured + stream emitted per-stream state with `sourceStats`; `None` skips the check. Raises: AssertionError: Listing every failure, by stream, with the path of the offending value - inside the record. + inside the record. Record values are never printed. """ record_counts: Counter[str] = Counter() checkers: dict[str, _StreamSchemaChecker] = {} @@ -390,15 +635,28 @@ def assert_read_records( ) for configured_stream in configured_catalog.streams } + key_checkers: dict[str, _StreamPrimaryKeyChecker] = {} + if validate_primary_keys: + key_checkers = { + configured_stream.stream.name: _StreamPrimaryKeyChecker( + configured_stream.stream.name, configured_stream.stream.source_defined_primary_key + ) + for configured_stream in configured_catalog.streams + if configured_stream.stream.source_defined_primary_key + } for message in records: record = message.record - if record is None: + if message.type != Type.RECORD or record is None: continue + record_number = record_counts[record.stream] record_counts[record.stream] += 1 checker = checkers.get(record.stream) if checker is not None: checker.check(record.data) + key_checker = key_checkers.get(record.stream) + if key_checker is not None: + key_checker.check(record.data, record_number) failures: list[str] = [] if not record_counts: @@ -430,5 +688,28 @@ def assert_read_records( "`basic_read` entry of `acceptance-test-config.yml`.\n" + "\n".join(schema_failures) ) + key_failures = [ + description + for key_checker in key_checkers.values() + if (description := key_checker.describe_failures()) is not None + ] + if key_failures: + failures.append( + "Records have invalid primary keys. Destinations deduplicate on the primary key, so " + "every record needs a scalar value in at least one key field, and none in an object " + "or an array. Fix the stream's `primary_key` or the records; to skip this check for " + "a config, set `validate_primary_keys_data_type: false` (with a reason) in its " + "`basic_read` entry of `acceptance-test-config.yml`.\n" + "\n".join(key_failures) + ) + + if state_messages is not None: + state_failures = _find_state_message_failures(state_messages, configured_catalog) + if state_failures: + failures.append( + "The read's state messages are incomplete. To skip this check for a config, set " + "`validate_state_messages: false` (with a reason) in its `basic_read` entry of " + "`acceptance-test-config.yml`.\n" + "\n".join(state_failures) + ) + if failures: raise AssertionError("\n\n".join(failures)) diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index dd8b365db..ee3d566e4 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -261,6 +261,14 @@ def _run_docker_spec(connector_image: str, *, connector_name: str) -> ConnectorS "the check outcome is not asserted without it. The spec tests report the `spec` failure." ) +_READ_CHECK_OPT_OUTS = ( + "validate_schema", + "validate_primary_keys_data_type", + "validate_state_messages", + "expect_trace_message_on_failure", +) +"""The `basic_read` opt-outs that a config keeps if any of its entries declares them `false`.""" + class DockerConnectorTestSuite: """Base class for connector test suites.""" @@ -327,7 +335,9 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe `discovery` and `basic_read` runs both `discover` and `read` (see `ConnectorTestScenario.commands`). - A `validate_schema: false` opt-out declared by any entry for the config is kept. + An opt-out (`validate_schema`, `validate_primary_keys_data_type`, + `validate_state_messages` or `expect_trace_message_on_failure` set to `false`) declared + by any entry for the config is kept. A config is read once, however many read sections (`basic_read`, `full_refresh`) list it. It keeps a `configured_catalog_path` only when every read entry for it declares that same @@ -366,8 +376,11 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe update={ "empty_streams": list(set(all_empty_streams)), "status": existing_scenario.status or scenario.status, - "validate_schema": existing_scenario.validate_schema - and scenario.validate_schema, + **{ + opt_out: getattr(existing_scenario, opt_out) + and getattr(scenario, opt_out) + for opt_out in _READ_CHECK_OPT_OUTS + }, "sections": tuple( dict.fromkeys(existing_scenario.sections + scenario.sections) ), diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index 0543b9dfc..1f087786f 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -16,6 +16,7 @@ SyncMode, Type, ) +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput from airbyte_cdk.test.models import ( ConnectorTestScenario, ) @@ -26,8 +27,10 @@ ) from airbyte_cdk.test.standard_tests._job_runner import run_test_job from airbyte_cdk.test.standard_tests._read_assertions import ( + assert_error_trace_messages, assert_read_records, assert_schema_validation_opt_out_allowed, + make_invalid_config, ) from airbyte_cdk.test.standard_tests._spec_lint import assert_spec_is_valid, get_single_spec from airbyte_cdk.test.standard_tests.connector_base import ( @@ -173,9 +176,16 @@ def test_basic_read( `get_read_streams`). A read expected to succeed must return records. For a config listed under `basic_read`, - the records are also checked as in CAT's basic read test: every stream it reads must - return at least one record, and every record must match its stream's JSON schema (opt out - with `validate_schema: false`, which a connector at `test_strictness_level: high` cannot). + the read is also checked as in CAT's basic read test: + + - every stream it reads must return at least one record; + - every record must match its stream's JSON schema (opt out with + `validate_schema: false`, which a connector at `test_strictness_level: high` cannot); + - every record of a stream with a `source_defined_primary_key` must have a usable + primary key: no key field holds an object or an array, and not every key field is + missing or null (opt out with `validate_primary_keys_data_type: false`); + - every stream it reads must emit a `STREAM` (or `GLOBAL`) state message, and every + state message must carry `sourceStats` (opt out with `validate_state_messages: false`). """ skip_unless_runs(scenario, "read") if scenario.is_basic_read_config: @@ -231,6 +241,10 @@ def test_basic_read( configured_catalog=configured_catalog, require_records_per_stream=True, validate_schema=scenario.validate_schema, + validate_primary_keys=scenario.validate_primary_keys_data_type, + state_messages=( + result.state_messages if scenario.validate_state_messages else None + ), ) elif next(result.records_iterator, None) is None: raise AssertionError("Expected records but got none.") @@ -242,7 +256,8 @@ def test_fail_read_with_bad_catalog( """Standard test for `read` when passed a bad catalog file. Runs for configs listed under `basic_read` or `full_refresh` in - `acceptance-test-config.yml`. + `acceptance-test-config.yml`. The read must fail and emit a well-formed `ERROR` trace + message (see `assert_error_trace_messages`). """ skip_unless_runs(scenario, "read") invalid_configured_catalog = ConfiguredAirbyteCatalog( @@ -272,4 +287,83 @@ def test_fail_read_with_bad_catalog( catalog=asdict(invalid_configured_catalog), ) assert result.errors, "Expected errors but got none." - assert result.trace_messages, "Expected trace messages but got none." + assert_error_trace_messages(result.get_message_iterator()) + + def test_fail_read_with_invalid_config(self) -> None: + """Standard test for `read` with a config that the connector's spec rejects. + + Ported from CAT's `TestBasicRead.test_airbyte_trace_message_on_failure`: a read that + fails must report why in a well-formed `ERROR` trace message (see + `assert_error_trace_messages`). Where `test_fail_read_with_bad_catalog` breaks the + catalog, which the entrypoint rejects before any connector code runs, this test gives + the connector a config its own spec rejects: one that holds none of the spec's required + top-level properties and no value from any real config (`make_invalid_config`). No + credentials are needed, so it runs once per connector, and no request can be sent with + real credentials. + + Skipped when the spec requires no top-level property (no config is invalid for sure), + or when any `basic_read` entry of `acceptance-test-config.yml` sets + `expect_trace_message_on_failure: false`, as CAT did. + + A source that cannot even be created from the invalid config fails the same way + `source-declarative-manifest` reports it: as an uncaught error, turned into a trace + message. + """ + opted_out = sorted( + scenario.id + for scenario in self.get_scenarios() + if not scenario.expect_trace_message_on_failure + ) + if opted_out: + pytest.skip( + "`expect_trace_message_on_failure: false` is set for " + f"{', '.join(repr(scenario_id) for scenario_id in opted_out)} in " + "`acceptance-test-config.yml`." + ) + + connector_root = self.get_connector_root_dir() + spec_result = run_test_job( + self.create_connector(scenario=None), + "spec", + test_scenario=None, + connector_root=connector_root, + ) + spec = get_single_spec(spec_result, connector_name=self.connector_name) + invalid_config = make_invalid_config(spec.connectionSpecification) + if invalid_config is None: + pytest.skip( + "The connector's spec requires no top-level property, so no config is " + "certain to be invalid." + ) + + scenario = ConnectorTestScenario(config_dict=invalid_config, status="failed") + catalog = ConfiguredAirbyteCatalog( + streams=[ + ConfiguredAirbyteStream( + stream=AirbyteStream( + name="__AIRBYTE__stream_that_does_not_exist", + json_schema={ + "type": "object", + "properties": {"f1": {"type": "string"}}, + }, + supported_sync_modes=[SyncMode.full_refresh], + ), + sync_mode=SyncMode.full_refresh, + destination_sync_mode=DestinationSyncMode.append, + ) + ] + ) + try: + connector = self.create_connector(scenario) + except Exception as error: # noqa: BLE001 # Any error here is the read failing. + result = EntrypointOutput(messages=[], uncaught_exception=error) + else: + # Asserts that the read fails. + result = run_test_job( + connector, + "read", + connector_root=connector_root, + test_scenario=scenario, + catalog=catalog, + ) + assert_error_trace_messages(result.get_message_iterator()) diff --git a/cdk-migrations.md b/cdk-migrations.md index ae3781c9b..6503318b1 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -1,5 +1,41 @@ # CDK Migration Guide +## Upgrading to the next minor release (unreleased) + +This release of the CDK adds three checks of the Connector Acceptance Tests (CAT) basic read test to the Standard Tests. 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, so connector CI picks the new checks up with this release, without a per-connector CDK bump. + +### Every record has a usable primary key (`basic_read`) + +For every config listed under `basic_read` in `acceptance-test-config.yml` (and not declared `status: failed` or `status: exception`), the in-process `test_basic_read` now checks the primary key of every record of every stream that declares a `source_defined_primary_key`, as CAT's `validate_primary_keys_data_type` check did: + +- No key field may hold an object or an array. Strings, numbers and booleans are accepted, as in CAT. +- At least one key field must be present and not null. In a composite key some fields may be null, as in CAT. + +The failure message names each stream, its primary key, the number of records with an invalid key, and for each distinct problem the record number and the path of the key field (for example ``record #3: key field `$.id` is an object``). Record values are never printed. + +### Every stream emits state (`basic_read`) + +For the same configs, `test_basic_read` now checks the state messages of the read, replacing CAT's `validate_state_messages` check: + +- Every stream the read covers must be named by at least one `STREAM` state message, or by the `stream_states` of a `GLOBAL` state message. CAT only compared the number of state messages with the number of streams, so the many state messages of one stream could hide another stream that emitted none. +- No state message may use the deprecated `LEGACY` state type, or no type. +- Every state message must carry `sourceStats.recordCount`. + +The CDK emits a final state message for every stream it completes, including full-refresh streams, and its entrypoint adds `sourceStats`, so a CDK connector meets the last two checks whenever it reads every stream to the end. + +### A failed read reports a well-formed error trace + +A new per-connector test, `SourceTestSuiteBase.test_fail_read_with_invalid_config`, ports CAT's `test_airbyte_trace_message_on_failure`. It runs `read` with a config that the connector's own spec rejects: a config that holds none of the spec's required top-level properties and no value from any real config, so it needs no credentials and cannot send a request with real ones. The read must fail and emit at least one `TRACE` message of type `ERROR`, and every `ERROR` trace message must be well-formed: it carries an `error` object with a non-blank `message` and a `failure_type`, and a positive `emitted_at`. The test is skipped when the spec requires no top-level property, and when any `basic_read` entry of `acceptance-test-config.yml` sets `expect_trace_message_on_failure: false`, as in CAT. `test_fail_read_with_bad_catalog` now checks its error trace messages the same way, instead of only requiring some trace message. + +Migration steps: + +- A record with an invalid primary key: fix the stream's `primary_key` (or the record) so that every record has a scalar key. To skip the check for one config while the fix is pending, set `validate_primary_keys_data_type: false` (with a `bypass_reason`) in its `basic_read` entry. As in CAT, this is allowed at `test_strictness_level: high`. +- A stream without state: a stream that ends early, or a source that emits only `LEGACY` state, has to be fixed. To skip the check for one config while the fix is pending, set `validate_state_messages: false` (with a `bypass_reason`) in its `basic_read` entry. +- A failed read without a well-formed error trace: make the connector raise (or emit) an `AirbyteTracedException` for invalid configs. To skip the check, set `expect_trace_message_on_failure: false` (with a `bypass_reason`) in any `basic_read` entry; this skips the new test for the whole connector. +- A config listed under several sections keeps an opt-out that any of its entries declares, as for `validate_schema`. + +Rationale: Destinations deduplicate on the primary key; a record whose key is an object, an array or entirely missing cannot be deduplicated. A stream that emits no state cannot be resumed, and the platform cannot tell whether it completed. A failed read without a well-formed error trace gives users and the platform no reason for the failure. CAT checked all three for every `basic_read` config, and the Standard Tests that replaced it did not. + ## 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/read_test_helpers.py b/unit_tests/test/read_test_helpers.py new file mode 100644 index 000000000..b9895dfe1 --- /dev/null +++ b/unit_tests/test/read_test_helpers.py @@ -0,0 +1,38 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Message builders shared by the `read` Standard Test unit tests.""" + +from __future__ import annotations + +from airbyte_cdk.models import ( + AirbyteMessage, + AirbyteStateBlob, + AirbyteStateMessage, + AirbyteStateStats, + AirbyteStateType, + AirbyteStreamState, + StreamDescriptor, + Type, +) + + +def final_state( + stream: str, *, namespace: str | None = None, record_count: float | None = None +) -> AirbyteMessage: + """The final state the CDK emits once a full-refresh stream completes. + + With `record_count`, the message carries `sourceStats`, as the CDK entrypoint adds them; + without it, the entrypoint (when the message goes through one) sets them. + """ + return AirbyteMessage( + type=Type.STATE, + state=AirbyteStateMessage( + type=AirbyteStateType.STREAM, + stream=AirbyteStreamState( + stream_descriptor=StreamDescriptor(name=stream, namespace=namespace), + stream_state=AirbyteStateBlob(__ab_no_cursor_state_message=True), + ), + sourceStats=( + AirbyteStateStats(recordCount=record_count) if record_count is not None else None + ), + ), + ) diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py index 89fb820dd..7c3c4716f 100644 --- a/unit_tests/test/test_read_assertions.py +++ b/unit_tests/test/test_read_assertions.py @@ -46,6 +46,7 @@ ) from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite from airbyte_cdk.utils.connector_paths import ACCEPTANCE_TEST_CONFIG +from unit_tests.test.read_test_helpers import final_state MSG_MISSING_STREAMS = "Every configured stream must return at least one record" MSG_SCHEMA = "Records do not match their stream's JSON schema" @@ -565,6 +566,7 @@ def read( for configured_stream in catalog.streams: for data in self._records.get(configured_stream.stream.name, []): yield _record(configured_stream.stream.name, data) + yield final_state(configured_stream.stream.name) def _run_basic_read( diff --git a/unit_tests/test/test_read_keys_state_and_traces.py b/unit_tests/test/test_read_keys_state_and_traces.py new file mode 100644 index 000000000..b14c8fb16 --- /dev/null +++ b/unit_tests/test/test_read_keys_state_and_traces.py @@ -0,0 +1,740 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Unit tests for the primary key, state message and error trace checks of the `read` tests. + +These port three more checks of the Connector Acceptance Tests (CAT) basic read test: + +- every record of a stream with a `source_defined_primary_key` has a usable primary key + (`_validate_primary_keys_data_type`); +- every configured stream emits per-stream state with `sourceStats` (`_validate_state_messages`); +- a failing read emits a well-formed `ERROR` trace message + (`test_airbyte_trace_message_on_failure`). +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any, Iterable, Mapping + +import pytest +import yaml + +from airbyte_cdk.models import ( + AirbyteCatalog, + AirbyteConnectionStatus, + AirbyteErrorTraceMessage, + AirbyteGlobalState, + AirbyteMessage, + AirbyteRecordMessage, + AirbyteStateBlob, + AirbyteStateMessage, + AirbyteStateStats, + AirbyteStateType, + AirbyteStream, + AirbyteStreamState, + AirbyteTraceMessage, + ConfiguredAirbyteCatalog, + ConfiguredAirbyteStream, + ConnectorSpecification, + DestinationSyncMode, + FailureType, + Status, + StreamDescriptor, + SyncMode, + TraceType, + Type, +) +from airbyte_cdk.sources import Source +from airbyte_cdk.test.models import ConnectorTestScenario +from airbyte_cdk.test.standard_tests import SourceTestSuiteBase +from airbyte_cdk.test.standard_tests._job_runner import IConnector +from airbyte_cdk.test.standard_tests._read_assertions import ( + INVALID_CONFIG_MARKER, + MAX_PRIMARY_KEY_ERRORS_PER_STREAM, + assert_error_trace_messages, + assert_read_records, + make_invalid_config, +) +from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite +from airbyte_cdk.utils.connector_paths import ACCEPTANCE_TEST_CONFIG +from airbyte_cdk.utils.traced_exception import AirbyteTracedException +from unit_tests.test.read_test_helpers import final_state + +MSG_PRIMARY_KEY = "Records have invalid primary keys" +MSG_STATE = "The read's state messages are incomplete" +MSG_NO_ERROR_TRACE = "emitted no `TRACE` message of type `ERROR`" +MSG_MALFORMED_TRACE = "trace messages of the failed read are not well-formed" +SENTINEL = "sk_live_SENTINEL_5c4b3a" + +SCHEMA: dict[str, Any] = {"type": "object", "properties": {"id": {}, "org": {}, "name": {}}} + + +def _catalog( + primary_keys: Mapping[str, list[list[str]] | None], namespace: str | None = None +) -> ConfiguredAirbyteCatalog: + return ConfiguredAirbyteCatalog( + streams=[ + ConfiguredAirbyteStream( + stream=AirbyteStream( + name=name, + namespace=namespace, + json_schema=SCHEMA, + supported_sync_modes=[SyncMode.full_refresh], + source_defined_primary_key=primary_key, + ), + sync_mode=SyncMode.full_refresh, + destination_sync_mode=DestinationSyncMode.append_dedup, + ) + for name, primary_key in primary_keys.items() + ] + ) + + +def _record(stream: str, data: Mapping[str, Any]) -> AirbyteMessage: + return AirbyteMessage( + type=Type.RECORD, + record=AirbyteRecordMessage(stream=stream, data=dict(data), emitted_at=0), + ) + + +def _key_error( + primary_key: list[list[str]] | None, + *records: Mapping[str, Any], + validate_primary_keys: bool = True, +) -> str | None: + """Check `records` of a single stream; return the failure message, or None if they pass.""" + try: + assert_read_records( + records=[_record("s", data) for data in records], + configured_catalog=_catalog({"s": primary_key}), + require_records_per_stream=True, + validate_schema=False, + validate_primary_keys=validate_primary_keys, + ) + except AssertionError as error: + return str(error) + return None + + +# --- Primary keys ---------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "primary_key, data", + [ + pytest.param([["id"]], {"id": 1}, id="integer"), + pytest.param([["id"]], {"id": "a"}, id="string"), + pytest.param([["id"]], {"id": 1.5}, id="number"), + # CAT only forbade objects and arrays; a boolean is a scalar. + pytest.param([["id"]], {"id": True}, id="boolean"), + pytest.param([["id"]], {"id": ""}, id="empty_string"), + pytest.param([["id"]], {"id": 0}, id="zero"), + pytest.param([["id"], ["org"]], {"id": 1, "org": "x"}, id="composite"), + # As in CAT, a composite key may have some of its fields null or missing. + pytest.param([["id"], ["org"]], {"id": 1, "org": None}, id="composite_one_null"), + pytest.param([["id"], ["org"]], {"org": "x"}, id="composite_one_missing"), + pytest.param([["org", "id"]], {"org": {"id": 7}}, id="nested_key_path"), + ], +) +def test_usable_primary_keys_pass(primary_key: list[list[str]], data: Mapping[str, Any]) -> None: + assert _key_error(primary_key, data) is None + + +@pytest.mark.parametrize( + "primary_key, data, expected", + [ + pytest.param( + [["id"]], + {"name": "x"}, + "record #0: every key field is missing or null (`$.id` is missing)", + id="missing", + ), + pytest.param( + [["id"]], + {"id": None}, + "record #0: every key field is missing or null (`$.id` is null)", + id="null", + ), + pytest.param( + [["id"], ["org"]], + {"id": None}, + "every key field is missing or null (`$.id` is null, `$.org` is missing)", + id="composite_all_missing_or_null", + ), + pytest.param( + [["org", "id"]], + {"org": "not an object"}, + "every key field is missing or null (`$.org.id` is missing)", + id="nested_parent_not_an_object", + ), + pytest.param( + [["id"]], + {"id": {"value": SENTINEL}}, + "record #0: key field `$.id` is an object; a primary key field must hold a scalar", + id="object", + ), + pytest.param( + [["id"]], + {"id": [SENTINEL]}, + "record #0: key field `$.id` is an array", + id="array", + ), + pytest.param( + [["id"], ["org"]], + {"id": 1, "org": {"name": SENTINEL}}, + "key field `$.org` is an object", + id="composite_one_object", + ), + ], +) +def test_unusable_primary_keys_fail_with_stream_record_and_key_path( + primary_key: list[list[str]], data: Mapping[str, Any], expected: str +) -> None: + message = _key_error(primary_key, data) + assert message is not None + assert MSG_PRIMARY_KEY in message + assert "- Stream 's' (primary key " in message + assert "1 of 1 records have an invalid primary key." in message + assert expected in message + assert "validate_primary_keys_data_type: false" in message + assert SENTINEL not in message + + +def test_primary_key_errors_never_print_key_values() -> None: + message = _key_error([["id"], ["org"]], {"id": {"x": SENTINEL}, "org": [SENTINEL]}) + assert message is not None + assert "`$.id` is an object" in message + assert "`$.org` is an array" in message + assert SENTINEL not in message + + +def test_streams_without_a_primary_key_are_not_checked() -> None: + assert _key_error(None, {"id": None}) is None + assert _key_error([], {"id": {"x": 1}}) is None + + +def test_validate_primary_keys_false_skips_the_check() -> None: + assert _key_error([["id"]], {"id": None}, validate_primary_keys=False) is None + + +def test_primary_keys_are_not_checked_by_default() -> None: + """Callers that predate the check (the Docker read test) keep their behaviour.""" + assert_read_records( + records=[_record("s", {"id": None})], + configured_catalog=_catalog({"s": [["id"]]}), + require_records_per_stream=True, + validate_schema=False, + ) + + +def test_record_numbers_count_the_records_of_each_stream() -> None: + with pytest.raises(AssertionError) as error: + assert_read_records( + records=[ + _record("a", {"id": 1}), + _record("b", {"id": 1}), + _record("a", {"id": 2}), + _record("a", {"id": None}), + _record("b", {"id": None}), + ], + configured_catalog=_catalog({"a": [["id"]], "b": [["id"]]}), + require_records_per_stream=True, + validate_schema=False, + validate_primary_keys=True, + ) + message = str(error.value) + assert "- Stream 'a' (primary key `$.id`): 1 of 3 records" in message + assert "record #2: every key field is missing or null" in message + assert "- Stream 'b' (primary key `$.id`): 1 of 2 records" in message + assert "record #1: every key field is missing or null" in message + + +def test_each_primary_key_problem_is_reported_once_with_its_first_record() -> None: + message = _key_error([["id"]], {"id": 1}, {"id": None}, {"id": None}, {"id": [1]}, {"id": [2]}) + assert message is not None + assert "4 of 5 records have an invalid primary key." in message + assert message.count("is null") == 1 + assert "record #1: every key field is missing or null (`$.id` is null)" in message + assert message.count("is an array") == 1 + assert "record #3: key field `$.id` is an array" in message + + +def test_distinct_primary_key_errors_are_capped_per_stream() -> None: + fields = [f"k{index}" for index in range(MAX_PRIMARY_KEY_ERRORS_PER_STREAM + 2)] + message = _key_error([[name] for name in fields], {name: [1] for name in fields}) + assert message is not None + assert message.count("is an array") == MAX_PRIMARY_KEY_ERRORS_PER_STREAM + assert "...and 2 more distinct primary key error(s)." in message + + +def test_primary_key_failures_are_reported_with_presence_failures() -> None: + with pytest.raises(AssertionError) as error: + assert_read_records( + records=[_record("a", {"id": None})], + configured_catalog=_catalog({"a": [["id"]], "b": [["id"]]}), + require_records_per_stream=True, + validate_schema=False, + validate_primary_keys=True, + ) + assert "Every configured stream must return at least one record" in str(error.value) + assert MSG_PRIMARY_KEY in str(error.value) + + +# --- State messages -------------------------------------------------------------------------- + + +def _state_error( + state_messages: Iterable[AirbyteMessage], + streams: Iterable[str] = ("a", "b"), + namespace: str | None = None, +) -> str | None: + """Check a read of `streams` (one record each) with `state_messages`.""" + stream_names = list(streams) + try: + assert_read_records( + records=[_record(name, {"id": 1}) for name in stream_names], + configured_catalog=_catalog({name: None for name in stream_names}, namespace), + require_records_per_stream=True, + validate_schema=False, + state_messages=state_messages, + ) + except AssertionError as error: + return str(error) + return None + + +def _global_state(*streams: str) -> AirbyteMessage: + return AirbyteMessage( + type=Type.STATE, + state=AirbyteStateMessage( + type=AirbyteStateType.GLOBAL, + global_=AirbyteGlobalState( + shared_state=AirbyteStateBlob(cursor=1), + stream_states=[ + AirbyteStreamState( + stream_descriptor=StreamDescriptor(name=name), + stream_state=AirbyteStateBlob(cursor=1), + ) + for name in streams + ], + ), + sourceStats=AirbyteStateStats(recordCount=1.0), + ), + ) + + +def _legacy_state() -> AirbyteMessage: + return AirbyteMessage( + type=Type.STATE, + state=AirbyteStateMessage( + type=AirbyteStateType.LEGACY, + data={"a": {"cursor": 1}}, + sourceStats=AirbyteStateStats(recordCount=1.0), + ), + ) + + +def test_a_final_state_per_stream_passes() -> None: + assert ( + _state_error([final_state("a", record_count=1), final_state("b", record_count=1)]) is None + ) + + +def test_a_global_state_covers_its_streams() -> None: + assert _state_error([_global_state("a", "b")]) is None + + +def test_state_is_matched_by_namespace_and_name() -> None: + assert ( + _state_error( + [final_state(name, namespace="ns", record_count=1) for name in ("a", "b")], + namespace="ns", + ) + is None + ) + message = _state_error( + [final_state(name, record_count=1) for name in ("a", "b")], namespace="ns" + ) + assert message is not None + assert "'ns.a', 'ns.b'" in message + + +def test_a_stream_without_state_fails_and_is_named() -> None: + message = _state_error([final_state("a", record_count=1)]) + assert message is not None + assert MSG_STATE in message + assert ( + "1 of 2 configured streams emitted no `STREAM` or `GLOBAL` state message: 'b'." in message + ) + assert "validate_state_messages: false" in message + + +def test_many_states_from_one_stream_do_not_cover_another() -> None: + """CAT only compared counts: two states for `a` hid that `b` emitted none.""" + message = _state_error([final_state("a", record_count=1), final_state("a", record_count=0)]) + assert message is not None + assert "emitted no `STREAM` or `GLOBAL` state message: 'b'." in message + + +def test_a_read_without_state_fails() -> None: + message = _state_error([]) + assert message is not None + assert "2 of 2 configured streams emitted no" in message + + +def test_legacy_state_fails() -> None: + message = _state_error( + [final_state("a", record_count=1), final_state("b", record_count=1), _legacy_state()] + ) + assert message is not None + assert "1 of 3 state messages use the deprecated `LEGACY` state type" in message + assert "state message(s) #2." in message + + +def test_state_without_source_stats_fails() -> None: + message = _state_error([final_state("a", record_count=1), final_state("b")]) + assert message is not None + assert "1 of 2 state messages carry no `sourceStats.recordCount`" in message + assert "state message(s) #1." in message + + +def test_state_messages_none_skips_the_check() -> None: + assert_read_records( + records=[_record("a", {"id": 1})], + configured_catalog=_catalog({"a": None}), + require_records_per_stream=True, + validate_schema=False, + state_messages=None, + ) + + +def test_other_messages_among_the_state_messages_are_ignored() -> None: + assert ( + _state_error( + [_record("a", {"id": 1}), final_state("a", record_count=1), _global_state("b")] + ) + is None + ) + + +# --- Error trace messages -------------------------------------------------------------------- + + +def _error_trace( + *, + message: str | None = "The API rejected the credentials.", + failure_type: FailureType | None = FailureType.config_error, + emitted_at: float = 1_700_000_000_000.0, + with_error: bool = True, +) -> AirbyteMessage: + return AirbyteMessage( + type=Type.TRACE, + trace=AirbyteTraceMessage( + type=TraceType.ERROR, + emitted_at=emitted_at, + error=( + AirbyteErrorTraceMessage( + message=message, # type: ignore[arg-type] # A malformed trace. + internal_message=SENTINEL, + failure_type=failure_type, + ) + if with_error + else None + ), + ), + ) + + +def test_a_well_formed_error_trace_passes() -> None: + assert_error_trace_messages([_record("a", {"id": 1}), _error_trace()]) + + +def test_no_error_trace_fails() -> None: + log = AirbyteMessage( + type=Type.TRACE, trace=AirbyteTraceMessage(type=TraceType.ESTIMATE, emitted_at=1.0) + ) + with pytest.raises(AssertionError, match=MSG_NO_ERROR_TRACE): + assert_error_trace_messages([_record("a", {"id": 1}), log]) + + +@pytest.mark.parametrize( + "trace, expected", + [ + pytest.param(_error_trace(message=""), "a non-blank `error.message`", id="empty_message"), + pytest.param(_error_trace(message=" \n"), "a non-blank `error.message`", id="blank"), + pytest.param(_error_trace(message=None), "a non-blank `error.message`", id="no_message"), + pytest.param( + _error_trace(failure_type=None), "an `error.failure_type`", id="no_failure_type" + ), + pytest.param(_error_trace(emitted_at=0), "a positive `emitted_at`", id="no_emitted_at"), + pytest.param(_error_trace(with_error=False), "the `error` object", id="no_error_object"), + ], +) +def test_malformed_error_traces_fail(trace: AirbyteMessage, expected: str) -> None: + with pytest.raises(AssertionError, match=MSG_MALFORMED_TRACE) as error: + assert_error_trace_messages([_error_trace(), trace]) + message = str(error.value) + assert "1 of 2 `ERROR` trace messages" in message + assert f"- error trace #1 lacks {expected}" in message + assert SENTINEL not in message + + +def test_make_invalid_config() -> None: + assert make_invalid_config({"type": "object", "required": ["api_key"]}) == { + INVALID_CONFIG_MARKER: True + } + assert make_invalid_config({"type": "object", "properties": {"a": {}}}) is None + assert make_invalid_config({"type": "object", "required": []}) is None + + +# --- Through the source test suite ----------------------------------------------------------- + +SPEC_REQUIRING_API_KEY: dict[str, Any] = { + "type": "object", + "required": ["api_key"], + "properties": {"api_key": {"type": "string", "airbyte_secret": True}}, +} + + +class _Source(Source): + """A source over one `users` stream with primary key `id`. + + Behaviours: + - `records`: the given records, then the final state (unless `emit_state` is false). + - `skip_config_check`: like `records`, without validating the config against the spec. + - `malformed_trace`: without validating the config, yield an `ERROR` trace with a blank + message, then raise. + """ + + def __init__( + self, + records: list[Mapping[str, Any]] | None = None, + *, + behavior: str = "records", + spec: Mapping[str, Any] = SPEC_REQUIRING_API_KEY, + emit_state: bool = True, + ) -> None: + self._records = records or [] + self._behavior = behavior + self._spec = dict(spec) + self._emit_state = emit_state + self.check_config_against_spec = behavior == "records" + + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + return ConnectorSpecification(connectionSpecification=self._spec) + + def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus: + return AirbyteConnectionStatus(status=Status.SUCCEEDED) + + def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCatalog: + return AirbyteCatalog( + streams=[ + AirbyteStream( + name="users", + json_schema=SCHEMA, + supported_sync_modes=[SyncMode.full_refresh], + source_defined_primary_key=[["id"]], + ) + ] + ) + + def read( + self, + logger: logging.Logger, + config: Mapping[str, Any], + catalog: ConfiguredAirbyteCatalog, + state: list[AirbyteStateMessage] | None = None, + ) -> Iterable[AirbyteMessage]: + if self._behavior == "malformed_trace": + yield _error_trace(message=" ") + raise RuntimeError("The read failed.") + for configured_stream in catalog.streams: + if configured_stream.stream.name != "users": + continue + for data in self._records: + yield _record("users", data) + if self._emit_state: + yield final_state("users") + + +def _suite(tmp_path: Path, source: _Source | Exception) -> SourceTestSuiteBase: + class _Suite(SourceTestSuiteBase): + @classmethod + def get_connector_root_dir(cls) -> Path: + return tmp_path + + @classmethod + def create_connector(cls, scenario: ConnectorTestScenario | None) -> IConnector: + if isinstance(source, Exception): + if scenario is not None and scenario.config_dict == {INVALID_CONFIG_MARKER: True}: + raise source + return _Source() + return source + + return _Suite() + + +def _scenario(**fields: Any) -> ConnectorTestScenario: + return ConnectorTestScenario.model_validate( + {"config_dict": {"api_key": "x"}, "sections": ("basic_read",), **fields} + ) + + +@pytest.mark.parametrize( + "scenario, source, failure_match", + [ + pytest.param(_scenario(), _Source([{"id": 1}]), None, id="valid"), + pytest.param( + _scenario(), _Source([{"id": None, "name": "x"}]), MSG_PRIMARY_KEY, id="null_key" + ), + pytest.param( + _scenario(validate_primary_keys_data_type=False), + _Source([{"id": None, "name": "x"}]), + None, + id="null_key_opted_out", + ), + pytest.param(_scenario(), _Source([{"id": 1}], emit_state=False), MSG_STATE, id="no_state"), + pytest.param( + _scenario(validate_state_messages=False), + _Source([{"id": 1}], emit_state=False), + None, + id="no_state_opted_out", + ), + # A config read only because it is listed under `full_refresh` keeps the old floor. + pytest.param( + _scenario(sections=("full_refresh",)), + _Source([{"id": None, "name": "x"}], emit_state=False), + None, + id="full_refresh_only", + ), + # `validate_primary_keys_data_type: false` stays available at high strictness, as in CAT. + pytest.param( + _scenario(validate_primary_keys_data_type=False, test_strictness_level="high"), + _Source([{"id": None, "name": "x"}]), + None, + id="high_strictness_allows_primary_key_opt_out", + ), + ], +) +def test_basic_read( + tmp_path: Path, + scenario: ConnectorTestScenario, + source: _Source, + failure_match: str | None, +) -> None: + suite = _suite(tmp_path, source) + if failure_match is None: + suite.test_basic_read(scenario) + return + with pytest.raises(AssertionError, match=failure_match): + suite.test_basic_read(scenario) + + +def test_fail_read_with_invalid_config_passes_when_the_spec_rejects_it(tmp_path: Path) -> None: + _suite(tmp_path, _Source([{"id": 1}])).test_fail_read_with_invalid_config() + + +def test_fail_read_with_invalid_config_fails_when_the_read_succeeds(tmp_path: Path) -> None: + with pytest.raises(AssertionError, match="Expected exception but got none"): + _suite( + tmp_path, _Source([{"id": 1}], behavior="skip_config_check") + ).test_fail_read_with_invalid_config() + + +def test_fail_read_with_invalid_config_fails_on_a_malformed_trace(tmp_path: Path) -> None: + with pytest.raises(AssertionError, match=MSG_MALFORMED_TRACE) as error: + _suite(tmp_path, _Source(behavior="malformed_trace")).test_fail_read_with_invalid_config() + assert "1 of 2 `ERROR` trace messages" in str(error.value) + assert SENTINEL not in str(error.value) + + +def test_fail_read_with_invalid_config_when_the_source_cannot_be_created(tmp_path: Path) -> None: + _suite( + tmp_path, + AirbyteTracedException( + message="The config is invalid.", failure_type=FailureType.config_error + ), + ).test_fail_read_with_invalid_config() + _suite(tmp_path, KeyError("api_key")).test_fail_read_with_invalid_config() + + +def test_fail_read_with_invalid_config_skips_without_required_properties(tmp_path: Path) -> None: + source = _Source(spec={"type": "object", "properties": {}}) + with pytest.raises(pytest.skip.Exception, match="requires no top-level property"): + _suite(tmp_path, source).test_fail_read_with_invalid_config() + + +def test_fail_read_with_invalid_config_honours_the_opt_out(tmp_path: Path) -> None: + (tmp_path / ACCEPTANCE_TEST_CONFIG).write_text( + yaml.safe_dump( + { + "acceptance_tests": { + "basic_read": { + "tests": [ + { + "config_path": "secrets/config.json", + "expect_trace_message_on_failure": False, + } + ] + } + } + } + ) + ) + with pytest.raises(pytest.skip.Exception, match="expect_trace_message_on_failure: false"): + _suite( + tmp_path, _Source([{"id": 1}], behavior="skip_config_check") + ).test_fail_read_with_invalid_config() + + +def test_fail_read_with_bad_catalog_checks_the_trace(tmp_path: Path) -> None: + _suite(tmp_path, _Source([{"id": 1}])).test_fail_read_with_bad_catalog(_scenario()) + + +# --- Opt-outs in `acceptance-test-config.yml` ------------------------------------------------ + +OPT_OUTS = ( + "validate_primary_keys_data_type", + "validate_state_messages", + "expect_trace_message_on_failure", +) + + +@pytest.mark.parametrize("opt_out", OPT_OUTS) +@pytest.mark.parametrize( + "values", + [ + pytest.param((False, True), id="false_then_true"), + pytest.param((True, False), id="true_then_false"), + ], +) +def test_opt_outs_are_kept_through_scenario_dedup( + tmp_path: Path, opt_out: str, values: tuple[bool, bool] +) -> None: + (tmp_path / ACCEPTANCE_TEST_CONFIG).write_text( + yaml.safe_dump( + { + "acceptance_tests": { + "connection": { + "tests": [{"config_path": "secrets/config.json", "status": "succeed"}] + }, + "basic_read": { + "tests": [ + {"config_path": "secrets/config.json", opt_out: value} + for value in values + ] + }, + } + } + ) + ) + + class _Suite(DockerConnectorTestSuite): + @classmethod + def get_connector_root_dir(cls) -> Path: + return tmp_path + + [scenario] = _Suite.get_scenarios() + assert getattr(scenario, opt_out) is False + assert all(getattr(scenario, other) for other in OPT_OUTS if other != opt_out) + assert scenario.validate_schema is True diff --git a/unit_tests/test/test_scenario_sections.py b/unit_tests/test/test_scenario_sections.py index af3b23da4..ca2987889 100644 --- a/unit_tests/test/test_scenario_sections.py +++ b/unit_tests/test/test_scenario_sections.py @@ -23,11 +23,15 @@ AirbyteConnectionStatus, AirbyteMessage, AirbyteRecordMessage, + AirbyteStateBlob, AirbyteStateMessage, + AirbyteStateType, AirbyteStream, + AirbyteStreamState, ConfiguredAirbyteCatalog, ConnectorSpecification, Status, + StreamDescriptor, SyncMode, Type, ) @@ -589,6 +593,17 @@ def read( stream=name, data={"id": 1, "updated_at": "2026-01-01"}, emitted_at=0 ), ) + # The final state the CDK emits once a full-refresh stream completes. + yield AirbyteMessage( + type=Type.STATE, + state=AirbyteStateMessage( + type=AirbyteStateType.STREAM, + stream=AirbyteStreamState( + stream_descriptor=StreamDescriptor(name=name), + stream_state=AirbyteStateBlob(__ab_no_cursor_state_message=True), + ), + ), + ) def _read_suite( From cc674f576ae4376213cc92845f0888eeeee6f1c9 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 14 Oct 2026 17:10:08 +0300 Subject: [PATCH 2/2] docs(cdk-migrations): give opt-out reasons as YAML comments, not bypass_reason A `basic_read` entry has no `bypass_reason` field: pydantic ignores the key silently, and only `empty_streams` items carry one. Tell readers to give the reason in a YAML comment, as the 7.36.0 `validate_schema` step and the connector opt-outs do. Co-Authored-By: Claude Opus 5.5 --- cdk-migrations.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cdk-migrations.md b/cdk-migrations.md index 6503318b1..7e18492f8 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -29,9 +29,9 @@ A new per-connector test, `SourceTestSuiteBase.test_fail_read_with_invalid_confi Migration steps: -- A record with an invalid primary key: fix the stream's `primary_key` (or the record) so that every record has a scalar key. To skip the check for one config while the fix is pending, set `validate_primary_keys_data_type: false` (with a `bypass_reason`) in its `basic_read` entry. As in CAT, this is allowed at `test_strictness_level: high`. -- A stream without state: a stream that ends early, or a source that emits only `LEGACY` state, has to be fixed. To skip the check for one config while the fix is pending, set `validate_state_messages: false` (with a `bypass_reason`) in its `basic_read` entry. -- A failed read without a well-formed error trace: make the connector raise (or emit) an `AirbyteTracedException` for invalid configs. To skip the check, set `expect_trace_message_on_failure: false` (with a `bypass_reason`) in any `basic_read` entry; this skips the new test for the whole connector. +- A record with an invalid primary key: fix the stream's `primary_key` (or the record) so that every record has a scalar key. To skip the check for one config while the fix is pending, set `validate_primary_keys_data_type: false` in its `basic_read` entry, with a YAML comment giving the reason. As in CAT, this is allowed at `test_strictness_level: high`. +- A stream without state: a stream that ends early, or a source that emits only `LEGACY` state, has to be fixed. To skip the check for one config while the fix is pending, set `validate_state_messages: false` in its `basic_read` entry, with a YAML comment giving the reason. +- A failed read without a well-formed error trace: make the connector raise (or emit) an `AirbyteTracedException` for invalid configs. To skip the check, set `expect_trace_message_on_failure: false` in any `basic_read` entry, with a YAML comment giving the reason; this skips the new test for the whole connector. - A config listed under several sections keeps an opt-out that any of its entries declares, as for `validate_schema`. Rationale: Destinations deduplicate on the primary key; a record whose key is an object, an array or entirely missing cannot be deduplicated. A stream that emits no state cannot be resumed, and the platform cannot tell whether it completed. A failed read without a well-formed error trace gives users and the platform no reason for the failure. CAT checked all three for every `basic_read` config, and the Standard Tests that replaced it did not.