diff --git a/airbyte_cdk/test/models/scenario.py b/airbyte_cdk/test/models/scenario.py index c4a38c046..217c28549 100644 --- a/airbyte_cdk/test/models/scenario.py +++ b/airbyte_cdk/test/models/scenario.py @@ -103,6 +103,11 @@ 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 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, ...] = () @@ -179,6 +184,18 @@ def runs(self, *commands: ScenarioCommand) -> bool: """Whether any of `commands` runs for this scenario. See `commands`.""" return bool(self.commands & frozenset(commands)) + @property + def is_basic_read_config(self) -> bool: + """Whether this config is listed under `basic_read` (or the scenario was built by hand). + + Only the records of these configs are checked as in CAT's basic read test: every stream + not declared in `empty_streams` returns records, and every record matches its stream's + schema unless `validate_schema: false` is set. A config read only because it is listed + under `full_refresh` never had CAT's basic read checks, and has no `basic_read` entry to + declare those exceptions in, so its `read` only has to return some records. + """ + 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 000000000..c25ccdeb6 --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_read_assertions.py @@ -0,0 +1,434 @@ +# 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, 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 + +MAX_SCHEMA_ERRORS_PER_STREAM = 5 +"""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. +_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("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 + + +_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: + """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, limit: int = _MAX_MESSAGE_CHARS) -> str: + if len(text) <= limit: + return text + return text[: limit - 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 + + +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.""" + + 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 + # Set when the schema itself cannot be applied; the stream's records are not checked further. + invalid_schema: 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 + if report.invalid_schema is not None: + return + record_number = report.records_checked + report.records_checked += 1 + + 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: + 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(_describe_error(error))} " + 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_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 " + 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_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], + configured_catalog: ConfiguredAirbyteCatalog, + require_records_per_stream: bool, + validate_schema: bool, +) -> 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` + 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 not record_counts: + failures.append("Expected records but got none.") + 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 93100fd98..7b4b8b92c 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -191,6 +191,8 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe `discovery` and `basic_read` runs both `discover` and `read` (see `ConnectorTestScenario.commands`). + A `validate_schema: false` opt-out declared by any entry for the config is kept. + A config is read once, however many read sections (`basic_read`, `full_refresh`) list it. It keeps a `configured_catalog_path` only when every read entry for it declares that same catalog. If one of them declares none, or they declare different catalogs, the merged @@ -228,6 +230,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) ), @@ -278,6 +282,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 ( @@ -295,7 +300,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 7444d7502..1c11d59a0 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -19,6 +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, + assert_schema_validation_opt_out_allowed, +) from airbyte_cdk.test.standard_tests.connector_base import ( ConnectorTestSuiteBase, ) @@ -134,8 +138,18 @@ def test_basic_read( records from the streams of the config's `configured_catalog_path`, or from every discovered stream when it declares none, minus its `empty_streams` (see `get_read_streams`). + + A read expected to succeed must return records. For a config listed under `basic_read`, + the records are also checked as in CAT's basic read test: every stream it reads must + return at least one record, and every record must match its stream's JSON schema (opt out + with `validate_schema: false`, which a connector at `test_strictness_level: high` cannot). """ 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), @@ -173,7 +187,19 @@ def test_basic_read( catalog=configured_catalog, ) - if read_scenario.expected_outcome.expect_success() and not result.records: + if read_scenario.expected_outcome.expect_exception(): + # The read failed as expected (asserted by `run_test_job`); there are no records to check. + return + + 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_fail_read_with_bad_catalog( diff --git a/cdk-migrations.md b/cdk-migrations.md index b63c7dbca..74c144891 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. @@ -46,6 +46,24 @@ Migration steps: Rationale: A `check` that fails is reported as `status: FAILED` in a `CONNECTION_STATUS` message with exit code 0, so an exit-code or exception check cannot detect it. The Standard Tests replaced CAT as the only `check` coverage for most connectors, and without asserting the reported status a connector whose credentials expired, or whose custom components are rejected by the CDK baked into the base image, kept passing. The sections of `acceptance-test-config.yml` name the CAT test each config was written for; running every command for every config would add syncs nobody asked for, running `check` alone would test the wrong thing for most of them, and running nothing left expired secrets and stale configs undetected. Running each config with its own command, with the `succeed` default CAT applied, restores the CAT guarantees in both paths. +### Every stream returns records that match its schema (`basic_read`) + +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 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. + +Only the test harness changes, but 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, start failing. 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. 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. + ## Upgrading to 7.0.0 [Version 7.0.0](https://github.com/airbytehq/airbyte-python-cdk/releases/tag/v7.0.0) of the CDK migrates the CDK to the Concurrent CDK by removing some of the Declarative CDK concepts that are better expressed in the Concurrent CDK or that are outright incompatible with it. This changes mostly impact the Python implementations although the concept of CustomIncrementalSync has been removed from the declarative language as well. diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py new file mode 100644 index 000000000..89fb820dd --- /dev/null +++ b/unit_tests/test/test_read_assertions.py @@ -0,0 +1,884 @@ +# 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 json +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: Iterable[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", "expected type 'integer', got string", id="type" + ), + pytest.param( + 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, + {"id": 1, "updated_at": "2024-01-02"}, + "$.updated_at", + "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", + "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", + "fails `format` 'date', got string", + 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", + "expected type 'number', got string", + id="nested_array", + ), + pytest.param( + {"type": "object", "properties": {"first name": {"type": "string"}}}, + {"first name": 1}, + "$['first name']", + "expected type 'string', got integer", + 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_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 + + +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_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 + + +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. + + With `records=None`, the read raises instead. + """ + + def __init__( + self, + schemas: Mapping[str, Mapping[str, Any]], + records: Mapping[str, list[Mapping[str, Any]]] | None, + ) -> 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]: + 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) + + +def _run_basic_read( + tmp_path: Path, + scenario: ConnectorTestScenario, + records: Mapping[str, list[Mapping[str, Any]]] | None, +) -> 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"}] +READ_FAILS = None +MSG_STRICTNESS = "`validate_schema: false` is not allowed for a connector with" + +# (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 read only because it is listed under `full_refresh` has no `basic_read` entry to + # declare its empty streams or its `validate_schema` opt-out in, so only the "some records" + # floor applies. + pytest.param( + _scenario("connection", "full_refresh"), + {"users": VALID_USERS}, + None, + id="full_refresh_only_empty_stream", + ), + pytest.param( + _scenario("connection", "full_refresh"), + {}, + MSG_NO_RECORDS, + id="full_refresh_only_no_records", + ), + pytest.param( + _scenario("connection", "full_refresh"), + {"users": [{"id": "1"}]}, + None, + id="full_refresh_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` means `succeed`, the CAT default, so both checks apply. + pytest.param( + _scenario("basic_read", status=None), + {"users": VALID_USERS}, + MSG_MISSING_STREAMS, + id="no_status_empty_stream", + ), + pytest.param( + _scenario("basic_read", status=None), + {"users": [{"id": "1"}]}, + 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"), + {"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", + ), +] + + +@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]]] | None, + failure_match: str | None, +) -> None: + 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) + + +@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` ---------------------------------------- + + +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 + + 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( + "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( + 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]