Add ConfigResolver with necessary tests

packages/smithy-aws-core/src/smithy_aws_core/config/__init__.py

@@ -1,7 +1,10 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+from .resolver import get_shared_resolver, reset_shared_resolver
 from .sources import EnvironmentSource
 
 __all__ = [
     "EnvironmentSource",
+    "get_shared_resolver",
+    "reset_shared_resolver",
 ]

packages/smithy-aws-core/src/smithy_aws_core/config/resolver.py

@@ -0,0 +1,30 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from functools import lru_cache
+
+from smithy_core.config.resolver import ConfigResolver
+from smithy_core.interfaces.config import ConfigSource
+
+from .sources import EnvironmentSource
+
+
+@lru_cache(maxsize=1)
+def get_shared_resolver() -> ConfigResolver:
+    """Get or create the shared AWS configuration resolver.
+
+    This resolver is shared across all config objects and AWS service clients to avoid
+    redundant reads from environment variables, config files, etc.
+
+    :returns: The shared ConfigResolver instance
+    """
+    sources: list[ConfigSource] = [EnvironmentSource()]
+    return ConfigResolver(sources=sources)
+
+
+def reset_shared_resolver() -> None:
+    """Reset the shared resolver (only for testing).
+
+    This allows tests to start with a clean resolver state.
+    """
+    get_shared_resolver.cache_clear()

packages/smithy-aws-core/tests/unit/config/test_shared_resolver.py

