diff --git a/airbyte_cdk/test/models/outcome.py b/airbyte_cdk/test/models/outcome.py index b5f45e3915..fe53703afb 100644 --- a/airbyte_cdk/test/models/outcome.py +++ b/airbyte_cdk/test/models/outcome.py @@ -15,11 +15,23 @@ class ExpectedOutcome(Enum): """Enum to represent the expected outcome of a test scenario. - Class supports comparisons to a boolean or None. + The members mirror the `status` values of `acceptance-test-config.yml`: + + - `EXPECT_SUCCESS` (`status: succeed`): the connector must report success. For `check`, + this means a `CONNECTION_STATUS` message with status `SUCCEEDED`. + - `EXPECT_EXCEPTION` (`status: failed`): the connector must report a graceful failure. For + `check`, this means a `CONNECTION_STATUS` message with status `FAILED`. + - `EXPECT_UNCAUGHT_ERROR` (`status: exception`): the connector must raise (a `TRACE` error + is emitted) instead of reporting a `CONNECTION_STATUS` message. + - `ALLOW_ANY` (no `status` declared): the CAT default of `succeed` applies to every command + the scenario runs (`ConnectorTestScenario.with_default_success`). As an outcome it is the + internal "no expectation yet" state of intermediate steps, such as the `discover` step of + a read test, which accept any outcome. """ EXPECT_EXCEPTION = auto() EXPECT_SUCCESS = auto() + EXPECT_UNCAUGHT_ERROR = auto() ALLOW_ANY = auto() @classmethod @@ -32,7 +44,7 @@ def from_status_str(cls, status: str | None) -> ExpectedOutcome: return { "succeed": ExpectedOutcome.EXPECT_SUCCESS, "failed": ExpectedOutcome.EXPECT_EXCEPTION, - "exception": ExpectedOutcome.EXPECT_EXCEPTION, # same as 'failed' + "exception": ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, }[status] except KeyError as ex: raise ValueError( @@ -53,8 +65,20 @@ def from_expecting_exception_bool(cls, expecting_exception: bool | None) -> Expe ) def expect_exception(self) -> bool: - """Return whether the expectation is that an exception should be raised.""" - return self == ExpectedOutcome.EXPECT_EXCEPTION + """Return whether the expectation is that the connector run should not succeed. + + This is true for both `status: failed` (a graceful, reported failure) and + `status: exception` (an uncaught error). Use `expect_uncaught_error()` to distinguish + the two. + """ + return self in ( + ExpectedOutcome.EXPECT_EXCEPTION, + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + ) + + def expect_uncaught_error(self) -> bool: + """Return whether the expectation is that the connector raises (`status: exception`).""" + return self == ExpectedOutcome.EXPECT_UNCAUGHT_ERROR def expect_success(self) -> bool: """Return whether the expectation is that the test should succeed without exceptions.""" diff --git a/airbyte_cdk/test/models/scenario.py b/airbyte_cdk/test/models/scenario.py index ddd99ef74d..217c285491 100644 --- a/airbyte_cdk/test/models/scenario.py +++ b/airbyte_cdk/test/models/scenario.py @@ -13,15 +13,55 @@ import tempfile from contextlib import contextmanager, suppress from pathlib import Path # noqa: TC003 # Pydantic needs this (don't move to 'if typing' block) +from types import MappingProxyType from typing import TYPE_CHECKING, Any, Literal, cast import yaml -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, field_validator from airbyte_cdk.test.models.outcome import ExpectedOutcome if TYPE_CHECKING: - from collections.abc import Generator + from collections.abc import Generator, Mapping + + +ScenarioCommand = Literal["spec", "check", "discover", "read"] +"""A connector command the standard tests run for a scenario. See `SECTION_COMMANDS`.""" + +SECTION_COMMANDS: Mapping[str, frozenset[ScenarioCommand]] = MappingProxyType( + { + "spec": frozenset({"spec"}), + "connection": frozenset({"check"}), + "discovery": frozenset({"discover"}), + "basic_read": frozenset({"read"}), + "full_refresh": frozenset({"read"}), + } +) +"""The command that runs for a config listed under each section of `acceptance-test-config.yml`. + +Each section stands for the connector command the Connector Acceptance Tests ran its configs +with, and the standard tests run that command for the config: + +- `spec`: the config must validate against the connector's spec (`test_config_matches_spec`). +- `connection`: `check` must report the declared `status` (`test_check`). +- `discovery`: `discover` must succeed and return streams (`test_discover`). +- `basic_read`, `full_refresh`: a full-refresh `read` must succeed and return records + (`test_basic_read`, `test_fail_read_with_bad_catalog`). It reads the streams of the config's + `configured_catalog_path`, or every discovered stream when it declares none, minus its + `empty_streams`. + +A config listed under several sections runs the union of their commands. Configs listed only +under `incremental` are not collected: without input state, an incremental `read` checks the same +thing as a full-refresh one. +""" + +SCENARIO_SECTIONS: tuple[str, ...] = tuple(SECTION_COMMANDS) +"""Every section that `get_scenarios()` reads, in the order its entries are collected.""" + +ALL_COMMANDS: frozenset[ScenarioCommand] = frozenset( + command for commands in SECTION_COMMANDS.values() for command in commands +) +"""Every command; what a scenario built by hand (no `sections`) runs.""" class ConnectorTestScenario(BaseModel): @@ -63,6 +103,25 @@ 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, ...] = () + + @field_validator("sections") + @classmethod + def _sections_must_be_known(cls, sections: tuple[str, ...]) -> tuple[str, ...]: + unknown = [section for section in sections if section not in SECTION_COMMANDS] + if unknown: + raise ValueError( + f"Unknown `acceptance-test-config.yml` section(s) {unknown}; " + f"expected any of {list(SCENARIO_SECTIONS)}." + ) + return sections def get_config_dict( self, @@ -107,6 +166,36 @@ def expected_outcome(self) -> ExpectedOutcome: """ return ExpectedOutcome.from_status_str(self.status) + @property + def commands(self) -> frozenset[ScenarioCommand]: + """The connector commands the standard tests run for this scenario. + + A config runs the command of every `acceptance-test-config.yml` section that lists it + (`SECTION_COMMANDS`). A scenario built by hand, with no `sections`, runs every command. + """ + if not self.sections: + return ALL_COMMANDS + + return frozenset( + command for section in self.sections for command in SECTION_COMMANDS[section] + ) + + 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. @@ -188,6 +277,18 @@ def with_expecting_success(self) -> ConnectorTestScenario: status="succeed", ) + def with_default_success(self) -> ConnectorTestScenario: + """Return the scenario, expecting success when it declares no `status`. + + No `status` in `acceptance-test-config.yml` means `succeed`, the CAT default, for every + command a config runs. `ALLOW_ANY` is kept only for intermediate steps that defer their + expectation (see `without_expected_outcome`). + """ + if self.status is not None: + return self + + return self.with_expecting_success() + @property def requires_creds(self) -> bool: """Return True if the scenario requires credentials to run.""" diff --git a/airbyte_cdk/test/standard_tests/_assertions.py b/airbyte_cdk/test/standard_tests/_assertions.py new file mode 100644 index 0000000000..5d374a1fa0 --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_assertions.py @@ -0,0 +1,110 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Shared assertions for Airbyte Standard Tests. + +These assertions are shared between the in-process test runner (`_job_runner.run_test_job`) +and the Docker-based test suite (`docker_base.DockerConnectorTestSuite`), so that both paths +enforce the same expectations. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import jsonschema + +from airbyte_cdk.models import Status +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput +from airbyte_cdk.test.models import ExpectedOutcome + +if TYPE_CHECKING: + from collections.abc import Mapping + + +def assert_check_outcome( + *, + check_result: EntrypointOutput, + expected_outcome: ExpectedOutcome, + connector_name: str, +) -> None: + """Assert that the outcome of `check` matches the scenario's expected outcome. + + A failing `check` reports `status: FAILED` in a `CONNECTION_STATUS` message and still + exits 0, so exit-code checks alone do not catch it. We therefore assert the reported + status explicitly, following the `status` semantics of `acceptance-test-config.yml`: + + - `succeed` (`EXPECT_SUCCESS`): a `CONNECTION_STATUS` message with status `SUCCEEDED`. + - `failed` (`EXPECT_EXCEPTION`): a `CONNECTION_STATUS` message with status `FAILED`. + - `exception` (`EXPECT_UNCAUGHT_ERROR`): `check` raises, so a `TRACE` error must be present + and no `SUCCEEDED` status may be reported. A `CONNECTION_STATUS` message is not required. + - no `status` (`ALLOW_ANY`): treated as `succeed`, matching the CAT default. A config that + `check` is expected to reject must declare `status: failed` (or `exception`). + + When more than one `CONNECTION_STATUS` message is present, the last one wins. + """ + connection_statuses = [ + message.connectionStatus + for message in check_result.connection_status_messages + if message.connectionStatus is not None + ] + if expected_outcome.expect_uncaught_error(): + assert check_result.errors, ( + f"`check` for connector '{connector_name}' was expected to raise " + f"(`status: exception`), but emitted no TRACE error. " + f"Reported statuses: {connection_statuses}" + ) + assert not connection_statuses or connection_statuses[-1].status != Status.SUCCEEDED, ( + f"`check` for connector '{connector_name}' was expected to raise " + f"(`status: exception`), but reported: {connection_statuses[-1]}" + ) + return + + assert connection_statuses, ( + f"`check` for connector '{connector_name}' emitted no CONNECTION_STATUS message. " + f"A `check` implementation should report its outcome as a CONNECTION_STATUS message " + f"instead of raising. If raising is the expected outcome for this config, declare " + f"`status: exception` for it in `acceptance-test-config.yml`.\n" + + (f"Errors: {check_result.get_formatted_error_message()}\n" if check_result.errors else "") + + f"Logs: {check_result.logs}" + ) + reported_status = connection_statuses[-1].status + if expected_outcome.expect_exception(): + assert reported_status == Status.FAILED, ( + f"`check` for connector '{connector_name}' was expected to fail, but reported: " + f"{connection_statuses[-1]}" + ) + return + + # Both `EXPECT_SUCCESS` and `ALLOW_ANY` (no declared status) require a successful `check`. + assert reported_status == Status.SUCCEEDED, ( + f"`check` for connector '{connector_name}' did not succeed: {connection_statuses[-1]}" + ) + + +def assert_config_matches_spec( + *, + config: Mapping[str, Any], + spec_result: EntrypointOutput, + connector_name: str, + scenario_id: str, +) -> None: + """Assert that a config validates against the `connectionSpecification` the connector reports. + + This is the CAT `test_config_match_spec` check for configs listed under `spec` in + `acceptance-test-config.yml`: the connector's own spec must accept the config. The config's + values never appear in the failure message, since they may be secrets. + """ + specs = [message.spec for message in spec_result.spec_messages if message.spec is not None] + assert len(specs) == 1, ( + f"Expected exactly one SPEC message from connector '{connector_name}', got {len(specs)}." + + (f"\nErrors: {spec_result.get_formatted_error_message()}" if spec_result.errors else "") + ) + try: + jsonschema.validate(instance=config, schema=specs[0].connectionSpecification) + except jsonschema.ValidationError as exc: + # `exc.message` embeds the offending value; keep it (and the chained exception, whose + # text includes the instance) out of the test output. + message = exc.message.replace(repr(exc.instance), "") + raise AssertionError( + f"Config '{scenario_id}' does not match the spec of connector '{connector_name}' " + f"at `{exc.json_path}`: {message}" + ) from None diff --git a/airbyte_cdk/test/standard_tests/_job_runner.py b/airbyte_cdk/test/standard_tests/_job_runner.py index bc729190dc..3bb0ea4707 100644 --- a/airbyte_cdk/test/standard_tests/_job_runner.py +++ b/airbyte_cdk/test/standard_tests/_job_runner.py @@ -13,12 +13,12 @@ from airbyte_cdk.models import ( ConfiguredAirbyteCatalog, - Status, ) from airbyte_cdk.test import entrypoint_wrapper from airbyte_cdk.test.models import ( ConnectorTestScenario, ) +from airbyte_cdk.test.standard_tests._assertions import assert_check_outcome @runtime_checkable @@ -113,32 +113,25 @@ def run_test_job( if inspect_output: inspect_output(result) - if result.errors and test_scenario.expected_outcome.expect_success(): - raise result.as_exception() - if verb == "check": - # Check is expected to fail gracefully without an exception. - # Instead, we assert that we have a CONNECTION_STATUS message with - # a failure status. - assert len(result.connection_status_messages) == 1, ( - "Expected exactly one CONNECTION_STATUS message. Got " - f"{len(result.connection_status_messages)}:\n" - + "\n".join([str(msg) for msg in result.connection_status_messages]) - + result.get_formatted_error_message() + # Unless the scenario declares `status: exception`, `check` is expected to report its + # outcome as a CONNECTION_STATUS message, gracefully and without an exception. Any + # uncaught error is surfaced as-is (mirroring `raise_if_errors` on the Docker path), and + # the reported status is then validated with the same assertion as the Docker-based test + # suite so that both paths enforce the same expectations. + if result.errors and not test_scenario.expected_outcome.expect_exception(): + raise result.as_exception() + + assert_check_outcome( + check_result=result, + expected_outcome=test_scenario.expected_outcome, + connector_name=connector_root.absolute().name, ) - if test_scenario.expected_outcome.expect_exception(): - conn_status = result.connection_status_messages[0].connectionStatus - assert conn_status, ( - "Expected CONNECTION_STATUS message to be present. Got: \n" - + "\n".join([str(msg) for msg in result.connection_status_messages]) - ) - assert conn_status.status == Status.FAILED, ( - "Expected CONNECTION_STATUS message to be FAILED. Got: \n" - + "\n".join([str(msg) for msg in result.connection_status_messages]) - ) - return result + if result.errors and test_scenario.expected_outcome.expect_success(): + raise result.as_exception() + # For all other verbs, we assert check that an exception is raised (or not). if test_scenario.expected_outcome.expect_exception(): if not result.errors: 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..c25ccdeb61 --- /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/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py index 8c251c835c..de567f343e 100644 --- a/airbyte_cdk/test/standard_tests/connector_base.py +++ b/airbyte_cdk/test/standard_tests/connector_base.py @@ -15,6 +15,7 @@ from airbyte_cdk.test.models import ( ConnectorTestScenario, ) +from airbyte_cdk.test.standard_tests._assertions import assert_config_matches_spec from airbyte_cdk.test.standard_tests._job_runner import IConnector, run_test_job from airbyte_cdk.test.standard_tests._spec_lint import ( assert_no_secrets_in_output, @@ -23,6 +24,7 @@ from airbyte_cdk.test.standard_tests.docker_base import ( _SPEC_FAILED_SKIP_REASON, DockerConnectorTestSuite, + skip_unless_runs, ) if TYPE_CHECKING: @@ -148,7 +150,13 @@ def test_check( self, scenario: ConnectorTestScenario, ) -> None: - """Run `connection` acceptance tests.""" + """Run `connection` acceptance tests. + + Runs for configs listed under `connection` in `acceptance-test-config.yml`. Scenarios + declared with `status: exception` expect `check` to raise instead of reporting a status; + for those, only the presence of a trace error is asserted (in `run_test_job`). + """ + skip_unless_runs(scenario, "check") result: entrypoint_wrapper.EntrypointOutput = run_test_job( self.create_connector(scenario), "check", @@ -158,7 +166,36 @@ def test_check( scenario, output, verb="check" ), ) + if scenario.expected_outcome.expect_uncaught_error(): + # An uncaught error is the expected outcome; no CONNECTION_STATUS is required. + return + assert len(result.connection_status_messages) == 1, ( "Expected exactly one CONNECTION_STATUS message. " f"Got: {result.connection_status_messages!s}" ) + + def test_config_matches_spec( + self, + scenario: ConnectorTestScenario, + ) -> None: + """Validate the scenario's config against the connector's `spec`. + + Runs for configs listed under `spec` in `acceptance-test-config.yml` (the CAT + `test_config_match_spec` check): the connector's own `connectionSpecification` must + accept the config. + """ + skip_unless_runs(scenario, "spec") + connector_root = self.get_connector_root_dir() + spec_result: entrypoint_wrapper.EntrypointOutput = run_test_job( + self.create_connector(scenario), + "spec", + connector_root=connector_root, + test_scenario=scenario.without_expected_outcome(), + ) + assert_config_matches_spec( + config=scenario.get_config_dict(connector_root=connector_root, empty_if_missing=False), + spec_result=spec_result, + connector_name=connector_root.absolute().name, + scenario_id=scenario.id, + ) diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index 4f18344168..dd8b365dba 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -19,17 +19,21 @@ from boltons.typeutils import classproperty from airbyte_cdk.models import ( - AirbyteCatalog, + AirbyteStream, ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, ConnectorSpecification, DestinationSyncMode, - Status, SyncMode, ) from airbyte_cdk.models.connector_metadata import MetadataFile from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput -from airbyte_cdk.test.models import ConnectorTestScenario, ExpectedOutcome +from airbyte_cdk.test.models import ConnectorTestScenario +from airbyte_cdk.test.models.scenario import SCENARIO_SECTIONS, ScenarioCommand +from airbyte_cdk.test.standard_tests._assertions import ( + assert_check_outcome, + assert_config_matches_spec, +) from airbyte_cdk.test.standard_tests._spec_compatibility import ( DEPLOYMENT_MODE_ENV_VARS, REGISTRY_UNAVAILABLE_ERRORS, @@ -58,42 +62,88 @@ ) -def _assert_check_outcome( - *, - check_result: EntrypointOutput, - expected_outcome: ExpectedOutcome, - connector_name: str, -) -> None: - """Assert that the reported CONNECTION_STATUS matches the scenario's expected outcome. - - A failing `check` reports `status: FAILED` in a `CONNECTION_STATUS` message and still - exits 0, so exit-code checks alone do not catch it. We therefore assert the reported - status explicitly, in both directions: - - A scenario expecting success must report `SUCCEEDED`. - - A scenario expecting failure must not report `SUCCEEDED` (a graceful `FAILED` status - or an uncaught error with no status message both count as the expected failure). - - `ALLOW_ANY` scenarios accept either outcome. +def skip_unless_runs(scenario: ConnectorTestScenario, *commands: ScenarioCommand) -> None: + """Skip the current test unless one of `commands` runs for `scenario`. + + A config runs the command of every `acceptance-test-config.yml` section that lists it; see + `ConnectorTestScenario.commands`. """ - connection_statuses = [ - message.connectionStatus - for message in check_result.connection_status_messages - if message.connectionStatus is not None - ] - if expected_outcome.expect_exception(): - assert not connection_statuses or connection_statuses[-1].status != Status.SUCCEEDED, ( - f"`check` for connector '{connector_name}' was expected to fail, but reported: " - f"{connection_statuses[-1]}" - ) + if scenario.runs(*commands): return - assert connection_statuses, ( - f"`check` for connector '{connector_name}' emitted no CONNECTION_STATUS message. " - f"Logs: {check_result.logs}" + wanted = ", ".join(f"`{command}`" for command in commands) + verb = "does not run" if len(commands) == 1 else "do not run" + pytest.skip( + f"{wanted} {verb} for scenario '{scenario.id}': its config is listed under " + f"{', '.join(f'`{section}`' for section in scenario.sections)} in " + "`acceptance-test-config.yml`, which runs " + f"{', '.join(f'`{command}`' for command in sorted(scenario.commands))}." ) - if expected_outcome.expect_success(): - assert connection_statuses[-1].status == Status.SUCCEEDED, ( - f"`check` for connector '{connector_name}' did not succeed: {connection_statuses[-1]}" + + +def get_discovered_streams( + discover_result: EntrypointOutput, + connector_name: str, +) -> list[AirbyteStream]: + """Return the streams of a `discover` result, failing on a missing or empty catalog.""" + catalog_message = discover_result.catalog + assert catalog_message.catalog is not None, "Catalog message missing catalog." + if not catalog_message.catalog.streams: + raise ValueError( + f"Discovered catalog for connector '{connector_name}' is empty. " + "Please check the connector's discover implementation." ) + return list(catalog_message.catalog.streams) + + +def get_read_streams( + scenario: ConnectorTestScenario, + discovered_streams: list[AirbyteStream], + connector_root: Path, +) -> list[AirbyteStream]: + """Return the streams that the `read` of `scenario` configures, minus its `empty_streams`. + + When the scenario declares a `configured_catalog_path`, the streams named in that catalog are + read, in its order, as the Connector Acceptance Tests did: the catalog picks the streams, and + each stream keeps the definition (and JSON schema) that `discover` reported. A stream of the + catalog that `discover` does not report fails the test. Without a declared catalog, every + discovered stream is read. + """ + streams = discovered_streams + if scenario.configured_catalog_path: + catalog_path = scenario.configured_catalog_path + if not catalog_path.is_absolute(): + catalog_path = connector_root / catalog_path + if not catalog_path.is_file(): + raise AssertionError( + f"The configured catalog `{scenario.configured_catalog_path}` of config " + f"'{scenario.id}' does not exist in connector " + f"'{connector_root.absolute().name}'. Fix its `configured_catalog_path`, or remove " + "it to read every discovered stream." + ) + declared_catalog = orjson.loads(catalog_path.read_bytes()) + # A stream listed twice in the catalog is read once. + declared_names = list( + dict.fromkeys( + configured["stream"]["name"] for configured in declared_catalog["streams"] + ) + ) + discovered_by_name = {stream.name: stream for stream in discovered_streams} + undiscovered = [name for name in declared_names if name not in discovered_by_name] + if undiscovered: + raise AssertionError( + f"The configured catalog `{scenario.configured_catalog_path}` of config " + f"'{scenario.id}' lists streams that connector '{connector_root.absolute().name}' " + f"does not discover: {', '.join(repr(name) for name in undiscovered)}. Remove " + "them from the catalog, or point `configured_catalog_path` at a current catalog." + ) + streams = [discovered_by_name[name] for name in declared_names] + + if scenario.empty_streams: + empty_stream_names = {stream.name for stream in scenario.empty_streams} + streams = [stream for stream in streams if stream.name not in empty_stream_names] + + return streams _SPEC_FAILURE_OUTPUT_LINES = 50 @@ -270,9 +320,28 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe config will not error on the lack of data in the empty streams or lack of permissions to read them. We also carry over any explicitly declared `status`. Only the `connection` section declares one, - so without this an entry from another section (e.g. `spec`) would win and silently downgrade the - scenario to `ALLOW_ANY`, which accepts a failing `check`. + so without this an entry from another section (e.g. `spec`) would win and silently drop the + declared expectation. Conflicting non-null statuses for the same config raise a `ValueError`. + + The sections a config is listed under are unioned as well, so a config that appears under + `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 + scenario declares none, and every discovered stream is read: a superset of what each + entry asked for. """ + read_catalogs: dict[Path | None, set[Path | None]] = {} + for scenario in scenarios: + if scenario.runs("read"): + read_catalogs.setdefault(scenario.config_path, set()).add( + scenario.configured_catalog_path + ) + deduped_scenarios: list[ConnectorTestScenario] = [] for scenario in scenarios: @@ -297,6 +366,11 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe update={ "empty_streams": list(set(all_empty_streams)), "status": existing_scenario.status or scenario.status, + "validate_schema": existing_scenario.validate_schema + and scenario.validate_schema, + "sections": tuple( + dict.fromkeys(existing_scenario.sections + scenario.sections) + ), } ) deduped_scenarios.remove(existing_scenario) @@ -305,13 +379,30 @@ def _dedup_scenarios(scenarios: list[ConnectorTestScenario]) -> list[ConnectorTe else: # If a scenario does not exist with the config, add the new scenario to the list. deduped_scenarios.append(scenario) + + for index, scenario in enumerate(deduped_scenarios): + catalogs = read_catalogs.get(scenario.config_path) + if catalogs is None: + continue + catalog = next(iter(catalogs)) if len(catalogs) == 1 else None + if scenario.configured_catalog_path != catalog: + deduped_scenarios[index] = scenario.model_copy( + update={"configured_catalog_path": catalog} + ) return deduped_scenarios @classmethod def get_scenarios( cls, ) -> list[ConnectorTestScenario]: - """Get acceptance tests for a given category. + """Collect one scenario per config listed in `acceptance-test-config.yml`. + + Every section with a `tests` list contributes its configs (`SCENARIO_SECTIONS`), and each + config runs the command of the section(s) that list it: `spec` validates the config + against the spec, `connection` runs `check` against the declared `status`, `discovery` + runs `discover`, and `basic_read` and `full_refresh` run a full-refresh `read` of the + streams of the config's `configured_catalog_path`, or of every discovered stream when it + declares none (`SECTION_COMMANDS`). The `incremental` section is not collected. This has to be a separate function because pytest does not allow parametrization of fixtures with arguments from the test class itself. @@ -327,9 +418,9 @@ def get_scenarios( ) return [] + test_strictness_level = all_tests_config.get("test_strictness_level") test_scenarios: list[ConnectorTestScenario] = [] - # we look in the basic_read section to find any empty streams - for category in ["spec", "connection", "basic_read"]: + for category in SCENARIO_SECTIONS: if ( category not in all_tests_config["acceptance_tests"] or "tests" not in all_tests_config["acceptance_tests"][category] @@ -345,7 +436,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) + scenario = ConnectorTestScenario.model_validate( + { + **test, + "sections": (category,), + "test_strictness_level": test_strictness_level, + } + ) test_scenarios.append(scenario) @@ -353,6 +450,33 @@ def get_scenarios( return deduped_test_scenarios + @classmethod + def _connector_image( + cls, + connector_image_override: str | None, + connector_base_image_override: str | None, + ) -> str: + """Return the image under test: the override if given, else a fresh `dev-latest` build. + + Note: + - It is expected for docker image caches to be reused between test runs. + - In the rare case that image caches need to be cleared, please clear + the local docker image cache using `docker image prune -a` command. + """ + if connector_image_override: + return connector_image_override + + connector_root = cls.get_connector_root_dir().absolute() + metadata = MetadataFile.from_file(connector_root / "metadata.yaml") + return build_connector_image( + connector_name=connector_root.name, + connector_directory=connector_root, + metadata=metadata, + tag="dev-latest", + no_verify=False, + base_image_override=connector_base_image_override, + ) + @pytest.mark.skipif( shutil.which("docker") is None, reason="docker CLI not found in PATH, skipping docker image tests", @@ -370,19 +494,9 @@ def test_docker_image_build_and_spec( what catches an image that emits no spec on that path. """ connector_root = self.get_connector_root_dir().absolute() - metadata = MetadataFile.from_file(connector_root / "metadata.yaml") - - connector_image: str | None = connector_image_override - if not connector_image: - tag = "dev-latest" - connector_image = build_connector_image( - connector_name=connector_root.absolute().name, - connector_directory=connector_root, - metadata=metadata, - tag=tag, - no_verify=False, - base_image_override=connector_base_image_override, - ) + connector_image = self._connector_image( + connector_image_override, connector_base_image_override + ) spec = _run_docker_spec(connector_image, connector_name=connector_root.name) assert_spec_is_valid(spec, connector_name=connector_root.name) @@ -557,27 +671,14 @@ def test_docker_image_build_and_check( ) -> None: """Run `docker_image` acceptance tests. - This test builds the connector image and runs the `check` command inside the container. - - Note: - - It is expected for docker image caches to be reused between test runs. - - In the rare case that image caches need to be cleared, please clear - the local docker image cache using `docker image prune -a` command. + This test builds the connector image and runs the `check` command inside the container, + for configs listed under `connection` in `acceptance-test-config.yml`. """ - tag = "dev-latest" + skip_unless_runs(scenario, "check") connector_root = self.get_connector_root_dir() - metadata = MetadataFile.from_file(connector_root / "metadata.yaml") - connector_image: str | None = connector_image_override - if not connector_image: - tag = "dev-latest" - connector_image = build_connector_image( - connector_name=connector_root.absolute().name, - connector_directory=connector_root, - metadata=metadata, - tag=tag, - no_verify=False, - base_image_override=connector_base_image_override, - ) + connector_image = self._connector_image( + connector_image_override, connector_base_image_override + ) container_config_path = "/secrets/config.json" with scenario.with_temp_config_file( @@ -621,12 +722,100 @@ def test_docker_image_build_and_check( # container, in both directions (e.g. it fails if bundled custom components are rejected # by the CDK baked into the base image, and it fails if a `check` that is expected to # fail starts succeeding). - _assert_check_outcome( + assert_check_outcome( check_result=check_result, expected_outcome=scenario.expected_outcome, connector_name=connector_name, ) + @pytest.mark.skipif( + shutil.which("docker") is None, + reason="docker CLI not found in PATH, skipping docker image tests", + ) + @pytest.mark.image_tests + def test_docker_image_build_and_config_matches_spec( + self, + scenario: ConnectorTestScenario, + connector_image_override: str | None, + connector_base_image_override: str | None, + ) -> None: + """Validate the scenario's config against the `spec` the image reports. + + Runs for configs listed under `spec` in `acceptance-test-config.yml` (the CAT + `test_config_match_spec` check). + """ + skip_unless_runs(scenario, "spec") + connector_root = self.get_connector_root_dir() + connector_image = self._connector_image( + connector_image_override, connector_base_image_override + ) + + spec_result = run_docker_airbyte_command( + [ + "docker", + "run", + "--rm", + connector_image, + "spec", + ], + raise_if_errors=True, + ) + assert_config_matches_spec( + config=scenario.get_config_dict(connector_root=connector_root, empty_if_missing=False), + spec_result=spec_result, + connector_name=connector_root.absolute().name, + scenario_id=scenario.id, + ) + + @pytest.mark.skipif( + shutil.which("docker") is None, + reason="docker CLI not found in PATH, skipping docker image tests", + ) + @pytest.mark.image_tests + def test_docker_image_build_and_discover( + self, + scenario: ConnectorTestScenario, + connector_image_override: str | None, + connector_base_image_override: str | None, + ) -> None: + """Run `discover` inside the connector image and require a non-empty catalog. + + Runs for configs listed under `discovery` in `acceptance-test-config.yml`. + """ + if self.is_destination_connector(): + pytest.skip("Skipping discover test for destination connector.") + + skip_unless_runs(scenario, "discover") + + if scenario.expected_outcome.expect_exception(): + pytest.skip("Skipping (expected to fail).") + + connector_root = self.get_connector_root_dir() + connector_image = self._connector_image( + connector_image_override, connector_base_image_override + ) + + container_config_path = "/secrets/config.json" + with scenario.with_temp_config_file( + connector_root=connector_root, + ) as temp_config_file: + discover_result = run_docker_airbyte_command( + [ + "docker", + "run", + "--rm", + "-v", + f"{temp_config_file}:{container_config_path}:rw", + connector_image, + "discover", + "--config", + container_config_path, + ], + raise_if_errors=True, + ) + + get_discovered_streams(discover_result, connector_root.absolute().name) + @pytest.mark.skipif( shutil.which("docker") is None, reason="docker CLI not found in PATH, skipping docker image tests", @@ -642,17 +831,20 @@ def test_docker_image_build_and_read( ) -> None: """Read from the connector's Docker image. - This test builds the connector image and runs the `read` command inside the container. + This test builds the connector image and runs a full-refresh `read` command inside the + container, for configs listed under `basic_read` or `full_refresh` in + `acceptance-test-config.yml`. The read covers the streams of the config's + `configured_catalog_path`, or every discovered stream when it declares none, minus its + `empty_streams` (see `get_read_streams`), narrowed further by `--read-from-streams`. Note: - - It is expected for docker image caches to be reused between test runs. - - In the rare case that image caches need to be cleared, please clear - the local docker image cache using `docker image prune -a` command. - If the --connector-image arg is provided, it will be used instead of building the image. """ if self.is_destination_connector(): pytest.skip("Skipping read test for destination connector.") + skip_unless_runs(scenario, "read") + if scenario.expected_outcome.expect_exception(): pytest.skip("Skipping (expected to fail).") @@ -678,21 +870,12 @@ def test_docker_image_build_and_read( f"(not in --read-scenarios={read_scenarios})." ) - tag = "dev-latest" connector_root = self.get_connector_root_dir() connector_name = connector_root.absolute().name metadata = MetadataFile.from_file(connector_root / "metadata.yaml") - connector_image: str | None = connector_image_override - if not connector_image: - tag = "dev-latest" - connector_image = build_connector_image( - connector_name=connector_name, - connector_directory=connector_root, - metadata=metadata, - tag=tag, - no_verify=False, - base_image_override=connector_base_image_override, - ) + connector_image = self._connector_image( + connector_image_override, connector_base_image_override + ) container_config_path = "/secrets/config.json" container_catalog_path = "/secrets/catalog.json" @@ -722,42 +905,28 @@ def test_docker_image_build_and_read( raise_if_errors=True, ) - catalog_message = discover_result.catalog # Get catalog message - assert catalog_message.catalog is not None, "Catalog message missing catalog." - discovered_catalog: AirbyteCatalog = catalog_message.catalog - if not discovered_catalog.streams: - raise ValueError( - f"Discovered catalog for connector '{connector_name}' is empty. " - "Please check the connector's discover implementation." - ) - - streams_list = [stream.name for stream in discovered_catalog.streams] + read_streams = get_read_streams( + scenario, + get_discovered_streams(discover_result, connector_name), + connector_root, + ) if read_from_streams == "default" and metadata.data.suggestedStreams: - # set `streams_list` to be the intersection of discovered and suggested streams. - streams_list = list(set(streams_list) & set(metadata.data.suggestedStreams.streams)) + # Read only the suggested streams. + suggested_streams = set(metadata.data.suggestedStreams.streams) + read_streams = [s for s in read_streams if s.name in suggested_streams] if isinstance(read_from_streams, list): - # If `read_from_streams` is a list, we filter the discovered streams. - streams_list = list(set(streams_list) & set(read_from_streams)) - - if scenario.empty_streams: - # Filter out streams marked as empty in the scenario. - empty_stream_names = [stream.name for stream in scenario.empty_streams] - streams_list = [s for s in streams_list if s.name not in empty_stream_names] + # If `read_from_streams` is a list, read only those streams. + read_streams = [s for s in read_streams if s.name in read_from_streams] configured_catalog: ConfiguredAirbyteCatalog = ConfiguredAirbyteCatalog( streams=[ ConfiguredAirbyteStream( stream=stream, - sync_mode=( - stream.supported_sync_modes[0] - if stream.supported_sync_modes - else SyncMode.full_refresh - ), + sync_mode=SyncMode.full_refresh, destination_sync_mode=DestinationSyncMode.append, ) - for stream in discovered_catalog.streams - if stream.name in streams_list + for stream in read_streams ] ) configured_catalog_path = temp_dir / "catalog.json" diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index 6146a3e354..0543b9dfc4 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -25,10 +25,19 @@ format_catalog_problems, ) 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._spec_lint import assert_spec_is_valid, get_single_spec from airbyte_cdk.test.standard_tests.connector_base import ( ConnectorTestSuiteBase, ) +from airbyte_cdk.test.standard_tests.docker_base import ( + get_discovered_streams, + get_read_streams, + skip_unless_runs, +) if TYPE_CHECKING: from airbyte_cdk.test import entrypoint_wrapper @@ -47,10 +56,15 @@ def test_check( ) -> None: """Run standard `check` tests on the connector. - Assert that the connector returns a single CONNECTION_STATUS message. - This test is designed to validate the connector's ability to establish a connection - and return its status with the expected message type. + Runs for configs listed under `connection` in `acceptance-test-config.yml`. Assert that + the connector returns a single CONNECTION_STATUS message whose status matches the + scenario's expectation. This test is designed to validate the connector's ability to + establish a connection and return its status with the expected message type. + + Scenarios declared with `status: exception` expect `check` to raise instead of reporting + a status; for those, only the presence of a trace error is asserted (in `run_test_job`). """ + skip_unless_runs(scenario, "check") result: entrypoint_wrapper.EntrypointOutput = run_test_job( self.create_connector(scenario), "check", @@ -60,6 +74,10 @@ def test_check( scenario, output, verb="check" ), ) + if scenario.expected_outcome.expect_uncaught_error(): + # An uncaught error is the expected outcome; no CONNECTION_STATUS is required. + return + num_status_messages = len(result.connection_status_messages) assert num_status_messages == 1, ( f"Expected exactly one CONNECTION_STATUS message. Got {num_status_messages}: \n" @@ -72,10 +90,12 @@ def test_discover( ) -> None: """Standard test for `discover`. - Assert that the connector emits exactly one catalog and that the catalog passes the - checks in `CATALOG_CHECKS`. Problems found by checks that are not enforced yet are - reported as a `DiscoveredCatalogWarning` instead of failing the test. + Runs for configs listed under `discovery` in `acceptance-test-config.yml`. Assert + that the connector emits exactly one catalog and that the catalog passes the checks in + `CATALOG_CHECKS`. Problems found by checks that are not enforced yet are reported as a + `DiscoveredCatalogWarning` instead of failing the test. """ + skip_unless_runs(scenario, "discover") if scenario.expected_outcome.expect_exception(): # If the scenario expects an exception, we can't ensure it specifically would fail # in discover, because some discover implementations do not need to make a connection. @@ -83,11 +103,12 @@ def test_discover( pytest.skip("Skipping discover test for scenario that expects an exception.") return + connector_root = self.get_connector_root_dir() result = run_test_job( self.create_connector(scenario), "discover", - connector_root=self.get_connector_root_dir(), - test_scenario=scenario, + connector_root=connector_root, + test_scenario=scenario.with_default_success(), ) if result.errors and not scenario.expected_outcome.expect_success(): # The scenario allows `discover` to fail, so there is no catalog to check. @@ -143,26 +164,42 @@ def test_basic_read( ) -> None: """Run standard `read` test on the connector. - 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. + Runs for configs listed under `basic_read` or `full_refresh` in + `acceptance-test-config.yml`. 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 full-refresh `read` job to fetch + 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), "discover", - connector_root=self.get_connector_root_dir(), + connector_root=connector_root, test_scenario=scenario.without_expected_outcome(), ) if scenario.expected_outcome.expect_exception() and discover_result.errors: # Failed as expected; we're done. return - streams = discover_result.catalog.catalog.streams # type: ignore [reportOptionalMemberAccess, union-attr] - - if scenario.empty_streams: - # Filter out streams marked as empty in the scenario. - empty_stream_names = [stream.name for stream in scenario.empty_streams] - streams = [s for s in streams if s.name not in empty_stream_names] + if discover_result.errors: + raise discover_result.as_exception() + streams = get_read_streams( + scenario, + get_discovered_streams(discover_result, connector_root.absolute().name), + connector_root, + ) configured_catalog = ConfiguredAirbyteCatalog( streams=[ @@ -174,22 +211,40 @@ def test_basic_read( for stream in streams ] ) + read_scenario = scenario.with_default_success() result = run_test_job( self.create_connector(scenario), "read", - test_scenario=scenario, - connector_root=self.get_connector_root_dir(), + test_scenario=read_scenario, + connector_root=connector_root, catalog=configured_catalog, ) - if 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( self, scenario: ConnectorTestScenario, ) -> None: - """Standard test for `read` when passed a bad catalog file.""" + """Standard test for `read` when passed a bad catalog file. + + Runs for configs listed under `basic_read` or `full_refresh` in + `acceptance-test-config.yml`. + """ + skip_unless_runs(scenario, "read") invalid_configured_catalog = ConfiguredAirbyteCatalog( streams=[ # Create ConfiguredAirbyteStream which is deliberately invalid diff --git a/cdk-migrations.md b/cdk-migrations.md index e29bbf266f..ae3781c9b6 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -1,8 +1,68 @@ # CDK Migration Guide -## Upgrading to the next minor release (unreleased) +## Upgrading to 7.36.0 -The next minor release of the CDK adds checks to the standard tests. Every Python and manifest-only connector that runs the standard tests, for example through `airbyte-cdk connector test` or `airbyte-cdk image test`, picks them up when it moves to this release. There is no change to how connectors run. +This release of the CDK adds checks to the Standard Tests (`airbyte-cdk connector test` and `airbyte-cdk image test`) and changes which commands they run for the configs listed in `acceptance-test-config.yml`. There is no change to how connectors run. The release is a minor one, since only the test harness changes, but the connector test suites of connectors that do not satisfy the new expectations start failing. The connector monorepo installs the `airbyte-cdk` CLI unpinned (`uv tool install --upgrade 'airbyte-cdk[dev]'`), so connector CI picks the new behaviour up with this release, without a per-connector CDK bump. + +### Each config runs the command of the section that lists it + +Previously only the `spec`, `connection` and `basic_read` sections of `acceptance-test-config.yml` were read, every config listed there ran every test (`check`, `discover` and `read`), and configs listed only under `discovery` or `full_refresh` were never tested. Now every section with a `tests` list contributes its configs, except `incremental`, and each config runs the command the Connector Acceptance Tests (CAT) ran it with: + +| Section | What the config must do | Tests | +|---|---|---| +| `spec` | Validate against the connector's `connectionSpecification` (the CAT `test_config_match_spec` check). | `test_config_matches_spec`, `test_docker_image_build_and_config_matches_spec` | +| `connection` | Make `check` report the declared `status` (see below). | `test_check`, `test_docker_image_build_and_check` | +| `discovery` | Make `discover` succeed and return at least one stream. | `test_discover`, `test_docker_image_build_and_discover` | +| `basic_read`, `full_refresh` | Make a full-refresh `read` succeed and return records. The read covers the streams of the config's `configured_catalog_path`, or every discovered stream when none is declared, minus `empty_streams`. | `test_basic_read`, `test_fail_read_with_bad_catalog`, `test_docker_image_build_and_read` | + +The `incremental` section is still not collected: without input state, an incremental `read` checks the same thing as a full-refresh one, so a config listed only there stays untested. A config listed under several sections runs the union of their commands, and no others: a config listed only under `spec` is no longer run through `check`, and a config listed only under `connection` is no longer read. `ConnectorTestScenario.sections` records where a config is listed and `ConnectorTestScenario.commands` the commands that run for it; a test that does not apply to a config skips with a reason naming its sections. The Docker read test keeps the `--read-scenarios` default (the `config`, `valid_config` and `default` configs), so other configs listed under the read sections are read by `airbyte-cdk connector test` (what the monorepo runs for manifest-only connectors) but not by `airbyte-cdk image test` (what it runs for Python connectors). The Docker read test also no longer fails with an `AttributeError` when the scenario declares `empty_streams`. + +A config's `configured_catalog_path` is now honoured, as it was in CAT: the read covers the streams named in that catalog, each with the definition and JSON schema that `discover` reports, with every stream read in full-refresh mode whatever sync mode the catalog declares. Previously the read covered every discovered stream and ignored the catalog. A config read through several sections is read once: it keeps its catalog only when every read entry declares the same one, and otherwise (one entry declares none, or the entries declare different catalogs) every discovered stream is read. A stream named in the catalog that `discover` does not report fails the read test, as it did in CAT. The Docker read test now configures every stream in full-refresh mode too, instead of the first mode the stream supports. + +### `check` must report the declared status + +Previously only an exception or a non-zero exit failed the `check` tests, so a `check` that reported `status: FAILED` passed. Both test paths now assert the reported `CONNECTION_STATUS` against the `status` declared for the config under `connection`, with the same semantics as CAT: + +| `status` in `acceptance-test-config.yml` | What `check` must do | +|---|---| +| `succeed` | Report a `CONNECTION_STATUS` message with status `SUCCEEDED`. | +| `failed` | Report a `CONNECTION_STATUS` message with status `FAILED` (gracefully, without raising). | +| `exception` | Raise: a `TRACE` error must be emitted and no `SUCCEEDED` status may be reported. A `CONNECTION_STATUS` message is not required. | +| not declared | Treated as `succeed`, matching the CAT default. | + +The `succeed` default applies to every command a config runs, not only `check`: for a config without a `status`, `discover` and `read` errors, and a `read` without records, now fail the test (previously they were tolerated unless `status: succeed` was declared explicitly). + +Migration steps: + +- A config listed under `spec` must validate against the connector's spec: remove the keys the spec no longer declares, or declare them in the spec. +- A config listed under `connection` without a `status` must now pass `check`. If the config is meant to fail `check`, declare `status: failed`; if its secret has expired, refresh it; if it is no longer used, remove it. +- A `check` that raises (uncaught exception, non-zero exit) now fails the test for every config that is not declared `status: failed` or `status: exception`, on the in-process path as well as the Docker path. +- A `status: failed` config whose `check` raises instead of reporting `FAILED` must either be fixed to report its failure gracefully, or be declared as `status: exception` if raising is the intended behaviour. +- A `status: exception` config whose `check` reports a `CONNECTION_STATUS` without raising must be declared as `status: failed` (or `succeed`). +- A config listed under `discovery`, `basic_read` or `full_refresh` must make that command succeed: `discover` must return streams, and a full-refresh `read` must return records. List the streams that legitimately return nothing, or that the config's credentials cannot read, under `empty_streams` (with a `bypass_reason`); move a config whose read is expected to fail out of the read sections, as the read tests carry no failure expectation. +- A config's `configured_catalog_path` now decides which streams are read. Remove the streams that the connector no longer discovers from the catalog. To read every discovered stream instead, drop `configured_catalog_path` from the config's read entries; to skip a stream, list it under `empty_streams`. +- Configs listed only under `discovery` or `full_refresh` were never tested before, so expect new failures from expired secrets and stale configs there. Remove the configs that are no longer needed. +- Python code that inspects `ExpectedOutcome` directly: `ExpectedOutcome.from_status_str("exception")` now returns the new member `EXPECT_UNCAUGHT_ERROR` instead of `EXPECT_EXCEPTION`. `expect_exception()` still returns `True` for both members; use `expect_uncaught_error()` to distinguish them. + +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. ### New mandatory checks in `test_discover` diff --git a/unit_tests/resources/source_pokeapi_w_components_py/acceptance-test-config.yml b/unit_tests/resources/source_pokeapi_w_components_py/acceptance-test-config.yml index e707a9099c..283898b359 100644 --- a/unit_tests/resources/source_pokeapi_w_components_py/acceptance-test-config.yml +++ b/unit_tests/resources/source_pokeapi_w_components_py/acceptance-test-config.yml @@ -11,6 +11,9 @@ acceptance_tests: tests: - config_path: "valid_config.yaml" status: "succeed" + # Fails spec validation (`pokemon_name` is not in the enum), so `check` must report FAILED. + - config_path: "integration_tests/invalid_config.json" + status: "failed" discovery: tests: - config_path: "valid_config.yaml" diff --git a/unit_tests/resources/source_pokeapi_w_components_py/integration_tests/configured_catalog.json b/unit_tests/resources/source_pokeapi_w_components_py/integration_tests/configured_catalog.json new file mode 100644 index 0000000000..81e82dbd2f --- /dev/null +++ b/unit_tests/resources/source_pokeapi_w_components_py/integration_tests/configured_catalog.json @@ -0,0 +1,13 @@ +{ + "streams": [ + { + "stream": { + "name": "pokemon", + "json_schema": {}, + "supported_sync_modes": ["full_refresh"] + }, + "sync_mode": "full_refresh", + "destination_sync_mode": "overwrite" + } + ] +} diff --git a/unit_tests/resources/source_pokeapi_w_components_py/integration_tests/invalid_config.json b/unit_tests/resources/source_pokeapi_w_components_py/integration_tests/invalid_config.json new file mode 100644 index 0000000000..e0daf447a7 --- /dev/null +++ b/unit_tests/resources/source_pokeapi_w_components_py/integration_tests/invalid_config.json @@ -0,0 +1,3 @@ +{ + "pokemon_name": "not-a-pokemon" +} diff --git a/unit_tests/test/test_check_assertions.py b/unit_tests/test/test_check_assertions.py new file mode 100644 index 0000000000..a1a50c433b --- /dev/null +++ b/unit_tests/test/test_check_assertions.py @@ -0,0 +1,586 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Unit tests for the shared `check` assertion used by the Standard Tests. + +The assertion is exercised directly (as the Docker-based suite calls it), through the +in-process job runner, and through the `test_check` methods of the test-suite base classes, +since all paths must enforce the same expectations: + +- `status: succeed` -> a CONNECTION_STATUS message with status SUCCEEDED. +- `status: failed` -> a CONNECTION_STATUS message with status FAILED. +- `status: exception` -> an uncaught error (TRACE error), no SUCCEEDED status. +- no `status` -> treated as `succeed` (the CAT default). +""" + +from __future__ import annotations + +import json +import logging +import shutil +from pathlib import Path +from typing import Any, Iterable, Literal, Mapping + +import pytest + +from airbyte_cdk.models import ( + AirbyteCatalog, + AirbyteConnectionStatus, + AirbyteMessage, + AirbyteStateMessage, + ConfiguredAirbyteCatalog, + ConnectorSpecification, + FailureType, + Status, +) +from airbyte_cdk.sources import Source +from airbyte_cdk.test.entrypoint_wrapper import AirbyteEntrypointException, EntrypointOutput +from airbyte_cdk.test.models import ConnectorTestScenario, ExpectedOutcome +from airbyte_cdk.test.standard_tests import ConnectorTestSuiteBase, SourceTestSuiteBase +from airbyte_cdk.test.standard_tests._assertions import assert_check_outcome +from airbyte_cdk.test.standard_tests._job_runner import IConnector, run_test_job +from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite +from airbyte_cdk.utils.traced_exception import AirbyteTracedException + +POKEAPI_RESOURCE_DIR = Path(__file__).parent.parent / "resources" / "source_pokeapi_w_components_py" + +# Substrings of the assertion messages, so that each failing case fails for the expected reason. +MSG_NO_STATUS = "emitted no CONNECTION_STATUS message" +MSG_NOT_SUCCEEDED = "did not succeed" +MSG_NOT_FAILED = "was expected to fail, but reported" +MSG_NO_TRACE = "emitted no TRACE error" +MSG_RAISE_BUT_SUCCEEDED = "was expected to raise .*, but reported" + +STATUS_STR_BY_OUTCOME: dict[ExpectedOutcome, Literal["succeed", "failed", "exception"] | None] = { + ExpectedOutcome.EXPECT_SUCCESS: "succeed", + ExpectedOutcome.EXPECT_EXCEPTION: "failed", + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR: "exception", + ExpectedOutcome.ALLOW_ANY: None, +} + + +def _check_output(*statuses: str, error: bool = False) -> EntrypointOutput: + """Build an EntrypointOutput with one CONNECTION_STATUS message per status. + + When `error` is set, a TRACE error message is appended, as the entrypoint emits for an + uncaught exception. + """ + messages = [ + json.dumps({"type": "CONNECTION_STATUS", "connectionStatus": {"status": status}}) + for status in statuses + ] + if error: + messages.append( + json.dumps( + { + "type": "TRACE", + "trace": { + "type": "ERROR", + "emitted_at": 0, + "error": {"message": "Uncaught error during check."}, + }, + } + ) + ) + return EntrypointOutput(messages=messages, command=["docker", "run", "..."]) + + +def _assert_outcome( + check_result: EntrypointOutput, + expected_outcome: ExpectedOutcome, + failure_match: str | None, +) -> None: + """Call `assert_check_outcome`, expecting it to pass or to fail with `failure_match`.""" + if failure_match is None: + assert_check_outcome( + check_result=check_result, + expected_outcome=expected_outcome, + connector_name="source-test", + ) + return + + with pytest.raises(AssertionError, match=failure_match): + assert_check_outcome( + check_result=check_result, + expected_outcome=expected_outcome, + connector_name="source-test", + ) + + +# (expected outcome, reported status, TRACE error present, expected failure message or None) +OUTCOME_MATRIX = [ + # `status: succeed`: only a reported SUCCEEDED passes. + pytest.param(ExpectedOutcome.EXPECT_SUCCESS, "SUCCEEDED", False, None, id="success_succeeded"), + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, "FAILED", False, MSG_NOT_SUCCEEDED, id="success_failed" + ), + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, None, False, MSG_NO_STATUS, id="success_no_status" + ), + pytest.param(ExpectedOutcome.EXPECT_SUCCESS, None, True, MSG_NO_STATUS, id="success_raised"), + # `status: failed`: only a reported FAILED passes. A `check` that raises without reporting + # a status is not a graceful failure; `status: exception` exists for that outcome. + pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, "FAILED", False, None, id="failure_failed"), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, "FAILED", True, None, id="failure_failed_with_trace" + ), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, "SUCCEEDED", False, MSG_NOT_FAILED, id="failure_succeeded" + ), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, None, False, MSG_NO_STATUS, id="failure_no_status" + ), + pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, None, True, MSG_NO_STATUS, id="failure_raised"), + # `status: exception`: a TRACE error is required and no SUCCEEDED may be reported. A FAILED + # status next to the trace (e.g. a `config_error`) is accepted. + pytest.param(ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, None, True, None, id="exception_raised"), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, "FAILED", True, None, id="exception_raised_failed" + ), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + "SUCCEEDED", + True, + MSG_RAISE_BUT_SUCCEEDED, + id="exception_raised_succeeded", + ), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + "SUCCEEDED", + False, + MSG_NO_TRACE, + id="exception_succeeded", + ), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, "FAILED", False, MSG_NO_TRACE, id="exception_failed" + ), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, None, False, MSG_NO_TRACE, id="exception_no_status" + ), + # No declared status: treated as `succeed`, matching the CAT default. + pytest.param(ExpectedOutcome.ALLOW_ANY, "SUCCEEDED", False, None, id="allow_any_succeeded"), + pytest.param( + ExpectedOutcome.ALLOW_ANY, "FAILED", False, MSG_NOT_SUCCEEDED, id="allow_any_failed" + ), + pytest.param(ExpectedOutcome.ALLOW_ANY, None, False, MSG_NO_STATUS, id="allow_any_no_status"), + pytest.param(ExpectedOutcome.ALLOW_ANY, None, True, MSG_NO_STATUS, id="allow_any_raised"), +] + + +@pytest.mark.parametrize("expected_outcome, status, error, failure_match", OUTCOME_MATRIX) +def test_assert_check_outcome( + expected_outcome: ExpectedOutcome, + status: str | None, + error: bool, + failure_match: str | None, +) -> None: + check_result = _check_output(*([status] if status else []), error=error) + _assert_outcome(check_result, expected_outcome, failure_match) + + +def test_assert_check_outcome_no_status_message_includes_errors_and_logs() -> None: + """The 'no status' failure must show why, including the trace error and the logs.""" + check_result = EntrypointOutput( + messages=[ + json.dumps({"type": "LOG", "log": {"level": "INFO", "message": "Starting check..."}}), + json.dumps( + { + "type": "TRACE", + "trace": { + "type": "ERROR", + "emitted_at": 0, + "error": {"message": "Connection refused by host"}, + }, + } + ), + ], + command=["docker", "run", "..."], + ) + with pytest.raises(AssertionError) as exc_info: + assert_check_outcome( + check_result=check_result, + expected_outcome=ExpectedOutcome.EXPECT_SUCCESS, + connector_name="source-test", + ) + message = str(exc_info.value) + assert "Connection refused by host" in message + assert "Starting check..." in message + assert "status: exception" in message + + +@pytest.mark.parametrize( + "expected_outcome, statuses, failure_match", + [ + # The last CONNECTION_STATUS message wins. + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, + ["FAILED", "SUCCEEDED"], + None, + id="success_last_status_wins", + ), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, + ["FAILED", "SUCCEEDED"], + MSG_NOT_FAILED, + id="failure_last_status_wins", + ), + ], +) +def test_assert_check_outcome_uses_last_status( + expected_outcome: ExpectedOutcome, + statuses: list[str], + failure_match: str | None, +) -> None: + _assert_outcome(_check_output(*statuses), expected_outcome, failure_match) + + +class _FakeSource(Source): + """A source whose `check` behaves as instructed. + + Behaviours: + - `succeeded` / `failed`: report that CONNECTION_STATUS. + - `raise`: raise a plain exception (uncaught; TRACE error, no status). + - `raise_config_error`: raise an `AirbyteTracedException` with `config_error`, which the + entrypoint turns into a TRACE error plus a FAILED status. + - `raise_system_error`: raise an `AirbyteTracedException` with `system_error`, which the + entrypoint re-raises after emitting the TRACE error (no status). + """ + + def __init__(self, behavior: str) -> None: + self._behavior = behavior + + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + return ConnectorSpecification( + connectionSpecification={"type": "object", "properties": {}}, + ) + + def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus: + if self._behavior == "raise": + raise RuntimeError("Uncaught error during check.") + if self._behavior == "raise_config_error": + raise AirbyteTracedException( + message="Invalid credentials.", + failure_type=FailureType.config_error, + ) + if self._behavior == "raise_system_error": + raise AirbyteTracedException( + message="Upstream is down.", + failure_type=FailureType.system_error, + ) + return AirbyteConnectionStatus(status=Status(self._behavior.upper())) + + def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCatalog: + return AirbyteCatalog(streams=[]) + + def read( + self, + logger: logging.Logger, + config: Mapping[str, Any], + catalog: ConfiguredAirbyteCatalog, + state: list[AirbyteStateMessage] | None = None, + ) -> Iterable[AirbyteMessage]: + yield from [] + + +# (expected outcome, `check` behaviour, expected exception type or None, failure message) +IN_PROCESS_MATRIX = [ + # `status: succeed` + pytest.param(ExpectedOutcome.EXPECT_SUCCESS, "succeeded", None, None, id="success_succeeded"), + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, + "failed", + AssertionError, + MSG_NOT_SUCCEEDED, + id="success_failed", + ), + # An uncaught error is surfaced as-is, as on the Docker path (`raise_if_errors=True`). + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, + "raise", + AirbyteEntrypointException, + "Uncaught error during check", + id="success_raised", + ), + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, + "raise_config_error", + AirbyteEntrypointException, + "Invalid credentials", + id="success_raised_config_error", + ), + # `status: failed` + pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, "failed", None, None, id="failure_failed"), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, + "raise_config_error", + None, + None, + id="failure_raised_config_error", + ), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, + "succeeded", + AssertionError, + MSG_NOT_FAILED, + id="failure_succeeded", + ), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, + "raise", + AssertionError, + MSG_NO_STATUS, + id="failure_raised", + ), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, + "raise_system_error", + AssertionError, + MSG_NO_STATUS, + id="failure_raised_system_error", + ), + # `status: exception` + pytest.param(ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, "raise", None, None, id="exception_raised"), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + "raise_system_error", + None, + None, + id="exception_raised_system_error", + ), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + "raise_config_error", + None, + None, + id="exception_raised_config_error", + ), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + "succeeded", + AssertionError, + MSG_NO_TRACE, + id="exception_succeeded", + ), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + "failed", + AssertionError, + MSG_NO_TRACE, + id="exception_failed", + ), + # No declared status: treated as `succeed`. + pytest.param(ExpectedOutcome.ALLOW_ANY, "succeeded", None, None, id="allow_any_succeeded"), + pytest.param( + ExpectedOutcome.ALLOW_ANY, + "failed", + AssertionError, + MSG_NOT_SUCCEEDED, + id="allow_any_failed", + ), + pytest.param( + ExpectedOutcome.ALLOW_ANY, + "raise", + AirbyteEntrypointException, + "Uncaught error during check", + id="allow_any_raised", + ), +] + + +def _scenario_for(expected_outcome: ExpectedOutcome) -> ConnectorTestScenario: + return ConnectorTestScenario( + config_dict={"dummy_setting": "dummy_value"}, + status=STATUS_STR_BY_OUTCOME[expected_outcome], + ) + + +@pytest.mark.parametrize( + "expected_outcome, behavior, expected_exception, failure_match", IN_PROCESS_MATRIX +) +def test_run_test_job_check_asserts_reported_status( + expected_outcome: ExpectedOutcome, + behavior: str, + expected_exception: type[Exception] | None, + failure_match: str | None, + tmp_path: Path, +) -> None: + """The in-process path must enforce the same expectations as the Docker path.""" + scenario = _scenario_for(expected_outcome) + source = _FakeSource(behavior) + + if expected_exception is None: + run_test_job(source, "check", connector_root=tmp_path, test_scenario=scenario) + return + + with pytest.raises(expected_exception, match=failure_match): + run_test_job(source, "check", connector_root=tmp_path, test_scenario=scenario) + + +class _SuiteWithFakeConnector: + """Mixin that points a test-suite base class at a fake connector and a temp root dir.""" + + connector_root_dir: Path + connector_behavior: str + + @classmethod + def get_connector_root_dir(cls) -> Path: + return cls.connector_root_dir + + @classmethod + def create_connector(cls, scenario: ConnectorTestScenario | None) -> IConnector: + return _FakeSource(cls.connector_behavior) + + +class _FakeSourceSuite(_SuiteWithFakeConnector, SourceTestSuiteBase): + pass + + +class _FakeConnectorSuite(_SuiteWithFakeConnector, ConnectorTestSuiteBase): + pass + + +@pytest.mark.parametrize("suite_class", [_FakeSourceSuite, _FakeConnectorSuite]) +@pytest.mark.parametrize( + "expected_outcome, behavior, expected_exception, failure_match", IN_PROCESS_MATRIX +) +def test_suite_test_check_asserts_reported_status( + suite_class: type[_SuiteWithFakeConnector], + expected_outcome: ExpectedOutcome, + behavior: str, + expected_exception: type[Exception] | None, + failure_match: str | None, + tmp_path: Path, +) -> None: + """`test_check` of the suite base classes must enforce the shared expectations too. + + In particular, a `status: exception` scenario must not be failed by the "exactly one + CONNECTION_STATUS message" assertion that follows the job run. + """ + suite_class.connector_root_dir = tmp_path + suite_class.connector_behavior = behavior + scenario = _scenario_for(expected_outcome) + suite = suite_class() + + if expected_exception is None: + suite.test_check(scenario) # type: ignore[attr-defined] + return + + with pytest.raises(expected_exception, match=failure_match): + suite.test_check(scenario) # type: ignore[attr-defined] + + +# (expected outcome, container output, expected exception type or None, failure message) +DOCKER_MATRIX = [ + pytest.param(ExpectedOutcome.EXPECT_SUCCESS, "SUCCEEDED", False, None, None, id="success"), + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, + "FAILED", + False, + AssertionError, + MSG_NOT_SUCCEEDED, + id="success_failed", + ), + # The container raised: `raise_if_errors` surfaces it before the status assertion. + pytest.param( + ExpectedOutcome.EXPECT_SUCCESS, + None, + True, + AirbyteEntrypointException, + "Uncaught error during check", + id="success_raised", + ), + pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, "FAILED", False, None, None, id="failure"), + pytest.param( + ExpectedOutcome.EXPECT_EXCEPTION, + None, + True, + AssertionError, + MSG_NO_STATUS, + id="failure_raised", + ), + pytest.param(ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, None, True, None, None, id="exception"), + pytest.param( + ExpectedOutcome.EXPECT_UNCAUGHT_ERROR, + "SUCCEEDED", + False, + AssertionError, + MSG_NO_TRACE, + id="exception_succeeded", + ), + pytest.param( + ExpectedOutcome.ALLOW_ANY, + "FAILED", + False, + AssertionError, + MSG_NOT_SUCCEEDED, + id="allow_any_failed", + ), + pytest.param( + ExpectedOutcome.ALLOW_ANY, + None, + False, + AssertionError, + MSG_NO_STATUS, + id="allow_any_no_status", + ), +] + + +@pytest.mark.parametrize( + "expected_outcome, status, error, expected_exception, failure_match", DOCKER_MATRIX +) +def test_docker_image_build_and_check_asserts_reported_status( + expected_outcome: ExpectedOutcome, + status: str | None, + error: bool, + expected_exception: type[Exception] | None, + failure_match: str | None, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The Docker-path `check` test must assert the reported status without needing Docker. + + `run_docker_airbyte_command` is replaced by a stub returning canned container output, and + the image build is skipped via `connector_image_override`. + """ + shutil.copy(POKEAPI_RESOURCE_DIR / "metadata.yaml", tmp_path / "metadata.yaml") + recorded_calls: list[dict[str, Any]] = [] + + def _fake_run_docker_airbyte_command( + cmd: list[str], + *, + raise_if_errors: bool = False, + ) -> EntrypointOutput: + recorded_calls.append({"cmd": cmd, "raise_if_errors": raise_if_errors}) + result = _check_output(*([status] if status else []), error=error) + if raise_if_errors: + result.raise_if_errors() + return result + + monkeypatch.setattr( + "airbyte_cdk.test.standard_tests.docker_base.run_docker_airbyte_command", + _fake_run_docker_airbyte_command, + ) + + class _Suite(DockerConnectorTestSuite): + @classmethod + def get_connector_root_dir(cls) -> Path: + return tmp_path + + scenario = _scenario_for(expected_outcome) + + def _run() -> None: + _Suite().test_docker_image_build_and_check( + scenario, + connector_image_override="airbyte/source-test:dev", + connector_base_image_override=None, + ) + + if expected_exception is None: + _run() + else: + with pytest.raises(expected_exception, match=failure_match): + _run() + + assert len(recorded_calls) == 1 + assert recorded_calls[0]["cmd"][-4:-1] == ["airbyte/source-test:dev", "check", "--config"] + # Only `failed` and `exception` scenarios may let a trace error through to the assertion. + assert recorded_calls[0]["raise_if_errors"] is (not expected_outcome.expect_exception()) diff --git a/unit_tests/test/test_docker_base_check_outcome.py b/unit_tests/test/test_docker_base_check_outcome.py deleted file mode 100644 index 8874881e88..0000000000 --- a/unit_tests/test/test_docker_base_check_outcome.py +++ /dev/null @@ -1,76 +0,0 @@ -# Copyright (c) 2025 Airbyte, Inc., all rights reserved. -"""Unit tests for `_assert_check_outcome` in `airbyte_cdk.test.standard_tests.docker_base`.""" - -import json - -import pytest - -from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput -from airbyte_cdk.test.models import ExpectedOutcome -from airbyte_cdk.test.standard_tests.docker_base import _assert_check_outcome - - -def _check_output(*statuses: str) -> EntrypointOutput: - """Build an EntrypointOutput with one CONNECTION_STATUS message per given status.""" - return EntrypointOutput( - messages=[ - json.dumps({"type": "CONNECTION_STATUS", "connectionStatus": {"status": status}}) - for status in statuses - ], - command=["docker", "run", "..."], - ) - - -@pytest.mark.parametrize( - "expected_outcome, statuses, should_pass", - [ - # Scenarios expecting success: only SUCCEEDED passes. - pytest.param(ExpectedOutcome.EXPECT_SUCCESS, ["SUCCEEDED"], True, id="success_succeeded"), - pytest.param(ExpectedOutcome.EXPECT_SUCCESS, ["FAILED"], False, id="success_failed"), - pytest.param(ExpectedOutcome.EXPECT_SUCCESS, [], False, id="success_no_status"), - # Scenarios expecting failure: a SUCCEEDED status is a regression and must fail; - # a FAILED status or no status at all (uncaught error) both count as failing as expected. - pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, ["FAILED"], True, id="failure_failed"), - pytest.param( - ExpectedOutcome.EXPECT_EXCEPTION, ["SUCCEEDED"], False, id="failure_succeeded" - ), - pytest.param(ExpectedOutcome.EXPECT_EXCEPTION, [], True, id="failure_no_status"), - # ALLOW_ANY scenarios: either reported outcome passes, but `check` must still - # emit a CONNECTION_STATUS message. - pytest.param(ExpectedOutcome.ALLOW_ANY, ["SUCCEEDED"], True, id="allow_any_succeeded"), - pytest.param(ExpectedOutcome.ALLOW_ANY, ["FAILED"], True, id="allow_any_failed"), - pytest.param(ExpectedOutcome.ALLOW_ANY, [], False, id="allow_any_no_status"), - # The last CONNECTION_STATUS message wins. - pytest.param( - ExpectedOutcome.EXPECT_SUCCESS, - ["FAILED", "SUCCEEDED"], - True, - id="success_last_status_wins", - ), - pytest.param( - ExpectedOutcome.EXPECT_EXCEPTION, - ["FAILED", "SUCCEEDED"], - False, - id="failure_last_status_wins", - ), - ], -) -def test_assert_check_outcome( - expected_outcome: ExpectedOutcome, - statuses: list[str], - should_pass: bool, -) -> None: - check_result = _check_output(*statuses) - if should_pass: - _assert_check_outcome( - check_result=check_result, - expected_outcome=expected_outcome, - connector_name="source-test", - ) - else: - with pytest.raises(AssertionError): - _assert_check_outcome( - check_result=check_result, - expected_outcome=expected_outcome, - connector_name="source-test", - ) diff --git a/unit_tests/test/test_read_assertions.py b/unit_tests/test/test_read_assertions.py new file mode 100644 index 0000000000..89fb820ddd --- /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] diff --git a/unit_tests/test/test_scenario_sections.py b/unit_tests/test/test_scenario_sections.py new file mode 100644 index 0000000000..af3b23da41 --- /dev/null +++ b/unit_tests/test/test_scenario_sections.py @@ -0,0 +1,957 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Each config in `acceptance-test-config.yml` runs the command of the section that lists it. + +`spec` validates the config against the spec, `connection` runs `check`, `discovery` runs +`discover`, and `basic_read` and `full_refresh` run a full-refresh `read`. The `incremental` +section is not collected. A config listed under several sections runs the union of their commands. +""" + +from __future__ import annotations + +import json +import logging +import shutil +from pathlib import Path +from typing import Any, Callable, Iterable, Mapping, TypeVar, cast + +import pytest +import yaml +from pydantic import ValidationError + +from airbyte_cdk.models import ( + AirbyteCatalog, + AirbyteConnectionStatus, + AirbyteMessage, + AirbyteRecordMessage, + AirbyteStateMessage, + AirbyteStream, + ConfiguredAirbyteCatalog, + ConnectorSpecification, + Status, + SyncMode, + Type, +) +from airbyte_cdk.sources import Source +from airbyte_cdk.test.entrypoint_wrapper import AirbyteEntrypointException, EntrypointOutput +from airbyte_cdk.test.models.scenario import ( + ALL_COMMANDS, + SCENARIO_SECTIONS, + SECTION_COMMANDS, + ConnectorTestScenario, + ScenarioCommand, +) +from airbyte_cdk.test.standard_tests import ConnectorTestSuiteBase, SourceTestSuiteBase +from airbyte_cdk.test.standard_tests._assertions import assert_config_matches_spec +from airbyte_cdk.test.standard_tests._job_runner import IConnector +from airbyte_cdk.test.standard_tests.docker_base import ( + DockerConnectorTestSuite, + skip_unless_runs, +) +from airbyte_cdk.utils.connector_paths import ACCEPTANCE_TEST_CONFIG + +POKEAPI_RESOURCE_DIR = Path(__file__).parent.parent / "resources" / "source_pokeapi_w_components_py" +IMAGE = "airbyte/source-test:dev" + +SPEC_WITH_LONG_API_KEY: dict[str, Any] = { + "type": "object", + "required": ["api_key"], + "properties": {"api_key": {"type": "string", "minLength": 20}}, +} +VALID_CONFIG = {"api_key": "x" * 20} +# Too short for the spec: validation fails on the value itself, which must not leak. +SECRET_CONFIG = {"api_key": "hunter2-secret"} + +USERS = AirbyteStream( + name="users", + json_schema={"type": "object"}, + supported_sync_modes=[SyncMode.full_refresh, SyncMode.incremental], + source_defined_cursor=True, + default_cursor_field=["updated_at"], +) +# Incremental with a user-defined cursor: the catalog cannot know it, so it is read full refresh. +EVENTS = AirbyteStream( + name="events", + json_schema={"type": "object"}, + supported_sync_modes=[SyncMode.full_refresh, SyncMode.incremental], +) +STATIC = AirbyteStream( + name="static", + json_schema={"type": "object"}, + supported_sync_modes=[SyncMode.full_refresh], +) + + +class _Reached(Exception): + """Raised by a fake connector or Docker command to prove a test body ran past its skips.""" + + +_SuiteT = TypeVar("_SuiteT", bound=DockerConnectorTestSuite) + + +def _suite_class( + tmp_path: Path, + acceptance_tests: dict[str, Any], + *, + base: type[_SuiteT], + make_connector: Callable[[], IConnector] | None = None, +) -> type[_SuiteT]: + """Build a suite class over a temporary connector dir holding `acceptance_tests`. + + `create_connector` returns `make_connector()`, or raises `_Reached` when none is given. + """ + connector_root = tmp_path / "source-test" + connector_root.mkdir(exist_ok=True) + (connector_root / ACCEPTANCE_TEST_CONFIG).write_text( + yaml.safe_dump({"acceptance_tests": acceptance_tests}) + ) + shutil.copy(POKEAPI_RESOURCE_DIR / "metadata.yaml", connector_root / "metadata.yaml") + + def get_connector_root_dir(cls: type) -> Path: + return connector_root + + def create_connector(self: Any, scenario: ConnectorTestScenario | None) -> IConnector: + if make_connector is None: + raise _Reached(scenario.id if scenario else None) + return make_connector() + + return cast( + type[_SuiteT], + type( + "_Suite", + (base,), + { + "get_connector_root_dir": classmethod(get_connector_root_dir), + "create_connector": create_connector, + }, + ), + ) + + +def _scenario( + *sections: str, + config: dict[str, Any] | None = None, + name: str = "config", + **extra: Any, +) -> ConnectorTestScenario: + """A scenario over an inline config, listed under `sections` (none: built by hand). + + The config defaults to `VALID_CONFIG`: an empty config dict counts as no config at all. + """ + return ConnectorTestScenario( + config_path=Path(f"secrets/{name}.json"), + config_dict=VALID_CONFIG if config is None else config, + sections=tuple(sections), + **extra, + ) + + +def _entry(config_path: str, **extra: Any) -> dict[str, Any]: + return {"config_path": config_path, **extra} + + +ALL_SECTIONS_CONFIG: dict[str, Any] = { + "spec": {"tests": [_entry("secrets/config.json")]}, + "connection": { + "tests": [ + _entry("secrets/config.json", status="succeed"), + _entry("integration_tests/invalid_config.json", status="failed"), + _entry("secrets/config_iam_role.json", status="succeed"), + {"status": "succeed"}, # no config_path: ignored + ] + }, + "discovery": { + "tests": [ + _entry( + "secrets/config.json", + backward_compatibility_tests_config={"disable_for_version": "1.0.0"}, + ), + _entry("secrets/config_old.json"), + ] + }, + "basic_read": { + "tests": [ + _entry( + "secrets/config.json", + empty_streams=[{"name": "empty_stream", "bypass_reason": "no data"}], + ), + _entry("secrets/config_read_only.json"), + ] + }, + "full_refresh": {"tests": [_entry("secrets/config_full_refresh.json")]}, + "incremental": { + "tests": [ + _entry( + "secrets/config_incremental.json", + configured_catalog_path="integration_tests/incremental_catalog.json", + future_state={"future_state_path": "integration_tests/abnormal_state.json"}, + ), + _entry("secrets/config_read_only.json"), + ] + }, +} + + +# --- Section to command mapping ------------------------------------------------------------- + + +def test_every_section_maps_to_the_command_cat_ran_it_with() -> None: + assert dict(SECTION_COMMANDS) == { + "spec": {"spec"}, + "connection": {"check"}, + "discovery": {"discover"}, + "basic_read": {"read"}, + "full_refresh": {"read"}, + } + assert SCENARIO_SECTIONS == tuple(SECTION_COMMANDS) + assert ALL_COMMANDS == {"spec", "check", "discover", "read"} + + +def test_get_scenarios_collects_every_section(tmp_path: Path) -> None: + suite = _suite_class(tmp_path, ALL_SECTIONS_CONFIG, base=DockerConnectorTestSuite) + scenarios = {str(s.config_path): s for s in suite.get_scenarios()} + + assert sorted(scenarios) == [ + "integration_tests/invalid_config.json", + "secrets/config.json", + "secrets/config_full_refresh.json", + "secrets/config_old.json", + "secrets/config_read_only.json", + ], ( + "iam_role configs, entries without config_path and the `incremental` section are " + "skipped; the rest are deduplicated" + ) + + full = scenarios["secrets/config.json"] + # Sections are recorded in collection order (`SCENARIO_SECTIONS`), not file order. + assert full.sections == ("spec", "connection", "discovery", "basic_read") + assert full.commands == {"spec", "check", "discover", "read"} + assert full.status == "succeed" + assert [s.name for s in full.empty_streams or []] == ["empty_stream"] + + invalid = scenarios["integration_tests/invalid_config.json"] + assert invalid.sections == ("connection",) + assert invalid.commands == {"check"} + assert invalid.status == "failed" + + read_only = scenarios["secrets/config_read_only.json"] + assert read_only.sections == ("basic_read",), "its `incremental` entry is ignored" + assert read_only.commands == {"read"} + + for config_path, section, command in [ + ("secrets/config_old.json", "discovery", "discover"), + ("secrets/config_full_refresh.json", "full_refresh", "read"), + ]: + scenario = scenarios[config_path] + assert scenario.sections == (section,) + assert scenario.commands == {command} + assert scenario.status is None, "no section but `connection` declares a status" + + +def test_dedup_scenarios_unions_sections() -> None: + deduped = DockerConnectorTestSuite._dedup_scenarios( + [ + ConnectorTestScenario(config_path=Path("secrets/config.json"), sections=("discovery",)), + ConnectorTestScenario( + config_path=Path("secrets/config.json"), sections=("connection",) + ), + ConnectorTestScenario(config_path=Path("secrets/config.json"), sections=("discovery",)), + ] + ) + assert len(deduped) == 1 + assert deduped[0].sections == ("discovery", "connection") + assert deduped[0].commands == {"discover", "check"} + + +CATALOG_A = Path("integration_tests/configured_catalog.json") +CATALOG_B = Path("integration_tests/full_refresh_catalog.json") + + +@pytest.mark.parametrize( + "entries, expected", + [ + pytest.param( + [("basic_read", CATALOG_A), ("full_refresh", CATALOG_A)], + CATALOG_A, + id="same_catalog_everywhere", + ), + pytest.param([("full_refresh", CATALOG_A)], CATALOG_A, id="single_read_entry"), + pytest.param( + [("spec", None), ("connection", None), ("basic_read", CATALOG_A)], + CATALOG_A, + id="non_read_entries_do_not_count", + ), + pytest.param( + [("basic_read", None), ("full_refresh", CATALOG_A)], + None, + id="one_read_entry_without_catalog_reads_everything", + ), + pytest.param( + [("full_refresh", CATALOG_A), ("basic_read", None)], + None, + id="order_does_not_matter", + ), + pytest.param( + [("basic_read", CATALOG_A), ("full_refresh", CATALOG_B)], + None, + id="different_catalogs_read_everything", + ), + pytest.param([("connection", CATALOG_A)], CATALOG_A, id="not_read_left_alone"), + ], +) +def test_dedup_scenarios_merges_the_read_catalogs( + entries: list[tuple[str, Path | None]], expected: Path | None +) -> None: + deduped = DockerConnectorTestSuite._dedup_scenarios( + [ + ConnectorTestScenario( + config_path=Path("secrets/config.json"), + configured_catalog_path=catalog, + sections=(section,), + ) + for section, catalog in entries + ] + ) + assert len(deduped) == 1 + assert deduped[0].configured_catalog_path == expected + + +@pytest.mark.parametrize( + "sections, expected", + [ + pytest.param((), ALL_COMMANDS, id="hand_built_scenario_runs_everything"), + pytest.param(("spec",), {"spec"}, id="spec"), + pytest.param(("connection",), {"check"}, id="connection"), + pytest.param(("discovery",), {"discover"}, id="discovery"), + pytest.param(("basic_read",), {"read"}, id="basic_read"), + pytest.param(("full_refresh",), {"read"}, id="full_refresh"), + pytest.param(("discovery", "basic_read"), {"discover", "read"}, id="mixed"), + pytest.param(SCENARIO_SECTIONS, ALL_COMMANDS, id="every_section"), + ], +) +def test_commands_property(sections: tuple[str, ...], expected: set[ScenarioCommand]) -> None: + scenario = _scenario(*sections) + assert scenario.commands == expected + for command in ALL_COMMANDS: + assert scenario.runs(command) is (command in expected) + assert scenario.runs(*ALL_COMMANDS) + # Derived scenarios keep the sections, so the expectation helpers cannot change the commands. + for derived in ( + scenario.without_expected_outcome(), + scenario.with_expecting_failure(), + scenario.with_expecting_success(), + scenario.with_default_success(), + ): + assert derived.sections == scenario.sections + assert derived.commands == expected + + +@pytest.mark.parametrize("section", ["sequential_reads", "incremental"]) +def test_unknown_section_is_rejected(section: str) -> None: + with pytest.raises(ValidationError, match="Unknown `acceptance-test-config.yml` section"): + _scenario(section) + + +@pytest.mark.parametrize( + "status, expected", + [ + pytest.param(None, "succeed", id="no_status_defaults_to_succeed"), + pytest.param("succeed", "succeed", id="succeed"), + pytest.param("failed", "failed", id="failed"), + pytest.param("exception", "exception", id="exception"), + ], +) +def test_with_default_success(status: str | None, expected: str) -> None: + scenario = _scenario("discovery", status=status) + assert scenario.with_default_success().status == expected + assert scenario.with_default_success().expected_outcome.expect_success() is ( + expected == "succeed" + ) + + +def test_skip_unless_runs_names_the_commands_and_sections() -> None: + discovery_only = _scenario("discovery", name="config_old") + + with pytest.raises(pytest.skip.Exception) as exc_info: + skip_unless_runs(discovery_only, "read") + assert str(exc_info.value) == ( + "`read` does not run for scenario 'config_old': its config is listed under " + "`discovery` in `acceptance-test-config.yml`, which runs `discover`." + ) + + skip_unless_runs(discovery_only, "read", "discover") # no skip + for command in ALL_COMMANDS: + skip_unless_runs(_scenario(), command) # hand-built scenarios never skip + + +# --- Which test runs for which section -------------------------------------------------------- + +SUITE_TEST_COMMANDS: dict[str, frozenset[ScenarioCommand]] = { + "test_check": frozenset({"check"}), + "test_config_matches_spec": frozenset({"spec"}), + "test_discover": frozenset({"discover"}), + "test_basic_read": frozenset({"read"}), + "test_fail_read_with_bad_catalog": frozenset({"read"}), +} + + +@pytest.mark.parametrize("section", SCENARIO_SECTIONS) +@pytest.mark.parametrize( + "test_name, commands", SUITE_TEST_COMMANDS.items(), ids=SUITE_TEST_COMMANDS +) +def test_source_suite_runs_each_test_for_its_sections( + tmp_path: Path, test_name: str, commands: frozenset[ScenarioCommand], section: str +) -> None: + suite = _suite_class(tmp_path, ALL_SECTIONS_CONFIG, base=SourceTestSuiteBase)() + scenario = _scenario(section) + + if SECTION_COMMANDS[section] & commands: + with pytest.raises(_Reached): + getattr(suite, test_name)(scenario) + else: + with pytest.raises(pytest.skip.Exception, match=r"do(es)? not run for scenario"): + getattr(suite, test_name)(scenario) + + +@pytest.mark.parametrize("test_name", SUITE_TEST_COMMANDS) +def test_source_suite_runs_every_test_for_hand_built_scenarios( + tmp_path: Path, test_name: str +) -> None: + suite = _suite_class(tmp_path, ALL_SECTIONS_CONFIG, base=SourceTestSuiteBase)() + with pytest.raises(_Reached): + getattr(suite, test_name)(_scenario()) + + +@pytest.mark.parametrize("test_name", ["test_check", "test_config_matches_spec"]) +def test_connector_suite_gates_its_tests(tmp_path: Path, test_name: str) -> None: + suite = _suite_class(tmp_path, ALL_SECTIONS_CONFIG, base=ConnectorTestSuiteBase)() + with pytest.raises(_Reached): + getattr(suite, test_name)(_scenario("connection", "spec")) + with pytest.raises(pytest.skip.Exception, match=r"do(es)? not run for scenario"): + getattr(suite, test_name)(_scenario("discovery")) + + +DOCKER_TEST_COMMANDS: dict[str, frozenset[ScenarioCommand]] = { + "test_docker_image_build_and_check": frozenset({"check"}), + "test_docker_image_build_and_config_matches_spec": frozenset({"spec"}), + "test_docker_image_build_and_discover": frozenset({"discover"}), + "test_docker_image_build_and_read": frozenset({"read"}), +} + + +def _docker_kwargs(test_name: str) -> dict[str, Any]: + kwargs: dict[str, Any] = { + "connector_image_override": IMAGE, # skips the image build + "connector_base_image_override": None, + } + if test_name == "test_docker_image_build_and_read": + kwargs |= {"read_from_streams": "all", "read_scenarios": "all"} + return kwargs + + +@pytest.mark.parametrize("section", SCENARIO_SECTIONS) +@pytest.mark.parametrize( + "test_name, commands", DOCKER_TEST_COMMANDS.items(), ids=DOCKER_TEST_COMMANDS +) +def test_docker_suite_runs_each_test_for_its_sections( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + test_name: str, + commands: frozenset[ScenarioCommand], + section: str, +) -> None: + def _reached(cmd: list[str], *, raise_if_errors: bool = False) -> EntrypointOutput: + raise _Reached(cmd) + + monkeypatch.setattr( + "airbyte_cdk.test.standard_tests.docker_base.run_docker_airbyte_command", _reached + ) + suite = _suite_class(tmp_path, ALL_SECTIONS_CONFIG, base=DockerConnectorTestSuite)() + scenario = _scenario(section) + + if SECTION_COMMANDS[section] & commands: + with pytest.raises(_Reached): + getattr(suite, test_name)(scenario, **_docker_kwargs(test_name)) + else: + with pytest.raises(pytest.skip.Exception, match=r"do(es)? not run for scenario"): + getattr(suite, test_name)(scenario, **_docker_kwargs(test_name)) + + +# --- `spec` sections: the config must validate against the spec ------------------------------- + + +class _SpecSource(Source): + """A source whose spec requires a 20-character `api_key`; the other commands are inert.""" + + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + return ConnectorSpecification(connectionSpecification=SPEC_WITH_LONG_API_KEY) + + 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=[]) + + def read( + self, + logger: logging.Logger, + config: Mapping[str, Any], + catalog: ConfiguredAirbyteCatalog, + state: list[AirbyteStateMessage] | None = None, + ) -> Iterable[AirbyteMessage]: + yield from [] + + +@pytest.mark.parametrize("base", [SourceTestSuiteBase, ConnectorTestSuiteBase]) +def test_config_matches_spec_validates_the_config_without_leaking_it( + tmp_path: Path, base: type[ConnectorTestSuiteBase] +) -> None: + suite = _suite_class(tmp_path, {}, base=base, make_connector=_SpecSource)() + + suite.test_config_matches_spec(_scenario("spec", config=VALID_CONFIG)) + + with pytest.raises( + AssertionError, + match=r"Config 'config' does not match the spec of connector 'source-test' at `\$\.api_key`", + ) as exc_info: + suite.test_config_matches_spec(_scenario("spec", config=SECRET_CONFIG)) + assert "hunter2" not in str(exc_info.value), "the offending config value must be redacted" + assert " is too short" in str(exc_info.value) + + with pytest.raises(AssertionError, match="'api_key' is a required property"): + suite.test_config_matches_spec(_scenario("spec", config={})) + + +def test_assert_config_matches_spec_requires_a_spec_message() -> None: + with pytest.raises(AssertionError, match="(?i)spec"): + assert_config_matches_spec( + config=VALID_CONFIG, + spec_result=EntrypointOutput(messages=[], command=["spec"]), + connector_name="source-test", + scenario_id="config", + ) + + +# --- `discovery` and `basic_read` sections ------------------------------------------------------- + + +class _ReadSource(Source): + """A source over `streams` whose `read` behaves as instructed. + + Behaviours: + - `records`: one record per configured stream. + - `no_records`: nothing at all. + - `raise`: an uncaught error during `read`. + - `discover_raises`: an uncaught error during `discover`. + + Every configured catalog passed to `read` is appended to `read_catalogs`. + """ + + def __init__( + self, + behavior: str, + streams: list[AirbyteStream], + read_catalogs: list[ConfiguredAirbyteCatalog], + ) -> None: + self._behavior = behavior + self._streams = streams + self._read_catalogs = read_catalogs + + 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: + if self._behavior == "discover_raises": + raise RuntimeError("Discover exploded.") + return AirbyteCatalog(streams=list(self._streams)) + + def read( + self, + logger: logging.Logger, + config: Mapping[str, Any], + catalog: ConfiguredAirbyteCatalog, + state: list[AirbyteStateMessage] | None = None, + ) -> Iterable[AirbyteMessage]: + self._read_catalogs.append(catalog) + if self._behavior == "raise": + raise RuntimeError("Read exploded.") + if self._behavior == "no_records": + return + for configured_stream in catalog.streams: + name = configured_stream.stream.name + yield AirbyteMessage( + type=Type.RECORD, + record=AirbyteRecordMessage( + stream=name, data={"id": 1, "updated_at": "2026-01-01"}, emitted_at=0 + ), + ) + + +def _read_suite( + tmp_path: Path, + behavior: str, + streams: list[AirbyteStream], + read_catalogs: list[ConfiguredAirbyteCatalog] | None = None, +) -> SourceTestSuiteBase: + catalogs = [] if read_catalogs is None else read_catalogs + return _suite_class( + tmp_path, + {}, + base=SourceTestSuiteBase, + make_connector=lambda: _ReadSource(behavior, streams, catalogs), + )() + + +def _configured(catalog: ConfiguredAirbyteCatalog) -> list[tuple[str, SyncMode, list[str] | None]]: + return [(s.stream.name, s.sync_mode, s.cursor_field) for s in catalog.streams] + + +def test_discover_fails_on_errors_without_a_declared_status(tmp_path: Path) -> None: + suite = _read_suite(tmp_path, "discover_raises", [USERS]) + with pytest.raises(AirbyteEntrypointException, match="Discover exploded"): + suite.test_discover(_scenario("discovery")) + + +def test_discover_fails_on_an_empty_catalog(tmp_path: Path) -> None: + suite = _read_suite(tmp_path, "records", []) + with pytest.raises(AssertionError, match="The catalog does not contain any streams"): + suite.test_discover(_scenario("discovery")) + + +def test_basic_read_reads_every_stream_full_refresh(tmp_path: Path) -> None: + catalogs: list[ConfiguredAirbyteCatalog] = [] + suite = _read_suite(tmp_path, "records", [USERS, EVENTS, STATIC], catalogs) + suite.test_basic_read( + _scenario("basic_read", empty_streams=[{"name": "static", "bypass_reason": "no data"}]) + ) + assert _configured(catalogs[-1]) == [ + ("users", SyncMode.full_refresh, None), + ("events", SyncMode.full_refresh, None), + ] + + +def _write_declared_catalog(tmp_path: Path, *stream_names: str) -> Path: + """Write a configured catalog naming `stream_names` (incremental, with a stale schema).""" + catalog_path = Path("integration_tests/configured_catalog.json") + file_path = tmp_path / "source-test" / catalog_path + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text( + json.dumps( + { + "streams": [ + { + "stream": { + "name": name, + "json_schema": {"type": "object", "properties": {"stale": {}}}, + "supported_sync_modes": ["full_refresh", "incremental"], + }, + "sync_mode": "incremental", + "cursor_field": ["updated_at"], + "destination_sync_mode": "append", + } + for name in stream_names + ] + } + ) + ) + return catalog_path + + +@pytest.mark.parametrize("section", ["basic_read", "full_refresh"]) +def test_basic_read_reads_the_declared_catalog_full_refresh(tmp_path: Path, section: str) -> None: + """A declared `configured_catalog_path` picks the streams; `discover` supplies their schemas.""" + catalogs: list[ConfiguredAirbyteCatalog] = [] + suite = _read_suite(tmp_path, "records", [USERS, EVENTS, STATIC], catalogs) + catalog_path = _write_declared_catalog(tmp_path, "static", "users", "static", "events") + + suite.test_basic_read( + _scenario( + section, + configured_catalog_path=catalog_path, + empty_streams=[{"name": "events", "bypass_reason": "no data"}], + ) + ) + + assert _configured(catalogs[-1]) == [ + ("static", SyncMode.full_refresh, None), + ("users", SyncMode.full_refresh, None), + ], "the catalog's streams, once each, in its order, full refresh, minus `empty_streams`" + assert [s.stream for s in catalogs[-1].streams] == [STATIC, USERS], ( + "each stream keeps the definition `discover` reported, not the catalog's stale one" + ) + + +def test_basic_read_fails_on_a_declared_stream_that_is_not_discovered(tmp_path: Path) -> None: + catalogs: list[ConfiguredAirbyteCatalog] = [] + suite = _read_suite(tmp_path, "records", [USERS], catalogs) + catalog_path = _write_declared_catalog(tmp_path, "users", "removed_stream") + + with pytest.raises( + AssertionError, + match=( + r"The configured catalog `integration_tests/configured_catalog.json` of config " + r"'config' lists streams that connector 'source-test' does not discover: " + r"'removed_stream'\." + ), + ): + suite.test_basic_read(_scenario("basic_read", configured_catalog_path=catalog_path)) + assert catalogs == [], "nothing is read" + + +def test_basic_read_fails_on_a_missing_declared_catalog(tmp_path: Path) -> None: + catalogs: list[ConfiguredAirbyteCatalog] = [] + suite = _read_suite(tmp_path, "records", [USERS], catalogs) + + with pytest.raises( + AssertionError, + match=( + r"The configured catalog `integration_tests/missing.json` of config 'config' does " + r"not exist in connector 'source-test'" + ), + ): + suite.test_basic_read( + _scenario("basic_read", configured_catalog_path=Path("integration_tests/missing.json")) + ) + assert catalogs == [], "nothing is read" + + +@pytest.mark.parametrize( + "behavior, expected_exception, match", + [ + pytest.param( + "no_records", AssertionError, "Expected records but got none", id="no_records" + ), + pytest.param("raise", AirbyteEntrypointException, "Read exploded", id="read_raises"), + pytest.param( + "discover_raises", AirbyteEntrypointException, "Discover exploded", id="discover_raises" + ), + ], +) +def test_basic_read_fails_without_a_declared_status( + tmp_path: Path, behavior: str, expected_exception: type[Exception], match: str +) -> None: + """No `status` means `succeed`: a read that errors or returns nothing fails.""" + suite = _read_suite(tmp_path, behavior, [USERS]) + with pytest.raises(expected_exception, match=match): + suite.test_basic_read(_scenario("basic_read")) + + +# --- Docker path ------------------------------------------------------------------------------- + + +def _stream_json( + name: str, sync_modes: list[str], cursor: list[str] | None = None +) -> dict[str, Any]: + stream: dict[str, Any] = { + "name": name, + "json_schema": {"type": "object"}, + "supported_sync_modes": sync_modes, + } + if cursor: + stream |= {"source_defined_cursor": True, "default_cursor_field": cursor} + return stream + + +SPEC_MESSAGE = json.dumps( + {"type": "SPEC", "spec": {"connectionSpecification": SPEC_WITH_LONG_API_KEY}} +) +TRACE_ERROR_MESSAGE = json.dumps( + { + "type": "TRACE", + "trace": {"type": "ERROR", "emitted_at": 0, "error": {"message": "Command exploded."}}, + } +) + + +def _catalog_message(*streams: dict[str, Any]) -> str: + return json.dumps({"type": "CATALOG", "catalog": {"streams": list(streams)}}) + + +def _record_message(stream: str) -> str: + return json.dumps( + {"type": "RECORD", "record": {"stream": stream, "data": {"id": 1}, "emitted_at": 0}} + ) + + +def _stub_docker( + monkeypatch: pytest.MonkeyPatch, + responses: dict[str, list[str]], + read_catalogs: list[dict[str, Any]] | None = None, +) -> list[list[str]]: + """Answer each connector command run "in Docker" with canned messages; record the commands. + + The configured catalog mounted into each `read` is appended to `read_catalogs`. + """ + calls: list[list[str]] = [] + + def _fake_run_docker_airbyte_command( + cmd: list[str], *, raise_if_errors: bool = False + ) -> EntrypointOutput: + calls.append(cmd) + command = next(part for part in cmd if part in ("spec", "check", "discover", "read")) + if command == "read" and read_catalogs is not None: + mount = next(part for part in cmd if part.endswith(":/secrets/catalog.json")) + read_catalogs.append(json.loads(Path(mount.rsplit(":", 1)[0]).read_text())) + result = EntrypointOutput(messages=responses[command], command=cmd) + if raise_if_errors: + result.raise_if_errors() + return result + + monkeypatch.setattr( + "airbyte_cdk.test.standard_tests.docker_base.run_docker_airbyte_command", + _fake_run_docker_airbyte_command, + ) + return calls + + +def test_docker_config_matches_spec(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + calls = _stub_docker(monkeypatch, {"spec": [SPEC_MESSAGE]}) + suite = _suite_class(tmp_path, {}, base=DockerConnectorTestSuite)() + kwargs = _docker_kwargs("test_docker_image_build_and_config_matches_spec") + + suite.test_docker_image_build_and_config_matches_spec( + _scenario("spec", config=VALID_CONFIG), **kwargs + ) + assert calls[-1][-2:] == [IMAGE, "spec"] + + with pytest.raises(AssertionError, match="does not match the spec") as exc_info: + suite.test_docker_image_build_and_config_matches_spec( + _scenario("spec", config=SECRET_CONFIG), **kwargs + ) + assert "hunter2" not in str(exc_info.value) + + +def test_docker_discover(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + calls = _stub_docker( + monkeypatch, {"discover": [_catalog_message(_stream_json("users", ["full_refresh"]))]} + ) + suite = _suite_class(tmp_path, {}, base=DockerConnectorTestSuite)() + + suite.test_docker_image_build_and_discover( + _scenario("discovery"), **_docker_kwargs("test_docker_image_build_and_discover") + ) + assert calls[-1][-4:-1] == [IMAGE, "discover", "--config"] + + +@pytest.mark.parametrize( + "messages, expected_exception, match", + [ + pytest.param([_catalog_message()], ValueError, "is empty", id="empty_catalog"), + pytest.param([TRACE_ERROR_MESSAGE], AirbyteEntrypointException, "exploded", id="error"), + ], +) +def test_docker_discover_fails( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + messages: list[str], + expected_exception: type[Exception], + match: str, +) -> None: + _stub_docker(monkeypatch, {"discover": messages}) + suite = _suite_class(tmp_path, {}, base=DockerConnectorTestSuite)() + with pytest.raises(expected_exception, match=match): + suite.test_docker_image_build_and_discover( + _scenario("discovery"), **_docker_kwargs("test_docker_image_build_and_discover") + ) + + +FULL_REFRESH_CATALOG = [("users", "full_refresh", None), ("events", "full_refresh", None)] + + +@pytest.mark.parametrize( + "sections, expected", + [ + pytest.param(("basic_read",), FULL_REFRESH_CATALOG, id="basic_read"), + pytest.param(("full_refresh",), FULL_REFRESH_CATALOG, id="full_refresh"), + pytest.param(("discovery", "basic_read"), FULL_REFRESH_CATALOG, id="mixed"), + ], +) +def test_docker_read_configures_the_catalog_for_the_section( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + sections: tuple[str, ...], + expected: list[tuple[str, str, list[str] | None]], +) -> None: + catalogs: list[dict[str, Any]] = [] + _stub_docker( + monkeypatch, + { + "discover": [ + _catalog_message( + _stream_json("users", ["full_refresh", "incremental"], ["updated_at"]), + _stream_json("events", ["full_refresh", "incremental"]), + _stream_json("static", ["full_refresh"]), + ) + ], + "read": [_record_message("users"), _record_message("events")], + }, + read_catalogs=catalogs, + ) + suite = _suite_class(tmp_path, {}, base=DockerConnectorTestSuite)() + + suite.test_docker_image_build_and_read( + # `empty_streams` used to crash the Docker read (names compared as streams). + _scenario(*sections, empty_streams=[{"name": "static", "bypass_reason": "no data"}]), + **_docker_kwargs("test_docker_image_build_and_read"), + ) + + assert [ + (s["stream"]["name"], s["sync_mode"], s.get("cursor_field")) + for s in catalogs[-1]["streams"] + ] == expected + + +def test_docker_read_uses_the_declared_catalog( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + catalogs: list[dict[str, Any]] = [] + _stub_docker( + monkeypatch, + { + "discover": [ + _catalog_message( + _stream_json("users", ["incremental", "full_refresh"], ["updated_at"]), + _stream_json("events", ["full_refresh", "incremental"]), + _stream_json("static", ["full_refresh"]), + ) + ], + "read": [_record_message("users")], + }, + read_catalogs=catalogs, + ) + suite = _suite_class(tmp_path, {}, base=DockerConnectorTestSuite)() + catalog_path = _write_declared_catalog(tmp_path, "static", "users", "events") + kwargs = _docker_kwargs("test_docker_image_build_and_read") + + suite.test_docker_image_build_and_read( + _scenario( + "full_refresh", + configured_catalog_path=catalog_path, + empty_streams=[{"name": "static", "bypass_reason": "no data"}], + ), + **kwargs, + ) + assert [ + (s["stream"]["name"], s["sync_mode"], s.get("cursor_field"), s["stream"]["json_schema"]) + for s in catalogs[-1]["streams"] + ] == [ + ("users", "full_refresh", None, {"type": "object"}), + ("events", "full_refresh", None, {"type": "object"}), + ], "the catalog's streams, full refresh even when incremental comes first, discovered schemas" + + # `--read-from-streams` narrows the declared catalog further. + suite.test_docker_image_build_and_read( + _scenario("full_refresh", configured_catalog_path=catalog_path), + **(kwargs | {"read_from_streams": ["users", "not_in_catalog"]}), + ) + assert [s["stream"]["name"] for s in catalogs[-1]["streams"]] == ["users"] + + _write_declared_catalog(tmp_path, "users", "removed_stream") + with pytest.raises(AssertionError, match=r"does not discover: 'removed_stream'"): + suite.test_docker_image_build_and_read( + _scenario("basic_read", configured_catalog_path=catalog_path), **kwargs + ) diff --git a/unit_tests/test/test_standard_tests.py b/unit_tests/test/test_standard_tests.py index 8f11ade6e1..3235950611 100644 --- a/unit_tests/test/test_standard_tests.py +++ b/unit_tests/test/test_standard_tests.py @@ -1,4 +1,4 @@ -# Copyright (c) 2024 Airbyte, Inc., all rights reserved. +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. """Unit tests for FAST Airbyte Standard Tests.""" from pathlib import Path