diff --git a/airbyte_cdk/test/standard_tests/__init__.py b/airbyte_cdk/test/standard_tests/__init__.py index 50c2332280..e99712ae2d 100644 --- a/airbyte_cdk/test/standard_tests/__init__.py +++ b/airbyte_cdk/test/standard_tests/__init__.py @@ -25,9 +25,19 @@ class TestSuiteSourcePokeAPI(standard_tests.DeclarativeSourceTestSuite): - `SourceTestSuiteBase`: A test suite for sources. - `DestinationTestSuiteBase`: A test suite for destinations. +`SourceTestSuiteBase.test_discover` checks the discovered catalog. Some of these checks are not +enforced yet: their problems are reported as a `DiscoveredCatalogWarning` and do not fail the +test. To make them fail, turn the warning into an error: + +```bash +pytest -W error::airbyte_cdk.test.standard_tests.DiscoveredCatalogWarning +airbyte-cdk connector test --pytest-arg=-Werror::airbyte_cdk.test.standard_tests.DiscoveredCatalogWarning +``` + ''' from airbyte_cdk.test.models.scenario import ConnectorTestScenario +from airbyte_cdk.test.standard_tests._catalog_checks import DiscoveredCatalogWarning from airbyte_cdk.test.standard_tests.connector_base import ConnectorTestSuiteBase from airbyte_cdk.test.standard_tests.declarative_sources import ( DeclarativeSourceTestSuite, @@ -40,5 +50,6 @@ class TestSuiteSourcePokeAPI(standard_tests.DeclarativeSourceTestSuite): "ConnectorTestSuiteBase", "DeclarativeSourceTestSuite", "DestinationTestSuiteBase", + "DiscoveredCatalogWarning", "SourceTestSuiteBase", ] diff --git a/airbyte_cdk/test/standard_tests/_catalog_checks.py b/airbyte_cdk/test/standard_tests/_catalog_checks.py new file mode 100644 index 0000000000..34014477a2 --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_catalog_checks.py @@ -0,0 +1,413 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Assertions on the catalog a source returns from `discover`. + +These restore the `TestDiscovery` assertions of the retired Connector Acceptance Tests (CAT). +Each check takes the discovered catalog and returns a list of problems, one string per +offending stream or field, so that a single test run reports everything that is wrong. + +Checks marked as not enforced report their problems as a `DiscoveredCatalogWarning` instead of +failing the test. They are not enforced yet because connectors in the fleet still violate them. +Run pytest with `-W error::airbyte_cdk.test.standard_tests.DiscoveredCatalogWarning` to make them +fail locally. +""" + +from __future__ import annotations + +from collections import Counter +from collections.abc import Callable, Iterator, Mapping, Sequence +from dataclasses import dataclass +from typing import Any + +import jsonschema +from jsonschema.protocols import Validator + +from airbyte_cdk.models import AirbyteCatalog + +# Keywords whose value maps names to subschemas. +_SCHEMA_MAP_KEYWORDS = frozenset( + {"properties", "patternProperties", "definitions", "$defs", "dependentSchemas"} +) +# Keywords whose value is a list of subschemas. +_SCHEMA_LIST_KEYWORDS = frozenset({"allOf", "anyOf", "oneOf", "prefixItems"}) +# Keywords whose value is a single subschema. `items` may also hold a list of subschemas. +_SCHEMA_KEYWORDS = frozenset( + { + "additionalItems", + "additionalProperties", + "contains", + "else", + "if", + "items", + "not", + "propertyNames", + "then", + "unevaluatedItems", + "unevaluatedProperties", + } +) + +UNSUPPORTED_KEYWORDS = ("allOf", "not") +"""JSON Schema keywords that Airbyte destinations do not support in stream schemas.""" + +SCALAR_PRIMARY_KEY_FORBIDDEN_TYPES = frozenset({"object", "array"}) + +SUPPORTED_AIRBYTE_TYPES: Mapping[str, frozenset[str]] = { + "timestamp_with_timezone": frozenset({"string"}), + "timestamp_without_timezone": frozenset({"string"}), + "time_with_timezone": frozenset({"string"}), + "time_without_timezone": frozenset({"string"}), + "integer": frozenset({"integer", "number"}), +} +"""Each `airbyte_type` from the Airbyte type system, mapped to the JSON types it can annotate. + +See https://docs.airbyte.com/platform/understanding-airbyte/supported-data-types +""" + +TEMPORAL_FORMATS = frozenset({"date", "date-time", "time"}) +"""String formats that the Airbyte type system maps to temporal types.""" + + +class DiscoveredCatalogWarning(UserWarning): + """A problem in the discovered catalog that the standard tests report but do not enforce. + + To make these problems fail the test, turn the warning into an error, either with + `pytest -W error::airbyte_cdk.test.standard_tests.DiscoveredCatalogWarning` or with + `airbyte-cdk connector test --pytest-arg=-Werror::airbyte_cdk.test.standard_tests.DiscoveredCatalogWarning`. + """ + + +def _pointer(path: Sequence[str]) -> str: + """Render a schema path as a JSON pointer fragment, for example `#/properties/id`.""" + escaped = (segment.replace("~", "~0").replace("/", "~1") for segment in path) + return "#/" + "/".join(escaped) if path else "#" + + +def _iter_subschemas( + schema: Any, + path: tuple[str, ...] = (), +) -> Iterator[tuple[tuple[str, ...], Mapping[str, Any]]]: + """Yield every subschema of `schema` with its path, starting with `schema` itself. + + Only keywords that hold subschemas are followed. Property names, and values such as + `default`, `examples` and `enum`, are never mistaken for keywords. + """ + if not isinstance(schema, Mapping): + # Boolean schemas have no keywords. + return + + yield path, schema + for keyword, value in schema.items(): + if keyword in _SCHEMA_MAP_KEYWORDS and isinstance(value, Mapping): + for name, subschema in value.items(): + yield from _iter_subschemas(subschema, (*path, keyword, str(name))) + elif keyword == "dependencies" and isinstance(value, Mapping): + # Values are either subschemas or lists of property names. + for name, subschema in value.items(): + yield from _iter_subschemas(subschema, (*path, keyword, str(name))) + elif isinstance(value, list) and (keyword in _SCHEMA_LIST_KEYWORDS or keyword == "items"): + for index, subschema in enumerate(value): + yield from _iter_subschemas(subschema, (*path, keyword, str(index))) + elif keyword in _SCHEMA_KEYWORDS: + yield from _iter_subschemas(value, (*path, keyword)) + + +def _json_types(schema: Mapping[str, Any]) -> set[str]: + """Return the JSON types a schema declares, as a set.""" + declared = schema.get("type") + if isinstance(declared, str): + return {declared} + if isinstance(declared, list): + return {item for item in declared if isinstance(item, str)} + return set() + + +def _json_types_with_alternatives(schema: Mapping[str, Any]) -> set[str]: + """Return the JSON types a schema declares, including those of its `anyOf`/`oneOf` branches. + + A value typed as `{"anyOf": [{"type": "string"}, {"type": "null"}]}` can be any of the + branch types, so they are added to the types declared on the node itself. + """ + types = _json_types(schema) + for keyword in ("anyOf", "oneOf"): + branches = schema.get(keyword) + if isinstance(branches, list): + for branch in branches: + if isinstance(branch, Mapping): + types |= _json_types_with_alternatives(branch) + return types + + +def _meta_validator(schema: Mapping[str, Any]) -> Validator: + """Return a validator for the meta-schema of the draft that `schema` declares in `$schema`. + + Draft 7 is used when `$schema` is absent, not a string, or names an unknown draft. + """ + validator_cls: type[Validator] = jsonschema.Draft7Validator + if isinstance(schema.get("$schema"), str): + validator_cls = jsonschema.validators.validator_for(schema, default=validator_cls) + return validator_cls(validator_cls.META_SCHEMA) + + +_MISSING = object() + + +def _find_field(schema: Mapping[str, Any], field_path: Sequence[str]) -> Any: + """Return the subschema of a (possibly nested) field, or `_MISSING` if the schema lacks it.""" + node: Any = schema + for segment in field_path: + properties = node.get("properties") if isinstance(node, Mapping) else None + if not isinstance(properties, Mapping) or segment not in properties: + return _MISSING + node = properties[segment] + return node + + +def _field_name(field_path: Sequence[str]) -> str: + return ".".join(field_path) + + +def check_catalog_has_streams(catalog: AirbyteCatalog) -> list[str]: + """The catalog declares at least one stream.""" + if catalog.streams: + return [] + return ["The catalog does not contain any streams."] + + +def check_stream_names_are_unique(catalog: AirbyteCatalog) -> list[str]: + """No two streams share the same namespace and name.""" + counts = Counter((stream.namespace, stream.name) for stream in catalog.streams) + return [ + f"Stream '{name}'" + + (f" in namespace '{namespace}'" if namespace else "") + + f" is declared {count} times." + for (namespace, name), count in counts.items() + if count > 1 + ] + + +def check_streams_declare_sync_modes(catalog: AirbyteCatalog) -> list[str]: + """Every stream declares at least one supported sync mode.""" + return [ + f"Stream '{stream.name}' does not declare any supported sync modes." + for stream in catalog.streams + if not stream.supported_sync_modes + ] + + +def check_schemas_are_valid_json_schema(catalog: AirbyteCatalog) -> list[str]: + """Every stream schema is a valid JSON Schema. + + Each schema is validated against the meta-schema of the draft it declares in `$schema`, + or Draft 7 when it declares none. Every violation is reported, ordered by location, rather + than the single best match that `check_schema` raises, so that fixing one does not reveal + another on the next run. + """ + problems: list[str] = [] + for stream in catalog.streams: + errors = sorted( + _meta_validator(stream.json_schema).iter_errors(stream.json_schema), + key=lambda error: [str(part) for part in error.absolute_path], + ) + problems.extend( + f"Stream '{stream.name}' has an invalid JSON schema at " + f"{_pointer([str(part) for part in error.absolute_path])}: {error.message}" + for error in errors + ) + return problems + + +def check_cursor_fields_exist_in_schema(catalog: AirbyteCatalog) -> list[str]: + """A declared default cursor field exists in the stream schema.""" + problems = [] + for stream in catalog.streams: + if not stream.default_cursor_field: + continue + if _find_field(stream.json_schema, stream.default_cursor_field) is _MISSING: + problems.append( + f"Stream '{stream.name}' declares cursor field " + f"'{_field_name(stream.default_cursor_field)}', " + "which is not a property in the stream schema." + ) + return problems + + +def check_primary_keys_exist_in_schema(catalog: AirbyteCatalog) -> list[str]: + """Every field of a declared primary key exists in the stream schema.""" + problems = [] + for stream in catalog.streams: + for key_path in stream.source_defined_primary_key or []: + if _find_field(stream.json_schema, key_path) is _MISSING: + problems.append( + f"Stream '{stream.name}' declares primary key field " + f"'{_field_name(key_path)}', which is not a property in the stream schema." + ) + return problems + + +def check_primary_keys_have_scalar_types(catalog: AirbyteCatalog) -> list[str]: + """No field of a declared primary key is typed as an object or an array. + + Fields missing from the schema are reported by `check_primary_keys_exist_in_schema`. + """ + problems = [] + for stream in catalog.streams: + for key_path in stream.source_defined_primary_key or []: + field_schema = _find_field(stream.json_schema, key_path) + if not isinstance(field_schema, Mapping): + continue + forbidden = _json_types(field_schema) & SCALAR_PRIMARY_KEY_FORBIDDEN_TYPES + if forbidden: + problems.append( + f"Stream '{stream.name}' declares primary key field " + f"'{_field_name(key_path)}' with type {sorted(_json_types(field_schema))}. " + "Primary key fields must not be objects or arrays." + ) + return problems + + +def check_refs_are_resolved(catalog: AirbyteCatalog) -> list[str]: + """Stream schemas contain no unresolved `$ref`.""" + return [ + f"Stream '{stream.name}' has an unresolved $ref '{subschema['$ref']}' at {_pointer(path)}." + for stream in catalog.streams + for path, subschema in _iter_subschemas(stream.json_schema) + if "$ref" in subschema + ] + + +def check_no_unsupported_keywords(catalog: AirbyteCatalog) -> list[str]: + """Stream schemas use none of the keywords in `UNSUPPORTED_KEYWORDS`.""" + return [ + f"Stream '{stream.name}' uses the unsupported keyword '{keyword}' at {_pointer(path)}." + for stream in catalog.streams + for path, subschema in _iter_subschemas(stream.json_schema) + for keyword in UNSUPPORTED_KEYWORDS + if keyword in subschema + ] + + +def check_additional_properties_not_false(catalog: AirbyteCatalog) -> list[str]: + """No stream schema sets `additionalProperties` to false. + + A closed schema turns the removal of a property into a breaking change for existing + connections. See https://github.com/airbytehq/airbyte/issues/14196. + """ + return [ + f"Stream '{stream.name}' sets additionalProperties to false at {_pointer(path)}." + for stream in catalog.streams + for path, subschema in _iter_subschemas(stream.json_schema) + if subschema.get("additionalProperties") is False + ] + + +def check_supported_data_types(catalog: AirbyteCatalog) -> list[str]: + """Stream schemas stay within the Airbyte type system. + + The top level is an object, every `airbyte_type` is a known one and annotates a compatible + JSON type, and temporal formats annotate strings. The types of `anyOf`/`oneOf` branches + count as types of the annotated node, and an annotation on a node that declares no type at + all is not checked. Unknown JSON type names are already rejected by + `check_schemas_are_valid_json_schema`. + """ + problems = [] + for stream in catalog.streams: + top_level_types = _json_types(stream.json_schema) + if "object" not in top_level_types: + problems.append( + f"Stream '{stream.name}' has a top-level schema of type " + f"{sorted(top_level_types)}; it must be an object." + ) + for path, subschema in _iter_subschemas(stream.json_schema): + json_types = _json_types_with_alternatives(subschema) + airbyte_type = subschema.get("airbyte_type") + if airbyte_type is not None: + compatible_types = ( + SUPPORTED_AIRBYTE_TYPES.get(airbyte_type) + if isinstance(airbyte_type, str) + else None + ) + if compatible_types is None: + problems.append( + f"Stream '{stream.name}' uses the unknown airbyte_type " + f"'{airbyte_type}' at {_pointer(path)}." + ) + elif json_types and not json_types & compatible_types: + problems.append( + f"Stream '{stream.name}' uses airbyte_type '{airbyte_type}' on type " + f"{sorted(json_types)} at {_pointer(path)}; it requires one of " + f"{sorted(compatible_types)}." + ) + string_format = subschema.get("format") + if ( + isinstance(string_format, str) + and string_format in TEMPORAL_FORMATS + and json_types + and "string" not in json_types + ): + problems.append( + f"Stream '{stream.name}' uses format '{string_format}' on type " + f"{sorted(json_types)} at {_pointer(path)}; it requires a string." + ) + return problems + + +@dataclass(frozen=True) +class CatalogCheck: + """A named assertion on the discovered catalog.""" + + name: str + run: Callable[[AirbyteCatalog], list[str]] + enforced: bool = True + + +CATALOG_CHECKS: tuple[CatalogCheck, ...] = ( + CatalogCheck("catalog has streams", check_catalog_has_streams), + CatalogCheck("stream names are unique", check_stream_names_are_unique), + CatalogCheck("streams declare sync modes", check_streams_declare_sync_modes), + CatalogCheck("refs are resolved", check_refs_are_resolved), + CatalogCheck("no unsupported keywords", check_no_unsupported_keywords), + CatalogCheck("additionalProperties is not false", check_additional_properties_not_false), + # Not enforced until the manifest-only connectors that violate them are fixed. + CatalogCheck( + "schemas are valid JSON schema", check_schemas_are_valid_json_schema, enforced=False + ), + CatalogCheck( + "cursor fields exist in schema", check_cursor_fields_exist_in_schema, enforced=False + ), + CatalogCheck( + "primary keys exist in schema", check_primary_keys_exist_in_schema, enforced=False + ), + CatalogCheck( + "primary keys have scalar types", check_primary_keys_have_scalar_types, enforced=False + ), + CatalogCheck("supported data types", check_supported_data_types, enforced=False), +) + + +def find_catalog_problems( + catalog: AirbyteCatalog, + checks: Sequence[CatalogCheck] = CATALOG_CHECKS, +) -> dict[CatalogCheck, list[str]]: + """Run each check against the catalog and return the problems of every failing check. + + A check that raises is reported as a problem of that check, so that a check which is not + enforced yet cannot fail the test by crashing on a schema it did not anticipate. + """ + results: dict[CatalogCheck, list[str]] = {} + for check in checks: + try: + problems = check.run(catalog) + except Exception as error: # Reported, not raised; see the docstring. + problems = [f"The check could not run: {type(error).__name__}: {error}"] + if problems: + results[check] = problems + return results + + +def format_catalog_problems(problems: Mapping[CatalogCheck, list[str]]) -> str: + """Render problems grouped by check, for an assertion or warning message.""" + sections = [ + f"[{check.name}]\n" + "\n".join(f" - {problem}" for problem in check_problems) + for check, check_problems in problems.items() + ] + return "\n".join(sections) diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index ce9b861726..18f3fa1792 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -1,6 +1,7 @@ # Copyright (c) 2024 Airbyte, Inc., all rights reserved. """Base class for source test suites.""" +import warnings from dataclasses import asdict from typing import TYPE_CHECKING @@ -18,6 +19,11 @@ from airbyte_cdk.test.models import ( ConnectorTestScenario, ) +from airbyte_cdk.test.standard_tests._catalog_checks import ( + DiscoveredCatalogWarning, + find_catalog_problems, + format_catalog_problems, +) from airbyte_cdk.test.standard_tests._job_runner import run_test_job from airbyte_cdk.test.standard_tests.connector_base import ( ConnectorTestSuiteBase, @@ -60,7 +66,12 @@ def test_discover( self, scenario: ConnectorTestScenario, ) -> None: - """Standard test for `discover`.""" + """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. + """ 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. @@ -68,12 +79,36 @@ def test_discover( pytest.skip("Skipping discover test for scenario that expects an exception.") return - run_test_job( + result = run_test_job( self.create_connector(scenario), "discover", connector_root=self.get_connector_root_dir(), test_scenario=scenario, ) + if result.errors and not scenario.expected_outcome.expect_success(): + # The scenario allows `discover` to fail, so there is no catalog to check. + return + + catalog_messages = result.get_message_by_types([Type.CATALOG]) + assert len(catalog_messages) == 1, ( + f"Expected exactly one CATALOG message. Got {len(catalog_messages)}." + ) + catalog = catalog_messages[0].catalog + assert catalog is not None, "The CATALOG message does not contain a catalog." + + problems = find_catalog_problems(catalog) + advisory = {check: found for check, found in problems.items() if not check.enforced} + if advisory: + warnings.warn( + "The discovered catalog has problems that will fail this test in a future " + f"CDK release:\n{format_catalog_problems(advisory)}", + category=DiscoveredCatalogWarning, + stacklevel=1, + ) + enforced = {check: found for check, found in problems.items() if check.enforced} + assert not enforced, ( + f"The discovered catalog has problems:\n{format_catalog_problems(enforced)}" + ) def test_spec(self) -> None: """Standard test for `spec`. diff --git a/cdk-migrations.md b/cdk-migrations.md index 3445a04b88..249b6256b2 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -1,5 +1,36 @@ # CDK Migration Guide +## Upgrading to the next minor release (unreleased) + +The next minor release of the CDK adds catalog checks to the `test_discover` standard test (`SourceTestSuiteBase.test_discover`). Every source connector that runs the standard tests, for example through `airbyte-cdk connector test`, picks them up when it moves to this release. There is no change to how connectors run. + +### New mandatory checks in `test_discover` + +`discover` must emit exactly one `CATALOG` message, and the catalog must pass these checks, or the test fails: + +- The catalog contains at least one stream. +- No two streams share the same namespace and name. +- Every stream declares at least one supported sync mode. +- Stream schemas contain no unresolved `$ref`. +- Stream schemas do not use the `allOf` or `not` keywords. +- No stream schema sets `additionalProperties` to `false`. + +The test reports every problem it finds in one run, grouped by check. + +These checks are reported as a `DiscoveredCatalogWarning` and do not fail the test yet; a later release will enforce them: + +- Stream schemas are valid JSON Schema, for the draft they declare in `$schema` (Draft 7 if none). +- The default cursor field exists in the stream schema. +- Every primary key field exists in the stream schema. +- No primary key field is typed as an object or an array. +- Stream schemas stay within the Airbyte type system (top-level object, known `airbyte_type` values on compatible types, temporal formats on strings). + +To make the warnings fail the test now, run `airbyte-cdk connector test --pytest-arg=-Werror::airbyte_cdk.test.standard_tests.DiscoveredCatalogWarning`, or pass `-W error::airbyte_cdk.test.standard_tests.DiscoveredCatalogWarning` to `pytest`. + +Migration steps: fix the stream schema, primary key or cursor field that the failure or warning names. + +Rationale: these checks restore the `TestDiscovery` assertions of the retired Connector Acceptance Tests, so that catalog problems are caught in the connector's own tests instead of in a destination or in a customer's sync. + ## Upgrading to 7.0.0 [Version 7.0.0](https://github.com/airbytehq/airbyte-python-cdk/releases/tag/v7.0.0) of the CDK migrates the CDK to the Concurrent CDK by removing some of the Declarative CDK concepts that are better expressed in the Concurrent CDK or that are outright incompatible with it. This changes mostly impact the Python implementations although the concept of CustomIncrementalSync has been removed from the declarative language as well. diff --git a/unit_tests/resources/source_pokeapi_w_components_py/manifest.yaml b/unit_tests/resources/source_pokeapi_w_components_py/manifest.yaml index af19485fad..14b636253e 100644 --- a/unit_tests/resources/source_pokeapi_w_components_py/manifest.yaml +++ b/unit_tests/resources/source_pokeapi_w_components_py/manifest.yaml @@ -976,5 +976,11 @@ schemas: pokemon: type: object $schema: http://json-schema.org/draft-07/schema# - properties: {} + properties: + id: + type: integer + name: + type: + - "null" + - string additionalProperties: true diff --git a/unit_tests/test/test_catalog_checks.py b/unit_tests/test/test_catalog_checks.py new file mode 100644 index 0000000000..b63ae73a82 --- /dev/null +++ b/unit_tests/test/test_catalog_checks.py @@ -0,0 +1,829 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Unit tests for the catalog checks run by the `discover` standard test.""" + +from __future__ import annotations + +import copy +import json +import warnings +from pathlib import Path +from typing import Any, Callable + +import pytest +import yaml + +from airbyte_cdk.models import AirbyteCatalog, AirbyteStream, SyncMode +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput +from airbyte_cdk.test.models.scenario import ConnectorTestScenario +from airbyte_cdk.test.standard_tests import ( + DeclarativeSourceTestSuite, + DiscoveredCatalogWarning, + SourceTestSuiteBase, + source_base, +) +from airbyte_cdk.test.standard_tests._catalog_checks import ( + CATALOG_CHECKS, + CatalogCheck, + check_additional_properties_not_false, + check_catalog_has_streams, + check_cursor_fields_exist_in_schema, + check_no_unsupported_keywords, + check_primary_keys_exist_in_schema, + check_primary_keys_have_scalar_types, + check_refs_are_resolved, + check_schemas_are_valid_json_schema, + check_stream_names_are_unique, + check_streams_declare_sync_modes, + check_supported_data_types, + find_catalog_problems, +) + +_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "updated_at": {"type": "string", "format": "date-time"}, + "author": { + "type": ["null", "object"], + "properties": {"login": {"type": "string"}}, + }, + "tags": {"type": "array", "items": {"type": "string"}}, + }, +} + + +def _stream( + name: str = "items", + json_schema: dict[str, Any] | None = None, + **kwargs: Any, +) -> AirbyteStream: + return AirbyteStream( + name=name, + json_schema=_SCHEMA if json_schema is None else json_schema, + supported_sync_modes=kwargs.pop("supported_sync_modes", [SyncMode.full_refresh]), + **kwargs, + ) + + +def _catalog(*streams: AirbyteStream) -> AirbyteCatalog: + return AirbyteCatalog(streams=list(streams)) + + +def _with_property(name: str, property_schema: dict[str, Any]) -> dict[str, Any]: + return {**_SCHEMA, "properties": {**_SCHEMA["properties"], name: property_schema}} + + +Check = Callable[[AirbyteCatalog], list[str]] + + +@pytest.mark.parametrize( + "check, catalog", + [ + pytest.param(check_catalog_has_streams, _catalog(_stream()), id="has_streams"), + pytest.param( + check_stream_names_are_unique, + _catalog(_stream("a"), _stream("b"), _stream("a", namespace="other")), + id="unique_names_namespace_disambiguates", + ), + pytest.param(check_streams_declare_sync_modes, _catalog(_stream()), id="sync_modes"), + pytest.param( + check_schemas_are_valid_json_schema, + _catalog(_stream(json_schema=_with_property("v", {"type": ["null", "string"]}))), + id="valid_json_schema", + ), + pytest.param( + check_cursor_fields_exist_in_schema, + _catalog( + _stream(default_cursor_field=["updated_at"]), + _stream("nested", default_cursor_field=["author", "login"]), + _stream("no_cursor"), + ), + id="cursor_exists_top_level_and_nested", + ), + pytest.param( + check_primary_keys_exist_in_schema, + _catalog(_stream(source_defined_primary_key=[["id"], ["author", "login"]])), + id="primary_keys_exist", + ), + pytest.param( + check_primary_keys_have_scalar_types, + _catalog( + _stream(source_defined_primary_key=[["id"]]), + _stream( + "untyped", + json_schema=_with_property("id", {}), + source_defined_primary_key=[["id"]], + ), + _stream("missing", source_defined_primary_key=[["not_in_schema"]]), + ), + id="primary_keys_scalar_untyped_or_missing_is_skipped", + ), + pytest.param( + check_refs_are_resolved, + _catalog( + _stream( + json_schema=_with_property( + "$ref", {"type": "string", "examples": [{"$ref": "data"}]} + ) + ) + ), + id="property_named_ref_and_ref_in_examples_are_not_refs", + ), + pytest.param( + check_no_unsupported_keywords, + _catalog( + _stream( + json_schema={ + "type": "object", + "properties": { + "not": {"type": "boolean"}, + "allOf": {"type": "string", "default": {"not": "data"}}, + "choice": {"anyOf": [{"type": "string"}, {"type": "integer"}]}, + }, + } + ) + ), + id="properties_named_like_keywords_are_allowed", + ), + pytest.param( + check_additional_properties_not_false, + _catalog( + _stream(json_schema={**_SCHEMA, "additionalProperties": True}), + _stream("open_object", json_schema={**_SCHEMA, "additionalProperties": {}}), + ), + id="additional_properties_true_or_schema", + ), + pytest.param( + check_supported_data_types, + _catalog( + _stream( + json_schema={ + "type": ["null", "object"], + "properties": { + "created_at": { + "type": ["null", "string"], + "format": "date-time", + "airbyte_type": "timestamp_without_timezone", + }, + "opens_at": { + "type": "string", + "format": "time", + "airbyte_type": "time_with_timezone", + }, + "day": {"type": "string", "format": "date"}, + "count": {"type": "number", "airbyte_type": "integer"}, + "email": {"type": "string", "format": "email"}, + "size": {"type": "integer", "format": "int64"}, + }, + } + ) + ), + id="supported_types_formats_and_airbyte_types", + ), + ], +) +def test_check_passes(check: Check, catalog: AirbyteCatalog) -> None: + assert check(catalog) == [] + + +@pytest.mark.parametrize( + "check, catalog, expected_problems", + [ + pytest.param( + check_catalog_has_streams, + _catalog(), + ["The catalog does not contain any streams."], + id="empty_catalog", + ), + pytest.param( + check_stream_names_are_unique, + _catalog( + _stream("a"), _stream("a"), _stream("b", namespace="n"), _stream("b", namespace="n") + ), + [ + "Stream 'a' is declared 2 times.", + "Stream 'b' in namespace 'n' is declared 2 times.", + ], + id="duplicate_names", + ), + pytest.param( + check_streams_declare_sync_modes, + _catalog(_stream(supported_sync_modes=[])), + ["Stream 'items' does not declare any supported sync modes."], + id="no_sync_modes", + ), + pytest.param( + check_schemas_are_valid_json_schema, + _catalog(_stream(json_schema=_with_property("v", {"type": "any"}))), + [ + ( + "Stream 'items' has an invalid JSON schema at #/properties/v/type: " + "'any' is not valid under any of the given schemas" + ) + ], + id="unknown_type_name", + ), + pytest.param( + check_schemas_are_valid_json_schema, + _catalog( + _stream( + json_schema={ + "type": "object", + "required": ["id", "id"], + "properties": {"zip": None, "address": None}, + } + ) + ), + [ + ( + "Stream 'items' has an invalid JSON schema at #/properties/address: " + "None is not of type 'object', 'boolean'" + ), + ( + "Stream 'items' has an invalid JSON schema at #/properties/zip: " + "None is not of type 'object', 'boolean'" + ), + ( + "Stream 'items' has an invalid JSON schema at #/required: " + "['id', 'id'] has non-unique elements" + ), + ], + id="every_violation_reported_in_path_order", + ), + pytest.param( + check_cursor_fields_exist_in_schema, + _catalog( + _stream(default_cursor_field=["modified_at"]), + _stream("nested", default_cursor_field=["author", "updated_at"]), + _stream("open", json_schema={"type": "object"}, default_cursor_field=["id"]), + ), + [ + ( + "Stream 'items' declares cursor field 'modified_at', " + "which is not a property in the stream schema." + ), + ( + "Stream 'nested' declares cursor field 'author.updated_at', " + "which is not a property in the stream schema." + ), + ( + "Stream 'open' declares cursor field 'id', " + "which is not a property in the stream schema." + ), + ], + id="cursor_missing", + ), + pytest.param( + check_primary_keys_exist_in_schema, + _catalog(_stream(source_defined_primary_key=[["id"], ["uuid"]])), + [ + ( + "Stream 'items' declares primary key field 'uuid', " + "which is not a property in the stream schema." + ) + ], + id="primary_key_missing", + ), + pytest.param( + check_primary_keys_have_scalar_types, + _catalog( + _stream(source_defined_primary_key=[["author"]]), + _stream( + "array_typed", + json_schema=_with_property("id", {"type": "array"}), + source_defined_primary_key=[["id"]], + ), + ), + [ + ( + "Stream 'items' declares primary key field 'author' with type " + "['null', 'object']. Primary key fields must not be objects or arrays." + ), + ( + "Stream 'array_typed' declares primary key field 'id' with type " + "['array']. Primary key fields must not be objects or arrays." + ), + ], + id="primary_key_object_or_array", + ), + pytest.param( + check_refs_are_resolved, + _catalog( + _stream( + json_schema=_with_property( + "owner", {"type": "array", "items": {"$ref": "#/definitions/user"}} + ) + ) + ), + [ + ( + "Stream 'items' has an unresolved $ref '#/definitions/user' " + "at #/properties/owner/items." + ) + ], + id="unresolved_ref", + ), + pytest.param( + check_no_unsupported_keywords, + _catalog( + _stream( + json_schema=_with_property( + "v", + { + "allOf": [{"type": "string"}], + "items": [{"not": {"type": "null"}}], + }, + ) + ) + ), + [ + "Stream 'items' uses the unsupported keyword 'allOf' at #/properties/v.", + "Stream 'items' uses the unsupported keyword 'not' at #/properties/v/items/0.", + ], + id="all_of_and_not", + ), + pytest.param( + check_additional_properties_not_false, + _catalog( + _stream( + json_schema=_with_property( + "a/b", {"type": "object", "additionalProperties": False} + ) + ) + ), + ["Stream 'items' sets additionalProperties to false at #/properties/a~1b."], + id="nested_additional_properties_false", + ), + pytest.param( + check_supported_data_types, + _catalog( + _stream( + json_schema={ + "type": "array", + "items": { + "type": "object", + "properties": { + "big": {"type": "string", "airbyte_type": "big_integer"}, + "ts": { + "type": "integer", + "airbyte_type": "timestamp_with_timezone", + }, + "day": {"type": ["null", "integer"], "format": "date"}, + }, + }, + } + ) + ), + [ + "Stream 'items' has a top-level schema of type ['array']; it must be an object.", + ( + "Stream 'items' uses the unknown airbyte_type 'big_integer' " + "at #/items/properties/big." + ), + ( + "Stream 'items' uses airbyte_type 'timestamp_with_timezone' on type ['integer'] " + "at #/items/properties/ts; it requires one of ['string']." + ), + ( + "Stream 'items' uses format 'date' on type ['integer', 'null'] " + "at #/items/properties/day; it requires a string." + ), + ], + id="unsupported_types", + ), + ], +) +def test_check_fails(check: Check, catalog: AirbyteCatalog, expected_problems: list[str]) -> None: + assert check(catalog) == expected_problems + + +def test_primary_key_typed_as_plain_string_object_is_caught() -> None: + """CAT built `set("object")`, so a PK typed `"object"` (not a list) slipped through.""" + catalog = _catalog( + _stream( + json_schema=_with_property("id", {"type": "object"}), + source_defined_primary_key=[["id"]], + ) + ) + assert check_primary_keys_have_scalar_types(catalog) != [] + + +def test_field_with_invalid_subschema_still_exists() -> None: + """A `null` property schema is reported as invalid JSON schema, not as a missing field.""" + catalog = _catalog( + _stream(json_schema=_with_property("ts", None), default_cursor_field=["ts"]) # type: ignore[arg-type] + ) + assert check_cursor_fields_exist_in_schema(catalog) == [] + assert check_schemas_are_valid_json_schema(catalog) != [] + + +def test_supported_data_types_tolerates_non_string_annotations() -> None: + catalog = _catalog( + _stream( + json_schema=_with_property( + "v", {"type": "integer", "format": ["date"], "airbyte_type": {"x": 1}} + ) + ) + ) + assert check_supported_data_types(catalog) == [ + "Stream 'items' uses the unknown airbyte_type '{'x': 1}' at #/properties/v." + ] + + +_REF = {"$ref": "#/definitions/user"} + + +@pytest.mark.parametrize( + "keyword_schema, pointer", + [ + *( + pytest.param({keyword: {"user": _REF}}, f"#/{keyword}/user", id=keyword) + for keyword in ( + "properties", + "patternProperties", + "definitions", + "$defs", + "dependentSchemas", + ) + ), + *( + pytest.param({keyword: [{}, _REF]}, f"#/{keyword}/1", id=f"{keyword}_list") + for keyword in ("allOf", "anyOf", "oneOf", "prefixItems", "items") + ), + *( + pytest.param({keyword: _REF}, f"#/{keyword}", id=keyword) + for keyword in ( + "additionalItems", + "additionalProperties", + "contains", + "else", + "if", + "items", + "not", + "propertyNames", + "then", + "unevaluatedItems", + "unevaluatedProperties", + ) + ), + pytest.param( + {"dependencies": {"user": _REF, "id": ["user"]}}, + "#/dependencies/user", + id="dependencies_schema_and_property_list", + ), + ], +) +def test_schema_walker_follows_keyword(keyword_schema: dict[str, Any], pointer: str) -> None: + catalog = _catalog(_stream(json_schema={"type": "object", **keyword_schema})) + assert check_refs_are_resolved(catalog) == [ + f"Stream 'items' has an unresolved $ref '#/definitions/user' at {pointer}." + ] + + +@pytest.mark.parametrize( + "dialect", + [ + pytest.param("http://json-schema.org/draft-04/schema#", id="draft_04"), + pytest.param("http://json-schema.org/draft-04/schema", id="draft_04_without_fragment"), + ], +) +def test_valid_json_schema_honours_declared_draft_04(dialect: str) -> None: + """Draft 4 `exclusiveMinimum` is a boolean, which Draft 7 rejects.""" + schema = _with_property("n", {"type": "number", "minimum": 0, "exclusiveMinimum": True}) + assert check_schemas_are_valid_json_schema(_catalog(_stream(json_schema=schema))) != [] + catalog = _catalog(_stream(json_schema={"$schema": dialect, **schema})) + assert check_schemas_are_valid_json_schema(catalog) == [] + + +def test_valid_json_schema_honours_declared_draft_2020_12() -> None: + """Draft 7 ignores `prefixItems`; Draft 2020-12 requires it to be a list.""" + schema = _with_property("pair", {"type": "array", "prefixItems": {"type": "string"}}) + assert check_schemas_are_valid_json_schema(_catalog(_stream(json_schema=schema))) == [] + catalog = _catalog( + _stream(json_schema={"$schema": "https://json-schema.org/draft/2020-12/schema", **schema}) + ) + assert check_schemas_are_valid_json_schema(catalog) == [ + "Stream 'items' has an invalid JSON schema at #/properties/pair/prefixItems: " + "{'type': 'string'} is not of type 'array'" + ] + + +@pytest.mark.parametrize( + "dialect", + [ + pytest.param("https://example.com/my-dialect", id="unknown_uri"), + pytest.param(7, id="not_a_string"), + ], +) +def test_valid_json_schema_falls_back_to_draft_7(dialect: Any) -> None: + schema = {"$schema": dialect, **_with_property("v", {"type": "any"})} + problems = check_schemas_are_valid_json_schema(_catalog(_stream(json_schema=schema))) + assert ( + "Stream 'items' has an invalid JSON schema at #/properties/v/type: " + "'any' is not valid under any of the given schemas" + ) in problems + + +@pytest.mark.parametrize( + "property_schema", + [ + pytest.param( + {"anyOf": [{"type": "string"}, {"type": "null"}], "format": "date-time"}, + id="format_beside_any_of", + ), + pytest.param( + {"oneOf": [{"type": "null"}, {"type": ["string"]}], "format": "date"}, + id="format_beside_one_of", + ), + pytest.param( + { + "anyOf": [{"type": "null"}, {"oneOf": [{"type": "string"}]}], + "airbyte_type": "timestamp_with_timezone", + }, + id="airbyte_type_beside_nested_alternatives", + ), + pytest.param({"format": "date"}, id="format_on_untyped_node"), + pytest.param({"airbyte_type": "integer"}, id="airbyte_type_on_untyped_node"), + ], +) +def test_supported_data_types_reads_types_of_alternatives( + property_schema: dict[str, Any], +) -> None: + catalog = _catalog(_stream(json_schema=_with_property("v", property_schema))) + assert check_supported_data_types(catalog) == [] + + +def test_supported_data_types_still_rejects_incompatible_alternatives() -> None: + catalog = _catalog( + _stream( + json_schema=_with_property( + "v", + { + "anyOf": [{"type": "integer"}, {"type": "null"}], + "format": "date", + "airbyte_type": "time_without_timezone", + }, + ) + ) + ) + assert check_supported_data_types(catalog) == [ + ( + "Stream 'items' uses airbyte_type 'time_without_timezone' on type " + "['integer', 'null'] at #/properties/v; it requires one of ['string']." + ), + ( + "Stream 'items' uses format 'date' on type ['integer', 'null'] " + "at #/properties/v; it requires a string." + ), + ] + + +def test_find_catalog_problems_returns_only_failing_checks() -> None: + catalog = _catalog(_stream(default_cursor_field=["missing"])) + problems = find_catalog_problems(catalog) + assert [check.run for check in problems] == [check_cursor_fields_exist_in_schema] + + +def test_find_catalog_problems_reports_a_crashing_check() -> None: + def crash(catalog: AirbyteCatalog) -> list[str]: + raise KeyError("boom") + + check = CatalogCheck("crashes", crash, enforced=False) + problems = find_catalog_problems(_catalog(_stream()), checks=[check]) + assert problems == {check: ["The check could not run: KeyError: 'boom'"]} + + +def test_every_check_has_a_unique_name() -> None: + names = [check.name for check in CATALOG_CHECKS] + assert len(names) == len(set(names)) + + +def test_catalog_checks_enforcement_wiring() -> None: + """Pin which checks fail `test_discover` and which only warn. + + Promoting a check to enforced changes the result of `test_discover` for every connector, + so it must be a deliberate edit of this test. + """ + assert {check.name: check.enforced for check in CATALOG_CHECKS} == { + "catalog has streams": True, + "stream names are unique": True, + "streams declare sync modes": True, + "refs are resolved": True, + "no unsupported keywords": True, + "additionalProperties is not false": True, + "schemas are valid JSON schema": False, + "cursor fields exist in schema": False, + "primary keys exist in schema": False, + "primary keys have scalar types": False, + "supported data types": False, + } + + +def _catalog_message(*streams: AirbyteStream) -> str: + return json.dumps( + { + "type": "CATALOG", + "catalog": { + "streams": [ + { + "name": stream.name, + "json_schema": stream.json_schema, + "supported_sync_modes": [ + mode.value for mode in stream.supported_sync_modes + ], + "default_cursor_field": stream.default_cursor_field, + } + for stream in streams + ] + }, + } + ) + + +class _FakeSourceSuite(SourceTestSuiteBase): + @classmethod + def create_connector(cls, scenario: ConnectorTestScenario | None) -> Any: + return object() + + @classmethod + def get_connector_root_dir(cls) -> Path: + return Path() + + +@pytest.fixture +def discover_output(monkeypatch: pytest.MonkeyPatch) -> Callable[[EntrypointOutput], None]: + def _set(output: EntrypointOutput) -> None: + monkeypatch.setattr(source_base, "run_test_job", lambda *args, **kwargs: output) + + return _set + + +def test_discover_fails_on_enforced_problem( + discover_output: Callable[[EntrypointOutput], None], +) -> None: + discover_output(EntrypointOutput(messages=[_catalog_message(_stream("a"), _stream("a"))])) + with pytest.raises(AssertionError, match=r"\[stream names are unique\]\n - Stream 'a'"): + _FakeSourceSuite().test_discover(ConnectorTestScenario(status="succeed")) + + +def test_discover_warns_on_advisory_problem( + discover_output: Callable[[EntrypointOutput], None], + monkeypatch: pytest.MonkeyPatch, +) -> None: + advisory = CatalogCheck("cursor check", check_cursor_fields_exist_in_schema, enforced=False) + monkeypatch.setattr( + source_base, + "find_catalog_problems", + lambda catalog: find_catalog_problems(catalog, checks=[advisory]), + ) + discover_output( + EntrypointOutput(messages=[_catalog_message(_stream(default_cursor_field=["x"]))]) + ) + with pytest.warns(DiscoveredCatalogWarning, match=r"\[cursor check\]"): + _FakeSourceSuite().test_discover(ConnectorTestScenario(status="succeed")) + + +def test_discover_fails_without_catalog( + discover_output: Callable[[EntrypointOutput], None], +) -> None: + log = json.dumps({"type": "LOG", "log": {"level": "INFO", "message": "no catalog"}}) + discover_output(EntrypointOutput(messages=[log])) + with pytest.raises(AssertionError, match="Expected exactly one CATALOG message. Got 0."): + _FakeSourceSuite().test_discover(ConnectorTestScenario(status="succeed")) + + +def test_discover_skips_checks_when_scenario_allows_failure( + discover_output: Callable[[EntrypointOutput], None], +) -> None: + discover_output(EntrypointOutput(messages=[], uncaught_exception=ValueError("boom"))) + _FakeSourceSuite().test_discover(ConnectorTestScenario()) + + +def test_discover_runs_checks_when_scenario_allows_failure_and_discover_succeeds( + discover_output: Callable[[EntrypointOutput], None], +) -> None: + """Most scenarios declare no `status`; their catalog must still be checked.""" + assert ConnectorTestScenario().expected_outcome.expect_success() is False + discover_output(EntrypointOutput(messages=[_catalog_message(_stream("a"), _stream("a"))])) + with pytest.raises(AssertionError, match=r"\[stream names are unique\]"): + _FakeSourceSuite().test_discover(ConnectorTestScenario()) + + +def test_discover_fails_on_more_than_one_catalog( + discover_output: Callable[[EntrypointOutput], None], +) -> None: + message = _catalog_message(_stream()) + discover_output(EntrypointOutput(messages=[message, message])) + with pytest.raises(AssertionError, match="Expected exactly one CATALOG message. Got 2."): + _FakeSourceSuite().test_discover(ConnectorTestScenario(status="succeed")) + + +def test_discover_is_skipped_when_scenario_expects_failure( + discover_output: Callable[[EntrypointOutput], None], +) -> None: + discover_output(EntrypointOutput(messages=[_catalog_message(_stream("a"), _stream("a"))])) + with pytest.raises(pytest.skip.Exception): + _FakeSourceSuite().test_discover(ConnectorTestScenario(status="failed")) + + +def test_discover_separates_advisory_and_enforced_problems( + discover_output: Callable[[EntrypointOutput], None], +) -> None: + discover_output( + EntrypointOutput( + messages=[ + _catalog_message( + _stream("a", default_cursor_field=["missing"]), + _stream("a", default_cursor_field=["missing"]), + ) + ] + ) + ) + with ( + pytest.warns(DiscoveredCatalogWarning) as warning_records, + pytest.raises(AssertionError) as assertion, + ): + _FakeSourceSuite().test_discover(ConnectorTestScenario(status="succeed")) + + [warning_record] = warning_records + warning_text = str(warning_record.message) + assert "[cursor fields exist in schema]" in warning_text + assert "[stream names are unique]" not in warning_text + assertion_text = str(assertion.value) + assert "[stream names are unique]" in assertion_text + assert "[cursor fields exist in schema]" not in assertion_text + + +_MANIFEST: dict[str, Any] = { + "version": "6.0.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": ["items"]}, + "spec": { + "type": "Spec", + "connection_specification": { + "type": "object", + "properties": {"api_key": {"type": "string"}}, + }, + }, + "streams": [ + { + "type": "DeclarativeStream", + "name": "items", + "primary_key": ["id"], + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.example.invalid", + "path": "/items", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + }, + "incremental_sync": { + "type": "DatetimeBasedCursor", + "cursor_field": "updated_at", + "datetime_format": "%Y-%m-%d", + "start_datetime": "2024-01-01", + }, + "schema_loader": {"type": "InlineSchemaLoader", "schema": _SCHEMA}, + } + ], +} + + +def _declarative_suite(tmp_path: Path, manifest: dict[str, Any]) -> DeclarativeSourceTestSuite: + (tmp_path / "manifest.yaml").write_text(yaml.safe_dump(manifest)) + + class _Suite(DeclarativeSourceTestSuite): + @classmethod + def get_connector_root_dir(cls) -> Path: + return tmp_path + + return _Suite() + + +_SCENARIO = ConnectorTestScenario(config_dict={"api_key": "unused"}, status="succeed") + + +def test_declarative_discover_passes_for_valid_catalog(tmp_path: Path) -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error", DiscoveredCatalogWarning) + _declarative_suite(tmp_path, _MANIFEST).test_discover(_SCENARIO) + + +def test_declarative_discover_warns_on_missing_cursor(tmp_path: Path) -> None: + manifest = copy.deepcopy(_MANIFEST) + manifest["streams"][0]["incremental_sync"]["cursor_field"] = "modified_at" + with pytest.warns(DiscoveredCatalogWarning, match="declares cursor field 'modified_at'"): + _declarative_suite(tmp_path, manifest).test_discover(_SCENARIO) + + +def test_declarative_discover_fails_on_closed_schema(tmp_path: Path) -> None: + manifest = copy.deepcopy(_MANIFEST) + manifest["streams"][0]["schema_loader"]["schema"] = {**_SCHEMA, "additionalProperties": False} + with pytest.raises(AssertionError, match="sets additionalProperties to false at #."): + _declarative_suite(tmp_path, manifest).test_discover(_SCENARIO)