@@ -0,0 +1,66 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import os
+from concurrent.futures import Future, ThreadPoolExecutor
+from unittest.mock import patch
+
+from smithy_aws_core.config import get_shared_resolver, reset_shared_resolver
+from smithy_core.config.resolver import ConfigResolver
+
+
+class TestGetSharedResolver:
+    def setup_method(self):
+        # Reset the shared resolver before each test
+        reset_shared_resolver()
+
+    def test_returns_config_resolver_instance(self):
+        resolver = get_shared_resolver()
+
+        assert isinstance(resolver, ConfigResolver)
+
+    def test_returns_same_instance_on_repeated_calls(self):
+        resolver1 = get_shared_resolver()
+        resolver2 = get_shared_resolver()
+        resolver3 = get_shared_resolver()
+
+        assert resolver1 is resolver2
+        assert resolver2 is resolver3
+
+    def test_resolves_from_environment_variables(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            resolver = get_shared_resolver()
+            value, source = resolver.get("region")
+
+            assert value == "us-west-2"
+            assert source == "environment"
+
+    def test_reset_clears_singleton(self):
+        resolver1 = get_shared_resolver()
+
+        reset_shared_resolver()
+
+        resolver2 = get_shared_resolver()
+
+        # After reset, it should get a new instance
+        assert resolver1 is not resolver2
+
+    def test_multiple_thread_calls_return_same_instance(self) -> None:
+        results: list[ConfigResolver] = []
+
+        # Multiple thread calls should use the same resolver instance
+        def get_resolver() -> None:
+            resolver = get_shared_resolver()
+            results.append(resolver)
+
+        # Create 10 threads that all call get_shared_resolver concurrently
+        with ThreadPoolExecutor(max_workers=10) as executor:
+            futures: list[Future[None]] = [
+                executor.submit(get_resolver) for _ in range(10)
+            ]
+            for future in futures:
+                future.result()
+
+        first_resolver: ConfigResolver = results[0]
+        assert len(results) == 10
+        # All threads should have gotten the same resolver instance
+        assert all(resolver is first_resolver for resolver in results)

packages/smithy-core/src/smithy_core/config/__init__.py

@@ -0,0 +1,10 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from smithy_core.interfaces.config import ConfigSource
+
+from .resolver import ConfigResolver
+
+__all__ = [
+    "ConfigResolver",
+    "ConfigSource",
+]

packages/smithy-core/src/smithy_core/config/resolver.py

@@ -0,0 +1,36 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from typing import Any
+
+from smithy_core.interfaces.config import ConfigSource
+
+
+class ConfigResolver:
+    """Resolves configuration values from multiple sources.
+
+    The resolver iterates through sources in precedence order, returning
+    the first non-None value found for a given configuration key.
+    """
+
+    def __init__(self, sources: list[ConfigSource]) -> None:
+        """Initialize the resolver with sources in precedence order.
+
+        :param sources: List of configuration sources in precedence order. The first
+            source in the list has the highest priority. The list is copied to
+            prevent external modification.
+        """
+        self._sources = list(sources)
+
+    def get(self, key: str) -> tuple[Any, str]:
+        """Resolve a configuration value from sources by iterating through them in precedence order.
+
+        :param key: The configuration key to resolve (e.g., 'retry_mode')
+
+        :returns: A tuple of (value, source_name). If no source provides a value,
+            returns (None, 'unresolved').
+        """
+        for source in self._sources:
+            value = source.get(key)
+            if value is not None:
+                return (value, source.name)
+        return (None, "unresolved")

packages/smithy-core/tests/unit/config/__init__.py

@@ -0,0 +1,2 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0

packages/smithy-core/tests/unit/config/test_resolver.py

@@ -0,0 +1,122 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from typing import Any
+
+from smithy_core.config.resolver import ConfigResolver
+
+
+class StubSource:
+    """A simple ConfigSource implementation for testing.
+
+    Returns values from a provided dictionary, or None if the key
+    is not present.
+    """
+
+    def __init__(self, source_name: str, data: dict[str, Any] | None = None):
+        self._name = source_name
+        self._data = data or {}
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def get(self, key: str) -> Any | None:
+        return self._data.get(key)
+
+
+class TestConfigResolver:
+    def test_returns_value_from_single_source(self):
+        source = StubSource("environment", {"region": "us-west-2"})
+        resolver = ConfigResolver(sources=[source])
+
+        result = resolver.get("region")
+
+        assert result == ("us-west-2", "environment")
+
+    def test_returns_unresolved_when_source_has_no_value(self):
+        source = StubSource("environment", {})
+        resolver = ConfigResolver(sources=[source])
+
+        result = resolver.get("region")
+
+        assert result == (None, "unresolved")
+
+    def test_returns_unresolved_with_empty_source_list(self):
+        resolver = ConfigResolver(sources=[])
+
+        result = resolver.get("region")
+
+        assert result == (None, "unresolved")
+
+    def test_first_source_takes_precedence(self):
+        first_priority_source = StubSource("source_one", {"region": "us-east-1"})
+        second_priority_source = StubSource("source_two", {"region": "eu-west-1"})
+        resolver = ConfigResolver(
+            sources=[first_priority_source, second_priority_source]
+        )
+
+        result = resolver.get("region")
+
+        assert result == ("us-east-1", "source_one")
+
+    def test_skips_source_returning_none_and_uses_next(self):
+        empty_source = StubSource("source_one", {})
+        fallback_source = StubSource("source_two", {"region": "ap-south-1"})
+        resolver = ConfigResolver(sources=[empty_source, fallback_source])
+
+        result = resolver.get("region")
+
+        assert result == ("ap-south-1", "source_two")
+
+    def test_stops_at_first_non_none_value(self):
+        first_source = StubSource("source_one", {"region": "us-west-2"})
+        second_source = StubSource("source_two", {"region": "eu-west-1"})
+        third_source = StubSource("source_three", {"region": "us-east-1"})
+        resolver = ConfigResolver(sources=[first_source, second_source, third_source])
+
+        result = resolver.get("region")
+
+        assert result == ("us-west-2", "source_one")
+
+    def test_resolves_different_keys_from_different_sources(self):
+        instance = StubSource("source_one", {"region": "us-west-2"})
+        environment = StubSource("source_two", {"retry_mode": "adaptive"})
+        resolver = ConfigResolver(sources=[instance, environment])
+
+        region = resolver.get("region")
+        retry_mode = resolver.get("retry_mode")
+
+        assert region == ("us-west-2", "source_one")
+        assert retry_mode == ("adaptive", "source_two")
+
+    def test_returns_non_string_values(self):
+        source = StubSource(
+            "default",
+            {
+                "max_retries": 3,
+                "use_ssl": True,
+            },
+        )
+        resolver = ConfigResolver(sources=[source])
+
+        assert resolver.get("max_retries") == (3, "default")
+        assert resolver.get("use_ssl") == (True, "default")
+
+    def test_get_is_idempotent(self):
+        source = StubSource("environment", {"region": "us-west-2"})
+        resolver = ConfigResolver(sources=[source])
+
+        result1 = resolver.get("region")
+        result2 = resolver.get("region")
+        result3 = resolver.get("region")
+
+        assert result1 == result2 == result3 == ("us-west-2", "environment")
+
+    def test_treats_empty_string_as_valid_value(self):
+        source = StubSource("test", {"region": ""})
+        resolver = ConfigResolver(sources=[source])
+
+        value, source_name = resolver.get("region")
+
+        assert value == ""
+        assert source_name == "test"