diff --git a/airbyte_cdk/test/standard_tests/_spec_compatibility.py b/airbyte_cdk/test/standard_tests/_spec_compatibility.py new file mode 100644 index 0000000000..c62f34f5e9 --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_spec_compatibility.py @@ -0,0 +1,1135 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Backward-compatibility checks between a connector and its published version. + +A connector's spec is the contract with every config that users have already saved. When a new +version is rolled out, the platform validates those saved configs against the new spec, so a +change that rejects a config the previous version accepted breaks existing connections at +upgrade time. `compare_specs` classifies every difference between two specs as breaking or +compatible, and the helpers below fetch the previously published spec from the connector +registry and decide whether a breaking change has been declared through the breaking-change +process. + +The comparison rules are ported from the spec comparator of the regression tests in +`airbyte-ops-mcp`. Two differences keep this check, which blocks a merge, focused on what existing +connections depend on: only the config-relevant parts of the spec are compared (see +`COMPARED_SPEC_KEYS`), and changes to the OAuth consent flow are compatible, because they apply +to new authorizations only. +""" + +from __future__ import annotations + +from collections.abc import Hashable, Mapping +from dataclasses import dataclass, field +from typing import Any, Literal + +import requests +from packaging.version import InvalidVersion, Version +from requests.adapters import HTTPAdapter, Retry + +DeploymentMode = Literal["oss", "cloud"] + +REGISTRY_ENTRY_URL_TEMPLATE = ( + "https://connectors.airbyte.com/files/metadata/{docker_repository}/{version}/{registry}.json" +) +"""Public URL of a published registry entry. `version` is a version tag or `latest`.""" + +REGISTRY_TIMEOUT_SECONDS = (5.0, 10.0) +"""Connect and read timeouts of one registry request. + +With the retries of `registry_retry`, an unreachable registry is given up on in about 30 seconds +rather than holding every image-test run for minutes. +""" + +REGISTRY_UNAVAILABLE_ERRORS: tuple[type[requests.RequestException], ...] = ( + requests.ConnectionError, + requests.Timeout, + requests.exceptions.RetryError, +) +"""Errors that mean the registry could not be reached or kept failing, rather than answering. + +The test is skipped, with a warning, when one of these persists after retries: a registry outage +or a network without access to it must not fail every connector's tests. Any other error, such as +a client error status or an entry that is not a registry entry, still fails the test. +""" + +BREAKING_CHANGES_DOCS_URL = ( + "https://docs.airbyte.com/platform/connector-development/connector-breaking-changes" +) + +DEPLOYMENT_MODE_ENV_VARS: dict[DeploymentMode, dict[str, str]] = { + "oss": {"DEPLOYMENT_MODE": "oss", "AIRBYTE_EDITION": "COMMUNITY"}, + "cloud": {"DEPLOYMENT_MODE": "cloud", "AIRBYTE_EDITION": "CLOUD"}, +} +"""Environment the registry publisher sets when it runs `spec` for each registry. + +Legacy Java and Python CDKs read `DEPLOYMENT_MODE`; the bulk CDK reads `AIRBYTE_EDITION`. +Running the current image with the same variables makes the two specs comparable. +""" + +COMPARED_SPEC_KEYS = ( + "connectionSpecification", + "advanced_auth", + "supported_destination_sync_modes", +) +"""The parts of a spec that existing connections depend on. + +`connectionSpecification` validates saved configs, `advanced_auth` says where the platform reads +and writes OAuth values in a config, and `supported_destination_sync_modes` lists the modes that +existing connections may use. The other keys are documentation or legacy flags. +""" + +MAX_REPORTED_CHANGES = 20 +"""How many changes of one kind `format_spec_change_summary` lists before it truncates.""" + + +@dataclass +class SpecComparison: + """The differences between two specs, split by whether they break saved configs.""" + + breaking: list[str] = field(default_factory=list) + compatible: list[str] = field(default_factory=list) + + @property + def is_backward_compatible(self) -> bool: + return not self.breaking + + +@dataclass(frozen=True) +class PublishedSpec: + """A spec as the connector registry published it.""" + + version: str + spec: dict[str, Any] + url: str + + +def compare_specs(previous: Mapping[str, Any], current: Mapping[str, Any]) -> SpecComparison: + """Compare two connector specs, allowing backward-compatible changes. + + The rule is compatibility, not equality: a new optional property passes, because no saved + config becomes invalid, while a removed property, a narrowed type, a tightened constraint or + a property that has become required fails, because saved configs can. Only JSON Schema + validation keywords and the protocol keys that locate values in a config can break one; + every other key (titles, descriptions, `order`, `airbyte_hidden`, `$schema` and so on) is an + annotation, and changing it is compatible. Default changes are compatible too. + + Args: + previous: The `spec` object the published version emitted, as plain JSON. + current: The `spec` object the version under test emits, as plain JSON. + """ + comparison = SpecComparison() + _diff_node(_compared_part(previous), _compared_part(current), "", None, None, comparison) + return comparison + + +def _compared_part(spec: Mapping[str, Any]) -> dict[str, Any]: + return {key: spec[key] for key in COMPARED_SPEC_KEYS if spec.get(key) is not None} + + +# Keys whose value maps config field names to schemas. Their keys are names a connector chose, +# so a field called `description` is a field, not documentation. +_PROPERTY_MAP_KEYS = frozenset({"$defs", "definitions", "patternProperties", "properties"}) + +# Keys that constrain what a config may contain. Adding one narrows the set of valid configs; +# removing one only widens it. +_CONSTRAINT_KEYS = frozenset( + { + "additionalProperties", + "const", + "enum", + "exclusiveMaximum", + "exclusiveMinimum", + "format", + "maxContains", + "maxItems", + "maxLength", + "maxProperties", + "maximum", + "minContains", + "minItems", + "minLength", + "minProperties", + "minimum", + "multipleOf", + "pattern", + "required", + "type", + "uniqueItems", + } +) + +# The value a keyword takes when it is absent. A keyword set to one of these accepts exactly what +# its absence accepts, so adding or removing it changes nothing, and moving a value keyword to one +# of these relaxes it. For a boolean keyword, any other value is the strict one. The draft-04 +# boolean form of `exclusiveMaximum`/`exclusiveMinimum` defaults to false. +_KEYWORD_DEFAULTS: dict[str, tuple[Any, ...]] = { + "additionalItems": (True, {}), + "additionalProperties": (True, {}), + "exclusiveMaximum": (False,), + "exclusiveMinimum": (False,), + "items": (True, {}), + "minContains": (1,), + "minItems": (0,), + "minLength": (0,), + "minProperties": (0,), + "patternProperties": ({},), + "properties": ({},), + "required": ([],), + "uniqueItems": (False,), +} + +# Keywords with a default whose value describes fields rather than constrains a value. +_FIELD_KEYWORDS = frozenset({"items", "patternProperties", "properties"}) + +# Keywords whose value is a schema or a boolean. `false` rejects every value they apply to. +_SCHEMA_OR_FALSE_KEYS = frozenset({"additionalItems", "additionalProperties", "items"}) + +_BOUNDS_RELAXED_BY_GROWING = frozenset( + {"exclusiveMaximum", "maxContains", "maxItems", "maxLength", "maxProperties", "maximum"} +) +_BOUNDS_RELAXED_BY_SHRINKING = frozenset( + {"exclusiveMinimum", "minContains", "minItems", "minLength", "minProperties", "minimum"} +) + +# Keys that decide the shape of a node rather than constrain a value. Removing one drops the +# fields it describes, so it is breaking like a removed property. +_STRUCTURE_KEYS = frozenset({"$ref", "allOf", "anyOf", "items", "oneOf"}) + +# Validation keywords this check does not model. Removing one only widens what a config may set; +# adding or changing one may narrow it in ways that are not compared here, so it is breaking. +# None of them appears in the connection specs of the connector fleet today. +_UNMODELED_VALIDATION_KEYS = frozenset( + { + "additionalItems", + "contains", + "dependencies", + "dependentRequired", + "dependentSchemas", + "else", + "if", + "not", + "prefixItems", + "propertyNames", + "then", + "unevaluatedItems", + "unevaluatedProperties", + } +) + +# Protocol keys that say where the platform reads and writes values in a saved config. Moving or +# removing one breaks existing connections even though no JSON Schema keyword changed. +_PROTOCOL_KEYS = frozenset( + { + *COMPARED_SPEC_KEYS, + "auth_flow_type", + "complete_oauth_output_specification", + "complete_oauth_server_input_specification", + "complete_oauth_server_output_specification", + "oauth_config_specification", + "oauth_user_input_from_connector_config_specification", + "path_in_connector_config", + "path_in_oauth_response", + "predicate_key", + "predicate_value", + } +) + +# Marks a field whose value the platform stores in its secret store. Marking more fields is +# compatible; unmarking one changes how values that were already saved as secrets are handled. +_SECRET_KEY = "airbyte_secret" + +# Keys of `advanced_auth` that define how a new OAuth consent is obtained (consent and token URLs, +# scopes, which outputs to extract). Saved configs and their tokens do not depend on them. +_OAUTH_CONSENT_FLOW_KEYS = frozenset({"oauth_connector_input_specification"}) + +_CONFIG_RELEVANT_KEYS = frozenset( + { + *_PROPERTY_MAP_KEYS, + *_CONSTRAINT_KEYS, + *_STRUCTURE_KEYS, + *_UNMODELED_VALIDATION_KEYS, + *_PROTOCOL_KEYS, + *_OAUTH_CONSENT_FLOW_KEYS, + _SECRET_KEY, + "default", + } +) +"""Every key whose change can matter to a saved config. Any other key is an annotation.""" + +# Keys whose list value is a set of allowed values. Every other list is positional: in +# `path_in_connector_config`, `["credentials", "client_id"]` is a different location from +# `["client_id"]`, not a wider one. +_SET_VALUED_KEYS = frozenset({"enum", "supported_destination_sync_modes"}) + +# Keys holding alternative shapes for the same node. The platform picks a branch by its +# discriminating `const`, not by its index, so branches are matched by what identifies them. +_BRANCH_KEYS = frozenset({"anyOf", "oneOf"}) + +_MAX_VALUE_CHARS = 120 + + +def _is_annotation(key: str) -> bool: + return key not in _CONFIG_RELEVANT_KEYS + + +def _is_keyword_default(key: str, value: Any) -> bool: + return any(_same_json(value, default) for default in _KEYWORD_DEFAULTS.get(key, ())) + + +def _same_json(left: Any, right: Any) -> bool: + """JSON equality, which unlike Python's does not equate `false` with `0` or `true` with `1`. + + The comparison recurses into objects and arrays, so `{"const": 0}` and `{"const": false}` + differ too. `1` and `1.0` are the same JSON number and stay equal. + """ + if isinstance(left, bool) or isinstance(right, bool): + return isinstance(left, bool) and isinstance(right, bool) and left == right + if isinstance(left, dict) and isinstance(right, dict): + return left.keys() == right.keys() and all( + _same_json(value, right[key]) for key, value in left.items() + ) + if isinstance(left, list) and isinstance(right, list): + return len(left) == len(right) and all(map(_same_json, left, right)) + if isinstance(left, (dict, list)) or isinstance(right, (dict, list)): + return False + return bool(left == right) + + +def _contains_json(values: list[Any], value: Any) -> bool: + return any(_same_json(value, candidate) for candidate in values) + + +def _is_secret(value: Any) -> bool: + # The platform reads the flag leniently, so the string "true" also marks a secret. + return value is True or (isinstance(value, str) and value.strip().lower() == "true") + + +def _diff_node( + previous: Any, + current: Any, + path: str, + key: str | None, + property_map: str | None, + comparison: SpecComparison, +) -> None: + """Compare two values found under `key`. + + `property_map` names the keyword (`properties`, `patternProperties`, ...) when the values + map field names to schemas, and is `None` when they are schemas or other spec objects. + """ + if _same_json(previous, current): + return + + if isinstance(previous, dict) and isinstance(current, dict): + _diff_object(previous, current, path, property_map, comparison) + return + + if isinstance(previous, list) and isinstance(current, list): + _diff_list(previous, current, path, key, comparison) + return + + comparison.breaking.append( + f"{_label(path)} changed from {_brief(previous)} to {_brief(current)}" + ) + + +def _diff_object( + previous: dict[str, Any], + current: dict[str, Any], + path: str, + property_map: str | None, + comparison: SpecComparison, +) -> None: + is_property_map = property_map is not None + required_handled = not is_property_map and _diff_required( + previous.get("required"), current.get("required"), path, comparison + ) + + for key in previous: + if required_handled and key == "required": + continue + + child_path = _child_path(path, key) + if key not in current: + _record_removed_key(key, previous[key], child_path, is_property_map, comparison) + continue + + if not _same_json(previous[key], current[key]): + _diff_member(key, previous[key], current[key], child_path, is_property_map, comparison) + + for key in current: + if key in previous or (required_handled and key == "required"): + continue + _record_added_key(key, current[key], _child_path(path, key), property_map, comparison) + + +def _record_removed_key( + key: str, + value: Any, + child_path: str, + is_property_map: bool, + comparison: SpecComparison, +) -> None: + label = _label(child_path) + + if is_property_map: + comparison.breaking.append(f"{label} was removed") + elif key in _OAUTH_CONSENT_FLOW_KEYS: + comparison.compatible.append(f"{label} was removed (OAuth consent flow)") + elif key == _SECRET_KEY: + _diff_secret(value, None, label, comparison) + elif _is_annotation(key): + comparison.compatible.append(f"{label} was removed (annotation)") + elif key == "default": + comparison.compatible.append(f"{label} was removed (changes behavior, not validity)") + elif _is_keyword_default(key, value): + comparison.compatible.append(f"{label} was removed; it was at its default value") + elif key in _CONSTRAINT_KEYS or key in _UNMODELED_VALIDATION_KEYS: + comparison.compatible.append(f"{label} was removed, widening what a config may set") + else: + comparison.breaking.append(f"{label} was removed") + + +def _record_added_key( + key: str, + value: Any, + child_path: str, + property_map: str | None, + comparison: SpecComparison, +) -> None: + label = _label(child_path) + + if property_map == "patternProperties" or ( + property_map is None and key == "patternProperties" and not _is_keyword_default(key, value) + ): + # A pattern applies to every field whose name matches it, declared or not, so it can + # reject the value of a field that saved configs already set. + comparison.breaking.append( + f"{label} was added; the existing fields whose names match it must now satisfy it" + ) + elif property_map is not None or key in _PROPERTY_MAP_KEYS: + comparison.compatible.append(f"{label} was added") + elif key == _SECRET_KEY: + comparison.compatible.append(f"{label} was added") + elif _is_annotation(key): + comparison.compatible.append(f"{label} was added (annotation)") + elif key == "default": + comparison.compatible.append(f"{label} was added (changes behavior, not validity)") + elif _is_keyword_default(key, value): + comparison.compatible.append( + f"{label} was added at its default value {_brief(value)}, which allows the same configs" + ) + elif key in _CONSTRAINT_KEYS: + comparison.breaking.append(f"{label} was added, narrowing what a config may set") + elif key in _STRUCTURE_KEYS or key in _UNMODELED_VALIDATION_KEYS: + comparison.breaking.append(f"{label} was added, changing the shape of this node") + else: + comparison.compatible.append(f"{label} was added") + + +def _diff_member( + key: str, + previous_value: Any, + current_value: Any, + child_path: str, + is_property_map: bool, + comparison: SpecComparison, +) -> None: + label = _label(child_path) + + if not is_property_map: + if key in _OAUTH_CONSENT_FLOW_KEYS: + comparison.compatible.append( + f"{label} changed (OAuth consent flow, applies to new authorizations only)" + ) + return + + if key == _SECRET_KEY: + _diff_secret(previous_value, current_value, label, comparison) + return + + if _is_annotation(key): + comparison.compatible.append(f"{label} changed (annotation)") + return + + if key == "default": + comparison.compatible.append( + f"{label} changed from {_brief(previous_value)} to {_brief(current_value)} " + "(changes behavior, not validity)" + ) + return + + if key == "type": + _diff_type(previous_value, current_value, label, comparison) + return + + if key in _UNMODELED_VALIDATION_KEYS and not _is_keyword_default(key, current_value): + comparison.breaking.append( + f"{label} changed; this check does not model `{key}`, so any change may narrow " + "what a config may set" + ) + return + + if key in _KEYWORD_DEFAULTS: + # Emptying `properties` or `items` drops the fields they describe, which the + # recursive comparison reports, so only value keywords relax to their default here. + if _is_keyword_default(key, current_value) and key not in _FIELD_KEYWORDS: + comparison.compatible.append( + f"{label} relaxed to its default {_brief(current_value)}" + ) + return + if isinstance(previous_value, bool) and isinstance(current_value, bool): + comparison.breaking.append(f"{label} tightened to {current_value!r}") + return + if key in _SCHEMA_OR_FALSE_KEYS and previous_value is False: + # `false` accepts nothing, so any schema accepts at least as much. + comparison.compatible.append( + f"{label} relaxed from False to {_brief(current_value)}" + ) + return + + if _is_number(previous_value) and _is_number(current_value): + if key == "multipleOf": + _diff_multiple_of(previous_value, current_value, label, comparison) + return + if _diff_bound(key, previous_value, current_value, label, comparison): + return + + _diff_node( + previous_value, + current_value, + child_path, + key, + key if not is_property_map and key in _PROPERTY_MAP_KEYS else None, + comparison, + ) + + +def _diff_secret( + previous_value: Any, + current_value: Any, + label: str, + comparison: SpecComparison, +) -> None: + """Compare `airbyte_secret` by direction. `current_value` is `None` when the key was removed. + + Values saved while a field was secret live in the platform's secret store, so a field that + stops being secret changes how those saved values are read and shown. A field that becomes + secret, or a flag rewritten without changing its meaning (`"true"` to `true`), is compatible. + """ + was_secret = _is_secret(previous_value) + is_secret = _is_secret(current_value) + change = "was removed" if current_value is None else "changed" + + if was_secret and not is_secret: + comparison.breaking.append( + f"{label} {change}, so the field is no longer a secret: values saved as secrets may " + "be exposed or no longer read from the secret store" + ) + elif is_secret and not was_secret: + comparison.compatible.append(f"{label} {change}, so the field is now a secret") + else: + comparison.compatible.append( + f"{label} {change} from {_brief(previous_value)} to {_brief(current_value)} " + "(same meaning)" + ) + + +def _diff_type( + previous_value: Any, + current_value: Any, + label: str, + comparison: SpecComparison, +) -> None: + """Compare two `type` declarations as the sets of values they allow. + + `"string"` becoming `["null", "string"]` accepts strictly more than before, and so does + `"integer"` becoming `"number"`, so comparing the raw values would wrongly call either a break. + """ + previous_types = _as_type_set(previous_value) + current_types = _as_type_set(current_value) + + removed = previous_types - current_types + added = current_types - previous_types + if "integer" in removed and "number" in current_types: + removed.discard("integer") + added.discard("number") + comparison.compatible.append(f"{label} widened from integer to number") + + if removed: + comparison.breaking.append(f"{label} no longer allows {', '.join(sorted(removed))}") + if added: + comparison.compatible.append(f"{label} also allows {', '.join(sorted(added))}") + + +def _as_type_set(value: Any) -> set[str]: + if isinstance(value, list): + return {str(entry) for entry in value} + return {str(value)} + + +def _diff_bound( + key: str, + previous_value: float, + current_value: float, + label: str, + comparison: SpecComparison, +) -> bool: + """Classify a moved numeric bound. Returns whether `key` is a bound at all.""" + if key in _BOUNDS_RELAXED_BY_GROWING: + relaxed = current_value > previous_value + elif key in _BOUNDS_RELAXED_BY_SHRINKING: + relaxed = current_value < previous_value + else: + return False + + verb = "relaxed" if relaxed else "tightened" + message = f"{label} {verb} from {previous_value!r} to {current_value!r}" + (comparison.compatible if relaxed else comparison.breaking).append(message) + return True + + +def _diff_multiple_of( + previous_value: float, + current_value: float, + label: str, + comparison: SpecComparison, +) -> None: + """A new `multipleOf` relaxes the old one when it divides it: every multiple of 4 is even.""" + quotient = previous_value / current_value if current_value else 0.5 + relaxed = current_value > 0 and abs(quotient - round(quotient)) < 1e-9 + verb = "relaxed" if relaxed else "changed" + message = f"{label} {verb} from {previous_value!r} to {current_value!r}" + (comparison.compatible if relaxed else comparison.breaking).append(message) + + +def _is_number(value: Any) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def _diff_list( + previous: list[Any], + current: list[Any], + path: str, + key: str | None, + comparison: SpecComparison, +) -> None: + if key in _BRANCH_KEYS: + _diff_branches(previous, current, path, exclusive=key == "oneOf", comparison=comparison) + return + + if key in _SET_VALUED_KEYS: + for entry in previous: + if not _contains_json(current, entry): + comparison.breaking.append(f"{_label(path)} no longer allows {_brief(entry)}") + for entry in current: + if not _contains_json(previous, entry): + comparison.compatible.append(f"{_label(path)} also allows {_brief(entry)}") + return + + for index, previous_entry in enumerate(previous): + child_path = f"{path}[{index}]" + if index >= len(current): + comparison.breaking.append(f"{_label(child_path)} was removed") + continue + _diff_node(previous_entry, current[index], child_path, key, None, comparison) + + for index in range(len(previous), len(current)): + comparison.breaking.append( + f"{_label(f'{path}[{index}]')} was added, changing an ordered list" + ) + + +def _diff_branches( + previous: list[Any], + current: list[Any], + path: str, + *, + exclusive: bool, + comparison: SpecComparison, +) -> None: + """Compare `oneOf`/`anyOf` branches, matched by what identifies them. + + Reordering auth methods is a common, harmless edit; compared by position it would read as + every field of both branches being replaced. A previous branch without a partner is reported + as removed, and a current one as added. Paths follow each version's own ordering. + + Under `oneOf` (`exclusive`), a config must match exactly one branch, so a branch that can + also match the configs of another previous branch rejects them. An added branch is + compatible only if `_excludes` proves it rejects the configs of every previous branch, and a + matched branch is breaking if it loses that proof for a previous branch it had it for, for + example when its discriminating `const` is removed. Under `anyOf` an added branch only widens. + """ + partners = _match_branches(previous, current) + + for index, branch in enumerate(previous): + partner = partners.get(index) + if partner is None: + comparison.breaking.append(f"{_label(f'{path}[{index}]')} was removed") + continue + branch_path = f"{path}[{index}]" + _diff_node(branch, current[partner], branch_path, None, None, comparison) + if exclusive: + overlapped = [ + _label(f"{path}[{other_index}]") + for other_index, other in enumerate(previous) + if other_index != index + and _excludes(branch, previous, other_index) + and not _excludes(current[partner], previous, other_index) + ] + if overlapped: + comparison.breaking.append( + f"{_label(branch_path)} may now also match the configs of " + f"{', '.join(overlapped)}, which `oneOf` then rejects" + ) + + matched = set(partners.values()) + for index in range(len(current)): + if index in matched: + continue + branch_path = _label(f"{path}[{index}]") + if exclusive and not all( + _excludes(current[index], previous, other_index) for other_index in range(len(previous)) + ): + comparison.breaking.append( + f"{branch_path} was added and may also match existing configs, which `oneOf` " + "then rejects; give it a required field, or a required discriminating `const`, " + "that the other branches do not share" + ) + else: + comparison.compatible.append(f"{branch_path} was added") + + +def _excludes(branch: Any, previous: list[Any], other_index: int) -> bool: + """Whether `branch` provably rejects every valid config of the previous branch `other_index`. + + It does when it requires a field that the other branch does not declare, so the other + branch's configs do not set it, or when both pin a field to different values. A pinned + field that neither branch requires still tells them apart if the other branch's configs + that leave it out were invalid already, because another previous branch matched them too. + Configs are assumed to set only the fields their branch declares, as the UI writes them. + """ + other = previous[other_index] + if not isinstance(branch, dict) or not isinstance(other, dict): + return False + required = _required_names(branch) + other_required = _required_names(other) + if any(name not in _property_schemas(other) for name in required): + return True + + other_pinned = _discriminators(other) + for name, value in _discriminators(branch).items(): + if name not in other_pinned or _same_json(value, other_pinned[name]): + continue + if name in required or name in other_required: + return True + if any( + _matches_configs_without(candidate, other, name) + for candidate_index, candidate in enumerate(previous) + if candidate_index != other_index + ): + return True + return False + + +def _matches_configs_without(candidate: Any, branch: dict[str, Any], name: str) -> bool: + """Whether `candidate` accepts every config of `branch` that does not set the field `name`. + + Only a candidate that constrains nothing but the fields it declares is considered. Each of + them must be one that `branch` does not declare either, or declares with the same schema. + """ + if not isinstance(candidate, dict): + return False + for key, value in candidate.items(): + if _is_annotation(key) or key in ("properties", "required"): + continue + if key == "type" and "object" in _as_type_set(value): + continue + if key == "additionalProperties" and _is_keyword_default(key, value): + continue + return False + + if not _required_names(candidate) <= _required_names(branch) - {name}: + return False + fields = _property_schemas(branch) + return all( + field_name == name or field_name not in fields or _same_json(schema, fields[field_name]) + for field_name, schema in _property_schemas(candidate).items() + ) + + +def _property_schemas(branch: dict[str, Any]) -> dict[str, Any]: + properties = branch.get("properties") + return properties if isinstance(properties, dict) else {} + + +def _required_names(branch: dict[str, Any]) -> set[str]: + required = branch.get("required") + if not isinstance(required, list) or not _is_name_list(required): + return set() + return set(required) + + +def _match_branches(previous: list[Any], current: list[Any]) -> dict[int, int]: + """Pair each previous branch with the current branch that is most likely the same one. + + Candidate pairs are ranked by, in order: being identical; the discriminating `const` values + they share, minus those they disagree on; the same title; how many fields and types they + share; and how close their positions are. Pairs are then taken greedily from the best. Two + branches that disagree on a discriminator and share none are different auth methods, never + a pair, so a renamed discriminator reads as one branch removed and one added. + """ + ranked: list[tuple[tuple[Any, ...], int, int]] = [] + for previous_index, previous_branch in enumerate(previous): + for current_index, current_branch in enumerate(current): + rank = _branch_pair_rank( + previous_branch, current_branch, abs(previous_index - current_index) + ) + if rank is not None: + ranked.append((rank, previous_index, current_index)) + + ranked.sort(key=lambda item: (item[0], -item[1], -item[2]), reverse=True) + + partners: dict[int, int] = {} + matched_current: set[int] = set() + for _, previous_index, current_index in ranked: + if previous_index in partners or current_index in matched_current: + continue + partners[previous_index] = current_index + matched_current.add(current_index) + return partners + + +def _branch_pair_rank(previous: Any, current: Any, distance: int) -> tuple[Any, ...] | None: + previous_discriminators = _discriminators(previous) + current_discriminators = _discriminators(current) + shared_names = previous_discriminators.keys() & current_discriminators.keys() + agreeing = sum( + _same_json(previous_discriminators[name], current_discriminators[name]) + for name in shared_names + ) + disagreeing = len(shared_names) - agreeing + if disagreeing and not agreeing: + return None + + previous_title = previous.get("title") if isinstance(previous, dict) else None + current_title = current.get("title") if isinstance(current, dict) else None + same_title = isinstance(previous_title, str) and previous_title == current_title + + return ( + _same_json(previous, current), + agreeing - disagreeing, + same_title, + _similarity(_branch_features(previous), _branch_features(current)), + -distance, + ) + + +def _discriminators(branch: Any) -> dict[str, Hashable]: + """The single-valued properties of a branch, such as `auth_type: {const: "oauth2.0"}`.""" + if not isinstance(branch, dict) or not isinstance(branch.get("properties"), dict): + return {} + return { + str(name): value + for name, schema in branch["properties"].items() + if isinstance(schema, dict) + for value in (_single_valued(schema),) + if value is not None + } + + +def _single_valued(schema: dict[str, Any]) -> Hashable | None: + value = schema.get("const") + if value is None: + enum = schema.get("enum") + value = enum[0] if isinstance(enum, list) and len(enum) == 1 else None + return value if isinstance(value, Hashable) else None + + +def _branch_features(branch: Any) -> set[str]: + if not isinstance(branch, dict): + return set() + properties = branch.get("properties") + features = ( + {f"property:{name}" for name in properties} if isinstance(properties, dict) else set() + ) + return features | {f"type:{name}" for name in _as_type_set(branch.get("type"))} + + +def _similarity(left: set[str], right: set[str]) -> float: + union = left | right + return len(left & right) / len(union) if union else 0.0 + + +def _diff_required( + previous_required: Any, + current_required: Any, + path: str, + comparison: SpecComparison, +) -> bool: + """Compare the `required` list of one schema node. Returns whether it was compared here. + + Requiring a property that was optional invalidates every saved config that omitted it, even + when the property is new and even when the node had no `required` list before. A `required` + value that is not a list of names is left to the generic comparison. + """ + if previous_required is None and current_required is None: + return False + + previous_names = [] if previous_required is None else previous_required + current_names = [] if current_required is None else current_required + if not _is_name_list(previous_names) or not _is_name_list(current_names): + return False + + node = _label(path) + for name in current_names: + if name not in previous_names: + comparison.breaking.append(f"{node}: `{name}` is now required") + for name in previous_names: + if name not in current_names: + comparison.compatible.append(f"{node}: `{name}` is no longer required") + return True + + +def _is_name_list(value: Any) -> bool: + return isinstance(value, list) and all(isinstance(entry, str) for entry in value) + + +def _child_path(path: str, key: str) -> str: + return f"{path}.{key}" if path else key + + +def _label(path: str) -> str: + return f"`{path}`" if path else "the spec" + + +def _brief(value: Any) -> str: + text = repr(value) + return text if len(text) <= _MAX_VALUE_CHARS else f"{text[:_MAX_VALUE_CHARS]}…" + + +def registry_retry() -> Retry: + """The retry policy of registry requests. + + Transient server errors and rate limits are retried twice with a short backoff. A + `Retry-After` header is not honored, so a rate-limited registry cannot stall the test. + """ + # GET is retried by default, so `allowed_methods`, which needs urllib3 1.26, is not passed. + return Retry( + total=2, + backoff_factor=0.5, + status_forcelist=(429, 500, 502, 503, 504), + respect_retry_after_header=False, + ) + + +def registry_session() -> requests.Session: + """A session that applies `registry_retry` to every registry request.""" + session = requests.Session() + adapter = HTTPAdapter(max_retries=registry_retry()) + session.mount("https://", adapter) + session.mount("http://", adapter) + return session + + +def fetch_published_spec( + docker_repository: str, + registry: DeploymentMode, + version: str = "latest", +) -> PublishedSpec | None: + """Fetch a published spec from the public connector registry. + + Args: + docker_repository: The connector's `dockerRepository`, e.g. `airbyte/source-faker`. + registry: Which registry entry to read. Each deployment mode has its own entry, because + a connector may emit a different spec in each. + version: A published version tag, or `latest` for the version most users run. During a + progressive rollout `latest` stays on the previous release until the rollout ends. + + Returns: + The published spec, or `None` if no entry was published for this registry, which is the + case for a connector that was never released or is disabled in that registry. + + Raises: + requests.RequestException: If the registry cannot be reached after retries, or answers + with an error status other than 404. `REGISTRY_UNAVAILABLE_ERRORS` lists the errors + that mean it could not be reached. + AssertionError: If the registry answers with something that is not a registry entry. + """ + url = REGISTRY_ENTRY_URL_TEMPLATE.format( + docker_repository=docker_repository, + version=version, + registry=registry, + ) + with registry_session() as session: + response = session.get(url, timeout=REGISTRY_TIMEOUT_SECONDS) + + if response.status_code == 404: + return None + response.raise_for_status() + + try: + entry = response.json() + except ValueError as error: + raise AssertionError( + f"The connector registry returned invalid JSON at {url}: {error}" + ) from error + if ( + not isinstance(entry, dict) + or not isinstance(entry.get("spec"), dict) + or entry.get("dockerImageTag") is None + ): + keys = sorted(entry) if isinstance(entry, dict) else type(entry).__name__ + raise AssertionError( + f"The connector registry entry at {url} has no `spec` object or no `dockerImageTag` " + f"(found: {keys}), so the published spec cannot be compared." + ) + return PublishedSpec(version=str(entry["dockerImageTag"]), spec=entry["spec"], url=url) + + +def is_newer_version(version: str, than: str) -> bool: + """Whether `version` is a later release than `than`. Unparsable versions are never newer. + + A connector whose `dockerImageTag` is older than the published `latest`, such as a branch + that has not merged a release from the base branch yet, has no earlier spec to compare + against: the published spec contains changes the branch has never seen. + """ + try: + return Version(version) > Version(than) + except InvalidVersion: + return False + + +def declared_breaking_changes( + metadata: Mapping[str, Any], + previous_version: str, + current_version: str, +) -> list[str]: + """List the breaking changes declared after `previous_version`, up to `current_version`. + + A declared breaking change covers every later version until the release that carries it + reaches `latest`, which matters during a progressive rollout: a patch on top of a major that + is still rolling out is compared against the version before that major. A pre-release such + as `2.0.0-rc.1` is covered by the breaking change declared for the release it precedes. + + Args: + metadata: The `data` section of the connector's `metadata.yaml`. + previous_version: The published version the spec is compared against. + current_version: The connector's `dockerImageTag`. + """ + releases = metadata.get("releases") or {} + breaking_changes = releases.get("breakingChanges") or {} + + try: + lower = Version(previous_version) + upper = Version(current_version) + except InvalidVersion: + return [current_version] if current_version in breaking_changes else [] + if upper.is_prerelease or upper.is_devrelease: + upper = Version(upper.base_version) + + declared: list[str] = [] + for version in breaking_changes: + try: + if lower < Version(str(version)) <= upper: + declared.append(str(version)) + except InvalidVersion: + continue + return declared + + +def format_breaking_spec_changes( + *, + connector_name: str, + current_version: str, + registry: DeploymentMode, + published: PublishedSpec, + comparison: SpecComparison, +) -> str: + """Explain a failed spec comparison, and how to resolve it through the breaking-change process.""" + findings = "\n".join(f" - {finding}" for finding in comparison.breaking) + return ( + f"The {registry.upper()} spec of `{connector_name}` {current_version} is not backward " + f"compatible with the published version {published.version} ({published.url}):\n" + f"{findings}\n\n" + "Existing connections keep their saved configs when they upgrade, and those configs are " + "validated against the new spec, so each change above can break an existing connection.\n" + "To resolve this, do one of the following:\n" + " - Make the change backward compatible, for example by keeping the old field or value, " + "or by making a new field optional.\n" + " - If the change is intentional, release it as a breaking change: bump the major " + "version and declare it under `releases.breakingChanges` in metadata.yaml, with a " + f"migration guide. See {BREAKING_CHANGES_DOCS_URL}\n" + " - If a config migration rewrites existing configs into the new shape, or a finding is " + "a false positive, waive this check until the next release by adding " + "`backward_compatibility_tests_config: {disable_for_version: " + f'"{published.version}"}}` to an entry of the `spec` section in ' + "acceptance-test-config.yml. The waiver expires when a newer version is published." + ) + + +def format_spec_change_summary( + *, + registry: DeploymentMode, + published: PublishedSpec, + comparison: SpecComparison, + waiver: str | None = None, +) -> str: + """Summarize a passed or waived spec comparison, so a reviewer can see what was judged safe. + + Each list is capped at `MAX_REPORTED_CHANGES` entries. + + Args: + waiver: Why the breaking changes, if any, were waived. + """ + lines = [ + f"{registry.upper()} spec compared with the published version {published.version}: " + f"{len(comparison.breaking)} breaking, {len(comparison.compatible)} compatible change(s)." + ] + if comparison.breaking: + lines.append(f"Breaking changes, waived ({waiver}):" if waiver else "Breaking changes:") + lines.extend(_capped(comparison.breaking)) + if comparison.compatible: + lines.append("Compatible changes:") + lines.extend(_capped(comparison.compatible)) + return "\n".join(lines) + + +def _capped(changes: list[str]) -> list[str]: + lines = [f" - {change}" for change in changes[:MAX_REPORTED_CHANGES]] + if len(changes) > MAX_REPORTED_CHANGES: + lines.append(f" … and {len(changes) - MAX_REPORTED_CHANGES} more") + return lines + + +def disabled_for_version(acceptance_test_config: Any) -> list[str]: + """Return the versions the spec backward-compatibility test is disabled for. + + Reads `backward_compatibility_tests_config.disable_for_version` from the entries of the + `spec` section of `acceptance-test-config.yml`, the same setting the legacy acceptance tests + honored. The test is skipped only while the published `latest` version equals one of these, + so the waiver expires on its own once the new version is released. + + Both layouts of the section are read: the current one, where the entries are under + `spec.tests`, and the legacy one, where `spec` is the list of entries itself. A config of + any other shape, including an empty file, has no waivers. + """ + if not isinstance(acceptance_test_config, Mapping): + return [] + acceptance_tests = acceptance_test_config.get("acceptance_tests") + if not isinstance(acceptance_tests, Mapping): + return [] + spec_section = acceptance_tests.get("spec") + tests = spec_section.get("tests") if isinstance(spec_section, Mapping) else spec_section + if not isinstance(tests, list): + return [] + + versions: list[str] = [] + for test in tests: + if not isinstance(test, Mapping): + continue + config = test.get("backward_compatibility_tests_config") + version = config.get("disable_for_version") if isinstance(config, Mapping) else None + if version is not None: + versions.append(str(version)) + return versions diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index aab32e154e..4f18344168 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -14,6 +14,7 @@ import orjson import pytest +import requests import yaml from boltons.typeutils import classproperty @@ -29,6 +30,18 @@ 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.standard_tests._spec_compatibility import ( + DEPLOYMENT_MODE_ENV_VARS, + REGISTRY_UNAVAILABLE_ERRORS, + DeploymentMode, + compare_specs, + declared_breaking_changes, + disabled_for_version, + fetch_published_spec, + format_breaking_spec_changes, + format_spec_change_summary, + is_newer_version, +) from airbyte_cdk.test.standard_tests._spec_lint import ( assert_no_secrets_in_output, assert_spec_is_valid, @@ -41,6 +54,7 @@ from airbyte_cdk.utils.docker import ( build_connector_image, run_docker_airbyte_command, + run_docker_command, ) @@ -82,6 +96,82 @@ def _assert_check_outcome( ) +_SPEC_FAILURE_OUTPUT_LINES = 50 +"""How many trailing lines of the connector's output a failed `spec` run reports.""" + + +def _run_spec_in_image(connector_image: str, deployment_mode: DeploymentMode) -> dict[str, Any]: + """Run `spec` in the image as the registry publisher does, and return the raw spec object. + + The raw JSON is compared rather than the parsed protocol model, because the registry stores + the raw output and parsing would drop or add keys that are not in the model. If `spec` exits + with an error or emits no SPEC message, the error names the exit code and includes what the + connector reported: its last error TRACE message, its last log lines and its stderr. + """ + env_args = [ + arg + for name, value in DEPLOYMENT_MODE_ENV_VARS[deployment_mode].items() + for arg in ("-e", f"{name}={value}") + ] + result = run_docker_command( + ["docker", "run", "--rm", *env_args, connector_image, "spec"], + capture_stdout=True, + capture_stderr=True, + raise_if_errors=False, + ) + + spec: dict[str, Any] | None = None + error_trace: dict[str, Any] | None = None + log_lines: list[str] = [] + for line in result.stdout.splitlines(): + try: + message = orjson.loads(line) + except orjson.JSONDecodeError: + log_lines.append(line) + continue + if not isinstance(message, dict): + continue + if message.get("type") == "SPEC" and spec is None: + spec = cast(dict[str, Any], message["spec"]) + elif message.get("type") == "LOG": + log_lines.append(str((message.get("log") or {}).get("message", ""))) + elif message.get("type") == "TRACE" and (message.get("trace") or {}).get("error"): + error_trace = message["trace"]["error"] + + if spec is not None and result.returncode == 0: + return spec + + problem = ( + f"exited with code {result.returncode}" + if result.returncode != 0 + else "emitted no SPEC message" + ) + details = [ + f"`spec` in image '{connector_image}' ({deployment_mode} mode) {problem}.", + ] + if error_trace: + details.append(f"Error reported by the connector: {error_trace.get('message')}") + if error_trace.get("internal_message"): + details.append(f"Internal message: {error_trace['internal_message']}") + if error_trace.get("stack_trace"): + details.append(f"Stack trace:\n{_tail(str(error_trace['stack_trace']))}") + if log_lines: + log_text = "\n".join(log_lines) + details.append(f"Last log lines:\n{_tail(log_text)}") + details.append(f"Stderr:\n{_tail(result.stderr or '') or '(empty)'}") + raise AssertionError("\n".join(details)) + + +def _tail(text: str) -> str: + lines = text.rstrip().splitlines() + if len(lines) <= _SPEC_FAILURE_OUTPUT_LINES: + return "\n".join(lines) + omitted = len(lines) - _SPEC_FAILURE_OUTPUT_LINES + return "\n".join( + [f"... ({omitted} earlier lines omitted)", *lines[-_SPEC_FAILURE_OUTPUT_LINES:]] + ) + + _docker_spec_cache: dict[str, ConnectorSpecification | Exception] = {} """The spec, or the error from running `spec`, of each connector image in this test session.""" @@ -297,6 +387,163 @@ def test_docker_image_build_and_spec( spec = _run_docker_spec(connector_image, connector_name=connector_root.name) assert_spec_is_valid(spec, connector_name=connector_root.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_spec_backward_compatibility( + self, + connector_image_override: str | None, + connector_base_image_override: str | None, + ) -> None: + """Fail if the image's spec rejects configs that the published version accepted. + + The spec is compared against the `latest` version in the public connector registry + (connectors.airbyte.com), once per registry the connector is published to, with `spec` + run in the matching deployment mode. A connector that was never published, or whose + version is older than the published one, has nothing to compare against. If the + registry cannot be reached after retries, the test is skipped with a warning rather + than failing every connector's tests during an outage. + + Breaking findings are waived, and the test is skipped, in two cases: + + - The change goes through the breaking-change process: `releases.breakingChanges` in + metadata.yaml declares a version after the published one, up to `dockerImageTag`. + - `backward_compatibility_tests_config.disable_for_version` in an entry of the `spec` + section of acceptance-test-config.yml names the published version. The waiver + expires when a newer version is published. A connector without that file, such as + one rebuilt on a new base image without a version bump, can add a file that holds + only this entry. + + The compatible changes, and the waived breaking ones, are reported as a warning, so they + appear in the warnings summary of the test run. + """ + connector_root = self.get_connector_root_dir().absolute() + metadata = MetadataFile.from_file(connector_root / "metadata.yaml") + metadata_data = metadata.data.model_dump() + current_version = metadata.data.dockerImageTag + + try: + published_specs = { + registry: published + for registry in DEPLOYMENT_MODE_ENV_VARS + if (published := fetch_published_spec(metadata.data.dockerRepository, registry)) + } + except REGISTRY_UNAVAILABLE_ERRORS as error: + message = ( + f"Skipping the spec backward-compatibility check of `{self.connector_name}`: the " + f"connector registry could not be reached ({error}). The spec was not compared " + "with the published version." + ) + warnings.warn(message, category=UserWarning, stacklevel=1) + pytest.skip(message) + except requests.RequestException as error: + pytest.fail( + f"Could not fetch the published spec of `{self.connector_name}` from the connector " + f"registry: {error}" + ) + if not published_specs: + pytest.skip(f"`{self.connector_name}` has no published version to compare against.") + + newer_versions = sorted( + { + published.version + for published in published_specs.values() + if is_newer_version(published.version, current_version) + } + ) + published_specs = { + registry: published + for registry, published in published_specs.items() + if published.version not in newer_versions + } + if not published_specs: + pytest.skip( + f"`{self.connector_name}` {current_version} is older than the published version " + f"{', '.join(newer_versions)}, so there is no earlier spec to compare against." + ) + + connector_image: str | None = connector_image_override + if not connector_image: + connector_image = 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, + ) + + failures: list[str] = [] + waivers: list[str] = [] + summaries: list[str] = [] + for registry, published in published_specs.items(): + comparison = compare_specs( + published.spec, _run_spec_in_image(connector_image, registry) + ) + if comparison.is_backward_compatible: + if comparison.compatible: + summaries.append( + format_spec_change_summary( + registry=registry, published=published, comparison=comparison + ) + ) + continue + + declared = declared_breaking_changes(metadata_data, published.version, current_version) + if declared: + waiver: str | None = f"declared as breaking in {', '.join(declared)}" + elif published.version in self._spec_compatibility_disabled_versions(): + waiver = f"waived by `disable_for_version: {published.version}`" + else: + waiver = None + + if waiver: + summaries.append( + format_spec_change_summary( + registry=registry, published=published, comparison=comparison, waiver=waiver + ) + ) + waivers.append( + f"{len(comparison.breaking)} breaking change(s) to the {registry.upper()} " + f"spec since {published.version}, {waiver}" + ) + else: + failures.append( + format_breaking_spec_changes( + connector_name=self.connector_name, + current_version=current_version, + registry=registry, + published=published, + comparison=comparison, + ) + ) + + if summaries: + warnings.warn( + f"Spec changes of `{self.connector_name}` {current_version} judged safe or " + "waived:\n" + "\n".join(summaries), + category=UserWarning, + stacklevel=1, + ) + if failures: + pytest.fail("\n\n".join(failures)) + if waivers: + pytest.skip("; ".join(waivers)) + + def _spec_compatibility_disabled_versions(self) -> list[str]: + """The `disable_for_version` waivers of acceptance-test-config.yml, if it has any. + + A missing, empty or malformed file has none. The file is only read once a comparison + is breaking, so its layout cannot fail a connector whose spec is compatible. + """ + try: + acceptance_test_config = self.acceptance_test_config + except (FileNotFoundError, ValueError, TypeError, yaml.YAMLError): + return [] + return disabled_for_version(acceptance_test_config) + @pytest.mark.skipif( shutil.which("docker") is None, reason="docker CLI not found in PATH, skipping docker image tests", diff --git a/cdk-migrations.md b/cdk-migrations.md index 5a87684f58..e29bbf266f 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -50,6 +50,27 @@ Migration steps: stop printing the value; a `CONNECTION_STATUS` message that the Rationale: these checks restore the `test_secret_is_properly_marked` assertion of the retired Connector Acceptance Tests, and add the leak check that `test_secret_never_in_the_output` never implemented. +### New mandatory test: `test_docker_image_spec_backward_compatibility` + +`DockerConnectorTestSuite.test_docker_image_spec_backward_compatibility` fails when the connector's spec rejects configs that the published version accepted. It replaces the spec half of the `test_backward_compatibility` test of the retired Connector Acceptance Tests (CAT). + +The test downloads the `latest` spec of each registry the connector is published to (OSS and Cloud) from `connectors.airbyte.com`, runs `spec` in the image with that registry's deployment mode, and compares the two. Removed properties, newly required properties, added or tightened constraints, narrowed types or enums, removed destination sync modes, moved OAuth output paths, unmarked secrets and `oneOf` branches that overlap existing configs are breaking. Annotations, defaults, relaxed constraints, new optional properties and newly secret fields are compatible. + +The test is skipped, rather than failing, when: + +- the connector has no published version, or its `dockerImageTag` is older than the published one; +- the registry cannot be reached after retries. The skip reason and a warning say that the spec was not compared. + +The compatible changes, and any waived breaking ones, are reported as a warning in the test run's warnings summary. + +Migration steps: when the test fails, do one of the following. + +- Make the change backward compatible, for example by keeping the old field or value, or by making a new field optional. +- Release the change as a breaking change: bump the major version, declare it under `releases.breakingChanges` in `metadata.yaml`, and write a migration guide. The test is waived for every version after the published one, up to the declared one, so a patch on top of a major that is still rolling out is waived too. +- If a config migration rewrites existing configs into the new shape, or a finding is a false positive, add `backward_compatibility_tests_config: {disable_for_version: ""}` to an entry of the `spec` section of `acceptance-test-config.yml`. This is the key CAT honored, in the current (`spec.tests`) and the legacy list layout. The waiver applies only while the published version equals the given one, so it expires with the next release. A connector without `acceptance-test-config.yml`, such as one rebuilt on a new base image without a version bump, can add a file that holds only this entry. + +Rationale: the platform validates saved configs against the new spec when a version rolls out, so a spec change that rejects them breaks existing connections at upgrade time. CAT caught this before merge; the Standard Tests did not. + ## 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 14b636253e..cdb562ad8f 100644 --- a/unit_tests/resources/source_pokeapi_w_components_py/manifest.yaml +++ b/unit_tests/resources/source_pokeapi_w_components_py/manifest.yaml @@ -80,10 +80,10 @@ spec: - raichu - sandshrew - sandslash - - nidoranf + - nidoran-f - nidorina - nidoqueen - - nidoranm + - nidoran-m - nidorino - nidoking - clefairy diff --git a/unit_tests/resources/source_pokeapi_w_components_py/metadata.yaml b/unit_tests/resources/source_pokeapi_w_components_py/metadata.yaml index b16ba57848..54d9a27a34 100644 --- a/unit_tests/resources/source_pokeapi_w_components_py/metadata.yaml +++ b/unit_tests/resources/source_pokeapi_w_components_py/metadata.yaml @@ -20,7 +20,7 @@ data: connectorSubtype: api connectorType: source definitionId: 6371b14b-bc68-4236-bfbd-468e8df8e968 - dockerImageTag: 0.3.24 + dockerImageTag: 0.3.74 dockerRepository: airbyte/source-pokeapi githubIssueLabel: source-pokeapi icon: pokeapi.svg diff --git a/unit_tests/test/test_spec_compatibility.py b/unit_tests/test/test_spec_compatibility.py new file mode 100644 index 0000000000..522495803f --- /dev/null +++ b/unit_tests/test/test_spec_compatibility.py @@ -0,0 +1,2166 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Unit tests for the spec backward-compatibility check.""" + +from __future__ import annotations + +import http.server +import json +import socket +import subprocess +import threading +import time +from pathlib import Path +from typing import Any + +import pytest +import requests +import requests_mock +import yaml + +from airbyte_cdk.test.standard_tests import _spec_compatibility, docker_base +from airbyte_cdk.test.standard_tests._spec_compatibility import ( + BREAKING_CHANGES_DOCS_URL, + MAX_REPORTED_CHANGES, + REGISTRY_TIMEOUT_SECONDS, + REGISTRY_UNAVAILABLE_ERRORS, + PublishedSpec, + SpecComparison, + compare_specs, + declared_breaking_changes, + disabled_for_version, + fetch_published_spec, + format_breaking_spec_changes, + format_spec_change_summary, + is_newer_version, + registry_retry, + registry_session, +) +from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite + + +def _spec( + properties: dict[str, Any], + required: list[str] | None = None, + **extra: Any, +) -> dict[str, Any]: + return { + "connectionSpecification": { + "type": "object", + "required": required if required is not None else ["api_key"], + "properties": properties, + }, + **extra, + } + + +_BASE_PROPERTIES: dict[str, Any] = { + "api_key": {"type": "string", "title": "API key", "airbyte_secret": True}, + "start_date": {"type": "string", "format": "date-time"}, +} + + +# compare_specs + + +def test_unchanged_spec_has_nothing_to_report() -> None: + comparison = compare_specs(_spec(_BASE_PROPERTIES), _spec(_BASE_PROPERTIES)) + + assert comparison.is_backward_compatible + assert not comparison.breaking + assert not comparison.compatible + + +def test_added_optional_property_is_compatible() -> None: + comparison = compare_specs( + _spec(_BASE_PROPERTIES), + _spec({**_BASE_PROPERTIES, "page_size": {"type": "integer"}}), + ) + + assert comparison.is_backward_compatible + assert any("page_size" in change for change in comparison.compatible) + + +def test_removed_property_is_breaking() -> None: + comparison = compare_specs( + _spec(_BASE_PROPERTIES), + _spec({"api_key": _BASE_PROPERTIES["api_key"]}), + ) + + assert comparison.breaking == ["`connectionSpecification.properties.start_date` was removed"] + + +def test_changed_property_type_is_breaking_and_dropped_format_is_compatible() -> None: + comparison = compare_specs( + _spec(_BASE_PROPERTIES), + _spec({**_BASE_PROPERTIES, "start_date": {"type": "integer"}}), + ) + + assert ( + "`connectionSpecification.properties.start_date.type` no longer allows string" + in comparison.breaking + ) + assert ( + "`connectionSpecification.properties.start_date.format` was removed, " + "widening what a config may set" + ) in comparison.compatible + + +def test_newly_required_property_is_breaking_even_when_new() -> None: + comparison = compare_specs( + _spec(_BASE_PROPERTIES), + _spec({**_BASE_PROPERTIES, "region": {"type": "string"}}, required=["api_key", "region"]), + ) + + assert comparison.breaking == ["`connectionSpecification`: `region` is now required"] + + +def test_first_required_list_is_breaking() -> None: + previous = _spec(_BASE_PROPERTIES) + del previous["connectionSpecification"]["required"] + + comparison = compare_specs(previous, _spec(_BASE_PROPERTIES, required=["start_date"])) + + assert comparison.breaking == ["`connectionSpecification`: `start_date` is now required"] + + +def test_dropped_requirement_is_compatible() -> None: + comparison = compare_specs( + _spec(_BASE_PROPERTIES, required=["api_key", "start_date"]), + _spec(_BASE_PROPERTIES, required=["api_key"]), + ) + + assert comparison.is_backward_compatible + assert any("no longer required" in change for change in comparison.compatible) + + +@pytest.mark.parametrize( + "added", + [ + pytest.param({"enum": ["us", "eu"]}, id="enum"), + pytest.param({"pattern": "^[0-9]+$"}, id="pattern"), + pytest.param({"const": "fixed"}, id="const"), + pytest.param({"maxLength": 8}, id="maxLength"), + pytest.param({"format": "date-time"}, id="format"), + pytest.param({"additionalProperties": False}, id="additionalProperties"), + ], +) +def test_added_constraint_is_breaking(added: dict[str, Any]) -> None: + comparison = compare_specs( + _spec({"region": {"type": "string"}}), + _spec({"region": {"type": "string", **added}}), + ) + + assert len(comparison.breaking) == 1 + assert "narrowing what a config may set" in comparison.breaking[0] + + +def test_becoming_nullable_is_compatible() -> None: + comparison = compare_specs( + _spec({"start_date": {"type": "string"}}), + _spec({"start_date": {"type": ["null", "string"]}}), + ) + + assert comparison.is_backward_compatible + assert "also allows null" in comparison.compatible[0] + + +def test_becoming_non_nullable_is_breaking() -> None: + comparison = compare_specs( + _spec({"start_date": {"type": ["null", "string"]}}), + _spec({"start_date": {"type": "string"}}), + ) + + assert comparison.breaking == [ + "`connectionSpecification.properties.start_date.type` no longer allows null" + ] + + +def test_changed_default_is_compatible() -> None: + comparison = compare_specs( + _spec({"page_size": {"type": "integer", "default": 100}}), + _spec({"page_size": {"type": "integer", "default": 500}}), + ) + + assert comparison.is_backward_compatible + assert "changes behavior, not validity" in comparison.compatible[0] + + +@pytest.mark.parametrize( + "key, previous_bound, current_bound, compatible", + [ + pytest.param("maximum", 100, 1000, True, id="raised-maximum-relaxes"), + pytest.param("maximum", 1000, 100, False, id="lowered-maximum-tightens"), + pytest.param("minLength", 8, 1, True, id="lowered-minimum-relaxes"), + pytest.param("minLength", 1, 8, False, id="raised-minimum-tightens"), + ], +) +def test_moved_bound_is_judged_by_its_direction( + key: str, previous_bound: int, current_bound: int, compatible: bool +) -> None: + comparison = compare_specs( + _spec({"page_size": {"type": "integer", key: previous_bound}}), + _spec({"page_size": {"type": "integer", key: current_bound}}), + ) + + assert comparison.is_backward_compatible is compatible + + +@pytest.mark.parametrize( + "key, previous_value, current_value, compatible", + [ + pytest.param("additionalProperties", False, True, True, id="unsealed-relaxes"), + pytest.param("additionalProperties", True, False, False, id="sealed-tightens"), + pytest.param("uniqueItems", True, False, True, id="duplicates-allowed-relaxes"), + pytest.param("uniqueItems", False, True, False, id="uniqueness-tightens"), + ], +) +def test_flipped_boolean_constraint_is_judged_by_its_direction( + key: str, previous_value: bool, current_value: bool, compatible: bool +) -> None: + comparison = compare_specs( + _spec({"options": {"type": "object", key: previous_value}}), + _spec({"options": {"type": "object", key: current_value}}), + ) + + assert comparison.is_backward_compatible is compatible + + +def test_narrowed_enum_is_breaking_and_widened_enum_is_compatible() -> None: + previous = _spec({"region": {"type": "string", "enum": ["us", "eu"]}}) + + narrowed = compare_specs(previous, _spec({"region": {"type": "string", "enum": ["us"]}})) + widened = compare_specs( + previous, _spec({"region": {"type": "string", "enum": ["us", "eu", "apac"]}}) + ) + + assert narrowed.breaking == [ + "`connectionSpecification.properties.region.enum` no longer allows 'eu'" + ] + assert widened.is_backward_compatible + + +def test_enum_of_objects_has_set_semantics() -> None: + one = _spec({"mode": {"enum": [{"kind": "full"}]}}) + two = _spec({"mode": {"enum": [{"kind": "full"}, {"kind": "incremental"}]}}) + + assert compare_specs(one, two).is_backward_compatible + assert not compare_specs(two, one).is_backward_compatible + + +def test_removed_destination_sync_mode_is_breaking() -> None: + previous = _spec({}, supported_destination_sync_modes=["append", "overwrite"]) + current = _spec({}, supported_destination_sync_modes=["overwrite"]) + + assert compare_specs(previous, current).breaking == [ + "`supported_destination_sync_modes` no longer allows 'append'" + ] + + +def test_documentation_changes_are_compatible() -> None: + comparison = compare_specs( + _spec(_BASE_PROPERTIES, documentationUrl="https://example.com/v1"), + _spec( + { + **_BASE_PROPERTIES, + "api_key": { + "type": "string", + "title": "API token", + "description": "The token to authenticate with.", + "order": 0, + "airbyte_secret": True, + }, + }, + documentationUrl="https://example.com/v2", + ), + ) + + assert comparison.is_backward_compatible + + +def test_config_field_named_description_is_still_a_property() -> None: + comparison = compare_specs( + _spec({**_BASE_PROPERTIES, "description": {"type": "string"}}), + _spec(_BASE_PROPERTIES), + ) + + assert comparison.breaking == ["`connectionSpecification.properties.description` was removed"] + + +def test_config_field_named_properties_keeps_its_own_documentation() -> None: + comparison = compare_specs( + _spec({"properties": {"type": "object", "title": "Extra properties"}}), + _spec({"properties": {"type": "object", "title": "Additional properties"}}), + ) + + assert comparison.is_backward_compatible + + +def test_finding_does_not_quote_a_whole_subtree() -> None: + wide = {f"field_{index}": {"type": "string", "description": "x" * 40} for index in range(30)} + + comparison = compare_specs( + _spec({"cfg": {"properties": wide}}), + _spec({"cfg": {"properties": [1, 2]}}), + ) + + assert len(comparison.breaking) == 1 + assert len(comparison.breaking[0]) < 400 + assert comparison.breaking[0].endswith("to [1, 2]") + + +_OAUTH_BRANCH: dict[str, Any] = { + "title": "OAuth", + "properties": { + "auth_type": {"type": "string", "const": "oauth2.0"}, + "client_id": {"type": "string"}, + }, + "required": ["auth_type", "client_id"], +} +_API_KEY_BRANCH: dict[str, Any] = { + "title": "API key", + "properties": { + "auth_type": {"type": "string", "const": "api_key"}, + "api_key": {"type": "string"}, + }, + "required": ["auth_type", "api_key"], +} + + +def test_reordered_oneof_branches_are_unchanged() -> None: + comparison = compare_specs( + _spec({"credentials": {"oneOf": [_API_KEY_BRANCH, _OAUTH_BRANCH]}}), + _spec({"credentials": {"oneOf": [_OAUTH_BRANCH, _API_KEY_BRANCH]}}), + ) + + assert comparison.is_backward_compatible + assert not comparison.compatible + + +def test_removed_oneof_branch_is_breaking_when_the_rest_are_reordered() -> None: + token = {"properties": {"auth_type": {"const": "token"}}} + + comparison = compare_specs( + _spec({"credentials": {"oneOf": [_API_KEY_BRANCH, _OAUTH_BRANCH, token]}}), + _spec({"credentials": {"oneOf": [_OAUTH_BRANCH, _API_KEY_BRANCH]}}), + ) + + assert comparison.breaking == [ + "`connectionSpecification.properties.credentials.oneOf[2]` was removed" + ] + + +def test_new_required_field_inside_a_matched_branch_is_breaking() -> None: + oauth_with_scope = { + **_OAUTH_BRANCH, + "properties": {**_OAUTH_BRANCH["properties"], "scope": {"type": "string"}}, + "required": [*_OAUTH_BRANCH["required"], "scope"], + } + + comparison = compare_specs( + _spec({"credentials": {"oneOf": [_API_KEY_BRANCH, _OAUTH_BRANCH]}}), + _spec({"credentials": {"oneOf": [oauth_with_scope, _API_KEY_BRANCH]}}), + ) + + assert comparison.breaking == [ + "`connectionSpecification.properties.credentials.oneOf[1]`: `scope` is now required" + ] + + +def test_branch_key_does_not_depend_on_property_order() -> None: + previous = _spec( + { + "credentials": { + "oneOf": [ + { + "properties": { + "auth_type": {"const": "oauth2.0"}, + "flavor": {"const": "std"}, + } + } + ] + } + } + ) + current = _spec( + { + "credentials": { + "oneOf": [ + { + "properties": { + "flavor": {"const": "std"}, + "auth_type": {"const": "oauth2.0"}, + "scopes": {"type": "string"}, + } + } + ] + } + } + ) + + comparison = compare_specs(previous, current) + + assert comparison.is_backward_compatible + assert any("scopes` was added" in change for change in comparison.compatible) + + +def test_swapped_auth_method_reads_as_one_removed_and_one_added() -> None: + def branch(auth_type: str) -> dict[str, Any]: + return {"properties": {"auth_type": {"const": auth_type}}, "required": ["auth_type"]} + + comparison = compare_specs( + _spec({"credentials": {"oneOf": [branch("api_key")]}}), + _spec({"credentials": {"oneOf": [branch("oauth2.0")]}}), + ) + + assert comparison.breaking == [ + "`connectionSpecification.properties.credentials.oneOf[0]` was removed" + ] + assert comparison.compatible == [ + "`connectionSpecification.properties.credentials.oneOf[0]` was added" + ] + + +def test_moved_oauth_output_path_is_breaking() -> None: + def advanced_auth(path: list[str]) -> dict[str, Any]: + return { + "auth_flow_type": "oauth2.0", + "oauth_config_specification": { + "complete_oauth_output_specification": { + "properties": {"access_token": {"path_in_connector_config": path}} + } + }, + } + + comparison = compare_specs( + _spec({}, advanced_auth=advanced_auth(["access_token"])), + _spec({}, advanced_auth=advanced_auth(["credentials", "access_token"])), + ) + + assert not comparison.is_backward_compatible + + +def _declarative_oauth(consent_flow: dict[str, Any] | None) -> dict[str, Any]: + oauth_config_specification: dict[str, Any] = { + "complete_oauth_output_specification": { + "properties": { + "refresh_token": {"path_in_connector_config": ["credentials", "refresh_token"]} + } + }, + } + if consent_flow is not None: + oauth_config_specification["oauth_connector_input_specification"] = consent_flow + return {"auth_flow_type": "oauth2.0", "oauth_config_specification": oauth_config_specification} + + +@pytest.mark.parametrize( + "current_consent_flow", + [ + pytest.param( + { + "consent_url": "https://example.com/v2/authorize?scope={{ scope }}", + "access_token_url": "https://example.com/v2/token", + "scope": "read write", + "extract_output": ["access_token", "refresh_token"], + }, + id="urls-scope-and-outputs-changed", + ), + pytest.param(None, id="consent-flow-removed"), + ], +) +def test_oauth_consent_flow_changes_are_compatible( + current_consent_flow: dict[str, Any] | None, +) -> None: + previous_consent_flow = { + "consent_url": "https://example.com/authorize?scope={{ scope }}", + "access_token_url": "https://example.com/token", + "scope": "read", + "extract_output": ["refresh_token"], + } + + comparison = compare_specs( + _spec({}, advanced_auth=_declarative_oauth(previous_consent_flow)), + _spec({}, advanced_auth=_declarative_oauth(current_consent_flow)), + ) + + assert comparison.is_backward_compatible + assert len(comparison.compatible) == 1 + assert "oauth_connector_input_specification" in comparison.compatible[0] + + +def test_spec_keys_existing_connections_do_not_depend_on_are_ignored() -> None: + comparison = compare_specs( + _spec( + _BASE_PROPERTIES, + supportsIncremental=True, + supportsNormalization=True, + protocol_version="0.2.0", + supported_sync_modes=["full_refresh", "incremental"], + ), + _spec(_BASE_PROPERTIES, supportsIncremental=False), + ) + + assert comparison.is_backward_compatible + assert not comparison.compatible + + +# Annotations: every key that is not a validation keyword or a config-locating protocol key + + +def _convex_published_spec() -> dict[str, Any]: + """The shape of destination-convex 0.2.19 as the registry publishes it.""" + return { + "connectionSpecification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "Destination Convex", + "type": "object", + "required": ["deployment_url", "access_key"], + "additionalProperties": False, + "properties": { + "deployment_url": { + "type": "string", + "description": "URL of the Convex deployment that is the destination", + "examples": ["https://murky-swan-635.convex.cloud"], + }, + "access_key": { + "type": "string", + "description": "API access key used to send data to a Convex deployment.", + "airbyte_secret": "true", + }, + }, + }, + "documentationUrl": "https://docs.airbyte.com/integrations/destinations/convex", + "supported_destination_sync_modes": ["overwrite", "append", "append_dedup"], + "supportsIncremental": True, + } + + +def test_string_airbyte_secret_rewritten_as_a_boolean_is_compatible() -> None: + previous = _convex_published_spec() + current = _convex_published_spec() + current["connectionSpecification"]["properties"]["access_key"]["airbyte_secret"] = True + + comparison = compare_specs(previous, current) + + assert comparison.is_backward_compatible + assert comparison.compatible == [ + "`connectionSpecification.properties.access_key.airbyte_secret` changed from 'true' to " + "True (same meaning)" + ] + + +@pytest.mark.parametrize( + "previous_value, current_value", + [ + pytest.param(False, True, id="false-to-true"), + pytest.param(None, True, id="added"), + pytest.param(False, None, id="non-secret-flag-removed"), + pytest.param("true", True, id="string-to-boolean"), + ], +) +def test_marking_more_values_secret_is_compatible(previous_value: Any, current_value: Any) -> None: + def spec(value: Any) -> dict[str, Any]: + schema: dict[str, Any] = {"type": "string"} + if value is not None: + schema["airbyte_secret"] = value + return _spec({"token": schema}) + + assert compare_specs(spec(previous_value), spec(current_value)).is_backward_compatible + + +@pytest.mark.parametrize( + "current_value", + [pytest.param(False, id="true-to-false"), pytest.param(None, id="removed")], +) +@pytest.mark.parametrize("previous_value", [True, "true"]) +def test_unmarking_a_secret_is_breaking(previous_value: Any, current_value: Any) -> None: + current_schema: dict[str, Any] = {"type": "string"} + if current_value is not None: + current_schema["airbyte_secret"] = current_value + + comparison = compare_specs( + _spec({"token": {"type": "string", "airbyte_secret": previous_value}}), + _spec({"token": current_schema}), + ) + + assert len(comparison.breaking) == 1 + assert "no longer a secret" in comparison.breaking[0] + + +@pytest.mark.parametrize( + "previous_schema, current_schema", + [ + pytest.param({"airbyte_hidden": True}, {}, id="airbyte_hidden-removed"), + pytest.param({}, {"airbyte_hidden": True}, id="airbyte_hidden-added"), + pytest.param({"multiline": True}, {}, id="multiline-removed"), + pytest.param({"always_show": True}, {"always_show": False}, id="always_show-changed"), + pytest.param({"order": 1}, {"order": 3}, id="order-changed"), + pytest.param({"group": "auth"}, {}, id="group-removed"), + pytest.param({"examples": ["a"]}, {"examples": ["a", "b"]}, id="examples-changed"), + pytest.param({"title": "Token"}, {}, id="title-removed"), + pytest.param({"x-display": {"width": 2}}, {"x-display": {"width": 3}}, id="vendor-key"), + pytest.param({"deprecated": False}, {"deprecated": True}, id="deprecated"), + ], +) +def test_annotation_changes_are_compatible( + previous_schema: dict[str, Any], current_schema: dict[str, Any] +) -> None: + comparison = compare_specs( + _spec({"token": {"type": "string", **previous_schema}}), + _spec({"token": {"type": "string", **current_schema}}), + ) + + assert comparison.is_backward_compatible + assert all("(annotation)" in change for change in comparison.compatible) + + +@pytest.mark.parametrize( + "current_schema_keyword", + [ + pytest.param({"$schema": "https://json-schema.org/draft-07/schema#"}, id="changed"), + pytest.param({}, id="removed"), + ], +) +def test_schema_keyword_changes_are_compatible(current_schema_keyword: dict[str, Any]) -> None: + previous = _spec(_BASE_PROPERTIES) + previous["connectionSpecification"]["$schema"] = "http://json-schema.org/draft-07/schema#" + current = _spec(_BASE_PROPERTIES) + current["connectionSpecification"].update(current_schema_keyword) + + assert compare_specs(previous, current).is_backward_compatible + + +def test_connection_specification_groups_are_annotations() -> None: + previous = _spec(_BASE_PROPERTIES) + previous["connectionSpecification"]["groups"] = [{"id": "auth", "title": "Auth"}] + current = _spec(_BASE_PROPERTIES) + current["connectionSpecification"]["groups"] = [{"id": "auth", "title": "Authentication"}] + + assert compare_specs(previous, current).is_backward_compatible + + +@pytest.mark.parametrize( + "key, previous_value, current_value", + [ + pytest.param("predicate_key", ["credentials", "auth_type"], ["auth_type"], id="moved"), + pytest.param("predicate_value", "oauth2.0", "oauth", id="changed"), + pytest.param("auth_flow_type", "oauth2.0", None, id="removed"), + ], +) +def test_advanced_auth_protocol_keys_are_still_breaking( + key: str, previous_value: Any, current_value: Any +) -> None: + def advanced_auth(value: Any) -> dict[str, Any]: + auth: dict[str, Any] = {"auth_flow_type": "oauth2.0", "predicate_value": "oauth2.0"} + if value is None: + del auth[key] + else: + auth[key] = value + return auth + + comparison = compare_specs( + _spec({}, advanced_auth=advanced_auth(previous_value)), + _spec({}, advanced_auth=advanced_auth(current_value)), + ) + + assert not comparison.is_backward_compatible + + +@pytest.mark.parametrize( + "previous_schema, current_schema, compatible", + [ + pytest.param({}, {"not": {"const": "x"}}, False, id="added"), + pytest.param( + {"dependencies": {"a": ["b"]}}, {"dependencies": {"a": ["c"]}}, False, id="changed" + ), + pytest.param( + {"if": {"required": ["a"]}, "then": {"required": ["b"]}}, {}, True, id="removed" + ), + ], +) +def test_unmodeled_validation_keywords( + previous_schema: dict[str, Any], current_schema: dict[str, Any], compatible: bool +) -> None: + comparison = compare_specs( + _spec({"options": {"type": "object", **previous_schema}}), + _spec({"options": {"type": "object", **current_schema}}), + ) + + assert comparison.is_backward_compatible is compatible + + +# Keywords at their default value + + +@pytest.mark.parametrize( + "added", + [ + pytest.param({"additionalProperties": True}, id="additionalProperties-true"), + pytest.param({"additionalProperties": {}}, id="additionalProperties-empty-schema"), + pytest.param({"uniqueItems": False}, id="uniqueItems-false"), + pytest.param({"minLength": 0}, id="minLength-0"), + pytest.param({"minItems": 0}, id="minItems-0"), + pytest.param({"minProperties": 0}, id="minProperties-0"), + pytest.param({"items": {}}, id="items-empty-schema"), + pytest.param({"required": []}, id="empty-required"), + pytest.param({"exclusiveMinimum": False}, id="draft-04-exclusiveMinimum-false"), + ], +) +def test_keyword_added_at_its_default_is_compatible(added: dict[str, Any]) -> None: + previous = _spec({"region": {"type": "string"}}) + current = _spec({"region": {"type": "string", **added}}) + + assert compare_specs(previous, current).is_backward_compatible + assert compare_specs(current, previous).is_backward_compatible + + +def test_root_additional_properties_true_after_a_manifest_migration_is_compatible() -> None: + current = _spec(_BASE_PROPERTIES) + current["connectionSpecification"]["additionalProperties"] = True + + comparison = compare_specs(_spec(_BASE_PROPERTIES), current) + + assert comparison.is_backward_compatible + assert comparison.compatible == [ + "`connectionSpecification.additionalProperties` was added at its default value True, " + "which allows the same configs" + ] + + +@pytest.mark.parametrize( + "added", + [ + pytest.param({"minLength": 1}, id="minLength-1"), + pytest.param({"uniqueItems": True}, id="uniqueItems-true"), + pytest.param( + {"additionalProperties": {"type": "string"}}, id="additionalProperties-schema" + ), + pytest.param({"exclusiveMaximum": True}, id="draft-04-exclusiveMaximum-true"), + pytest.param({"minItems": False}, id="false-is-not-zero"), + ], +) +def test_keyword_added_away_from_its_default_is_breaking(added: dict[str, Any]) -> None: + comparison = compare_specs( + _spec({"region": {"type": "string"}}), + _spec({"region": {"type": "string", **added}}), + ) + + assert not comparison.is_backward_compatible + + +@pytest.mark.parametrize( + "previous_value, current_value, compatible", + [ + pytest.param({"type": "string"}, True, True, id="schema-relaxed-to-true"), + pytest.param({"type": "string"}, {}, True, id="schema-relaxed-to-empty-schema"), + pytest.param(True, {"type": "string"}, False, id="true-tightened-to-schema"), + pytest.param(False, {}, True, id="false-relaxed-to-empty-schema"), + ], +) +def test_additional_properties_schema_is_judged_by_its_direction( + previous_value: Any, current_value: Any, compatible: bool +) -> None: + comparison = compare_specs( + _spec({"options": {"type": "object", "additionalProperties": previous_value}}), + _spec({"options": {"type": "object", "additionalProperties": current_value}}), + ) + + assert comparison.is_backward_compatible is compatible + + +@pytest.mark.parametrize( + "key, previous_value", + [ + pytest.param("properties", {"client_id": {"type": "string"}}, id="properties-emptied"), + pytest.param( + "items", + {"type": "object", "properties": {"name": {"type": "string"}}}, + id="items-emptied", + ), + ], +) +def test_emptying_a_field_keyword_still_removes_its_fields(key: str, previous_value: Any) -> None: + comparison = compare_specs( + _spec({"options": {"type": "object", key: previous_value}}), + _spec({"options": {"type": "object", key: {}}}), + ) + + assert len(comparison.breaking) == 1 + assert comparison.breaking[0].endswith("was removed") + + +def test_draft_04_exclusive_bound_flag_is_judged_by_its_direction() -> None: + def spec(exclusive: bool) -> dict[str, Any]: + return _spec( + {"page_size": {"type": "integer", "maximum": 10, "exclusiveMaximum": exclusive}} + ) + + assert not compare_specs(spec(False), spec(True)).is_backward_compatible + assert compare_specs(spec(True), spec(False)).is_backward_compatible + + +# Integer and number + + +@pytest.mark.parametrize( + "previous_type, current_type, compatible", + [ + pytest.param("integer", "number", True, id="integer-widened-to-number"), + pytest.param(["integer", "null"], ["null", "number"], True, id="nullable-widened"), + pytest.param("number", "integer", False, id="number-narrowed-to-integer"), + pytest.param("integer", "string", False, id="integer-to-string"), + ], +) +def test_integer_and_number(previous_type: Any, current_type: Any, compatible: bool) -> None: + comparison = compare_specs( + _spec({"page_size": {"type": previous_type}}), + _spec({"page_size": {"type": current_type}}), + ) + + assert comparison.is_backward_compatible is compatible + + +def test_integer_widened_to_number_is_reported_once() -> None: + comparison = compare_specs( + _spec({"page_size": {"type": "integer"}}), + _spec({"page_size": {"type": "number"}}), + ) + + assert comparison.compatible == [ + "`connectionSpecification.properties.page_size.type` widened from integer to number" + ] + + +# oneOf/anyOf branch identity + + +def _credentials(*branches: dict[str, Any]) -> dict[str, Any]: + return _spec({"credentials": {"type": "object", "oneOf": list(branches)}}) + + +def _branch( + title: str, *fields: str, required: list[str] | None = None, **consts: str +) -> dict[str, Any]: + properties: dict[str, Any] = { + name: {"type": "string", "const": value} for name, value in consts.items() + } + properties.update({name: {"type": "string"} for name in fields}) + branch: dict[str, Any] = {"title": title, "type": "object", "properties": properties} + if required is not None: + branch["required"] = required + return branch + + +def test_optional_discriminator_added_beside_the_existing_one_is_compatible() -> None: + previous = _credentials( + _branch("OAuth", "client_id", option_title="OAuth Credentials"), + _branch("API key", "api_key", option_title="API Key Credentials"), + ) + with_auth_type = [ + _branch("OAuth", "client_id", option_title="OAuth Credentials"), + _branch("API key", "api_key", option_title="API Key Credentials"), + ] + with_auth_type[0]["properties"]["auth_type"] = {"const": "oauth2.0", "default": "oauth2.0"} + with_auth_type[1]["properties"]["auth_type"] = {"const": "api_key", "default": "api_key"} + + comparison = compare_specs(previous, _credentials(*reversed(with_auth_type))) + + assert comparison.is_backward_compatible + assert not any("was removed" in change for change in comparison.compatible) + + +def test_renamed_title_of_a_branch_without_a_discriminator_is_compatible() -> None: + comparison = compare_specs( + _credentials(_branch("Basic", "username", "password"), _branch("Token", "token")), + _credentials( + _branch("Username and password", "username", "password"), _branch("Token", "token") + ), + ) + + assert comparison.is_backward_compatible + assert comparison.compatible == [ + "`connectionSpecification.properties.credentials.oneOf[0].title` changed (annotation)" + ] + + +def test_reordered_branches_without_a_discriminator_are_matched_by_their_fields() -> None: + basic = _branch("Basic", "username", "password") + token = _branch("Token", "token") + renamed_token = _branch("Access token", "token") + + comparison = compare_specs(_credentials(basic, token), _credentials(renamed_token, basic)) + + assert comparison.is_backward_compatible + + +def test_removed_branch_without_a_discriminator_is_breaking() -> None: + basic = _branch("Basic", "username", "password") + token = _branch("Token", "token") + key = _branch("Key", "api_key") + + comparison = compare_specs(_credentials(basic, token, key), _credentials(key, basic)) + + assert comparison.breaking == [ + "`connectionSpecification.properties.credentials.oneOf[1]` was removed" + ] + + +def test_branches_sharing_a_discriminator_are_told_apart_by_title() -> None: + app = _branch("OAuth app", "client_id", required=["client_id"], auth_type="oauth2.0") + token = _branch("OAuth token", "access_token", required=["access_token"], auth_type="oauth2.0") + key = _branch("API key", "api_key", required=["api_key"], auth_type="api_key") + + comparison = compare_specs(_credentials(app, token), _credentials(token, app, key)) + + assert comparison.is_backward_compatible + assert comparison.compatible == [ + "`connectionSpecification.properties.credentials.oneOf[2]` was added" + ] + + +def test_renamed_discriminator_is_breaking_even_with_the_same_title() -> None: + comparison = compare_specs( + _credentials(_branch("OAuth", "client_id", required=["auth_type"], auth_type="oauth")), + _credentials(_branch("OAuth", "client_id", required=["auth_type"], auth_type="oauth2.0")), + ) + + assert comparison.breaking == [ + "`connectionSpecification.properties.credentials.oneOf[0]` was removed" + ] + + +@pytest.mark.parametrize( + "current_branches, compatible", + [ + pytest.param([{"type": "integer"}, {"type": "string"}], True, id="reordered"), + pytest.param([{"type": "string"}], False, id="branch-removed"), + ], +) +def test_anyof_type_branches(current_branches: list[Any], compatible: bool) -> None: + comparison = compare_specs( + _spec({"page_size": {"anyOf": [{"type": "string"}, {"type": "integer"}]}}), + _spec({"page_size": {"anyOf": current_branches}}), + ) + + assert comparison.is_backward_compatible is compatible + + +# oneOf branches that overlap + +_KEY_BRANCH = { + "properties": {"auth_type": {"const": "key"}, "api_key": {"type": "string"}}, + "required": ["api_key"], +} +_TOKEN_BRANCH = { + "properties": {"auth_type": {"const": "tok"}, "token": {"type": "string"}}, + "required": ["token"], +} + + +def _one_of(*branches: dict[str, Any], keyword: str = "oneOf") -> dict[str, Any]: + return _spec({"credentials": {"type": "object", keyword: list(branches)}}) + + +def test_added_oneof_branch_that_matches_existing_configs_is_breaking() -> None: + # `{"api_key": "k"}` matched only the first branch; it matches the new one too, and a config + # that matches two `oneOf` branches is rejected. + loose = {"properties": {"auth_type": {"const": "none"}}} + + comparison = compare_specs( + _one_of(_KEY_BRANCH, _TOKEN_BRANCH), _one_of(_KEY_BRANCH, _TOKEN_BRANCH, loose) + ) + + assert len(comparison.breaking) == 1 + assert comparison.breaking[0].startswith( + "`connectionSpecification.properties.credentials.oneOf[2]` was added and may also match " + "existing configs" + ) + + +@pytest.mark.parametrize( + "added", + [ + pytest.param( + {"properties": {"refresh_token": {"type": "string"}}, "required": ["refresh_token"]}, + id="requires-a-field-no-other-branch-declares", + ), + pytest.param( + {"properties": {"auth_type": {"const": "none"}}, "required": ["auth_type"]}, + id="requires-its-own-discriminator", + ), + ], +) +def test_added_oneof_branch_that_rejects_existing_configs_is_compatible( + added: dict[str, Any], +) -> None: + comparison = compare_specs( + _one_of(_KEY_BRANCH, _TOKEN_BRANCH), _one_of(_KEY_BRANCH, _TOKEN_BRANCH, added) + ) + + assert comparison.is_backward_compatible + assert comparison.compatible == [ + "`connectionSpecification.properties.credentials.oneOf[2]` was added" + ] + + +def test_added_oneof_branch_pinned_apart_from_branches_that_overlap_without_it_is_compatible() -> ( + None +): + # No branch requires `filetype`, but a config that leaves it out matches the `jsonl` branch + # as well as its own, so the configs that were valid all set it, and the new branch rejects + # them. This is the shape of the file-based `format` options. + def file_format(filetype: str, **fields: Any) -> dict[str, Any]: + return { + "title": filetype, + "type": "object", + "properties": {"filetype": {"type": "string", "const": filetype}, **fields}, + } + + previous = [ + file_format("avro", double_as_string={"type": "boolean"}), + file_format("csv", delimiter={"type": "string"}), + file_format("jsonl"), + ] + + comparison = compare_specs(_one_of(*previous), _one_of(*previous, file_format("unstructured"))) + + assert comparison.is_backward_compatible + + +def test_added_anyof_branch_only_widens() -> None: + loose = {"properties": {"auth_type": {"const": "none"}}} + + comparison = compare_specs( + _one_of(_KEY_BRANCH, _TOKEN_BRANCH, keyword="anyOf"), + _one_of(_KEY_BRANCH, _TOKEN_BRANCH, loose, keyword="anyOf"), + ) + + assert comparison.is_backward_compatible + + +def test_oneof_branch_widened_into_another_branch_is_breaking() -> None: + # Without its discriminator and its required field, the first branch also matches every + # config of the second one. + widened_key = {"properties": {"auth_type": {}, "api_key": {"type": "string"}}} + tokens = { + "properties": {"auth_type": {"const": "tok"}, "token": {"type": "string"}}, + "required": ["auth_type", "token"], + } + keys = {**_KEY_BRANCH, "required": ["auth_type", "api_key"]} + + comparison = compare_specs(_one_of(keys, tokens), _one_of(widened_key, tokens)) + + assert ( + "`connectionSpecification.properties.credentials.oneOf[0]` may now also match the configs " + "of `connectionSpecification.properties.credentials.oneOf[1]`, which `oneOf` then rejects" + ) in comparison.breaking + + +def test_removed_discriminator_of_a_oneof_branch_is_breaking() -> None: + keys = {**_KEY_BRANCH, "required": ["auth_type"]} + tokens = {**_TOKEN_BRANCH, "required": ["auth_type"]} + keys_without_const = {"properties": {"auth_type": {}, "api_key": {"type": "string"}}} + + comparison = compare_specs(_one_of(keys, tokens), _one_of(keys_without_const, tokens)) + + assert comparison.breaking == [ + "`connectionSpecification.properties.credentials.oneOf[0]` may now also match the configs " + "of `connectionSpecification.properties.credentials.oneOf[1]`, which `oneOf` then rejects" + ] + + +def test_oneof_branch_that_keeps_its_discriminator_is_compatible() -> None: + keys = {**_KEY_BRANCH, "required": ["auth_type", "api_key"]} + tokens = {**_TOKEN_BRANCH, "required": ["auth_type", "token"]} + keys_with_optional_key = {**keys, "required": ["auth_type"]} + + comparison = compare_specs(_one_of(keys, tokens), _one_of(keys_with_optional_key, tokens)) + + assert comparison.is_backward_compatible + + +# patternProperties + + +@pytest.mark.parametrize( + "previous_schema", + [ + pytest.param({}, id="keyword-added"), + pytest.param({"patternProperties": {}}, id="pattern-added-to-an-empty-map"), + pytest.param( + {"patternProperties": {"^y_": {"type": "string"}}}, id="pattern-added-beside-another" + ), + ], +) +def test_added_pattern_property_is_breaking(previous_schema: dict[str, Any]) -> None: + patterns = {**previous_schema.get("patternProperties", {}), "^x_": {"type": "integer"}} + + comparison = compare_specs( + _spec({"options": {"type": "object", **previous_schema}}), + _spec({"options": {"type": "object", "patternProperties": patterns}}), + ) + + assert not comparison.is_backward_compatible + + +def test_pattern_properties_added_at_its_default_is_compatible() -> None: + comparison = compare_specs( + _spec({"options": {"type": "object"}}), + _spec({"options": {"type": "object", "patternProperties": {}}}), + ) + + assert comparison.is_backward_compatible + + +def test_config_field_named_pattern_properties_is_a_field() -> None: + comparison = compare_specs( + _spec({"options": {"type": "object", "properties": {}}}), + _spec({"options": {"type": "object", "properties": {"patternProperties": {}}}}), + ) + + assert comparison.is_backward_compatible + + +# JSON equality and widenings the generic comparison cannot see + + +@pytest.mark.parametrize( + "previous_schema, current_schema", + [ + pytest.param({"enum": [1]}, {"enum": [True]}, id="enum-1-to-true"), + pytest.param({"enum": [True]}, {"enum": [1]}, id="enum-true-to-1"), + pytest.param({"const": 0}, {"const": False}, id="const-0-to-false"), + pytest.param({"enum": [{"a": 1}]}, {"enum": [{"a": True}]}, id="nested-enum"), + ], +) +def test_booleans_are_not_numbers( + previous_schema: dict[str, Any], current_schema: dict[str, Any] +) -> None: + comparison = compare_specs(_spec({"flag": previous_schema}), _spec({"flag": current_schema})) + + assert not comparison.is_backward_compatible + + +def test_integer_and_float_of_the_same_value_are_equal() -> None: + assert ( + compare_specs( + _spec({"page_size": {"type": "number", "maximum": 10}}), + _spec({"page_size": {"type": "number", "maximum": 10.0}}), + ) + == SpecComparison() + ) + + +@pytest.mark.parametrize( + "keyword, current_value", + [ + pytest.param("additionalProperties", {"type": "string"}, id="additionalProperties"), + pytest.param("items", {"type": "string"}, id="items"), + ], +) +def test_false_relaxed_to_a_schema_is_compatible(keyword: str, current_value: Any) -> None: + comparison = compare_specs( + _spec({"options": {"type": "object", keyword: False}}), + _spec({"options": {"type": "object", keyword: current_value}}), + ) + + assert comparison.is_backward_compatible + assert comparison.compatible == [ + f"`connectionSpecification.properties.options.{keyword}` relaxed from False to " + "{'type': 'string'}" + ] + + +@pytest.mark.parametrize( + "previous_value, current_value, compatible", + [ + pytest.param(4, 2, True, id="divisor-relaxes"), + pytest.param(0.5, 0.25, True, id="float-divisor-relaxes"), + pytest.param(2, 4, False, id="multiple-tightens"), + pytest.param(4, 3, False, id="unrelated-tightens"), + ], +) +def test_multiple_of_is_judged_by_divisibility( + previous_value: float, current_value: float, compatible: bool +) -> None: + comparison = compare_specs( + _spec({"page_size": {"type": "integer", "multipleOf": previous_value}}), + _spec({"page_size": {"type": "integer", "multipleOf": current_value}}), + ) + + assert comparison.is_backward_compatible is compatible + + +# declared_breaking_changes + + +_METADATA_WITH_BREAKING_CHANGES: dict[str, Any] = { + "releases": { + "breakingChanges": { + "1.0.0": {"message": "first", "upgradeDeadline": "2026-01-01"}, + "2.0.0": {"message": "second", "upgradeDeadline": "2026-06-01"}, + } + } +} + + +@pytest.mark.parametrize( + "previous_version, current_version, expected", + [ + pytest.param("1.9.0", "2.0.0", ["2.0.0"], id="major-under-test"), + pytest.param("1.9.0", "2.0.1", ["2.0.0"], id="patch-on-top-of-major-still-rolling-out"), + pytest.param("2.0.0", "2.0.1", [], id="major-already-published"), + pytest.param("1.2.0", "1.2.1", [], id="no-major-in-range"), + pytest.param("0.9.0", "2.0.0", ["1.0.0", "2.0.0"], id="several-majors-in-range"), + pytest.param("2.0.0", "2.0.0", [], id="version-not-bumped"), + pytest.param("1.9.0", "2.0.0-rc.1", ["2.0.0"], id="release-candidate-of-the-major"), + pytest.param("1.9.0", "2.0.1-rc.1", ["2.0.0"], id="release-candidate-after-the-major"), + pytest.param("1.9.0", "1.9.1-rc.1", [], id="release-candidate-before-the-major"), + pytest.param("2.0.0", "2.0.1-rc.1", [], id="release-candidate-after-a-published-major"), + pytest.param("not-a-version", "2.0.0", ["2.0.0"], id="unparsable-falls-back-to-exact"), + ], +) +def test_declared_breaking_changes( + previous_version: str, current_version: str, expected: list[str] +) -> None: + assert ( + declared_breaking_changes( + _METADATA_WITH_BREAKING_CHANGES, previous_version, current_version + ) + == expected + ) + + +def test_declared_breaking_changes_without_releases() -> None: + assert declared_breaking_changes({}, "1.0.0", "2.0.0") == [] + + +@pytest.mark.parametrize( + "version, than, expected", + [ + pytest.param("1.2.4", "1.2.3", True, id="newer"), + pytest.param("1.2.3", "1.2.3", False, id="same"), + pytest.param("1.2.3", "1.10.0", False, id="older-by-number-not-by-string"), + pytest.param("2.0.0", "2.0.0-rc.1", True, id="release-after-its-candidate"), + pytest.param("dev", "1.2.3", False, id="unparsable"), + ], +) +def test_is_newer_version(version: str, than: str, expected: bool) -> None: + assert is_newer_version(version, than) is expected + + +# disabled_for_version + + +def test_disabled_for_version_reads_every_spec_entry() -> None: + acceptance_test_config = { + "acceptance_tests": { + "spec": { + "tests": [ + {"spec_path": "manifest.yaml"}, + {"backward_compatibility_tests_config": {"disable_for_version": "1.2.3"}}, + {"backward_compatibility_tests_config": {"disable_for_version": 2.0}}, + ] + }, + "discovery": { + "tests": [{"backward_compatibility_tests_config": {"disable_for_version": "9.9.9"}}] + }, + } + } + + assert disabled_for_version(acceptance_test_config) == ["1.2.3", "2.0"] + + +def test_disabled_for_version_reads_the_legacy_list_layout() -> None: + acceptance_test_config = { + "acceptance_tests": { + "spec": [ + {"spec_path": "source_test/spec.yaml"}, + {"backward_compatibility_tests_config": {"disable_for_version": "1.2.3"}}, + ] + } + } + + assert disabled_for_version(acceptance_test_config) == ["1.2.3"] + + +@pytest.mark.parametrize( + "acceptance_test_config", + [ + pytest.param(None, id="empty-file"), + pytest.param([], id="list-file"), + pytest.param({}, id="empty"), + pytest.param({"acceptance_tests": None}, id="null-acceptance-tests"), + pytest.param({"acceptance_tests": []}, id="list-acceptance-tests"), + pytest.param({"acceptance_tests": {}}, id="no-spec-section"), + pytest.param({"acceptance_tests": {"spec": {"bypass_reason": "n/a"}}}, id="no-tests"), + pytest.param({"acceptance_tests": {"spec": {"tests": [None]}}}, id="null-entry"), + pytest.param({"acceptance_tests": {"spec": {"tests": "x"}}}, id="tests-not-a-list"), + pytest.param({"acceptance_tests": {"spec": [{"spec_path": "x"}]}}, id="legacy-no-waiver"), + pytest.param( + {"acceptance_tests": {"spec": [{"backward_compatibility_tests_config": "x"}]}}, + id="malformed-waiver", + ), + ], +) +def test_disabled_for_version_without_waivers(acceptance_test_config: Any) -> None: + assert disabled_for_version(acceptance_test_config) == [] + + +# fetch_published_spec + +_REGISTRY_URL = "https://connectors.airbyte.com/files/metadata/airbyte/source-test/latest/oss.json" + + +def test_fetch_published_spec_returns_the_latest_entry() -> None: + spec = _spec(_BASE_PROPERTIES) + with requests_mock.Mocker() as mocker: + mocker.get(_REGISTRY_URL, json={"dockerImageTag": "1.2.3", "spec": spec}) + + published = fetch_published_spec("airbyte/source-test", "oss") + + assert published == PublishedSpec(version="1.2.3", spec=spec, url=_REGISTRY_URL) + + +def test_fetch_published_spec_returns_none_when_not_published() -> None: + with requests_mock.Mocker() as mocker: + mocker.get(_REGISTRY_URL, status_code=404) + + assert fetch_published_spec("airbyte/source-test", "oss") is None + + +def test_fetch_published_spec_raises_when_the_registry_errors() -> None: + with requests_mock.Mocker() as mocker: + mocker.get(_REGISTRY_URL, status_code=403) + + with pytest.raises(requests.HTTPError) as error: + fetch_published_spec("airbyte/source-test", "oss") + + assert not isinstance(error.value, REGISTRY_UNAVAILABLE_ERRORS) + + +@pytest.mark.parametrize( + "response, expected_message", + [ + pytest.param({"text": ""}, "invalid JSON", id="not-json"), + pytest.param( + {"json": {"spec": {}}}, "no `spec` object or no `dockerImageTag`", id="no-tag" + ), + pytest.param( + {"json": {"dockerImageTag": "1.2.3"}}, + "no `spec` object or no `dockerImageTag`", + id="no-spec", + ), + pytest.param( + {"json": {"dockerImageTag": "1.2.3", "spec": "x"}}, + "no `spec` object or no `dockerImageTag`", + id="spec-not-an-object", + ), + pytest.param({"json": []}, "(found: list)", id="not-an-object"), + ], +) +def test_fetch_published_spec_rejects_a_malformed_entry( + response: dict[str, Any], expected_message: str +) -> None: + with requests_mock.Mocker() as mocker: + mocker.get(_REGISTRY_URL, **response) + + with pytest.raises(AssertionError) as error: + fetch_published_spec("airbyte/source-test", "oss") + + assert expected_message in str(error.value) + assert _REGISTRY_URL in str(error.value) + + +@pytest.mark.parametrize("scheme", ["https", "http"]) +def test_registry_session_retries_every_scheme(scheme: str) -> None: + with registry_session() as session: + adapter = session.get_adapter(f"{scheme}://connectors.airbyte.com/files/metadata") + + assert adapter.max_retries.total == registry_retry().total # type: ignore[attr-defined] + assert set(adapter.max_retries.status_forcelist) >= {500, 503} # type: ignore[attr-defined] + + +def test_registry_retry_does_not_need_urllib3_1_26() -> None: + # `allowed_methods` was added in urllib3 1.26, which the CDK does not require. GET is + # retried by the default policy of every urllib3 version. + retry = registry_retry() + + assert retry.is_retry("GET", 503) + + +class _ScriptedRegistry: + """A local HTTP server answering each request with the next scripted status code. + + `requests_mock` replaces the transport adapter, so it cannot exercise the retry policy that + `fetch_published_spec` mounts; a real server can. + """ + + def __init__(self, statuses: list[int]) -> None: + self.statuses = statuses + self.requests = 0 + registry = self + + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - the name http.server dispatches to + status = registry.statuses[min(registry.requests, len(registry.statuses) - 1)] + registry.requests += 1 + body = json.dumps({"dockerImageTag": "1.2.3", "spec": {}}).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_: Any) -> None: + pass + + self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + + @property + def url_template(self) -> str: + port = self.server.server_address[1] + return f"http://127.0.0.1:{port}/{{docker_repository}}/{{version}}/{{registry}}.json" + + def __enter__(self) -> _ScriptedRegistry: + self.thread.start() + return self + + def __exit__(self, *_: Any) -> None: + self.server.shutdown() + self.server.server_close() + + +@pytest.mark.parametrize( + "statuses, expected_requests", + [ + pytest.param([503, 503, 200], 3, id="recovers-after-two-server-errors"), + pytest.param([429, 200], 2, id="recovers-after-a-rate-limit"), + pytest.param([200], 1, id="first-try"), + ], +) +def test_fetch_published_spec_retries_transient_errors( + monkeypatch: pytest.MonkeyPatch, statuses: list[int], expected_requests: int +) -> None: + with _ScriptedRegistry(statuses) as registry: + monkeypatch.setattr( + _spec_compatibility, "REGISTRY_ENTRY_URL_TEMPLATE", registry.url_template + ) + + published = fetch_published_spec("airbyte/source-test", "oss") + + assert published is not None + assert published.version == "1.2.3" + assert registry.requests == expected_requests + + +@pytest.mark.parametrize( + "statuses, error, expected_requests", + [ + pytest.param([503], requests.exceptions.RetryError, 3, id="gives-up-after-two-retries"), + pytest.param([403], requests.HTTPError, 1, id="client-error-is-not-retried"), + ], +) +def test_fetch_published_spec_raises_when_retries_do_not_help( + monkeypatch: pytest.MonkeyPatch, + statuses: list[int], + error: type[Exception], + expected_requests: int, +) -> None: + with _ScriptedRegistry(statuses) as registry: + monkeypatch.setattr( + _spec_compatibility, "REGISTRY_ENTRY_URL_TEMPLATE", registry.url_template + ) + + with pytest.raises(error): + fetch_published_spec("airbyte/source-test", "oss") + + assert registry.requests == expected_requests + + +def test_fetch_published_spec_does_not_retry_a_missing_entry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + with _ScriptedRegistry([404]) as registry: + monkeypatch.setattr( + _spec_compatibility, "REGISTRY_ENTRY_URL_TEMPLATE", registry.url_template + ) + + assert fetch_published_spec("airbyte/source-test", "oss") is None + + assert registry.requests == 1 + + +def test_unreachable_registry_fails_fast(monkeypatch: pytest.MonkeyPatch) -> None: + with socket.socket() as closed: + closed.bind(("127.0.0.1", 0)) + port = closed.getsockname()[1] + monkeypatch.setattr( + _spec_compatibility, + "REGISTRY_ENTRY_URL_TEMPLATE", + f"http://127.0.0.1:{port}/{{docker_repository}}/{{version}}/{{registry}}.json", + ) + + started = time.monotonic() + with pytest.raises(requests.ConnectionError): + fetch_published_spec("airbyte/source-test", "oss") + + assert time.monotonic() - started < 5 + + +def test_registry_wait_is_bounded() -> None: + retry = registry_retry() + assert isinstance(retry.total, int) + attempts = retry.total + 1 + backoff = sum( + retry.backoff_factor * 2 ** (consecutive_errors - 1) + for consecutive_errors in range(2, attempts) + ) + + assert retry.respect_retry_after_header is False + assert set(retry.status_forcelist) >= {429, 500, 502, 503, 504} + assert attempts * sum(REGISTRY_TIMEOUT_SECONDS) + backoff <= 60 + assert attempts * REGISTRY_TIMEOUT_SECONDS[0] + backoff <= 20 + + +def test_failure_message_points_at_the_breaking_change_process() -> None: + published = PublishedSpec(version="1.2.3", spec={}, url=_REGISTRY_URL) + comparison = compare_specs( + _spec(_BASE_PROPERTIES), _spec({"api_key": _BASE_PROPERTIES["api_key"]}) + ) + + message = format_breaking_spec_changes( + connector_name="source-test", + current_version="1.2.4", + registry="oss", + published=published, + comparison=comparison, + ) + + assert "The OSS spec of `source-test` 1.2.4" in message + assert "published version 1.2.3" in message + assert " - `connectionSpecification.properties.start_date` was removed" in message + assert BREAKING_CHANGES_DOCS_URL in message + assert "`releases.breakingChanges`" in message + assert 'disable_for_version: "1.2.3"' in message + + +def test_change_summary_lists_compatible_and_waived_changes() -> None: + published = PublishedSpec(version="1.2.3", spec={}, url=_REGISTRY_URL) + comparison = SpecComparison(breaking=["`a` was removed"], compatible=["`b` was added"]) + + summary = format_spec_change_summary( + registry="cloud", published=published, comparison=comparison, waiver="declared in 2.0.0" + ) + + assert summary.splitlines() == [ + "CLOUD spec compared with the published version 1.2.3: 1 breaking, 1 compatible change(s).", + "Breaking changes, waived (declared in 2.0.0):", + " - `a` was removed", + "Compatible changes:", + " - `b` was added", + ] + + +def test_change_summary_is_capped() -> None: + published = PublishedSpec(version="1.2.3", spec={}, url=_REGISTRY_URL) + changes = [f"`field_{index}` was added" for index in range(MAX_REPORTED_CHANGES + 5)] + + summary = format_spec_change_summary( + registry="oss", published=published, comparison=SpecComparison(compatible=changes) + ) + + lines = summary.splitlines() + assert len(lines) == 2 + MAX_REPORTED_CHANGES + 1 + assert lines[-1] == " … and 5 more" + assert "Breaking" not in summary + + +# DockerConnectorTestSuite.test_docker_image_spec_backward_compatibility + +_PREVIOUS_SPEC = _spec(_BASE_PROPERTIES) +_BREAKING_SPEC = _spec({"api_key": _BASE_PROPERTIES["api_key"]}) + + +def _make_suite( + tmp_path: Path, + *, + docker_image_tag: str = "1.2.4", + breaking_changes: dict[str, Any] | None = None, + acceptance_test_config: dict[str, Any] | None = None, + acceptance_test_config_text: str | None = None, +) -> type[DockerConnectorTestSuite]: + connector_root = tmp_path / "source-test" + connector_root.mkdir() + data: dict[str, Any] = { + "dockerRepository": "airbyte/source-test", + "dockerImageTag": docker_image_tag, + "tags": ["language:manifest-only"], + } + if breaking_changes: + data["releases"] = {"breakingChanges": breaking_changes} + (connector_root / "metadata.yaml").write_text(yaml.safe_dump({"data": data})) + if acceptance_test_config is not None: + acceptance_test_config_text = yaml.safe_dump(acceptance_test_config) + if acceptance_test_config_text is not None: + (connector_root / "acceptance-test-config.yml").write_text(acceptance_test_config_text) + + return type( + "TestSuite", + (DockerConnectorTestSuite,), + {"get_connector_root_dir": classmethod(lambda cls: connector_root)}, + ) + + +def _patch_specs( + monkeypatch: pytest.MonkeyPatch, + *, + published: dict[str, dict[str, Any]], + current: dict[str, dict[str, Any]], + versions: dict[str, str] | None = None, +) -> list[str]: + """Serve `published` from the registry and `current` from the image; record modes run. + + Each registry publishes version 1.2.3 unless `versions` says otherwise. + """ + modes_run: list[str] = [] + + def fake_fetch(docker_repository: str, registry: str) -> PublishedSpec | None: + assert docker_repository == "airbyte/source-test" + if registry not in published: + return None + version = (versions or {}).get(registry, "1.2.3") + return PublishedSpec(version=version, spec=published[registry], url=f"{registry}.json") + + def fake_run_spec(connector_image: str, deployment_mode: str) -> dict[str, Any]: + assert connector_image == "airbyte/source-test:dev" + modes_run.append(deployment_mode) + return current[deployment_mode] + + monkeypatch.setattr(docker_base, "fetch_published_spec", fake_fetch) + monkeypatch.setattr(docker_base, "_run_spec_in_image", fake_run_spec) + return modes_run + + +class _Skipped(Exception): + """The test under test skipped. + + A `pytest.skip` that escapes a unit test skips the unit test itself, so a test that expects + a failure would silently pass when the test under test skips instead. `_run_test` turns the + skip into this error, which a test expecting a failure does not catch. + """ + + +def _run_test(suite: type[DockerConnectorTestSuite]) -> None: + try: + suite().test_docker_image_spec_backward_compatibility( + connector_image_override="airbyte/source-test:dev", + connector_base_image_override=None, + ) + except pytest.skip.Exception as skip: + raise _Skipped(skip.msg) from skip + + +def test_compatible_spec_passes_in_every_published_registry( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + modes_run = _patch_specs( + monkeypatch, + published={"oss": _PREVIOUS_SPEC, "cloud": _PREVIOUS_SPEC}, + current={"oss": _PREVIOUS_SPEC, "cloud": _PREVIOUS_SPEC}, + ) + + _run_test(_make_suite(tmp_path)) + + assert modes_run == ["oss", "cloud"] + + +def test_breaking_spec_fails_with_the_findings( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs( + monkeypatch, + published={"oss": _PREVIOUS_SPEC}, + current={"oss": _BREAKING_SPEC}, + ) + + with pytest.raises(pytest.fail.Exception) as error: + _run_test(_make_suite(tmp_path)) + + assert "`connectionSpecification.properties.start_date` was removed" in str(error.value) + assert BREAKING_CHANGES_DOCS_URL in str(error.value) + + +def test_breaking_change_only_in_cloud_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs( + monkeypatch, + published={"oss": _PREVIOUS_SPEC, "cloud": _PREVIOUS_SPEC}, + current={"oss": _PREVIOUS_SPEC, "cloud": _BREAKING_SPEC}, + ) + + with pytest.raises(pytest.fail.Exception) as error: + _run_test(_make_suite(tmp_path)) + + assert "The CLOUD spec of `source-test` 1.2.4" in str(error.value) + assert "The OSS spec" not in str(error.value) + + +def test_registry_without_an_entry_is_not_compared( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + modes_run = _patch_specs( + monkeypatch, + published={"oss": _PREVIOUS_SPEC}, + current={"oss": _PREVIOUS_SPEC, "cloud": _BREAKING_SPEC}, + ) + + _run_test(_make_suite(tmp_path)) + + assert modes_run == ["oss"] + + +def test_unpublished_connector_is_skipped(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + modes_run = _patch_specs(monkeypatch, published={}, current={}) + + with pytest.raises(_Skipped, match="no published version"): + _run_test(_make_suite(tmp_path)) + + assert modes_run == [] + + +def test_version_older_than_the_published_one_is_skipped( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + modes_run = _patch_specs( + monkeypatch, + published={"oss": _PREVIOUS_SPEC}, + current={"oss": _BREAKING_SPEC}, + ) + + with pytest.raises(_Skipped, match="older than the published version 1.2.3"): + _run_test(_make_suite(tmp_path, docker_image_tag="1.2.2")) + + assert modes_run == [] + + +def test_declared_breaking_change_waives_the_findings( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite( + tmp_path, + docker_image_tag="2.0.0", + breaking_changes={"2.0.0": {"message": "Removes `start_date`."}}, + ) + + with pytest.raises(_Skipped, match="declared as breaking in 2.0.0"): + _run_test(suite) + + +def test_breaking_change_declared_for_an_older_version_does_not_waive( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite(tmp_path, breaking_changes={"1.0.0": {"message": "Old change."}}) + + with pytest.raises(pytest.fail.Exception): + _run_test(suite) + + +def test_disable_for_version_matching_the_published_version_waives_the_findings( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite( + tmp_path, + acceptance_test_config={ + "acceptance_tests": { + "spec": { + "tests": [ + {"backward_compatibility_tests_config": {"disable_for_version": "1.2.3"}} + ] + } + } + }, + ) + + with pytest.raises(_Skipped, match="disable_for_version: 1.2.3"): + _run_test(suite) + + +def test_stale_disable_for_version_does_not_waive( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite( + tmp_path, + acceptance_test_config={ + "acceptance_tests": { + "spec": { + "tests": [ + {"backward_compatibility_tests_config": {"disable_for_version": "0.1.0"}} + ] + } + } + }, + ) + + with pytest.raises(pytest.fail.Exception): + _run_test(suite) + + +def test_release_candidate_of_a_declared_major_is_waived( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite( + tmp_path, + docker_image_tag="2.0.0-rc.1", + breaking_changes={"2.0.0": {"message": "Removes `start_date`."}}, + ) + + with pytest.raises(_Skipped, match="declared as breaking in 2.0.0"): + _run_test(suite) + + +def test_compatible_changes_are_reported_as_a_warning_on_pass( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + extended = _spec({**_BASE_PROPERTIES, "page_size": {"type": "integer"}}) + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": extended}) + + with pytest.warns(UserWarning, match="judged safe or waived") as warned: + _run_test(_make_suite(tmp_path)) + + message = str(warned[0].message) + assert "OSS spec compared with the published version 1.2.3" in message + assert "`connectionSpecification.properties.page_size` was added" in message + + +def test_nothing_is_reported_for_an_unchanged_spec( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, recwarn: pytest.WarningsRecorder +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _PREVIOUS_SPEC}) + + _run_test(_make_suite(tmp_path)) + + assert not [w for w in recwarn if "judged safe" in str(w.message)] + + +def test_waived_changes_are_reported_as_a_warning( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite( + tmp_path, + docker_image_tag="2.0.0", + breaking_changes={"2.0.0": {"message": "Removes `start_date`."}}, + ) + + with pytest.warns(UserWarning, match="judged safe or waived") as warned: + with pytest.raises(_Skipped): + _run_test(suite) + + message = str(warned[0].message) + assert "Breaking changes, waived (declared as breaking in 2.0.0):" in message + assert "`connectionSpecification.properties.start_date` was removed" in message + + +@pytest.mark.parametrize( + "error", + [ + pytest.param(requests.ConnectionError("connection refused"), id="connection-error"), + pytest.param(requests.exceptions.ProxyError("blocked by proxy"), id="proxy-error"), + pytest.param(requests.exceptions.ReadTimeout("read timed out"), id="timeout"), + pytest.param(requests.exceptions.RetryError("too many 503 responses"), id="retries"), + ], +) +def test_unreachable_registry_skips_with_a_warning( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, error: Exception +) -> None: + def unreachable(docker_repository: str, registry: str) -> PublishedSpec | None: + raise error + + monkeypatch.setattr(docker_base, "fetch_published_spec", unreachable) + + with pytest.warns(UserWarning, match="could not be reached") as warned: + with pytest.raises(_Skipped) as skipped: + _run_test(_make_suite(tmp_path)) + + assert str(error) in str(skipped.value) + assert "not compared with the published version" in str(warned[0].message) + + +def test_registry_client_error_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + def forbidden(docker_repository: str, registry: str) -> PublishedSpec | None: + raise requests.HTTPError("403 Client Error: Forbidden") + + monkeypatch.setattr(docker_base, "fetch_published_spec", forbidden) + + with pytest.raises(pytest.fail.Exception, match="403 Client Error"): + _run_test(_make_suite(tmp_path)) + + +def test_unreachable_registry_skips_the_real_fetch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + with socket.socket() as closed: + closed.bind(("127.0.0.1", 0)) + port = closed.getsockname()[1] + monkeypatch.setattr( + _spec_compatibility, + "REGISTRY_ENTRY_URL_TEMPLATE", + f"http://127.0.0.1:{port}/{{docker_repository}}/{{version}}/{{registry}}.json", + ) + + with pytest.warns(UserWarning, match="could not be reached"): + with pytest.raises(_Skipped): + _run_test(_make_suite(tmp_path)) + + +# Registries that publish different versions + + +def test_registry_ahead_of_the_image_is_not_compared_but_the_other_one_is( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + modes_run = _patch_specs( + monkeypatch, + published={"oss": _PREVIOUS_SPEC, "cloud": _PREVIOUS_SPEC}, + current={"oss": _BREAKING_SPEC, "cloud": _BREAKING_SPEC}, + versions={"oss": "1.3.0", "cloud": "1.2.3"}, + ) + + with pytest.raises(pytest.fail.Exception) as error: + _run_test(_make_suite(tmp_path, docker_image_tag="1.2.4")) + + assert modes_run == ["cloud"] + assert "The CLOUD spec of `source-test` 1.2.4" in str(error.value) + assert "published version 1.2.3" in str(error.value) + + +def test_waiver_for_one_registry_does_not_hide_a_break_in_the_other( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Cloud is still on 1.2.2, for example because it is pinned there, so a waiver for the OSS + # version 1.2.3 does not cover it. + _patch_specs( + monkeypatch, + published={"oss": _PREVIOUS_SPEC, "cloud": _PREVIOUS_SPEC}, + current={"oss": _BREAKING_SPEC, "cloud": _BREAKING_SPEC}, + versions={"oss": "1.2.3", "cloud": "1.2.2"}, + ) + suite = _make_suite( + tmp_path, + acceptance_test_config={ + "acceptance_tests": { + "spec": { + "tests": [ + {"backward_compatibility_tests_config": {"disable_for_version": "1.2.3"}} + ] + } + } + }, + ) + + with pytest.warns(UserWarning, match="disable_for_version: 1.2.3"): + with pytest.raises(pytest.fail.Exception) as error: + _run_test(suite) + + assert "The CLOUD spec of `source-test` 1.2.4" in str(error.value) + assert "The OSS spec" not in str(error.value) + + +def test_image_is_built_when_no_image_is_given( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _PREVIOUS_SPEC}) + builds: list[dict[str, Any]] = [] + + def fake_build(**kwargs: Any) -> str: + builds.append(kwargs) + return "airbyte/source-test:dev" + + monkeypatch.setattr(docker_base, "build_connector_image", fake_build) + + _make_suite(tmp_path)().test_docker_image_spec_backward_compatibility( + connector_image_override=None, + connector_base_image_override="airbyte/python-connector-base:9.9.9", + ) + + assert len(builds) == 1 + assert builds[0]["connector_name"] == "source-test" + assert builds[0]["tag"] == "dev-latest" + assert builds[0]["base_image_override"] == "airbyte/python-connector-base:9.9.9" + + +# acceptance-test-config.yml layouts + + +_LEGACY_LIST_LAYOUT = """\ +acceptance_tests: + spec: + - spec_path: source_test/spec.yaml + backward_compatibility_tests_config: + disable_for_version: "1.2.3" +""" + + +def test_legacy_list_layout_waiver_is_honored( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite(tmp_path, acceptance_test_config_text=_LEGACY_LIST_LAYOUT) + + with pytest.warns(UserWarning): + with pytest.raises(_Skipped, match="disable_for_version: 1.2.3"): + _run_test(suite) + + +@pytest.mark.parametrize( + "acceptance_test_config_text", + [ + pytest.param("", id="empty-file"), + pytest.param("acceptance_tests:\n spec:\n - spec_path: x\n", id="legacy-no-waiver"), + pytest.param("tests: {}\n", id="no-acceptance-tests"), + pytest.param("acceptance_tests: [\n", id="invalid-yaml"), + ], +) +def test_unusable_acceptance_test_config_does_not_crash_the_test( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, acceptance_test_config_text: str +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _BREAKING_SPEC}) + suite = _make_suite(tmp_path, acceptance_test_config_text=acceptance_test_config_text) + + with pytest.raises(pytest.fail.Exception, match="start_date` was removed"): + _run_test(suite) + + +def test_acceptance_test_config_is_not_read_for_a_compatible_spec( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_specs(monkeypatch, published={"oss": _PREVIOUS_SPEC}, current={"oss": _PREVIOUS_SPEC}) + suite = _make_suite(tmp_path, acceptance_test_config_text="acceptance_tests: [\n") + + _run_test(suite) + + +def test_run_spec_in_image_returns_the_raw_spec(monkeypatch: pytest.MonkeyPatch) -> None: + raw_spec = {"connectionSpecification": {"type": "object"}, "x_unknown_key": True} + commands: list[list[str]] = [] + + def fake_run_docker_command(cmd: list[str], **_: Any) -> subprocess.CompletedProcess[str]: + commands.append(cmd) + stdout = "\n".join( + [ + "not json", + json.dumps({"type": "LOG", "log": {"level": "INFO", "message": "hello"}}), + json.dumps({"type": "SPEC", "spec": raw_spec}), + ] + ) + return subprocess.CompletedProcess(cmd, 0, stdout=stdout, stderr="") + + monkeypatch.setattr(docker_base, "run_docker_command", fake_run_docker_command) + + assert docker_base._run_spec_in_image("image:tag", "cloud") == raw_spec + assert commands == [ + [ + "docker", + "run", + "--rm", + "-e", + "DEPLOYMENT_MODE=cloud", + "-e", + "AIRBYTE_EDITION=CLOUD", + "image:tag", + "spec", + ] + ] + + +def test_run_spec_in_image_without_a_spec_message_fails(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + docker_base, + "run_docker_command", + lambda cmd, **_: subprocess.CompletedProcess(cmd, 0, stdout="", stderr="boom"), + ) + + with pytest.raises(AssertionError, match="emitted no SPEC message") as error: + docker_base._run_spec_in_image("image:tag", "oss") + + assert "Stderr:\nboom" in str(error.value) + + +def test_failing_spec_reports_what_the_connector_printed(monkeypatch: pytest.MonkeyPatch) -> None: + trace = { + "type": "TRACE", + "trace": { + "type": "ERROR", + "emitted_at": 0, + "error": { + "message": "Cloud mode needs a secret store.", + "internal_message": "KeyError: 'SECRET_STORE'", + "stack_trace": "Traceback (most recent call last):\n KeyError: 'SECRET_STORE'", + }, + }, + } + stdout = "\n".join( + [ + json.dumps({"type": "LOG", "log": {"level": "INFO", "message": "starting spec"}}), + "plain stdout marker", + json.dumps(trace), + ] + ) + calls: list[dict[str, Any]] = [] + + def failing_run_docker_command( + cmd: list[str], **kwargs: Any + ) -> subprocess.CompletedProcess[str]: + calls.append(kwargs) + return subprocess.CompletedProcess(cmd, 1, stdout=stdout, stderr="stderr marker") + + monkeypatch.setattr(docker_base, "run_docker_command", failing_run_docker_command) + + with pytest.raises(AssertionError) as error: + docker_base._run_spec_in_image("image:tag", "cloud") + + message = str(error.value) + assert calls[0]["raise_if_errors"] is False + assert "`spec` in image 'image:tag' (cloud mode) exited with code 1." in message + assert "Error reported by the connector: Cloud mode needs a secret store." in message + assert "Internal message: KeyError: 'SECRET_STORE'" in message + assert "Traceback (most recent call last)" in message + assert "starting spec" in message + assert "plain stdout marker" in message + assert "Stderr:\nstderr marker" in message + + +def test_failing_spec_output_is_truncated(monkeypatch: pytest.MonkeyPatch) -> None: + stderr = "\n".join(f"line {index}" for index in range(200)) + monkeypatch.setattr( + docker_base, + "run_docker_command", + lambda cmd, **_: subprocess.CompletedProcess(cmd, 2, stdout="", stderr=stderr), + ) + + with pytest.raises(AssertionError) as error: + docker_base._run_spec_in_image("image:tag", "oss") + + message = str(error.value) + assert "... (150 earlier lines omitted)" in message + assert "line 199" in message + assert "line 149\n" not in message + + +def test_spec_message_with_a_non_zero_exit_still_fails(monkeypatch: pytest.MonkeyPatch) -> None: + stdout = json.dumps({"type": "SPEC", "spec": {"connectionSpecification": {}}}) + monkeypatch.setattr( + docker_base, + "run_docker_command", + lambda cmd, **_: subprocess.CompletedProcess(cmd, 1, stdout=stdout, stderr="crashed"), + ) + + with pytest.raises(AssertionError, match="exited with code 1"): + docker_base._run_spec_in_image("image:tag", "oss")