fix(low-code): interpolate PredicateValidator.value and allow plain strings in ValidateAdheresToSchema

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -5487,7 +5487,7 @@ definitions:
         enum: [PredicateValidator]
       value:
         title: Value
-        description: The value to be validated. Can be a literal value or interpolated from configuration.
+        description: The value to be validated. Can be a literal value or interpolated from configuration. Strings (including strings nested in arrays and objects) are interpolated against the config at validation time, after config transformations are applied, and rendered strings are parsed as literals so an expression rendering a list is validated as a list.
         type:
           - string
           - number
@@ -5501,6 +5501,7 @@ definitions:
           - "test-value"
           - "{{ config['api_version'] }}"
           - "{{ config['tenant_id'] }}"
+          - "{{ config['report_options_list'] | map(attribute='stream_name') | list }}"
           - 123
       validation_strategy:
         title: Validation Strategy

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2492,11 +2492,12 @@ class PredicateValidator(BaseModel):
     type: Literal["PredicateValidator"]
     value: Optional[Union[str, float, Dict[str, Any], List[Any], bool]] = Field(
         ...,
-        description="The value to be validated. Can be a literal value or interpolated from configuration.",
+        description="The value to be validated. Can be a literal value or interpolated from configuration. Strings (including strings nested in arrays and objects) are interpolated against the config at validation time, after config transformations are applied, and rendered strings are parsed as literals so an expression rendering a list is validated as a list.",
         examples=[
             "test-value",
             "{{ config['api_version'] }}",
             "{{ config['tenant_id'] }}",
+            "{{ config['report_options_list'] | map(attribute='stream_name') | list }}",
             123,
         ],
         title="Value",

airbyte_cdk/sources/declarative/validators/predicate_validator.py

@@ -3,24 +3,44 @@
 #
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Mapping, Optional
 
+from airbyte_cdk.sources.declarative.interpolation.jinja import JinjaInterpolation
 from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+from airbyte_cdk.sources.declarative.validators.validator import Validator
 
 
 @dataclass
-class PredicateValidator:
+class PredicateValidator(Validator):
     """
     Validator that applies a validation strategy to a value.
+
+    String values (including strings nested in lists and objects) are interpolated against the
+    config passed to `validate`, so the value can be derived from the (already transformed) config.
+    Rendered strings are parsed as Python literals when possible, so an expression rendering a list
+    reaches the strategy as a list rather than as its string representation.
     """
 
     value: Any
     strategy: ValidationStrategy
 
-    def validate(self) -> None:
+    def __post_init__(self) -> None:
+        self._interpolation = JinjaInterpolation()
+
+    def validate(self, input_data: Optional[Mapping[str, Any]] = None) -> None:
         """
-        Applies the validation strategy to the value.
+        Interpolates the value against the config and applies the validation strategy to it.
 
+        :param input_data: The config the value is interpolated against
         :raises ValueError: If validation fails
         """
-        self.strategy.validate(self.value)
+        self.strategy.validate(self._interpolate(self.value, input_data or {}))
+
+    def _interpolate(self, value: Any, config: Mapping[str, Any]) -> Any:
+        if isinstance(value, str):
+            return self._interpolation.eval(value, config)
+        if isinstance(value, list):
+            return [self._interpolate(item, config) for item in value]
+        if isinstance(value, dict):
+            return {key: self._interpolate(item, config) for key, item in value.items()}
+        return value

airbyte_cdk/sources/declarative/validators/validate_adheres_to_schema.py

@@ -30,8 +30,9 @@ def validate(self, value: Any) -> None:
         if isinstance(value, str):
             try:
                 value = json.loads(value)
-            except json.JSONDecodeError as e:
-                raise ValueError(f"Invalid JSON string: {value}") from e
+            except json.JSONDecodeError:
+                # Not a JSON document (eg a hostname), so validate the raw string itself.
+                pass
 
         try:
             jsonschema.validate(instance=value, schema=self.schema)

unit_tests/sources/declarative/spec/test_spec.py

@@ -3,6 +3,7 @@
 #
 
 import dataclasses
+from typing import Any, List, Mapping
 
 import pytest
 
@@ -36,9 +37,15 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     OauthConnectorInputSpecification as component_declarative_oauth_connector_input_spec,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    Spec as SpecModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     State as component_declarative_oauth_state,
 )
+from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
+    ModelToComponentFactory,
+)
 from airbyte_cdk.sources.declarative.spec.spec import ConfigMigration
 from airbyte_cdk.sources.declarative.spec.spec import Spec as component_spec
 from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition
@@ -315,3 +322,99 @@ def test_protocol_override_fields_in_sync(
         f"Upstream protocol added fields {missing} to {upstream_class_name} "
         f"that are missing from the airbyte_protocol.py override. Update the override to match."
     )
