add tests for dpath validator

Author: pnilan

airbyte_cdk/sources/declarative/validators/dpath_validator.py

@@ -39,11 +39,27 @@ def validate(self, input_data: dict[str, Any]) -> None:
         :param input_data: Dictionary containing the data to validate
         :raises ValueError: If the path doesn't exist or validation fails
         """
-        try:
-            path = [path.eval(input_data) for path in self._field_path]
-            value = dpath.util.get(input_data, path)
-            self.strategy.validate(value)
-        except KeyError:
-            raise ValueError(f"Path '{self.field_path}' not found in the input data")
-        except Exception as e:
-            raise ValueError(f"Error validating path '{self.field_path}': {e}")
+        path = [path.eval({}) for path in self._field_path]
+
+        if len(path) == 0:
+            raise ValueError("Field path is empty")
+
+        if "*" in path:
+            try:
+                values = dpath.values(input_data, path)
+            except KeyError as e:
+                raise KeyError(f"Error validating path '{self.field_path}': {e}")
+            for value in values:
+                try:
+                    self.strategy.validate(value)
+                except Exception as e:
+                    raise ValueError(f"Error validating value '{value}': {e}")
+        else:
+            try:
+                value = dpath.get(input_data, path)
+            except KeyError as e:
+                raise KeyError(f"Error validating path '{self.field_path}': {e}")
+            try:
+                self.strategy.validate(value)
+            except Exception as e:
+                raise ValueError(f"Error validating value '{value}': {e}")

unit_tests/sources/declarative/validators/test_dpath_validator.py

@@ -0,0 +1,92 @@
+import pytest
+from unittest import TestCase
+
+from airbyte_cdk.sources.declarative.validators.dpath_validator import DpathValidator
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+
+
+class MockValidationStrategy(ValidationStrategy):
+    def __init__(self, should_fail=False, error_message="Validation failed"):
+        self.should_fail = should_fail
+        self.error_message = error_message
+        self.validate_called = False
+        self.validated_value = None
+
+    def validate(self, value):
+        self.validate_called = True
+        self.validated_value = value
+        if self.should_fail:
+            raise ValueError(self.error_message)
+
+
+class TestDpathValidator(TestCase):
+    def test_given_valid_path_and_input_validate_is_successful(self):
+        strategy = MockValidationStrategy()
+        validator = DpathValidator(field_path=["user", "profile", "email"], strategy=strategy)
+
+        test_data = {"user": {"profile": {"email": "[email protected]", "name": "Test User"}}}
+
+        validator.validate(test_data)
+
+        assert strategy.validate_called
+        assert strategy.validated_value
+
+    def test_given_invalid_path_when_validate_then_raise_key_error(self):
+        strategy = MockValidationStrategy()
+        validator = DpathValidator(field_path=["user", "profile", "phone"], strategy=strategy)
+
+        test_data = {"user": {"profile": {"email": "[email protected]"}}}
+
+        with pytest.raises(KeyError) as context:
+            validator.validate(test_data)
+            assert "Error validating path" in str(context.exception)
+            assert not strategy.validate_called
+
+    def test_given_strategy_fails_when_validate_then_raise_value_error(self):
+        error_message = "Invalid email format"
+        strategy = MockValidationStrategy(should_fail=True, error_message=error_message)
+        validator = DpathValidator(field_path=["user", "email"], strategy=strategy)
+
+        test_data = {"user": {"email": "invalid-email"}}
+
+        with pytest.raises(ValueError) as context:
+            validator.validate(test_data)
+
+            assert "Error validating value" in str(context.exception)
+            assert error_message in str(context.exception)
+            assert strategy.validate_called
+            assert strategy.validated_value == "invalid-email"
+
+    def test_given_empty_path_list_when_validate_then_validate_raises_exception(self):
+        strategy = MockValidationStrategy()
+        validator = DpathValidator(field_path=[], strategy=strategy)
+        test_data = {"key": "value"}
+
+        with pytest.raises(ValueError):
+            validator.validate(test_data)
+
+    def test_given_empty_input_data_when_validate_then_validate_raises_exception(self):
+        strategy = MockValidationStrategy()
+        validator = DpathValidator(field_path=["data", "field"], strategy=strategy)
+
+        test_data = {}
+
+        with pytest.raises(KeyError):
+            validator.validate(test_data)
+
+    def test_path_with_wildcard_when_validate_then_validate_is_successful(self):
+        strategy = MockValidationStrategy()
+        validator = DpathValidator(field_path=["users", "*", "email"], strategy=strategy)
+
+        test_data = {
+            "users": {
+                "user1": {"email": "[email protected]", "name": "User One"},
+                "user2": {"email": "[email protected]", "name": "User Two"},
+            }
+        }
+
+        validator.validate(test_data)
+
+        assert strategy.validate_called
+        assert strategy.validated_value in ["[email protected]", "[email protected]"]
+        self.assertIn(strategy.validated_value, ["[email protected]", "[email protected]"])