test: add tests for lazy load

Author: fede-bello

tests/test_lazy_loading.py

@@ -0,0 +1,217 @@
+"""
+Test pydantic_settings LazyMapping core functionality.
+
+LazyMapping defers field value resolution until the fields are accessed,
+rather than eagerly evaluating all fields during settings initialization.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from pydantic_settings import PydanticBaseSettingsSource
+from pydantic_settings.sources.lazy import LazyMapping
+
+
+class TestLazyMapping:
+    """Test LazyMapping class behavior."""
+
+    def test_lazy_mapping_init(self):
+        """Test LazyMapping initialization."""
+        source = MagicMock(spec=PydanticBaseSettingsSource)
+        mapping = LazyMapping(source)
+        assert mapping._source is source
+        assert mapping._cached_values == {}
+
+    def test_lazy_mapping_getitem_with_caching(self):
+        """Test LazyMapping caches values after first access."""
+        source = MagicMock()
+
+        # Mock settings class with a field
+        settings_cls = MagicMock()
+        field_info = MagicMock()
+        settings_cls.model_fields = {'test_field': field_info}
+        source.settings_cls = settings_cls
+
+        # Mock the field resolution
+        source._get_resolved_field_value.return_value = ('test-value', None, False)
+        source.prepare_field_value.return_value = 'prepared-value'
+
+        mapping = LazyMapping(source)
+
+        # First access should call the resolution methods
+        value1 = mapping['test_field']
+        assert value1 == 'prepared-value'
+        assert source._get_resolved_field_value.call_count == 1
+
+        # Second access should use cached value
+        value2 = mapping['test_field']
+        assert value2 == 'prepared-value'
+        assert source._get_resolved_field_value.call_count == 1  # Not called again
+
+    def test_lazy_mapping_getitem_key_not_found(self):
+        """Test LazyMapping raises KeyError for missing keys."""
+        source = MagicMock()
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {}
+        source.settings_cls = settings_cls
+
+        mapping = LazyMapping(source)
+
+        with pytest.raises(KeyError):
+            _ = mapping['nonexistent']
+
+    def test_lazy_mapping_iter(self):
+        """Test LazyMapping iteration returns all field names."""
+        source = MagicMock()
+
+        # Mock settings class with multiple fields
+        field_info1 = MagicMock()
+        field_info2 = MagicMock()
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {'field1': field_info1, 'field2': field_info2}
+        source.settings_cls = settings_cls
+
+        # Mock alias names function
+        with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias:
+            mock_alias.side_effect = [(['alias1'], None), (['alias2'], None)]
+
+            mapping = LazyMapping(source)
+            keys = list(mapping)
+
+            assert 'alias1' in keys
+            assert 'field1' in keys or 'alias2' in keys
+            assert 'field2' in keys or 'alias2' in keys
+
+    def test_lazy_mapping_len(self):
+        """Test LazyMapping returns correct length."""
+        source = MagicMock(spec=PydanticBaseSettingsSource)
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {'field1': MagicMock(), 'field2': MagicMock(), 'field3': MagicMock()}
+        source.settings_cls = settings_cls
+
+        mapping = LazyMapping(source)
+        assert len(mapping) == 3
+
+    def test_lazy_mapping_copy(self):
+        """Test LazyMapping.copy() preserves cached values."""
+        source = MagicMock(spec=PydanticBaseSettingsSource)
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {'field1': MagicMock()}
+        source.settings_cls = settings_cls
+
+        mapping = LazyMapping(source)
+        mapping._cached_values['field1'] = 'cached-value'
+
+        copied = mapping.copy()
+
+        assert isinstance(copied, LazyMapping)
+        assert copied._source is source
+        assert copied._cached_values == {'field1': 'cached-value'}
+        # Ensure it's a shallow copy of the dict
+        assert copied._cached_values is not mapping._cached_values
+
+    def test_lazy_mapping_items(self):
+        """Test LazyMapping.items() yields accessible key-value pairs."""
+        source = MagicMock()
+
+        field_info = MagicMock()
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {'test_field': field_info}
+        source.settings_cls = settings_cls
+
+        source._get_resolved_field_value.return_value = ('value', None, False)
+        source.prepare_field_value.return_value = 'prepared-value'
+
+        with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias:
+            mock_alias.return_value = (['test_field'], None)
+
+            mapping = LazyMapping(source)
+            items = dict(mapping.items())
+
+            assert 'test_field' in items
+            assert items['test_field'] == 'prepared-value'
+
+
+class TestLazyLoadingWithDeepUpdate:
+    """Test LazyMapping behavior with deep_update to ensure lazy behavior is preserved."""
+
+    def test_lazy_mapping_deep_update_preserves_lazy_behavior(self):
+        """Test that deep_update with LazyMapping copy() preserves lazy behavior."""
+        from pydantic._internal._utils import deep_update
+
+        source = MagicMock()
+        settings_cls = MagicMock()
+        field_info = MagicMock()
+        settings_cls.model_fields = {'field1': field_info, 'field2': field_info}
+        source.settings_cls = settings_cls
+
+        source._get_resolved_field_value.return_value = ('value', None, False)
+        source.prepare_field_value.side_effect = lambda fname, fi, v, _: f'prepared-{fname}'
+
+        # Create LazyMapping
+        mapping2 = LazyMapping(source)
+
+        # Simulate deep_update behavior
+        result = deep_update({'field1': 'eager-value'}, mapping2.copy())
+
+        # The result should contain the lazily evaluated value from mapping2
+        # and the copy() should have returned a LazyMapping, not a regular dict
+        assert isinstance(result.get('field1'), str)
+
+    def test_lazy_mapping_copy_maintains_source_reference(self):
+        """Test that LazyMapping.copy() maintains reference to the same source."""
+        source = MagicMock()
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {}
+        source.settings_cls = settings_cls
+
+        mapping = LazyMapping(source)
+        copied = mapping.copy()
+
+        # Both should reference the same source object
+        assert mapping._source is copied._source
+
+
+class TestLazyLoadingWithConfigFiles:
+    """Test lazy loading compatibility with config file sources."""
+
+    def test_lazy_mapping_with_dict_like_interface(self):
+        """Test that LazyMapping implements proper dict-like interface."""
+        source = MagicMock()
+        field_info = MagicMock()
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {'field1': field_info}
+        source.settings_cls = settings_cls
+
+        source._get_resolved_field_value.return_value = ('test-value', None, False)
+        source.prepare_field_value.return_value = 'prepared-value'
+
+        mapping = LazyMapping(source)
+
+        # Test Mapping interface methods
+        assert 'field1' in list(mapping)
+        assert len(mapping) == 1
+        assert mapping['field1'] == 'prepared-value'
+
+    def test_lazy_mapping_error_on_missing_field(self):
+        """Test LazyMapping raises SettingsError for missing fields during resolution."""
+        from pydantic_settings.exceptions import SettingsError
+
+        source = MagicMock()
+        field_info = MagicMock()
+        settings_cls = MagicMock()
+        settings_cls.model_fields = {'field1': field_info}
+        source.settings_cls = settings_cls
+        source.__class__.__name__ = 'MockSource'
+
+        # Simulate an error during field value resolution
+        source._get_resolved_field_value.side_effect = ValueError('Connection failed')
+
+        mapping = LazyMapping(source)
+
+        # Accessing the field should raise SettingsError (not ValueError)
+        with pytest.raises(SettingsError, match='error getting value for field "field1"'):
+            _ = mapping['field1']

tests/test_lazy_mapping.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 from typing import Annotated, Any, Generic, Literal, TypeVar
 from unittest import mock
+from unittest.mock import patch
 
 import pytest
 from annotated_types import MinLen
@@ -50,6 +51,7 @@
     SettingsError,
 )
 from pydantic_settings.sources import DefaultSettingsSource
+from pydantic_settings.sources.lazy import LazyMapping
 
 try:
     import dotenv
@@ -3264,3 +3266,133 @@ class StrictSettings(BaseSettings, env_nested_delimiter='__', strict=True):
             'my_int': 1,
         },
     }
+
+
+class TestEnvSettingsSourceLazyLoading:
+    """Unit tests for lazy_load behavior in EnvSettingsSource."""
+
+    def test_returns_lazy_mapping_stored_when_lazy_load_true(self):
+        """Test EnvSettingsSource stores LazyMapping when lazy_load=True."""
+
+        class TestSettings(BaseSettings):
+            field: str = 'default'
+
+        source = EnvSettingsSource(TestSettings, lazy_load=True)
+        result = source()
+        # When lazy_load=True, __call__ returns empty dict
+        assert isinstance(result, dict)
+        assert len(result) == 0
+        # But LazyMapping is stored on the source for later retrieval
+        assert hasattr(source, '_lazy_mapping')
+        assert isinstance(source._lazy_mapping, LazyMapping)
+
+    def test_returns_dict_when_lazy_load_false(self):
+        """Test EnvSettingsSource returns dict when lazy_load=False."""
+
+        class TestSettings(BaseSettings):
+            field: str = 'default'
+
+        source = EnvSettingsSource(TestSettings, lazy_load=False)
+        result = source()
+        assert isinstance(result, dict)
+        # When lazy_load=False, there should be no _lazy_mapping
+        assert not hasattr(source, '_lazy_mapping') or source._lazy_mapping is None
+
+    def test_lazy_mapping_defers_resolution(self):
+        """Test LazyMapping defers field value resolution."""
+
+        class TestSettings(BaseSettings):
+            test_field: str = 'default'
+
+        source = EnvSettingsSource(TestSettings, lazy_load=True)
+
+        # Mock the internal resolution method
+        with patch.object(source, '_get_resolved_field_value', wraps=source._get_resolved_field_value) as mock_resolve:
+            source()
+            # Resolution should NOT be called during source()
+            assert mock_resolve.call_count == 0
+
+            # Now access a field through LazyMapping stored on source
+            lazy_mapping = source._lazy_mapping
+            try:
+                _ = lazy_mapping['test_field']
+                # Resolution SHOULD be called on field access
+                assert mock_resolve.call_count > 0
+            except KeyError:
+                # Field not found is ok, point is resolution was attempted
+                assert mock_resolve.call_count > 0
+
+
+def test_env_source_accepts_lazy_load():
+    """Test EnvSettingsSource accepts lazy_load parameter."""
+
+    class TestSettings(BaseSettings):
+        field: str = 'default'
+
+    # Should not raise any error
+    source = EnvSettingsSource(TestSettings, lazy_load=True)
+    assert hasattr(source, 'lazy_load')
+
+
+class TestDotEnvSettingsSourceLazyLoading:
+    """Unit tests for lazy_load behavior in DotEnvSettingsSource."""
+
+    def test_returns_lazy_mapping_when_lazy_load_true(self, tmp_path):
+        """Test DotEnvSettingsSource stores LazyMapping when lazy_load=True."""
+        env_file = tmp_path / '.env'
+        env_file.write_text('TEST_FIELD=value\n')
+
+        class TestSettings(BaseSettings):
+            test_field: str = 'default'
+
+        source = DotEnvSettingsSource(TestSettings, env_file=env_file, lazy_load=True)
+        result = source()
+        assert isinstance(result, dict)
+        assert len(result) == 0
+        assert hasattr(source, '_lazy_mapping')
+        assert isinstance(source._lazy_mapping, LazyMapping)
+
+    def test_returns_dict_when_lazy_load_false(self, tmp_path):
+        """Test DotEnvSettingsSource returns dict when lazy_load=False."""
+        env_file = tmp_path / '.env'
+        env_file.write_text('TEST_FIELD=value\n')
+
+        class TestSettings(BaseSettings):
+            test_field: str = 'default'
+
+        source = DotEnvSettingsSource(TestSettings, env_file=env_file, lazy_load=False)
+        result = source()
+        assert isinstance(result, dict)
+        assert not hasattr(source, '_lazy_mapping') or source._lazy_mapping is None
+
+    def test_lazy_mapping_defers_resolution(self, tmp_path):
+        """Test DotEnvSettingsSource LazyMapping defers resolution."""
+        env_file = tmp_path / '.env'
+        env_file.write_text('TEST_FIELD=test_value\n')
+
+        class TestSettings(BaseSettings):
+            test_field: str = 'default'
+
+        source = DotEnvSettingsSource(TestSettings, env_file=env_file, lazy_load=True)
+
+        with patch.object(source, '_get_resolved_field_value', wraps=source._get_resolved_field_value) as mock_resolve:
+            source()
+            assert mock_resolve.call_count == 0
+
+            lazy_mapping = source._lazy_mapping
+            try:
+                _ = lazy_mapping['test_field']
+                assert mock_resolve.call_count > 0
+            except KeyError:
+                assert mock_resolve.call_count > 0
+
+
+def test_dotenv_source_accepts_lazy_load():
+    """Test DotEnvSettingsSource accepts lazy_load parameter."""
+
+    class TestSettings(BaseSettings):
+        field: str = 'default'
+
+    # Should not raise any error
+    source = DotEnvSettingsSource(TestSettings, lazy_load=True)
+    assert hasattr(source, 'lazy_load')

tests/test_settings.py

@@ -158,3 +158,12 @@ def settings_customise_sources(
 
         assert settings.sql_server_user == 'test-user'
         assert settings.sql_server.password == 'test-password'
+
+
+@pytest.mark.skipif(not aws_secrets_manager, reason='pydantic-settings[aws-secrets-manager] is not installed')
+def test_aws_secrets_source_accepts_lazy_load():
+    """Test AWSSecretsManagerSettingsSource accepts lazy_load parameter."""
+    import inspect
+
+    sig = inspect.signature(AWSSecretsManagerSettingsSource.__init__)
+    assert 'lazy_load' in sig.parameters