+
+
+def _create_spec_with_validations(validations: List[Mapping[str, Any]]) -> component_spec:
+    return ModelToComponentFactory().create_component(
+        model_type=SpecModel,
+        component_definition={
+            "type": "Spec",
+            "connection_specification": {},
+            "config_normalization_rules": {
+                "type": "ConfigNormalizationRules",
+                "validations": validations,
+            },
+        },
+        config={},
+    )
+
+
+_UNIQUE_STREAM_NAMES_VALIDATION = {
+    "type": "PredicateValidator",
+    "value": "{{ config['report_options_list'] | map(attribute='stream_name') | list }}",
+    "validation_strategy": {
+        "type": "ValidateAdheresToSchema",
+        "base_schema": {"type": "array", "uniqueItems": True},
+    },
+}
+
+_JIRA_DOMAIN_VALIDATION = {
+    "type": "DpathValidator",
+    "field_path": ["domain"],
+    "validation_strategy": {
+        "type": "ValidateAdheresToSchema",
+        "base_schema": {"type": "string", "pattern": "^[a-zA-Z0-9-]+\\.atlassian\\.net$"},
+    },
+}
+
+
+def test_given_predicate_validator_with_unique_interpolated_list_when_validate_config_then_succeeds() -> (
+    None
+):
+    spec = _create_spec_with_validations([_UNIQUE_STREAM_NAMES_VALIDATION])
+
+    spec.validate_config(
+        {"report_options_list": [{"stream_name": "orders"}, {"stream_name": "returns"}]}
+    )
+
+
+def test_given_predicate_validator_with_duplicate_interpolated_list_when_validate_config_then_raises() -> (
+    None
+):
+    spec = _create_spec_with_validations([_UNIQUE_STREAM_NAMES_VALIDATION])
+
+    with pytest.raises(ValueError, match="non-unique elements"):
+        spec.validate_config(
+            {"report_options_list": [{"stream_name": "orders"}, {"stream_name": "orders"}]}
+        )
+
+
+def test_given_predicate_validator_reading_transformed_config_when_validate_config_then_uses_passed_config() -> (
+    None
+):
+    spec = _create_spec_with_validations(
+        [
+            {
+                "type": "PredicateValidator",
+                "value": "{{ config['added_field'] }}",
+                "validation_strategy": {
+                    "type": "ValidateAdheresToSchema",
+                    "base_schema": {"type": "string", "const": "expected"},
+                },
+            }
+        ]
+    )
+
+    spec.validate_config({"added_field": "expected"})
+    with pytest.raises(ValueError):
+        spec.validate_config({"added_field": "unexpected"})
+
+
+@pytest.mark.parametrize(
+    "domain, is_valid",
+    [
+        pytest.param("example.atlassian.net", True, id="valid_hostname"),
+        pytest.param("https://example.atlassian.net/", False, id="url_instead_of_hostname"),
+        pytest.param("example.com", False, id="non_atlassian_hostname"),
+    ],
+)
+def test_given_dpath_validator_with_plain_string_pattern_when_validate_config(
+    domain: str, is_valid: bool
+) -> None:
+    spec = _create_spec_with_validations([_JIRA_DOMAIN_VALIDATION])
+
+    if is_valid:
+        spec.validate_config({"domain": domain})
+    else:
+        with pytest.raises(ValueError, match="does not match"):
+            spec.validate_config({"domain": domain})

unit_tests/sources/declarative/validators/test_predicate_validator.py

@@ -53,3 +53,39 @@ def test_given_complex_object_when_validate_then_successful(self):
 
         assert strategy.validate_called
         assert strategy.validated_value == test_value
+
+    def test_given_interpolated_value_when_validate_then_value_is_evaluated_against_config(self):
+        strategy = MockValidationStrategy()
+        validator = PredicateValidator(value="{{ config['domain'] }}", strategy=strategy)
+
+        validator.validate({"domain": "example.atlassian.net"})
+
+        assert strategy.validated_value == "example.atlassian.net"
+
+    def test_given_interpolated_list_expression_when_validate_then_strategy_receives_list(self):
+        strategy = MockValidationStrategy()
+        validator = PredicateValidator(
+            value="{{ config['report_options_list'] | map(attribute='stream_name') | list }}",
+            strategy=strategy,
+        )
+
+        validator.validate({"report_options_list": [{"stream_name": "a"}, {"stream_name": "b"}]})
+
+        assert strategy.validated_value == ["a", "b"]
+
+    def test_given_literal_non_string_values_when_validate_then_values_are_preserved(self):
+        for literal in [123, 1.5, True, None, ["a", 1], {"key": [1, 2]}]:
+            strategy = MockValidationStrategy()
+            PredicateValidator(value=literal, strategy=strategy).validate({})
+            assert strategy.validated_value == literal
+
+    def test_given_nested_interpolated_values_when_validate_then_nested_strings_are_evaluated(self):
+        strategy = MockValidationStrategy()
+        validator = PredicateValidator(
+            value={"name": "{{ config['name'] }}", "ids": ["{{ config['id'] }}", 2]},
+            strategy=strategy,
+        )
+
+        validator.validate({"name": "test", "id": 1})
+
+        assert strategy.validated_value == {"name": "test", "ids": [1, 2]}

unit_tests/sources/declarative/validators/test_validate_adheres_to_schema.py

@@ -128,3 +128,20 @@ def test_given_json_string_when_validate_then_succeeds(self):
         validator = ValidateAdheresToSchema(schema=schema)
 
         validator.validate('{"id": 1, "name": "John Doe"}')
+
+    def test_given_non_json_string_matching_schema_when_validate_then_succeeds(self):
+        validator = ValidateAdheresToSchema(
+            schema={"type": "string", "pattern": "^[a-zA-Z0-9-]+\\.atlassian\\.net$"}
+        )
+
+        validator.validate("example.atlassian.net")
+
+    def test_given_non_json_string_not_matching_schema_when_validate_then_raises_error(self):
+        validator = ValidateAdheresToSchema(
+            schema={"type": "string", "pattern": "^[a-zA-Z0-9-]+\\.atlassian\\.net$"}
+        )
+
+        with pytest.raises(ValueError) as exc_info:
+            validator.validate("https://example.atlassian.net/")
+
+        assert "does not match" in str(exc_info.value)