fix: improve oneOf/anyOf config validation error messages

Replace validate() with Draft7Validator.iter_errors() to detect
oneOf/anyOf validation failures and produce per-branch error details
instead of surfacing a misleading error from the last non-matching
branch (e.g. "'Service' was expected" when the user intended OAuth).

The improved message lists each oneOf option by title with the specific
validation errors for that branch, making it actionable for users to
identify which required fields are missing in their chosen auth method.

Co-Authored-By: bot_apk <apk@cognition.ai>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/utils/schema_helpers.py

@@ -11,7 +11,7 @@
 from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Mapping, MutableMapping, Tuple, cast
 
 import jsonref
-from jsonschema import validate
+from jsonschema import Draft7Validator
 from jsonschema.exceptions import ValidationError
 from pydantic.v1 import BaseModel, Field
 from referencing import Registry, Resource
@@ -185,25 +185,66 @@ def _resolve_schema_references(self, raw_schema: dict[str, Any]) -> dict[str, An
             raise ValueError(f"Expected resolved to be a dict. Got {resolved}")
 
 
+def _format_oneof_error(error: ValidationError) -> str:
+    """Format a oneOf/anyOf validation error with per-branch details."""
+    field_path = ".".join(str(p) for p in error.absolute_path) if error.absolute_path else "root"
+    branch_schemas = error.schema.get(error.validator, [])
+    branch_messages: list[str] = []
+    errors_by_branch: dict[int, list[str]] = {}
+    for sub_error in error.context:
+        branch_index = sub_error.schema_path[0]
+        if isinstance(branch_index, int):
+            errors_by_branch.setdefault(branch_index, []).append(sub_error.message)
+
+    for i, branch_schema in enumerate(branch_schemas):
+        label = branch_schema.get("title", f"option {i + 1}")
+        sub_errors = errors_by_branch.get(i, [])
+        if sub_errors:
+            details = "; ".join(sub_errors)
+            branch_messages.append(f"{label}: {details}")
+        else:
+            branch_messages.append(f"{label}: unknown validation failure")
+
+    joined = " | ".join(branch_messages)
+    return f'Config validation error: the provided configuration for "{field_path}" does not match any supported option. {joined}.'
+
+
 def check_config_against_spec_or_exit(
     config: Mapping[str, Any], spec: ConnectorSpecification
 ) -> None:
-    """
-    Check config object against spec. In case of spec is invalid, throws
-    an exception with validation error description.
+    """Check config object against spec and raise on validation failure.
 
-    :param config - config loaded from file specified over command line
-    :param spec - spec object generated by connector
+    Uses iter_errors to detect oneOf/anyOf failures and produce a message
+    that lists per-branch validation details instead of surfacing a
+    misleading error from the last non-matching branch.
     """
     spec_schema = spec.connectionSpecification
-    try:
-        validate(instance=config, schema=spec_schema)
-    except ValidationError as validation_error:
-        raise AirbyteTracedException(
-            message="Config validation error: " + validation_error.message,
-            internal_message=validation_error.message,
-            failure_type=FailureType.config_error,
-        ) from None  # required to prevent logging config secrets from the ValidationError's stacktrace
+    validator = Draft7Validator(spec_schema)
+    errors = list(validator.iter_errors(config))
+    if not errors:
+        return
+
+    oneof_errors = [e for e in errors if e.validator in ("oneOf", "anyOf") and e.context]
+    if oneof_errors:
+        error = oneof_errors[0]
+        message = _format_oneof_error(error)
+        internal_parts = [
+            f"{sub.message} (path: {list(sub.absolute_path)})" for sub in error.context
+        ]
+        internal_message = (
+            f"oneOf validation failed at path {list(error.absolute_path)}: "
+            + "; ".join(internal_parts)
+        )
+    else:
+        error = errors[0]
+        message = "Config validation error: " + error.message
+        internal_message = error.message
+
+    raise AirbyteTracedException(
+        message=message,
+        internal_message=internal_message,
+        failure_type=FailureType.config_error,
+    ) from None  # required to prevent logging config secrets from the ValidationError's stacktrace
 
 
 class InternalConfig(BaseModel):