From 6a670caf328136274a13c88806eec419826f1258 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Thu, 8 Oct 2026 20:58:10 +0300 Subject: [PATCH 1/8] feat(standard-tests)!: assert per-stream records and record schemas in the basic read test `SourceTestSuiteBase.test_basic_read` only failed when the whole read returned no records, so one stream with records passed the test for every stream, and no record was ever checked against its schema. Port the two CAT basic read checks that ran by default: - every stream of a `basic_read` config returns at least one record, except the ones declared in `empty_streams` (which are still not read); - every record matches its stream's JSON schema, with CAT's validator (Draft 7, strict integers, lenient `date-time`, extra columns allowed) and its structure check, unless the config sets `validate_schema: false`. Failure messages name the stream, the number of failing records, and the path of each offending value inside the record. The assertions live in `_read_assertions.py` and take the read output and catalog, so the Docker read test can reuse them once it runs. Part of airbytehq/airbyte-internal-issues#17135. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/models/scenario.py | 13 + .../test/standard_tests/_read_assertions.py | 317 ++++++++++ .../test/standard_tests/docker_base.py | 4 + .../test/standard_tests/source_base.py | 19 +- cdk-migrations.md | 18 + unit_tests/test/test_read_assertions.py | 548 ++++++++++++++++++ 6 files changed, 918 insertions(+), 1 deletion(-) create mode 100644 airbyte_cdk/test/standard_tests/_read_assertions.py create mode 100644 unit_tests/test/test_read_assertions.py diff --git a/airbyte_cdk/test/models/scenario.py b/airbyte_cdk/test/models/scenario.py index 0b5764b033..832c05a7a8 100644 --- a/airbyte_cdk/test/models/scenario.py +++ b/airbyte_cdk/test/models/scenario.py @@ -73,6 +73,8 @@ def __hash__(self) -> int: expect_records: AcceptanceTestExpectRecords | None = None file_types: AcceptanceTestFileTypes | None = None status: Literal["succeed", "failed", "exception"] | None = None + # CAT's `basic_read` opt-out from validating records against the stream schemas. + validate_schema: bool = True # The sections of `acceptance-test-config.yml` that list this config. Populated by # `get_scenarios()`; empty for scenarios built by hand, which run every test. sections: tuple[str, ...] = () @@ -135,6 +137,17 @@ def check_only(self) -> bool: section in FULL_SUITE_SECTIONS for section in self.sections ) + @property + def is_basic_read_config(self) -> bool: + """Whether this config is listed under `basic_read` (or the scenario was built by hand). + + Only these configs must return records from every stream not declared in + `empty_streams`, as in CAT. A config listed only under `spec` or `connection` never had a + basic read test, and has nowhere to declare its empty streams, so its `read` only has to + return some records. + """ + return not self.sections or "basic_read" in self.sections + @property def id(self) -> str: """Return a unique identifier for the test scenario. diff --git a/airbyte_cdk/test/standard_tests/_read_assertions.py b/airbyte_cdk/test/standard_tests/_read_assertions.py new file mode 100644 index 0000000000..f5f75e6e04 --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_read_assertions.py @@ -0,0 +1,317 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Record assertions for the `read` standard tests. + +These port the record checks of the Connector Acceptance Tests (CAT) `TestBasicRead.test_read`, +which the standard tests did not reproduce: + +- 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. + +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. +""" + +from __future__ import annotations + +import re +from collections import Counter +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field +from typing import Any + +from jsonschema import Draft7Validator, FormatChecker, validators +from jsonschema.exceptions import FormatError, ValidationError + +from airbyte_cdk.models import AirbyteMessage, ConfiguredAirbyteCatalog +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_MESSAGE_CHARS = 300 + +# CAT's shape check for `date-time` values: a date, a ' ' or 'T' separator, a time, then anything +# (fraction, offset, zone name). The value must also parse as a datetime. +_DATETIME_SHAPE = re.compile(r"^\d{4}-\d?\d-\d?\d(\s|T)\d?\d:\d?\d:\d?\d(.\d+)?.*$") + +_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def _is_strict_integer(_checker: object, instance: object) -> bool: + # JSON Schema treats `1.0` as an integer; destinations do not, so neither do we. Booleans are + # `int` subclasses in Python but not integers in JSON. + return isinstance(instance, int) and not isinstance(instance, bool) + + +StrictIntegerDraft7Validator = validators.extend( + Draft7Validator, + type_checker=Draft7Validator.TYPE_CHECKER.redefine("integer", _is_strict_integer), +) +"""Draft 7 validator that only accepts `int` values as `integer`.""" + + +class RecordFormatChecker(FormatChecker): + """Format checker for records: CAT's lenient `date-time` check, defaults for other formats. + + The default `date-time` check requires strict RFC 3339. Sources emit many shapes that + destinations accept (a ' ' separator, no offset, a zone name), so `date-time` values only + need a date-and-time shape that parses as a datetime. + """ + + def check(self, instance: object, format: str) -> None: + if format == "date-time": + if isinstance(instance, str) and not _is_valid_datetime(instance): + raise FormatError(f"{instance!r} is not a valid 'date-time'") + return + + super().check(instance, format) + + +def _is_valid_datetime(value: str) -> bool: + return bool(_DATETIME_SHAPE.match(value)) and ab_datetime_try_parse(value) is not None + + +def format_record_path(path: Iterable[Any]) -> str: + """Render a path inside a record as a JSONPath, e.g. `$.items[0]['first name']`.""" + rendered = "$" + for part in path: + if isinstance(part, int): + rendered += f"[{part}]" + elif _IDENTIFIER.match(str(part)): + rendered += f".{part}" + else: + rendered += f"[{str(part)!r}]" + return rendered + + +def _truncate(text: str) -> str: + if len(text) <= _MAX_MESSAGE_CHARS: + return text + return text[: _MAX_MESSAGE_CHARS - 3] + "..." + + +def _resolve_local_ref(root: Mapping[str, Any], ref: str) -> Any: + if not ref.startswith("#/"): + return None + node: Any = root + for part in ref[2:].split("/"): + if not isinstance(node, Mapping) or part not in node: + return None + node = node[part] + return node + + +def get_expected_schema_paths(schema: Mapping[str, Any]) -> set[str]: + """Return the leaf property paths a record of this schema is expected to have. + + Paths look like `/a`, `/a/b` and `/list/[]/c`. A `oneOf`/`anyOf` contributes the paths of + every option. An object without `properties` is a leaf. Local `$ref`s are resolved, and a + recursive reference stops the descent. + + Ported from CAT's `get_expected_schema_structure`. + """ + paths: set[str] = set() + on_stack: set[int] = set() + + def _scan(subschema: Any, path: str) -> None: + if not isinstance(subschema, Mapping): + if path: + paths.add(path) + return + if "$ref" in subschema: + target = _resolve_local_ref(schema, subschema["$ref"]) + if target is None or id(target) in on_stack: + if path: + paths.add(path) + return + on_stack.add(id(target)) + _scan(target, path) + on_stack.discard(id(target)) + return + + options = subschema.get("oneOf") or subschema.get("anyOf") + if options: + for option in options: + _scan({"type": "object", **option} if isinstance(option, Mapping) else option, path) + return + + schema_type = subschema.get("type", ["object", "null"]) + if not isinstance(schema_type, list): + schema_type = [schema_type] + if "object" in schema_type and subschema.get("properties"): + for name, property_schema in subschema["properties"].items(): + _scan(property_schema, f"{path}/{name}") + return + if "array" in schema_type and "object" not in schema_type: + _scan(subschema.get("items", {}), f"{path}/[]") + return + if path: + paths.add(path) + + _scan(schema, "") + return paths + + +def get_record_paths(data: Any) -> set[str]: + """Return every key path in a record, with list items represented by their first element. + + Ported from CAT's `get_object_structure`. + """ + paths: set[str] = set() + + def _traverse(value: Any, path: str) -> None: + if path: + paths.add(path) + if isinstance(value, dict): + for key, nested in value.items(): + _traverse(nested, f"{path}/{key}") + elif isinstance(value, list) and value: + _traverse(value[0], f"{path}/[]") + + _traverse(data, "") + return paths + + +@dataclass +class _StreamSchemaReport: + """Schema problems found in one stream's records.""" + + records_checked: int = 0 + invalid_records: int = 0 + # Distinct errors keyed by the schema rule that failed, first occurrence wins. + errors: dict[tuple[Any, ...], str] = field(default_factory=dict) + unmatched_records: int = 0 + first_unmatched: str | None = None + + +class _StreamSchemaChecker: + def __init__(self, stream_name: str, json_schema: Mapping[str, Any]) -> None: + self._stream_name = stream_name + self._validator = StrictIntegerDraft7Validator( + json_schema, format_checker=RecordFormatChecker() + ) + self._expected_paths = get_expected_schema_paths(json_schema) + self.report = _StreamSchemaReport() + + def check(self, data: Mapping[str, Any]) -> None: + report = self.report + record_number = report.records_checked + report.records_checked += 1 + + errors: list[ValidationError] = list(self._validator.iter_errors(data)) + if errors: + report.invalid_records += 1 + for error in errors: + schema_rule = tuple(error.schema_path) + if schema_rule not in report.errors: + report.errors[schema_rule] = ( + f"record #{record_number} at `{format_record_path(error.absolute_path)}`: " + f"{_truncate(error.message)} " + f"(schema rule: `{'/'.join(str(part) for part in schema_rule)}`)" + ) + + # A schema with `additionalProperties` and no `required` accepts any object, so also + # require each record to share at least one field with the schema. + if self._expected_paths and not (get_record_paths(data) & self._expected_paths): + report.unmatched_records += 1 + if report.first_unmatched is None: + record_fields = sorted(data)[:10] if isinstance(data, Mapping) else [] + schema_fields = sorted(self._expected_paths)[:10] + report.first_unmatched = ( + f"record #{record_number} has top-level fields {record_fields}, none of which " + f"match the schema's fields (first 10: {schema_fields})" + ) + + def describe_failures(self) -> str | None: + report = self.report + lines: list[str] = [] + if report.invalid_records: + lines.append( + f"- Stream '{self._stream_name}': {report.invalid_records} of " + f"{report.records_checked} records do not match the stream's JSON schema." + ) + shown = list(report.errors.values())[:MAX_SCHEMA_ERRORS_PER_STREAM] + lines.extend(f" - {error}" for error in shown) + hidden = len(report.errors) - len(shown) + if hidden: + lines.append(f" - ...and {hidden} more distinct schema error(s).") + if report.unmatched_records: + lines.append( + f"- Stream '{self._stream_name}': {report.unmatched_records} of " + f"{report.records_checked} records share no field with the stream's JSON schema; " + f"{report.first_unmatched}." + ) + return "\n".join(lines) if lines else None + + +def assert_read_records( + *, + records: Iterable[AirbyteMessage], + configured_catalog: ConfiguredAirbyteCatalog, + require_records_per_stream: bool, + validate_schema: bool, +) -> None: + """Assert the records a `read` returned against the configured catalog. + + Args: + records: The `RECORD` messages from the read. + 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. + + Raises: + AssertionError: Listing every failure, by stream, with the path of the offending value + inside the record. + """ + record_counts: Counter[str] = Counter() + checkers: dict[str, _StreamSchemaChecker] = {} + if validate_schema: + checkers = { + configured_stream.stream.name: _StreamSchemaChecker( + configured_stream.stream.name, configured_stream.stream.json_schema + ) + for configured_stream in configured_catalog.streams + } + + for message in records: + record = message.record + if record is None: + continue + record_counts[record.stream] += 1 + checker = checkers.get(record.stream) + if checker is not None: + checker.check(record.data) + + failures: list[str] = [] + if require_records_per_stream: + streams_without_records = [ + configured_stream.stream.name + for configured_stream in configured_catalog.streams + if not record_counts[configured_stream.stream.name] + ] + if streams_without_records: + failures.append( + "Every configured stream must return at least one record, but these returned " + f"none: {', '.join(repr(name) for name in sorted(streams_without_records))}. " + "If a stream is expected to be empty for this config, list it under " + "`empty_streams` (with a `bypass_reason`) in the config's `basic_read` entry of " + "`acceptance-test-config.yml`." + ) + + schema_failures = [ + description + for checker in checkers.values() + if (description := checker.describe_failures()) is not None + ] + if schema_failures: + failures.append( + "Records do not match their stream's JSON schema. Fix the schema or the records; to " + "skip this check for a config, set `validate_schema: false` (with a reason) in its " + "`basic_read` entry of `acceptance-test-config.yml`.\n" + "\n".join(schema_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 171b11b25a..8c81e5e80e 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -116,6 +116,8 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe The sections a config is listed under are unioned as well, so a config that appears under `discovery` and `basic_read` keeps running every test, while one listed only under the check-only sections runs `check` alone (see `ConnectorTestScenario.check_only`). + + A `validate_schema: false` opt-out declared by any entry for the config is kept. """ deduped_scenarios: list[ConnectorTestScenario] = [] @@ -141,6 +143,8 @@ 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, "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 75a53efea3..25607c601b 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -19,6 +19,7 @@ ConnectorTestScenario, ) from airbyte_cdk.test.standard_tests._job_runner import run_test_job +from airbyte_cdk.test.standard_tests._read_assertions import assert_read_records from airbyte_cdk.test.standard_tests.connector_base import ( ConnectorTestSuiteBase, ) @@ -118,7 +119,11 @@ def test_basic_read( This test is designed to validate the connector's ability to read data from the source and return records. It first runs a `discover` job to obtain the catalog of streams, and then it runs a `read` job to fetch - records from those streams. + records from those streams, except the ones declared in `empty_streams`. + + Unless the scenario expects a failure, every record must match its stream's JSON schema + (opt out with `validate_schema: false`). A scenario expected to succeed must return + records, and for a config listed under `basic_read` every stream must return at least one. """ skip_if_check_only(scenario) discover_result = run_test_job( @@ -155,6 +160,18 @@ def test_basic_read( catalog=configured_catalog, ) + if scenario.expected_outcome.expect_exception(): + # The read failed as expected (asserted by `run_test_job`); there are no records to check. + return + + assert_read_records( + records=result.records_iterator, + configured_catalog=configured_catalog, + require_records_per_stream=( + scenario.expected_outcome.expect_success() and scenario.is_basic_read_config + ), + validate_schema=scenario.validate_schema, + ) if scenario.expected_outcome.expect_success() and not result.records: raise AssertionError("Expected records but got none.") diff --git a/cdk-migrations.md b/cdk-migrations.md index 3e1cae19a4..630b7abcd8 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -1,5 +1,23 @@ # CDK Migration Guide +## Upgrading to 7.36.0 + +Version 7.36.0 of the CDK adds record assertions to the in-process `read` Standard Test (`SourceTestSuiteBase.test_basic_read`, run by `airbyte-cdk connector test`). Previously the test only failed when the whole read returned no records, so one stream with records passed the test for every stream. It now restores two checks of the Connector Acceptance Tests (CAT) basic read test: + +- **Every stream returns records.** For a config listed under `basic_read` in `acceptance-test-config.yml` and expected to succeed, every stream that is not declared in that config's `empty_streams` must return at least one record. A config listed only under `spec` or `connection` keeps the old "some records" check, since it has nowhere to declare its empty streams. +- **Records match their schema.** Unless the config's `status` is `failed` or `exception`, every record is validated against its stream's JSON schema with the validator CAT used: Draft 7, integers must be integers (`1.0` and `true` are not), `date-time` values must look like a date and a time and parse as a datetime (a space separator and a missing offset are accepted), other `format`s use the `jsonschema` defaults, and extra columns are allowed. A record that shares no field with a schema that declares `properties` also fails. + +The failure message names each stream, the number of failing records, and for each distinct schema error the record number and the path of the offending value inside the record (for example ``record #3 at `$.items[0].price`: 'free' is not of type 'number'``). + +This change is breaking for the connector test suites of connectors whose test accounts have empty streams that are not declared, or whose records do not match their declared schemas. The connector monorepo installs the `airbyte-cdk` CLI unpinned, so connector CI picks the new assertions up with this release, without a per-connector CDK bump. + +Migration steps: + +- A stream with no records for a `basic_read` config: add data to the test account, or list the stream under `empty_streams` for that config, with a `bypass_reason`. Declared empty streams are not read. +- A schema error: fix the stream's schema (or the records) so they agree. To skip schema validation for one config while the fix is pending, set `validate_schema: false` in its `basic_read` entry, which CAT honoured too. + +Rationale: The Standard Tests replaced CAT as the only `read` coverage for most connectors, and CAT's per-stream and schema checks were not carried over. A stream that silently stopped returning records, or a schema that drifted from the API, kept passing as long as some other stream returned a record. + ## Upgrading to 7.35.0 [Version 7.34.0](https://github.com/airbytehq/airbyte-python-cdk/releases/tag/v7.34.0) of the CDK changes what the Standard Tests (`airbyte-cdk connector test` and `airbyte-cdk image test`, including the in-process `test_check` of `SourceTestSuiteBase` / `ConnectorTestSuiteBase` and the Docker-based `test_docker_image_build_and_check`) assert about the outcome of `check`. Previously only an exception or a non-zero exit failed the test, so a `check` that reported `status: FAILED` passed. Both test paths now assert the reported `CONNECTION_STATUS` against the `status` declared for the config in `acceptance-test-config.yml`, with the same semantics as the Connector Acceptance Tests (CAT): diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py new file mode 100644 index 0000000000..5e8f679de9 --- /dev/null +++ b/unit_tests/test/test_read_assertions.py @@ -0,0 +1,548 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Unit tests for the record assertions of the `read` Standard Tests. + +The assertions are exercised directly and through `SourceTestSuiteBase.test_basic_read` with a +fake source: + +- every configured stream of a `basic_read` config returns at least one record, except the + streams declared in `empty_streams` (which are not read); +- every record matches its stream's JSON schema, unless the config sets `validate_schema: false`. +""" + +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, + AirbyteMessage, + AirbyteRecordMessage, + AirbyteStateMessage, + AirbyteStream, + ConfiguredAirbyteCatalog, + ConfiguredAirbyteStream, + ConnectorSpecification, + DestinationSyncMode, + Status, + SyncMode, + 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 ( + MAX_SCHEMA_ERRORS_PER_STREAM, + assert_read_records, + format_record_path, + get_expected_schema_paths, +) +from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite +from airbyte_cdk.utils.connector_paths import ACCEPTANCE_TEST_CONFIG + +MSG_MISSING_STREAMS = "Every configured stream must return at least one record" +MSG_SCHEMA = "Records do not match their stream's JSON schema" +MSG_NO_RECORDS = "Expected records but got none" + +USERS_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "name": {"type": ["null", "string"]}, + "updated_at": {"type": ["null", "string"], "format": "date-time"}, + }, +} +EVENTS_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": {"id": {"type": "string"}}, +} + + +def _catalog(**schemas: Mapping[str, Any]) -> ConfiguredAirbyteCatalog: + return ConfiguredAirbyteCatalog( + streams=[ + ConfiguredAirbyteStream( + stream=AirbyteStream( + name=name, + json_schema=dict(schema), + supported_sync_modes=[SyncMode.full_refresh], + ), + sync_mode=SyncMode.full_refresh, + destination_sync_mode=DestinationSyncMode.append, + ) + for name, schema in schemas.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 _assert( + records: list[AirbyteMessage], + catalog: ConfiguredAirbyteCatalog, + *, + require_records_per_stream: bool = True, + validate_schema: bool = True, +) -> None: + assert_read_records( + records=records, + configured_catalog=catalog, + require_records_per_stream=require_records_per_stream, + validate_schema=validate_schema, + ) + + +def _schema_error( + schema: Mapping[str, Any], *records: Mapping[str, Any], validate_schema: bool = True +) -> str | None: + """Validate `records` of a single stream; return the failure message, or None if they pass.""" + try: + _assert( + [_record("s", data) for data in records], + _catalog(s=schema), + validate_schema=validate_schema, + ) + except AssertionError as error: + return str(error) + return None + + +# --- Per-stream record presence ------------------------------------------------------------ + + +def test_every_stream_with_records_passes() -> None: + _assert( + [_record("users", {"id": 1}), _record("events", {"id": "a"})], + _catalog(users=USERS_SCHEMA, events=EVENTS_SCHEMA), + ) + + +def test_stream_without_records_fails_and_is_named() -> None: + with pytest.raises(AssertionError, match=MSG_MISSING_STREAMS) as error: + _assert([_record("users", {"id": 1})], _catalog(users=USERS_SCHEMA, events=EVENTS_SCHEMA)) + assert "'events'" in str(error.value) + assert "'users'" not in str(error.value) + assert "empty_streams" in str(error.value) + + +def test_streams_without_records_are_allowed_when_not_required() -> None: + _assert( + [_record("users", {"id": 1})], + _catalog(users=USERS_SCHEMA, events=EVENTS_SCHEMA), + require_records_per_stream=False, + ) + + +def test_records_of_unconfigured_streams_do_not_count() -> None: + with pytest.raises(AssertionError, match=MSG_MISSING_STREAMS): + _assert( + [_record("users", {"id": 1}), _record("other", {"anything": True})], + _catalog(users=USERS_SCHEMA, events=EVENTS_SCHEMA), + ) + + +# --- Schema validation ----------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "schema, data", + [ + pytest.param(USERS_SCHEMA, {"id": 1, "name": None}, id="nullable_null"), + pytest.param(USERS_SCHEMA, {"id": 1, "extra": {"x": 1}}, id="extra_column_is_allowed"), + pytest.param(USERS_SCHEMA, {"id": 1, "updated_at": "2024-01-02T03:04:05Z"}, id="rfc3339"), + pytest.param( + USERS_SCHEMA, {"id": 1, "updated_at": "2024-01-02 03:04:05"}, id="space_no_offset" + ), + pytest.param( + USERS_SCHEMA, + {"id": 1, "updated_at": "2024-01-02T03:04:05.123456+05:30"}, + id="fraction_offset", + ), + pytest.param(USERS_SCHEMA, {"id": 1, "updated_at": None}, id="null_date_time"), + pytest.param( + {"type": "object", "properties": {"d": {"type": "string", "format": "date"}}}, + {"d": "2024-01-02"}, + id="date", + ), + pytest.param({"type": "object"}, {"anything": [1, 2]}, id="schema_without_properties"), + pytest.param({}, {"anything": 1}, id="empty_schema"), + ], +) +def test_valid_records_pass(schema: Mapping[str, Any], data: Mapping[str, Any]) -> None: + assert _schema_error(schema, data) is None + + +@pytest.mark.parametrize( + "schema, data, expected_path, expected_message", + [ + pytest.param(USERS_SCHEMA, {"id": "1"}, "$.id", "'1' is not of type 'integer'", id="type"), + pytest.param( + USERS_SCHEMA, {"id": 1.0}, "$.id", "1.0 is not of type 'integer'", id="float_integer" + ), + pytest.param( + USERS_SCHEMA, {"id": True}, "$.id", "True is not of type 'integer'", id="bool_integer" + ), + pytest.param( + USERS_SCHEMA, + {"id": 1, "updated_at": "2024-01-02"}, + "$.updated_at", + "is not a valid 'date-time'", + id="date_time_without_time", + ), + pytest.param( + USERS_SCHEMA, + {"id": 1, "updated_at": "2024-13-45T00:00:00Z"}, + "$.updated_at", + "is not a valid 'date-time'", + id="date_time_unparseable", + ), + pytest.param( + {"type": "object", "properties": {"d": {"type": "string", "format": "date"}}}, + {"d": "2024-01-02T00:00:00Z"}, + "$.d", + "is not a 'date'", + id="date_with_time", + ), + pytest.param( + { + "type": "object", + "properties": { + "items": { + "type": "array", + "items": {"type": "object", "properties": {"price": {"type": "number"}}}, + } + }, + }, + {"items": [{"price": 1}, {"price": "free"}]}, + "$.items[1].price", + "'free' is not of type 'number'", + id="nested_array", + ), + pytest.param( + {"type": "object", "properties": {"first name": {"type": "string"}}}, + {"first name": 1}, + "$['first name']", + "1 is not of type 'string'", + id="non_identifier_key", + ), + ], +) +def test_invalid_records_fail_with_stream_and_record_path( + schema: Mapping[str, Any], + data: Mapping[str, Any], + expected_path: str, + expected_message: str, +) -> None: + message = _schema_error(schema, {**data}, {**data}) + assert message is not None + assert MSG_SCHEMA in message + assert "Stream 's': 2 of 2 records do not match" in message + assert f"record #0 at `{expected_path}`" in message + assert expected_message in message + assert "validate_schema: false" in message + + +def test_validate_schema_false_skips_schema_checks() -> None: + assert _schema_error(USERS_SCHEMA, {"id": "not an integer"}, validate_schema=False) is None + + +def test_record_sharing_no_field_with_schema_fails() -> None: + # The schema accepts any object (no `required`, additional properties allowed), so only the + # structure check catches a record that has none of the declared fields. + message = _schema_error(USERS_SCHEMA, {"id": 1}, {"unexpected": 1}) + assert message is not None + assert "Stream 's': 1 of 2 records share no field with the stream's JSON schema" in message + assert "record #1 has top-level fields ['unexpected']" in message + + +def test_distinct_schema_errors_are_capped_per_stream() -> None: + properties = {f"f{index}": {"type": "integer"} for index in range(8)} + message = _schema_error( + {"type": "object", "properties": properties}, {name: "x" for name in properties} + ) + assert message is not None + assert message.count("record #0 at") == MAX_SCHEMA_ERRORS_PER_STREAM + assert f"...and {8 - MAX_SCHEMA_ERRORS_PER_STREAM} more distinct schema error(s)." in message + + +def test_each_schema_rule_is_reported_once_with_its_first_record() -> None: + message = _schema_error(USERS_SCHEMA, {"id": 1}, {"id": "a"}, {"id": "b"}) + assert message is not None + assert "2 of 3 records do not match" in message + assert "record #1 at `$.id`" in message + assert "record #2" not in message + + +def test_presence_and_schema_failures_are_reported_together() -> None: + with pytest.raises(AssertionError) as error: + _assert([_record("users", {"id": "1"})], _catalog(users=USERS_SCHEMA, events=EVENTS_SCHEMA)) + assert MSG_MISSING_STREAMS in str(error.value) + assert MSG_SCHEMA in str(error.value) + + +def test_recursive_schema_is_supported() -> None: + schema = { + "type": "object", + "definitions": { + "node": { + "type": "object", + "properties": { + "value": {"type": "integer"}, + "children": {"type": "array", "items": {"$ref": "#/definitions/node"}}, + }, + } + }, + "properties": {"root": {"$ref": "#/definitions/node"}}, + } + assert _schema_error(schema, {"root": {"value": 1, "children": [{"value": 2}]}}) is None + message = _schema_error(schema, {"root": {"value": 1, "children": [{"value": "2"}]}}) + assert message is not None + assert "`$.root.children[0].value`" in message + + +# --- Helpers --------------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "path, expected", + [ + pytest.param([], "$", id="root"), + pytest.param(["a", "b"], "$.a.b", id="keys"), + pytest.param(["a", 0, "b"], "$.a[0].b", id="index"), + pytest.param(["first name", "x-y"], "$['first name']['x-y']", id="quoted"), + ], +) +def test_format_record_path(path: list[Any], expected: str) -> None: + assert format_record_path(path) == expected + + +def test_get_expected_schema_paths() -> None: + schema = { + "$ref": "#/definitions/root", + "definitions": { + "root": { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "meta": {"type": ["null", "object"]}, + "tags": {"type": "array", "items": {"type": "string"}}, + "owner": { + "anyOf": [ + {"type": "null"}, + {"properties": {"login": {"type": "string"}}}, + ] + }, + }, + } + }, + } + assert get_expected_schema_paths(schema) == { + "/id", + "/meta", + "/tags/[]", + "/owner", + "/owner/login", + } + + +# --- Through `SourceTestSuiteBase.test_basic_read` -------------------------------------------- + + +class _FakeReadSource(Source): + """A source that discovers the given streams and reads the given records for each one.""" + + def __init__( + self, + schemas: Mapping[str, Mapping[str, Any]], + records: Mapping[str, list[Mapping[str, Any]]], + ) -> None: + self._schemas = schemas + self._records = records + + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + return ConnectorSpecification(connectionSpecification={"type": "object", "properties": {}}) + + 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=name, + json_schema=dict(schema), + supported_sync_modes=[SyncMode.full_refresh], + ) + for name, schema in self._schemas.items() + ] + ) + + def read( + self, + logger: logging.Logger, + config: Mapping[str, Any], + catalog: ConfiguredAirbyteCatalog, + state: list[AirbyteStateMessage] | None = None, + ) -> Iterable[AirbyteMessage]: + for configured_stream in catalog.streams: + for data in self._records.get(configured_stream.stream.name, []): + yield _record(configured_stream.stream.name, data) + + +def _run_basic_read( + tmp_path: Path, + scenario: ConnectorTestScenario, + records: Mapping[str, list[Mapping[str, Any]]], +) -> None: + source = _FakeReadSource({"users": USERS_SCHEMA, "events": EVENTS_SCHEMA}, records) + + class _Suite(SourceTestSuiteBase): + @classmethod + def get_connector_root_dir(cls) -> Path: + return tmp_path + + @classmethod + def create_connector(cls, scenario: ConnectorTestScenario | None) -> IConnector: + return source + + _Suite().test_basic_read(scenario) + + +def _scenario( + *sections: str, status: str | None = "succeed", **fields: Any +) -> ConnectorTestScenario: + return ConnectorTestScenario.model_validate( + {"config_dict": {"api_key": "x"}, "status": status, "sections": sections, **fields} + ) + + +VALID_USERS = [{"id": 1, "name": "a"}] +VALID_EVENTS = [{"id": "e1"}] + +# (scenario, records by stream, expected failure message or None) +BASIC_READ_MATRIX = [ + pytest.param( + _scenario("connection", "basic_read"), + {"users": VALID_USERS, "events": VALID_EVENTS}, + None, + id="basic_read_all_streams", + ), + pytest.param( + _scenario("connection", "basic_read"), + {"users": VALID_USERS}, + MSG_MISSING_STREAMS, + id="basic_read_empty_stream", + ), + pytest.param( + _scenario("connection", "basic_read", empty_streams=[{"name": "events"}]), + {"users": VALID_USERS, "events": [{"id": 1}]}, + None, + id="basic_read_declared_empty_stream_is_not_read", + ), + pytest.param( + _scenario(), + {"users": VALID_USERS}, + MSG_MISSING_STREAMS, + id="hand_built_scenario_is_a_basic_read_config", + ), + # A config never listed under `basic_read` has nowhere to declare its empty streams. + pytest.param( + _scenario("connection"), + {"users": VALID_USERS}, + None, + id="connection_only_empty_stream", + ), + pytest.param( + _scenario("connection"), + {}, + MSG_NO_RECORDS, + id="connection_only_no_records", + ), + pytest.param( + _scenario("connection"), + {"users": [{"id": "1"}]}, + MSG_SCHEMA, + id="connection_only_schema_mismatch", + ), + pytest.param( + _scenario("connection", "basic_read"), + {"users": [{"id": "1"}], "events": VALID_EVENTS}, + MSG_SCHEMA, + id="basic_read_schema_mismatch", + ), + pytest.param( + _scenario("connection", "basic_read", validate_schema=False), + {"users": [{"id": "1"}], "events": VALID_EVENTS}, + None, + id="basic_read_schema_mismatch_opted_out", + ), + # No declared `status`: any outcome is allowed, but emitted records must match the schema. + pytest.param( + _scenario("basic_read", status=None), + {"users": VALID_USERS}, + None, + id="no_status_empty_stream", + ), + pytest.param( + _scenario("basic_read", status=None), + {"users": [{"id": "1"}]}, + MSG_SCHEMA, + id="no_status_schema_mismatch", + ), +] + + +@pytest.mark.parametrize("scenario, records, failure_match", BASIC_READ_MATRIX) +def test_basic_read( + tmp_path: Path, + scenario: ConnectorTestScenario, + records: Mapping[str, list[Mapping[str, Any]]], + failure_match: str | None, +) -> None: + if failure_match is None: + _run_basic_read(tmp_path, scenario, records) + return + + with pytest.raises(AssertionError, match=failure_match): + _run_basic_read(tmp_path, scenario, records) + + +# --- `validate_schema` in `acceptance-test-config.yml` ---------------------------------------- + + +def test_validate_schema_opt_out_survives_scenario_dedup(tmp_path: Path) -> 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", "validate_schema": False}] + }, + } + } + ) + ) + + class _Suite(DockerConnectorTestSuite): + @classmethod + def get_connector_root_dir(cls) -> Path: + return tmp_path + + [scenario] = _Suite.get_scenarios() + assert scenario.validate_schema is False + assert scenario.is_basic_read_config + assert scenario.expected_outcome.expect_success() From 065803f22ce84c2abe589f806ba135dcceb61d55 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Thu, 8 Oct 2026 21:08:33 +0300 Subject: [PATCH 2/8] fix(standard-tests): name the JSON type in schema type errors instead of printing the value A type error's value can be a whole nested object, which made the failure line long and copied record values into public CI logs. Report "expected type 'number', got string" instead. Co-Authored-By: Claude Opus 5.5 --- .../test/standard_tests/_read_assertions.py | 34 ++++++++++++++++++- cdk-migrations.md | 2 +- unit_tests/test/test_read_assertions.py | 28 ++++++++++++--- 3 files changed, 57 insertions(+), 7 deletions(-) diff --git a/airbyte_cdk/test/standard_tests/_read_assertions.py b/airbyte_cdk/test/standard_tests/_read_assertions.py index f5f75e6e04..016c43abc8 100644 --- a/airbyte_cdk/test/standard_tests/_read_assertions.py +++ b/airbyte_cdk/test/standard_tests/_read_assertions.py @@ -86,6 +86,38 @@ def format_record_path(path: Iterable[Any]) -> str: return rendered +_JSON_TYPE_NAMES: tuple[tuple[type, str], ...] = ( + (bool, "boolean"), # before `int`: `bool` is an `int` subclass + (int, "integer"), + (float, "number"), + (str, "string"), + (list, "array"), + (Mapping, "object"), +) + + +def _json_type_name(value: object) -> str: + if value is None: + return "null" + for python_type, name in _JSON_TYPE_NAMES: + if isinstance(value, python_type): + return name + return type(value).__name__ + + +def _describe_error(error: ValidationError) -> str: + if error.validator == "type": + # The value can be a whole nested object, and record values end up in public CI logs: + # name its JSON type instead of printing it. + expected = error.validator_value + expected_types = [expected] if isinstance(expected, str) else list(expected) + return ( + f"expected type {' or '.join(repr(name) for name in expected_types)}, " + f"got {_json_type_name(error.instance)}" + ) + return _truncate(error.message) + + def _truncate(text: str) -> str: if len(text) <= _MAX_MESSAGE_CHARS: return text @@ -208,7 +240,7 @@ def check(self, data: Mapping[str, Any]) -> None: if schema_rule not in report.errors: report.errors[schema_rule] = ( f"record #{record_number} at `{format_record_path(error.absolute_path)}`: " - f"{_truncate(error.message)} " + f"{_describe_error(error)} " f"(schema rule: `{'/'.join(str(part) for part in schema_rule)}`)" ) diff --git a/cdk-migrations.md b/cdk-migrations.md index 630b7abcd8..fd1d10c599 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -7,7 +7,7 @@ Version 7.36.0 of the CDK adds record assertions to the in-process `read` Standa - **Every stream returns records.** For a config listed under `basic_read` in `acceptance-test-config.yml` and expected to succeed, every stream that is not declared in that config's `empty_streams` must return at least one record. A config listed only under `spec` or `connection` keeps the old "some records" check, since it has nowhere to declare its empty streams. - **Records match their schema.** Unless the config's `status` is `failed` or `exception`, every record is validated against its stream's JSON schema with the validator CAT used: Draft 7, integers must be integers (`1.0` and `true` are not), `date-time` values must look like a date and a time and parse as a datetime (a space separator and a missing offset are accepted), other `format`s use the `jsonschema` defaults, and extra columns are allowed. A record that shares no field with a schema that declares `properties` also fails. -The failure message names each stream, the number of failing records, and for each distinct schema error the record number and the path of the offending value inside the record (for example ``record #3 at `$.items[0].price`: 'free' is not of type 'number'``). +The failure message names each stream, the number of failing records, and for each distinct schema error the record number and the path of the offending value inside the record (for example ``record #3 at `$.items[0].price`: expected type 'number', got string``). A type error names the value's JSON type rather than printing the value. This change is breaking for the connector test suites of connectors whose test accounts have empty streams that are not declared, or whose records do not match their declared schemas. The connector monorepo installs the `airbyte-cdk` CLI unpinned, so connector CI picks the new assertions up with this release, without a per-connector CDK bump. diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py index 5e8f679de9..cc6b359788 100644 --- a/unit_tests/test/test_read_assertions.py +++ b/unit_tests/test/test_read_assertions.py @@ -186,12 +186,22 @@ def test_valid_records_pass(schema: Mapping[str, Any], data: Mapping[str, Any]) @pytest.mark.parametrize( "schema, data, expected_path, expected_message", [ - pytest.param(USERS_SCHEMA, {"id": "1"}, "$.id", "'1' is not of type 'integer'", id="type"), pytest.param( - USERS_SCHEMA, {"id": 1.0}, "$.id", "1.0 is not of type 'integer'", id="float_integer" + USERS_SCHEMA, {"id": "1"}, "$.id", "expected type 'integer', got string", id="type" ), pytest.param( - USERS_SCHEMA, {"id": True}, "$.id", "True is not of type 'integer'", id="bool_integer" + USERS_SCHEMA, + {"id": 1.0}, + "$.id", + "expected type 'integer', got number", + id="float_integer", + ), + pytest.param( + USERS_SCHEMA, + {"id": True}, + "$.id", + "expected type 'integer', got boolean", + id="bool_integer", ), pytest.param( USERS_SCHEMA, @@ -226,14 +236,14 @@ def test_valid_records_pass(schema: Mapping[str, Any], data: Mapping[str, Any]) }, {"items": [{"price": 1}, {"price": "free"}]}, "$.items[1].price", - "'free' is not of type 'number'", + "expected type 'number', got string", id="nested_array", ), pytest.param( {"type": "object", "properties": {"first name": {"type": "string"}}}, {"first name": 1}, "$['first name']", - "1 is not of type 'string'", + "expected type 'string', got integer", id="non_identifier_key", ), ], @@ -253,6 +263,14 @@ def test_invalid_records_fail_with_stream_and_record_path( assert "validate_schema: false" in message +def test_type_errors_name_the_type_instead_of_printing_the_value() -> None: + schema = {"type": "object", "properties": {"reason": {"type": ["null", "string"]}}} + message = _schema_error(schema, {"reason": {"text": "private value"}}) + assert message is not None + assert "record #0 at `$.reason`: expected type 'null' or 'string', got object" in message + assert "private value" not in message + + def test_validate_schema_false_skips_schema_checks() -> None: assert _schema_error(USERS_SCHEMA, {"id": "not an integer"}, validate_schema=False) is None From 34b6c917861910f84b3992ce1bcb4501c6b30e45 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Thu, 8 Oct 2026 22:02:02 +0300 Subject: [PATCH 3/8] fix(standard-tests): limit the record checks to basic_read configs A config listed only under `connection` (for example when `basic_read` is bypassed) was schema validated, but had no `basic_read` entry to put the `validate_schema: false` opt-out in, and CAT never ran a basic read with it. Both record checks now apply only to `basic_read` configs, as in CAT; the others keep the "some records" floor. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/models/scenario.py | 9 ++++---- .../test/standard_tests/source_base.py | 23 ++++++++++--------- cdk-migrations.md | 4 ++-- unit_tests/test/test_read_assertions.py | 5 ++-- 4 files changed, 22 insertions(+), 19 deletions(-) diff --git a/airbyte_cdk/test/models/scenario.py b/airbyte_cdk/test/models/scenario.py index 832c05a7a8..7b7ce2f872 100644 --- a/airbyte_cdk/test/models/scenario.py +++ b/airbyte_cdk/test/models/scenario.py @@ -141,10 +141,11 @@ def check_only(self) -> bool: def is_basic_read_config(self) -> bool: """Whether this config is listed under `basic_read` (or the scenario was built by hand). - Only these configs must return records from every stream not declared in - `empty_streams`, as in CAT. A config listed only under `spec` or `connection` never had a - basic read test, and has nowhere to declare its empty streams, so its `read` only has to - return some records. + 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 listed only under `spec` or + `connection` never had a basic read test, and has nowhere to declare those exceptions, + so its `read` only has to return some records. """ return not self.sections or "basic_read" in self.sections diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index 25607c601b..14c008b44e 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -121,9 +121,11 @@ def test_basic_read( obtain the catalog of streams, and then it runs a `read` job to fetch records from those streams, except the ones declared in `empty_streams`. - Unless the scenario expects a failure, every record must match its stream's JSON schema - (opt out with `validate_schema: false`). A scenario expected to succeed must return - records, and for a config listed under `basic_read` every stream must return at least one. + A scenario 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: unless the + scenario expects a failure, every record must match its stream's JSON schema (opt out + with `validate_schema: false`), and if it expects success, every stream must return at + least one record. """ skip_if_check_only(scenario) discover_result = run_test_job( @@ -164,14 +166,13 @@ def test_basic_read( # The read failed as expected (asserted by `run_test_job`); there are no records to check. return - assert_read_records( - records=result.records_iterator, - configured_catalog=configured_catalog, - require_records_per_stream=( - scenario.expected_outcome.expect_success() and scenario.is_basic_read_config - ), - validate_schema=scenario.validate_schema, - ) + if scenario.is_basic_read_config: + assert_read_records( + records=result.records_iterator, + configured_catalog=configured_catalog, + require_records_per_stream=scenario.expected_outcome.expect_success(), + validate_schema=scenario.validate_schema, + ) if scenario.expected_outcome.expect_success() and not result.records: raise AssertionError("Expected records but got none.") diff --git a/cdk-migrations.md b/cdk-migrations.md index fd1d10c599..7401631d41 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -2,9 +2,9 @@ ## Upgrading to 7.36.0 -Version 7.36.0 of the CDK adds record assertions to the in-process `read` Standard Test (`SourceTestSuiteBase.test_basic_read`, run by `airbyte-cdk connector test`). Previously the test only failed when the whole read returned no records, so one stream with records passed the test for every stream. It now restores two checks of the Connector Acceptance Tests (CAT) basic read test: +Version 7.36.0 of the CDK adds record assertions to the in-process `read` Standard Test (`SourceTestSuiteBase.test_basic_read`, run by `airbyte-cdk connector test`). Previously the test only failed when the whole read returned no records, so one stream with records passed the test for every stream. For every config listed under `basic_read` in `acceptance-test-config.yml`, it now restores two checks of the Connector Acceptance Tests (CAT) basic read test. A config listed only under `spec` or `connection` keeps the old "some records" check, as CAT never ran a basic read with it and it has nowhere to declare the exceptions below. -- **Every stream returns records.** For a config listed under `basic_read` in `acceptance-test-config.yml` and expected to succeed, every stream that is not declared in that config's `empty_streams` must return at least one record. A config listed only under `spec` or `connection` keeps the old "some records" check, since it has nowhere to declare its empty streams. +- **Every stream returns records.** When the config is expected to succeed (`status: succeed`, usually declared on its `connection` entry), every stream that is not declared in its `empty_streams` must return at least one record. - **Records match their schema.** Unless the config's `status` is `failed` or `exception`, every record is validated against its stream's JSON schema with the validator CAT used: Draft 7, integers must be integers (`1.0` and `true` are not), `date-time` values must look like a date and a time and parse as a datetime (a space separator and a missing offset are accepted), other `format`s use the `jsonschema` defaults, and extra columns are allowed. A record that shares no field with a schema that declares `properties` also fails. The failure message names each stream, the number of failing records, and for each distinct schema error the record number and the path of the offending value inside the record (for example ``record #3 at `$.items[0].price`: expected type 'number', got string``). A type error names the value's JSON type rather than printing the value. diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py index cc6b359788..bc1ca094b6 100644 --- a/unit_tests/test/test_read_assertions.py +++ b/unit_tests/test/test_read_assertions.py @@ -474,7 +474,8 @@ def _scenario( MSG_MISSING_STREAMS, id="hand_built_scenario_is_a_basic_read_config", ), - # A config never listed under `basic_read` has nowhere to declare its empty streams. + # A config never listed under `basic_read` has nowhere to declare its empty streams or its + # `validate_schema` opt-out, so only the "some records" floor applies. pytest.param( _scenario("connection"), {"users": VALID_USERS}, @@ -490,7 +491,7 @@ def _scenario( pytest.param( _scenario("connection"), {"users": [{"id": "1"}]}, - MSG_SCHEMA, + None, id="connection_only_schema_mismatch", ), pytest.param( From 5a2c471da55f25001f3d4085fc0a0872441c9433 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 14:49:10 +0300 Subject: [PATCH 4/8] fix(standard-tests): never print record values in schema failure messages Only top-level `type` errors were rewritten. Every other jsonschema message embeds the failing value: an `anyOf`/`oneOf` error printed the whole nested object, and `not`, `enum`, `const`, `pattern` and `format` printed the value too. These messages end up in public CI logs. Each schema error is now described by its validator keyword, the schema's own scalar constraint (a format name, a pattern, a length) and the JSON type of the value, next to the record number, the record path and the schema rule. For `anyOf`/`oneOf` the closest option's failure is named the same way, preferring options whose type the value has. A schema the validator cannot apply (an unknown `type`, an unresolvable `$ref`, an invalid `pattern`) used to abort the whole check with the raw jsonschema exception, whose text prints the instance. It now fails that stream with the schema problem and stops validating its records, while the other streams are still reported. Co-Authored-By: Claude Opus 5.5 --- .../test/standard_tests/_read_assertions.py | 95 +++++++++--- cdk-migrations.md | 2 +- unit_tests/test/test_read_assertions.py | 136 +++++++++++++++++- 3 files changed, 212 insertions(+), 21 deletions(-) diff --git a/airbyte_cdk/test/standard_tests/_read_assertions.py b/airbyte_cdk/test/standard_tests/_read_assertions.py index 016c43abc8..5b5f75c988 100644 --- a/airbyte_cdk/test/standard_tests/_read_assertions.py +++ b/airbyte_cdk/test/standard_tests/_read_assertions.py @@ -22,7 +22,8 @@ from typing import Any from jsonschema import Draft7Validator, FormatChecker, validators -from jsonschema.exceptions import FormatError, ValidationError +from jsonschema.exceptions import FormatError, UnknownType, ValidationError, best_match +from referencing.exceptions import Unresolvable from airbyte_cdk.models import AirbyteMessage, ConfiguredAirbyteCatalog from airbyte_cdk.utils.datetime_helpers import ab_datetime_try_parse @@ -31,6 +32,7 @@ """How many distinct schema errors are reported per stream. The total is always reported.""" _MAX_MESSAGE_CHARS = 300 +_MAX_CONSTRAINT_CHARS = 80 # CAT's shape check for `date-time` values: a date, a ' ' or 'T' separator, a time, then anything # (fraction, offset, zone name). The value must also parse as a datetime. @@ -63,7 +65,7 @@ class RecordFormatChecker(FormatChecker): def check(self, instance: object, format: str) -> None: if format == "date-time": if isinstance(instance, str) and not _is_valid_datetime(instance): - raise FormatError(f"{instance!r} is not a valid 'date-time'") + raise FormatError("is not a valid 'date-time'") return super().check(instance, format) @@ -106,22 +108,52 @@ def _json_type_name(value: object) -> str: def _describe_error(error: ValidationError) -> str: - if error.validator == "type": - # The value can be a whole nested object, and record values end up in public CI logs: - # name its JSON type instead of printing it. - expected = error.validator_value - expected_types = [expected] if isinstance(expected, str) else list(expected) - return ( - f"expected type {' or '.join(repr(name) for name in expected_types)}, " - f"got {_json_type_name(error.instance)}" - ) - return _truncate(error.message) + """Describe a schema error without printing any part of the record. + jsonschema's own messages embed the failing value (for an `anyOf` the whole nested object), + and these messages end up in public CI logs. So only the validator keyword, the schema's own + constraint and the JSON type of the value are reported. + """ + keyword = error.validator + got = _json_type_name(error.instance) + constraint = error.validator_value + + if keyword == "type": + expected_types = [constraint] if isinstance(constraint, str) else list(constraint) + return f"expected type {' or '.join(repr(name) for name in expected_types)}, got {got}" + + if keyword == "required" and isinstance(error.instance, Mapping): + # The names come from the schema, not from the record. + missing = [name for name in constraint if name not in error.instance] + return f"missing required field(s) {', '.join(repr(name) for name in missing)}" + + description = f"fails `{keyword}`" + if isinstance(constraint, (str, int, float, bool)): + # A scalar constraint (`format`, `pattern`, `maxLength`, `const`...) is schema content. + description += f" {_truncate(repr(constraint), _MAX_CONSTRAINT_CHARS)}" + description += f", got {got}" + + if error.context: + # `anyOf`/`oneOf`: name the closest option's failure, described the same way. An option + # whose type the value does not even have says the least, so it is the last resort. + candidates = [ + option_error + for option_error in error.context + if not (option_error.validator == "type" and not option_error.relative_path) + ] + closest = best_match(candidates or error.context) + if closest is not None: + description += ( + f"; closest option fails at `{format_record_path(closest.absolute_path)}`: " + f"{_describe_error(closest)}" + ) + return description -def _truncate(text: str) -> str: - if len(text) <= _MAX_MESSAGE_CHARS: + +def _truncate(text: str, limit: int = _MAX_MESSAGE_CHARS) -> str: + if len(text) <= limit: return text - return text[: _MAX_MESSAGE_CHARS - 3] + "..." + return text[: limit - 3] + "..." def _resolve_local_ref(root: Mapping[str, Any], ref: str) -> Any: @@ -206,6 +238,22 @@ def _traverse(value: Any, path: str) -> None: return paths +def _describe_schema_problem(error: Exception) -> str: + """Describe an error the validator raised on a broken schema, without the record value. + + The exceptions' own messages print the instance being checked (`While checking instance:`), + so only the schema part of the problem is reported. + """ + if isinstance(error, UnknownType): + return f"unknown type {_truncate(repr(error.type), _MAX_CONSTRAINT_CHARS)}" + if isinstance(error, Unresolvable): + return f"unresolvable `$ref` {_truncate(repr(error.ref), _MAX_CONSTRAINT_CHARS)}" + if isinstance(error, re.error): + pattern = error.pattern if isinstance(error.pattern, str) else None + return f"invalid `pattern` {_truncate(repr(pattern), _MAX_CONSTRAINT_CHARS)}: {error.msg}" + return f"the validator raised {type(error).__name__}" + + @dataclass class _StreamSchemaReport: """Schema problems found in one stream's records.""" @@ -216,6 +264,8 @@ class _StreamSchemaReport: errors: dict[tuple[Any, ...], str] = field(default_factory=dict) unmatched_records: int = 0 first_unmatched: str | None = None + # Set when the schema itself cannot be applied; the stream's records are not checked further. + invalid_schema: str | None = None class _StreamSchemaChecker: @@ -229,10 +279,16 @@ def __init__(self, stream_name: str, json_schema: Mapping[str, Any]) -> None: def check(self, data: Mapping[str, Any]) -> None: report = self.report + if report.invalid_schema is not None: + return record_number = report.records_checked report.records_checked += 1 - errors: list[ValidationError] = list(self._validator.iter_errors(data)) + try: + errors: list[ValidationError] = list(self._validator.iter_errors(data)) + except Exception as schema_problem: # noqa: BLE001 # Any error here is a schema problem. + report.invalid_schema = _describe_schema_problem(schema_problem) + return if errors: report.invalid_records += 1 for error in errors: @@ -240,7 +296,7 @@ def check(self, data: Mapping[str, Any]) -> None: if schema_rule not in report.errors: report.errors[schema_rule] = ( f"record #{record_number} at `{format_record_path(error.absolute_path)}`: " - f"{_describe_error(error)} " + f"{_truncate(_describe_error(error))} " f"(schema rule: `{'/'.join(str(part) for part in schema_rule)}`)" ) @@ -259,6 +315,11 @@ def check(self, data: Mapping[str, Any]) -> None: def describe_failures(self) -> str | None: report = self.report lines: list[str] = [] + if report.invalid_schema is not None: + lines.append( + f"- Stream '{self._stream_name}': the stream's JSON schema is invalid " + f"({report.invalid_schema}), so its records cannot be validated. Fix the schema." + ) if report.invalid_records: lines.append( f"- Stream '{self._stream_name}': {report.invalid_records} of " diff --git a/cdk-migrations.md b/cdk-migrations.md index 9d2d863ee7..fadb29d29c 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -7,7 +7,7 @@ Version 7.36.0 of the CDK adds record assertions to the in-process `read` Standa - **Every stream returns records.** Every stream that is not declared in the config's `empty_streams` must return at least one record. - **Records match their schema.** Every record is validated against its stream's JSON schema with the validator CAT used: Draft 7, integers must be integers (`1.0` and `true` are not), `date-time` values must look like a date and a time and parse as a datetime (a space separator and a missing offset are accepted), other `format`s use the `jsonschema` defaults, and extra columns are allowed. A record that shares no field with a schema that declares `properties` also fails. -The failure message names each stream, the number of failing records, and for each distinct schema error the record number and the path of the offending value inside the record (for example ``record #3 at `$.items[0].price`: expected type 'number', got string``). A type error names the value's JSON type rather than printing the value. +The failure message names each stream, the number of failing records, and for each distinct schema error the record number, the path of the offending value inside the record, the failing schema keyword and the schema rule (for example ``record #3 at `$.items[0].price`: expected type 'number', got string``). Record values are never printed, for any kind of schema error: the message names the value's JSON type, and for an `anyOf`/`oneOf` the closest option's failure. A stream schema that cannot be applied (an unknown `type`, an unresolvable `$ref`, an invalid `pattern`) fails the check for that stream with the schema problem. This change is breaking for the connector test suites of connectors whose test accounts have empty streams that are not declared, or whose records do not match their declared schemas. The connector monorepo installs the `airbyte-cdk` CLI unpinned, so connector CI picks the new assertions up with this release, without a per-connector CDK bump. diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py index e29ac8dea4..5e11bebef5 100644 --- a/unit_tests/test/test_read_assertions.py +++ b/unit_tests/test/test_read_assertions.py @@ -207,21 +207,21 @@ def test_valid_records_pass(schema: Mapping[str, Any], data: Mapping[str, Any]) USERS_SCHEMA, {"id": 1, "updated_at": "2024-01-02"}, "$.updated_at", - "is not a valid 'date-time'", + "fails `format` 'date-time', got string", id="date_time_without_time", ), pytest.param( USERS_SCHEMA, {"id": 1, "updated_at": "2024-13-45T00:00:00Z"}, "$.updated_at", - "is not a valid 'date-time'", + "fails `format` 'date-time', got string", id="date_time_unparseable", ), pytest.param( {"type": "object", "properties": {"d": {"type": "string", "format": "date"}}}, {"d": "2024-01-02T00:00:00Z"}, "$.d", - "is not a 'date'", + "fails `format` 'date', got string", id="date_with_time", ), pytest.param( @@ -271,6 +271,136 @@ def test_type_errors_name_the_type_instead_of_printing_the_value() -> None: assert "private value" not in message +SENTINEL = "sk_live_SENTINEL_9f8e7d" +"""A value that must never appear in a failure message: record values reach public CI logs.""" + +_NESTED_USER = { + "type": "object", + "properties": {"email": {"type": "string"}, "role": {"type": "string"}}, +} + + +@pytest.mark.parametrize( + "property_schema, value, expected_message", + [ + pytest.param( + {"anyOf": [{"type": "null"}, _NESTED_USER]}, + {"email": SENTINEL, "role": 1}, + "fails `anyOf`, got object; closest option fails at `$.user.role`: " + "expected type 'string', got integer", + id="anyOf_nested_object", + ), + pytest.param( + {"oneOf": [{"type": "string", "maxLength": 3}, {"type": "integer"}]}, + SENTINEL, + "fails `oneOf`, got string; closest option fails at `$.user`: " + "fails `maxLength` 3, got string", + id="oneOf_string", + ), + pytest.param( + { + "anyOf": [ + {"type": "null"}, + { + "type": "object", + "properties": { + "inner": { + "oneOf": [ + {"type": "integer"}, + {"type": "object", "properties": {"x": {"enum": [1, 2]}}}, + ] + } + }, + }, + ] + }, + {"inner": {"x": SENTINEL}}, + "fails `anyOf`, got object", + id="anyOf_of_oneOf", + ), + pytest.param({"not": {"type": "object"}}, {"secret": SENTINEL}, "fails `not`", id="not"), + pytest.param( + {"allOf": [{"type": "object"}, {"properties": {"k": {"type": "integer"}}}]}, + {"k": SENTINEL}, + "expected type 'integer', got string", + id="allOf", + ), + pytest.param({"enum": ["a", "b"]}, SENTINEL, "fails `enum`, got string", id="enum"), + pytest.param({"const": "a"}, SENTINEL, "fails `const` 'a', got string", id="const"), + pytest.param( + {"type": "string", "pattern": "^Bearer "}, + SENTINEL, + "fails `pattern` '^Bearer ', got string", + id="pattern", + ), + pytest.param( + {"type": "string", "format": "date-time"}, + SENTINEL, + "fails `format` 'date-time', got string", + id="format_date_time", + ), + pytest.param( + {"type": "string", "format": "uuid"}, + SENTINEL, + "fails `format` 'uuid', got string", + id="format_uuid", + ), + pytest.param( + {"type": "array", "uniqueItems": True}, + [SENTINEL, SENTINEL], + "fails `uniqueItems` True, got array", + id="uniqueItems", + ), + pytest.param( + {"type": "object", "required": ["id"], "additionalProperties": False}, + {SENTINEL.lower(): 1, "x": SENTINEL}, + "missing required field(s) 'id'", + id="required", + ), + ], +) +def test_schema_errors_never_print_record_values( + property_schema: Mapping[str, Any], value: Any, expected_message: str +) -> None: + schema = {"type": "object", "properties": {"user": property_schema}} + message = _schema_error(schema, {"user": value}) + assert message is not None + assert "record #0 at `$.user" in message + assert expected_message in message + assert "(schema rule: `properties/user/" in message + assert SENTINEL not in message + + +@pytest.mark.parametrize( + "property_schema, expected_problem", + [ + pytest.param({"type": "int"}, "unknown type 'int'", id="unknown_type"), + pytest.param( + {"$ref": "#/definitions/missing"}, + "unresolvable `$ref` '/definitions/missing'", + id="unresolvable_local_ref", + ), + pytest.param( + {"$ref": "https://example.invalid/schema.json"}, + "unresolvable `$ref` 'https://example.invalid/schema.json'", + id="unresolvable_remote_ref", + ), + pytest.param( + {"type": "string", "pattern": "["}, "invalid `pattern` '['", id="invalid_pattern" + ), + ], +) +def test_invalid_schema_fails_with_the_stream_and_schema_problem( + property_schema: Mapping[str, Any], expected_problem: str +) -> None: + schema = {"type": "object", "properties": {"token": property_schema}} + message = _schema_error(schema, {"token": SENTINEL}, {"token": SENTINEL}) + assert message is not None + assert MSG_SCHEMA in message + assert f"Stream 's': the stream's JSON schema is invalid ({expected_problem}" in message + assert SENTINEL not in message + + def test_validate_schema_false_skips_schema_checks() -> None: assert _schema_error(USERS_SCHEMA, {"id": "not an integer"}, validate_schema=False) is None From 82b04de8da0838987def20c7e71fda6a1a821699 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 14:49:43 +0300 Subject: [PATCH 5/8] fix(standard-tests): reject validate_schema: false at test_strictness_level: high CAT's basic read test failed a config that set `validate_schema: false` when the connector declared `test_strictness_level: high`, so certified connectors could not drop schema validation by config. The standard tests honoured the opt-out at every strictness level. `get_scenarios()` now carries the connector's top-level `test_strictness_level` into each scenario, and `test_basic_read` fails a `basic_read` config that combines it with `validate_schema: false` before reading, with a message that points at fixing the schema or lowering the strictness level. A config read only because it is listed under `full_refresh` is unaffected, as its records are not validated anyway. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/models/scenario.py | 3 + .../test/standard_tests/_read_assertions.py | 19 ++++ .../test/standard_tests/docker_base.py | 9 +- .../test/standard_tests/source_base.py | 12 ++- cdk-migrations.md | 2 +- unit_tests/test/test_read_assertions.py | 87 +++++++++++++++---- 6 files changed, 112 insertions(+), 20 deletions(-) diff --git a/airbyte_cdk/test/models/scenario.py b/airbyte_cdk/test/models/scenario.py index e0e6e9ea54..5da455c0e8 100644 --- a/airbyte_cdk/test/models/scenario.py +++ b/airbyte_cdk/test/models/scenario.py @@ -104,6 +104,9 @@ 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 + # 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 # The sections of `acceptance-test-config.yml` that list this config. Populated by # `get_scenarios()`; empty for scenarios built by hand, which run every command. sections: tuple[str, ...] = () diff --git a/airbyte_cdk/test/standard_tests/_read_assertions.py b/airbyte_cdk/test/standard_tests/_read_assertions.py index 5b5f75c988..7ef3d3e924 100644 --- a/airbyte_cdk/test/standard_tests/_read_assertions.py +++ b/airbyte_cdk/test/standard_tests/_read_assertions.py @@ -339,6 +339,25 @@ def describe_failures(self) -> str | None: return "\n".join(lines) if lines else None +def assert_schema_validation_opt_out_allowed( + *, validate_schema: bool, test_strictness_level: str | None +) -> None: + """Reject `validate_schema: false` for a connector at `test_strictness_level: high`. + + CAT refused the opt-out at high strictness (`TestBasicRead.test_read`), so certified + connectors cannot drop schema validation by config. + """ + if validate_schema or test_strictness_level != "high": + return + raise AssertionError( + "`validate_schema: false` is not allowed for a connector with " + "`test_strictness_level: high` in `acceptance-test-config.yml` (CAT refused it too). Fix " + "the stream schemas (or the records) so the records validate, and remove " + "`validate_schema: false` from the config's `basic_read` entry; or lower the connector's " + "`test_strictness_level`." + ) + + def assert_read_records( *, records: Iterable[AirbyteMessage], diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index d24a845867..4e5668ff07 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -233,6 +233,7 @@ def get_scenarios( ) return [] + test_strictness_level = all_tests_config.get("test_strictness_level") test_scenarios: list[ConnectorTestScenario] = [] for category in SCENARIO_SECTIONS: if ( @@ -250,7 +251,13 @@ def get_scenarios( # We skip iam_role tests for now, as they are not supported in the test suite. continue - scenario = ConnectorTestScenario.model_validate({**test, "sections": (category,)}) + scenario = ConnectorTestScenario.model_validate( + { + **test, + "sections": (category,), + "test_strictness_level": test_strictness_level, + } + ) test_scenarios.append(scenario) diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index 493dd164f7..a2a1083a14 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -19,7 +19,10 @@ ConnectorTestScenario, ) from airbyte_cdk.test.standard_tests._job_runner import run_test_job -from airbyte_cdk.test.standard_tests._read_assertions import assert_read_records +from airbyte_cdk.test.standard_tests._read_assertions import ( + assert_read_records, + assert_schema_validation_opt_out_allowed, +) from airbyte_cdk.test.standard_tests.connector_base import ( ConnectorTestSuiteBase, ) @@ -136,9 +139,14 @@ def test_basic_read( 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 must return at least one record, and every record must match its stream's JSON schema (opt out with - `validate_schema: false`). + `validate_schema: false`, which a connector at `test_strictness_level: high` cannot). """ skip_unless_runs(scenario, "read") + if scenario.is_basic_read_config: + assert_schema_validation_opt_out_allowed( + validate_schema=scenario.validate_schema, + test_strictness_level=scenario.test_strictness_level, + ) connector_root = self.get_connector_root_dir() discover_result = run_test_job( self.create_connector(scenario), diff --git a/cdk-migrations.md b/cdk-migrations.md index fadb29d29c..ec767d2fdf 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -14,7 +14,7 @@ This change is breaking for the connector test suites of connectors whose test a Migration steps: - A stream with no records for a `basic_read` config: add data to the test account, or list the stream under `empty_streams` for that config, with a `bypass_reason`. Declared empty streams are not read. -- A schema error: fix the stream's schema (or the records) so they agree. To skip schema validation for one config while the fix is pending, set `validate_schema: false` in its `basic_read` entry, which CAT honoured too. +- A schema error: fix the stream's schema (or the records) so they agree. To skip schema validation for one config while the fix is pending, set `validate_schema: false` in its `basic_read` entry. As in CAT, this is rejected for a connector at `test_strictness_level: high`, which has to fix the schema (or lower its strictness level). Rationale: The Standard Tests replaced CAT as the only `read` coverage for most connectors, and CAT's per-stream and schema checks were not carried over. A stream that silently stopped returning records, or a schema that drifted from the API, kept passing as long as some other stream returned a record. diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py index 5e11bebef5..1dac3916e6 100644 --- a/unit_tests/test/test_read_assertions.py +++ b/unit_tests/test/test_read_assertions.py @@ -577,6 +577,7 @@ def _scenario( VALID_USERS = [{"id": 1, "name": "a"}] VALID_EVENTS = [{"id": "e1"}] +MSG_STRICTNESS = "`validate_schema: false` is not allowed for a connector with" # (scenario, records by stream, expected failure message or None) BASIC_READ_MATRIX = [ @@ -650,6 +651,39 @@ def _scenario( MSG_SCHEMA, id="no_status_schema_mismatch", ), + # CAT refused `validate_schema: false` at `test_strictness_level: high`. + pytest.param( + _scenario("connection", "basic_read", validate_schema=False, test_strictness_level="high"), + {"users": VALID_USERS, "events": VALID_EVENTS}, + MSG_STRICTNESS, + id="high_strictness_rejects_validate_schema_false", + ), + pytest.param( + _scenario("connection", "basic_read", test_strictness_level="high"), + {"users": VALID_USERS, "events": VALID_EVENTS}, + None, + id="high_strictness_validates_schema", + ), + pytest.param( + _scenario("connection", "basic_read", test_strictness_level="high"), + {"users": [{"id": "1"}], "events": VALID_EVENTS}, + MSG_SCHEMA, + id="high_strictness_schema_mismatch", + ), + pytest.param( + _scenario("connection", "basic_read", validate_schema=False, test_strictness_level="low"), + {"users": [{"id": "1"}], "events": VALID_EVENTS}, + None, + id="low_strictness_allows_validate_schema_false", + ), + pytest.param( + _scenario( + "connection", "full_refresh", validate_schema=False, test_strictness_level="high" + ), + {"users": VALID_USERS}, + None, + id="high_strictness_full_refresh_only_ignores_validate_schema", + ), ] @@ -671,28 +705,49 @@ def test_basic_read( # --- `validate_schema` in `acceptance-test-config.yml` ---------------------------------------- -def test_validate_schema_opt_out_survives_scenario_dedup(tmp_path: Path) -> 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", "validate_schema": False}] - }, - } - } - ) - ) +def _get_scenarios(tmp_path: Path, config: Mapping[str, Any]) -> list[ConnectorTestScenario]: + (tmp_path / ACCEPTANCE_TEST_CONFIG).write_text(yaml.safe_dump(dict(config))) class _Suite(DockerConnectorTestSuite): @classmethod def get_connector_root_dir(cls) -> Path: return tmp_path - [scenario] = _Suite.get_scenarios() + return _Suite.get_scenarios() + + +def test_validate_schema_opt_out_survives_scenario_dedup(tmp_path: Path) -> None: + [scenario] = _get_scenarios( + tmp_path, + { + "acceptance_tests": { + "connection": { + "tests": [{"config_path": "secrets/config.json", "status": "succeed"}] + }, + "basic_read": { + "tests": [{"config_path": "secrets/config.json", "validate_schema": False}] + }, + } + }, + ) assert scenario.validate_schema is False assert scenario.is_basic_read_config assert scenario.expected_outcome.expect_success() + assert scenario.test_strictness_level is None + + +@pytest.mark.parametrize("level", ["high", "low"]) +def test_test_strictness_level_is_read_from_the_config(tmp_path: Path, level: str) -> None: + scenarios = _get_scenarios( + tmp_path, + { + "test_strictness_level": level, + "acceptance_tests": { + "connection": { + "tests": [{"config_path": "secrets/config.json", "status": "succeed"}] + }, + "basic_read": {"tests": [{"config_path": "secrets/other.json"}]}, + }, + }, + ) + assert [scenario.test_strictness_level for scenario in scenarios] == [level, level] From 21ea84b84a8e55fff4e66ce15ee32c83fa1f8d1b Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 14:49:50 +0300 Subject: [PATCH 6/8] fix(standard-tests): read the records once and cover expected-failure reads `test_basic_read` built the full list of record messages to check that the read returned any, then re-read and re-parsed the message file for the record assertions. `assert_read_records` now owns the "Expected records but got none." floor, so a `basic_read` config's records are parsed in one pass; a config read only for `full_refresh` stops at the first record. The early return for a read expected to fail (`status: failed` or `exception`) had no test. The basic read matrix now runs a source whose read raises under both statuses, a `failed` config whose read succeeds, and a `succeed` config whose read raises. The `validate_schema` dedup is tested with two `basic_read` entries for the same config, in both orders. Co-Authored-By: Claude Opus 5.5 --- .../test/standard_tests/_read_assertions.py | 5 ++ .../test/standard_tests/source_base.py | 6 +- unit_tests/test/test_read_assertions.py | 88 +++++++++++++++++-- 3 files changed, 90 insertions(+), 9 deletions(-) diff --git a/airbyte_cdk/test/standard_tests/_read_assertions.py b/airbyte_cdk/test/standard_tests/_read_assertions.py index 7ef3d3e924..c25ccdeb61 100644 --- a/airbyte_cdk/test/standard_tests/_read_assertions.py +++ b/airbyte_cdk/test/standard_tests/_read_assertions.py @@ -367,6 +367,9 @@ def assert_read_records( ) -> None: """Assert the records a `read` returned against the configured catalog. + The read must return at least one record, as any `read` expected to succeed must. The + records are consumed once, so a lazy iterator over a large read is fine. + Args: records: The `RECORD` messages from the read. configured_catalog: The catalog the read ran with. Streams declared in `empty_streams` @@ -398,6 +401,8 @@ def assert_read_records( checker.check(record.data) failures: list[str] = [] + if not record_counts: + failures.append("Expected records but got none.") if require_records_per_stream: streams_without_records = [ configured_stream.stream.name diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index a2a1083a14..b041dd7ec4 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -189,16 +189,16 @@ def test_basic_read( # The read failed as expected (asserted by `run_test_job`); there are no records to check. return - if not result.records: - raise AssertionError("Expected records but got none.") - if scenario.is_basic_read_config: + # Also asserts that the read returned records, in the same pass over them. assert_read_records( records=result.records_iterator, configured_catalog=configured_catalog, require_records_per_stream=True, validate_schema=scenario.validate_schema, ) + elif next(result.records_iterator, None) is None: + raise AssertionError("Expected records but got none.") def test_incremental_read( self, diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py index 1dac3916e6..b977e01f20 100644 --- a/unit_tests/test/test_read_assertions.py +++ b/unit_tests/test/test_read_assertions.py @@ -89,7 +89,7 @@ def _record(stream: str, data: Mapping[str, Any]) -> AirbyteMessage: def _assert( - records: list[AirbyteMessage], + records: Iterable[AirbyteMessage], catalog: ConfiguredAirbyteCatalog, *, require_records_per_stream: bool = True, @@ -401,6 +401,19 @@ def test_invalid_schema_fails_with_the_stream_and_schema_problem( assert SENTINEL not in message +def test_read_without_records_fails() -> None: + with pytest.raises(AssertionError, match=MSG_NO_RECORDS): + _assert([], _catalog(), require_records_per_stream=False) + + +def test_records_are_consumed_in_one_pass() -> None: + def _records() -> Iterable[AirbyteMessage]: + yield _record("users", {"id": 1}) + yield _record("events", {"id": "a"}) + + _assert(_records(), _catalog(users=USERS_SCHEMA, events=EVENTS_SCHEMA)) + + def test_validate_schema_false_skips_schema_checks() -> None: assert _schema_error(USERS_SCHEMA, {"id": "not an integer"}, validate_schema=False) is None @@ -508,12 +521,15 @@ def test_get_expected_schema_paths() -> None: class _FakeReadSource(Source): - """A source that discovers the given streams and reads the given records for each one.""" + """A source that discovers the given streams and reads the given records for each one. + + With `records=None`, the read raises instead. + """ def __init__( self, schemas: Mapping[str, Mapping[str, Any]], - records: Mapping[str, list[Mapping[str, Any]]], + records: Mapping[str, list[Mapping[str, Any]]] | None, ) -> None: self._schemas = schemas self._records = records @@ -543,6 +559,8 @@ def read( catalog: ConfiguredAirbyteCatalog, state: list[AirbyteStateMessage] | None = None, ) -> Iterable[AirbyteMessage]: + if self._records is None: + raise RuntimeError("The read failed.") for configured_stream in catalog.streams: for data in self._records.get(configured_stream.stream.name, []): yield _record(configured_stream.stream.name, data) @@ -551,7 +569,7 @@ def read( def _run_basic_read( tmp_path: Path, scenario: ConnectorTestScenario, - records: Mapping[str, list[Mapping[str, Any]]], + records: Mapping[str, list[Mapping[str, Any]]] | None, ) -> None: source = _FakeReadSource({"users": USERS_SCHEMA, "events": EVENTS_SCHEMA}, records) @@ -577,6 +595,7 @@ def _scenario( VALID_USERS = [{"id": 1, "name": "a"}] VALID_EVENTS = [{"id": "e1"}] +READ_FAILS = None MSG_STRICTNESS = "`validate_schema: false` is not allowed for a connector with" # (scenario, records by stream, expected failure message or None) @@ -651,6 +670,37 @@ def _scenario( MSG_SCHEMA, id="no_status_schema_mismatch", ), + pytest.param( + _scenario("connection", "basic_read"), + {}, + MSG_NO_RECORDS, + id="basic_read_no_records", + ), + # A read expected to fail has no records to check: it only has to fail. + pytest.param( + _scenario("connection", "basic_read", status="failed"), + READ_FAILS, + None, + id="status_failed_read_fails", + ), + pytest.param( + _scenario("connection", "basic_read", status="exception"), + READ_FAILS, + None, + id="status_exception_read_fails", + ), + pytest.param( + _scenario("connection", "basic_read", status="failed"), + {"users": VALID_USERS, "events": VALID_EVENTS}, + "Expected exception but got none", + id="status_failed_read_succeeds", + ), + pytest.param( + _scenario("connection", "basic_read", status="succeed"), + READ_FAILS, + "The read failed", + id="status_succeed_read_fails", + ), # CAT refused `validate_schema: false` at `test_strictness_level: high`. pytest.param( _scenario("connection", "basic_read", validate_schema=False, test_strictness_level="high"), @@ -691,14 +741,14 @@ def _scenario( def test_basic_read( tmp_path: Path, scenario: ConnectorTestScenario, - records: Mapping[str, list[Mapping[str, Any]]], + records: Mapping[str, list[Mapping[str, Any]]] | None, failure_match: str | None, ) -> None: if failure_match is None: _run_basic_read(tmp_path, scenario, records) return - with pytest.raises(AssertionError, match=failure_match): + with pytest.raises(Exception, match=failure_match): _run_basic_read(tmp_path, scenario, records) @@ -736,6 +786,32 @@ def test_validate_schema_opt_out_survives_scenario_dedup(tmp_path: Path) -> None assert scenario.test_strictness_level is None +@pytest.mark.parametrize( + "validate_schema_values", + [ + pytest.param((False, True), id="false_then_true"), + pytest.param((True, False), id="true_then_false"), + ], +) +def test_validate_schema_opt_out_is_kept_in_either_order( + tmp_path: Path, validate_schema_values: tuple[bool, bool] +) -> None: + [scenario] = _get_scenarios( + tmp_path, + { + "acceptance_tests": { + "basic_read": { + "tests": [ + {"config_path": "secrets/config.json", "validate_schema": value} + for value in validate_schema_values + ] + }, + } + }, + ) + assert scenario.validate_schema is False + + @pytest.mark.parametrize("level", ["high", "low"]) def test_test_strictness_level_is_read_from_the_config(tmp_path: Path, level: str) -> None: scenarios = _get_scenarios( From ea0e6b4a9320ffef22235122ec580b5f53afe887 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Mon, 12 Oct 2026 19:13:23 +0300 Subject: [PATCH 7/8] test(standard-tests): basic_read record checks follow the declared catalog With #1142's `configured_catalog_path` support, the per-stream record check covers the streams of the config's declared catalog, and records are validated against the schemas that `discover` reports. Pin both, and say so in the migration guide. Co-Authored-By: Claude Opus 5.5 --- cdk-migrations.md | 2 +- unit_tests/test/test_read_assertions.py | 55 +++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/cdk-migrations.md b/cdk-migrations.md index 2c085a3156..94c5fec362 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -50,7 +50,7 @@ Rationale: A `check` that fails is reported as `status: FAILED` in a `CONNECTION This release also adds record assertions to the in-process `read` Standard Test (`SourceTestSuiteBase.test_basic_read`, run by `airbyte-cdk connector test`). Previously the test only failed when the whole read returned no records, so one stream with records passed the test for every stream. For every config listed under `basic_read` in `acceptance-test-config.yml`, it now restores two checks of the Connector Acceptance Tests (CAT) basic read test, unless the config's `status` is `failed` or `exception`. A config read only because it is listed under `full_refresh` keeps the old "some records" check, as CAT's full refresh test did not run these checks and the config has no `basic_read` entry to declare the exceptions below. -- **Every stream returns records.** Every stream that is not declared in the config's `empty_streams` must return at least one record. +- **Every stream returns records.** Every stream the read covers (the streams of the config's `configured_catalog_path`, or every discovered stream when it declares none) that is not declared in the config's `empty_streams` must return at least one record. - **Records match their schema.** Every record is validated against its stream's JSON schema with the validator CAT used: Draft 7, integers must be integers (`1.0` and `true` are not), `date-time` values must look like a date and a time and parse as a datetime (a space separator and a missing offset are accepted), other `format`s use the `jsonschema` defaults, and extra columns are allowed. A record that shares no field with a schema that declares `properties` also fails. The failure message names each stream, the number of failing records, and for each distinct schema error the record number, the path of the offending value inside the record, the failing schema keyword and the schema rule (for example ``record #3 at `$.items[0].price`: expected type 'number', got string``). Record values are never printed, for any kind of schema error: the message names the value's JSON type, and for an `anyOf`/`oneOf` the closest option's failure. A stream schema that cannot be applied (an unknown `type`, an unresolvable `$ref`, an invalid `pattern`) fails the check for that stream with the schema problem. diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py index b977e01f20..89fb820ddd 100644 --- a/unit_tests/test/test_read_assertions.py +++ b/unit_tests/test/test_read_assertions.py @@ -11,6 +11,7 @@ from __future__ import annotations +import json import logging from pathlib import Path from typing import Any, Iterable, Mapping @@ -752,6 +753,60 @@ def test_basic_read( _run_basic_read(tmp_path, scenario, records) +@pytest.mark.parametrize( + "declared_streams, records, failure_match", + [ + pytest.param( + ["users"], {"users": VALID_USERS}, None, id="undeclared_empty_stream_not_read" + ), + pytest.param( + ["users", "events"], {"users": VALID_USERS}, MSG_MISSING_STREAMS, id="declared_empty" + ), + pytest.param( + ["users"], {"users": [{"id": "1"}]}, MSG_SCHEMA, id="validated_against_discovered" + ), + ], +) +def test_basic_read_checks_the_streams_of_the_declared_catalog( + tmp_path: Path, + declared_streams: list[str], + records: Mapping[str, list[Mapping[str, Any]]], + failure_match: str | None, +) -> None: + """With a `configured_catalog_path`, only its streams must return records. + + The records are validated against the schema `discover` reports, not the catalog's own. + """ + catalog_path = tmp_path / "integration_tests" / "configured_catalog.json" + catalog_path.parent.mkdir(parents=True) + catalog_path.write_text( + json.dumps( + { + "streams": [ + { + # A stale schema that accepts anything. + "stream": {"name": name, "json_schema": {}, "supported_sync_modes": []}, + "sync_mode": "full_refresh", + "destination_sync_mode": "overwrite", + } + for name in declared_streams + ] + } + ) + ) + scenario = _scenario( + "connection", + "basic_read", + configured_catalog_path="integration_tests/configured_catalog.json", + ) + if failure_match is None: + _run_basic_read(tmp_path, scenario, records) + return + + with pytest.raises(Exception, match=failure_match): + _run_basic_read(tmp_path, scenario, records) + + # --- `validate_schema` in `acceptance-test-config.yml` ---------------------------------------- From 71efdfdbd773950afae1db59b74d4c8b968866a4 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 13 Oct 2026 19:06:44 +0300 Subject: [PATCH 8/8] docs(cdk-migrations): title the standard-tests migration section 7.36.0 Co-Authored-By: Claude Opus 5.5 --- cdk-migrations.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdk-migrations.md b/cdk-migrations.md index 94c5fec362..74c144891d 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -1,6 +1,6 @@ # CDK Migration Guide -## Upgrading to the next minor release (unreleased) +## Upgrading to 7.36.0 This release of the CDK changes which commands the Standard Tests (`airbyte-cdk connector test` and `airbyte-cdk image test`) run for the configs listed in `acceptance-test-config.yml`, and what they assert about the outcome of `check`. The release is a minor one, since only the test harness changes, but the connector test suites of connectors whose configs 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.