From 24c682387013f5668a0dade74a487e8df23693e0 Mon Sep 17 00:00:00 2001
From: ubaskota <19787410+ubaskota@users.noreply.github.com>
Date: Mon, 13 Jul 2026 01:08:03 -0400
Subject: [PATCH 1/2] Add support for config resolver

# Conflicts:
#	packages/smithy-aws-core/src/smithy_aws_core/config/exceptions.py
---
 .../src/smithy_aws_core/config/aws_config.py  | 176 +++++++++++
 .../src/smithy_aws_core/config/context.py     | 139 +++++++++
 .../src/smithy_aws_core/config/resolvers.py   | 135 +++++++++
 .../src/smithy_aws_core/config/types.py       |  70 +++++
 .../src/smithy_aws_core/config/validators.py  |  42 +++
 .../tests/unit/config/test_resolver.py        | 285 ++++++++++++++++++
 .../tests/unit/config/test_validators.py      |  93 ++++++
 7 files changed, 940 insertions(+)
 create mode 100644 packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
 create mode 100644 packages/smithy-aws-core/src/smithy_aws_core/config/context.py
 create mode 100644 packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
 create mode 100644 packages/smithy-aws-core/src/smithy_aws_core/config/types.py
 create mode 100644 packages/smithy-aws-core/src/smithy_aws_core/config/validators.py
 create mode 100644 packages/smithy-aws-core/tests/unit/config/test_resolver.py
 create mode 100644 packages/smithy-aws-core/tests/unit/config/test_validators.py

diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py b/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
new file mode 100644
index 000000000..5fbcf601b
--- /dev/null
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
@@ -0,0 +1,176 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from typing import Any, ClassVar, Self
+
+from smithy_aws_core.config.context import FileSystem, SharedConfigContext
+from smithy_aws_core.config.exceptions import ConfigError
+from smithy_aws_core.config.resolvers import resolve_region, resolve_retry_config
+from smithy_aws_core.config.types import UNSET, ConfigSource, FieldSpec, Resolved
+from smithy_aws_core.config.validators import (
+    validate_region,
+    validate_retry_strategy_options,
+)
+from smithy_core.retries import RetryStrategyOptions
+
+
+@dataclass(kw_only=True)
+class AsyncAwsConfig:
+    """Base configuration class for all AWS services.
+
+    Fields are resolved asynchronously from multiple sources (env vars,
+    config files, defaults) through the resolve() classmethod.
+
+    Do not instantiate directly — use:
+        config = await AsyncAwsConfig.resolve()
+    """
+
+    region: str | None = None
+    retry_strategy_options: RetryStrategyOptions | None = None
+
+    _ctx: SharedConfigContext | None = field(default=None, repr=False, compare=False)
+    _sources: dict[str, ConfigSource] = field(  # type: ignore[assignment]
+        default_factory=dict,
+        repr=False,
+        compare=False,
+    )
+
+    _FIELDS: ClassVar[dict[str, FieldSpec]] = {
+        "region": FieldSpec(
+            default=None,
+            resolver=resolve_region,
+            validator=validate_region,
+        ),
+        "retry_strategy_options": FieldSpec(
+            default_factory=RetryStrategyOptions,
+            resolver=resolve_retry_config,
+            validator=validate_retry_strategy_options,
+        ),
+    }
+
+    def __post_init__(self) -> None:
+        """Block direct construction. Use resolve() instead."""
+        raise ConfigError(
+            f"{type(self).__name__} cannot be constructed directly. "
+            f"Use `await {type(self).__name__}.resolve(...)` instead."
+        )
+
+    @classmethod
+    async def resolve(
+        cls,
+        *,
+        profile: str | None = None,
+        env: Mapping[str, str] | None = None,
+        fs: FileSystem | None = None,
+        config_file_path: str | None = None,
+        credentials_file_path: str | None = None,
+        **overrides: Any,
+    ) -> Self:
+        """Resolve a config object from environment, config files, and defaults.
+
+        This is the only supported way to create a config instance.
+
+        :param profile: Override the active profile name.
+        :param env: Override the environment variable mapping.
+        :param fs: Override the filesystem abstraction.
+        :param config_file_path: Override path for config file.
+        :param credentials_file_path: Override path for credentials file.
+        :param overrides: Explicit field values that skip resolution.
+        :returns: A fully-resolved config instance.
+        """
+        ctx = SharedConfigContext(
+            profile_name=profile,
+            env=env,
+            fs=fs,
+            config_file_path=config_file_path,
+            credentials_file_path=credentials_file_path,
+        )
+
+        # Create the instance bypassing __post_init__ check
+        instance = cls._create_instance()
+        instance._ctx = ctx
+
+        # Resolve each field
+        await instance._resolve_fields(overrides)
+
+        return instance
+
+    def source_of(self, field_name: str) -> ConfigSource | None:
+        """Get the source that provided a field's value.
+
+        :param field_name: The config field name.
+        :returns: The ConfigSource, or None if not tracked.
+        """
+        return self._sources.get(field_name)
+
+    def resolution_context(self) -> SharedConfigContext | None:
+        """Get the resolution context used to create this config.
+
+        :returns: The SharedConfigContext, or None if not available.
+        """
+        return self._ctx
+
+    @classmethod
+    def _create_instance(cls) -> Self:
+        """Create an instance that bypasses construction blocking."""
+        instance = object.__new__(cls)
+        object.__setattr__(instance, "_sources", {})
+        object.__setattr__(instance, "_ctx", None)
+        for field_name in cls._FIELDS:
+            object.__setattr__(instance, field_name, UNSET)
+        return instance
+
+    async def _resolve_fields(self, overrides: dict[str, Any]) -> None:
+        """Run the resolution pipeline for all fields."""
+        unknown = set(overrides) - set(self._FIELDS)
+        if unknown:
+            raise ConfigError(
+                f"Unknown config field(s): {sorted(unknown)}. "
+                f"Valid fields are: {sorted(self._FIELDS)}"
+            )
+
+        for field_name, spec in self._FIELDS.items():
+            # check for overrides first
+            if field_name in overrides:
+                value = overrides[field_name]
+                setattr(self, field_name, value)
+                self._sources[field_name] = ConfigSource.OVERRIDE
+            # check in resolver
+            elif spec.resolver is not None:
+                result: Resolved[Any] = await spec.resolver(self._ctx)
+                if result.value is not UNSET:
+                    setattr(self, field_name, result.value)
+                    self._sources[field_name] = result.source
+                else:
+                    # If resolver returned UNSET, fall back to default
+                    self._apply_default(field_name, spec)
+
+            else:
+                # No resolver, use default directly
+                self._apply_default(field_name, spec)
+
+            # Run validator for all sources
+            if spec.validator is not None:
+                spec.validator(getattr(self, field_name))
+
+    def _apply_default(self, field_name: str, spec: FieldSpec) -> None:
+        """Apply the default value for a field."""
+        if spec.default_factory is not None:
+            value = spec.default_factory()
+        else:
+            value = spec.default
+        setattr(self, field_name, value)
+        self._sources[field_name] = ConfigSource.DEFAULT
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        """Track provenance when fields are set with plugins after construction"""
+        super().__setattr__(name, value)
+        # Mark as override only if the field is in _FIELDS and was already resolved
+        if (
+            name in self.__class__._FIELDS
+            and hasattr(self, "_sources")
+            and name in self._sources
+        ):
+            self._sources[name] = ConfigSource.OVERRIDE
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/context.py b/packages/smithy-aws-core/src/smithy_aws_core/config/context.py
new file mode 100644
index 000000000..fac549038
--- /dev/null
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/context.py
@@ -0,0 +1,139 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+import logging
+import os
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any, Protocol, runtime_checkable
+
+from smithy_aws_core.config import load_config
+from smithy_aws_core.config.merged_config import MergedConfig
+
+logger = logging.getLogger(__name__)
+
+
+@runtime_checkable
+class FileSystem(Protocol):
+    """Protocol for filesystem operations.
+
+    Abstraction over file I/O so tests can provide mock implementations
+    without touching the real filesystem.
+    """
+
+    async def read_file(self, path: str) -> str | None:
+        """Read a file's content as UTF-8.
+
+        :param path: Resolved file path.
+        :returns: File content, or None if the file is inaccessible.
+        """
+        ...
+
+
+class DefaultFileSystem:
+    """Default filesystem implementation using real disk I/O."""
+
+    async def read_file(self, path: str) -> str | None:
+        """Read a file asynchronously from disk.
+
+        Missing files and permission errors return None with a warning.
+        Encoding errors (invalid UTF-8) are raised to the caller.
+
+        :param path: Resolved file path.
+        :returns: File content, or None if the file is inaccessible.
+        """
+        try:
+            content: str = await asyncio.to_thread(
+                Path(path).read_text, encoding="utf-8"
+            )
+            return content
+        except FileNotFoundError:
+            return None
+        except (PermissionError, OSError) as e:
+            logger.warning("Unable to read config file '%s': %s", path, e)
+            return None
+
+
+class SharedConfigContext:
+    """Resolution context shared across resolvers during config construction.
+
+    Holds environment state and cached file data that resolvers use to
+    look up values. Created once per resolve() call.
+    """
+
+    def __init__(
+        self,
+        *,
+        profile_name: str | None = None,
+        env: Mapping[str, str] | None = None,
+        fs: FileSystem | None = None,
+        http_client: Any | None = None,
+        config_file_path: str | Path | None = None,
+        credentials_file_path: str | Path | None = None,
+    ) -> None:
+        """Initialize the resolution context.
+
+        :param profile_name: The active profile to use. Defaults to
+            AWS_PROFILE env var, then "default".
+        :param env: Environment variable mapping. Defaults to os.environ.
+        :param fs: Filesystem abstraction for reading config files.
+            Defaults to real disk I/O.
+        :param http_client: HTTP client for network-based resolvers
+            (e.g., IMDS).
+        :param config_file_path: Override path for config file.
+        :param credentials_file_path: Override path for credentials file.
+        """
+        self._env: Mapping[str, str] = env if env is not None else os.environ
+        self._fs: FileSystem = fs if fs is not None else DefaultFileSystem()
+        self._http_client: Any | None = http_client
+        self._profile_name: str = self._resolve_profile_name(profile_name)
+        self._config_file_path: Path | None = (
+            Path(config_file_path) if config_file_path is not None else None
+        )
+        self._credentials_file_path: Path | None = (
+            Path(credentials_file_path) if credentials_file_path is not None else None
+        )
+        self._cached_config_file: MergedConfig | None = None
+
+    @property
+    def env(self) -> Mapping[str, str]:
+        """The environment variable mapping."""
+        return self._env
+
+    @property
+    def profile_name(self) -> str:
+        """The active profile name."""
+        return self._profile_name
+
+    @property
+    def fs(self) -> FileSystem:
+        """The filesystem abstraction."""
+        return self._fs
+
+    @property
+    def http_client(self) -> Any | None:
+        """HTTP client for network-based resolvers."""
+        return self._http_client
+
+    async def parsed_profiles(self) -> MergedConfig:
+        """Get the parsed and merged config/credentials file data.
+
+        The result is cached after the first call so files are only
+        read from disk once per context.
+        """
+        if self._cached_config_file is None:
+            self._cached_config_file = await load_config(
+                config_file_path=self._config_file_path,
+                credentials_file_path=self._credentials_file_path,
+            )
+        return self._cached_config_file
+
+    def _resolve_profile_name(self, explicit_profile: str | None) -> str:
+        """Determine the active profile name.
+
+        Priority: explicit argument > AWS_PROFILE env var > "default"
+        """
+        if explicit_profile is not None:
+            return explicit_profile
+        return self._env.get("AWS_PROFILE", "default")
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py b/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
new file mode 100644
index 000000000..388390d03
--- /dev/null
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
@@ -0,0 +1,135 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Sequence
+from typing import Any
+
+from smithy_aws_core.config.context import SharedConfigContext
+from smithy_aws_core.config.exceptions import ConfigError
+from smithy_aws_core.config.types import UNSET, ConfigSource, Resolved
+from smithy_core.retries import RetryStrategyOptions
+
+
+async def _resolve_str(
+    ctx: SharedConfigContext,
+    *,
+    env_vars: Sequence[str] = (),
+    profile_keys: Sequence[str] = (),
+) -> Resolved[str | None]:
+    """Resolve a string value by checking providers in priority order.
+
+    Priority: env vars (first match) > config file (profile keys, first match) > unresolved.
+
+    :param ctx: The shared resolution context.
+    :param env_vars: Environment variable names to check, in order.
+    :param profile_keys: Config file profile keys to check, in order.
+    :returns: Resolved value with source, or Resolved(value=UNSET) if not found.
+    """
+    # Check environment variables first
+    for var_name in env_vars:
+        value: str | None = ctx.env.get(var_name)
+        if value:
+            return Resolved(value=value, source=ConfigSource.ENV)
+
+    # Check config file profile keys
+    if profile_keys:
+        config_file = await ctx.parsed_profiles()
+        for key in profile_keys:
+            value = config_file.get(ctx.profile_name, key)
+            if value:
+                return Resolved(value=value, source=ConfigSource.PROFILE)
+
+    return Resolved(value=UNSET, source=ConfigSource.DEFAULT)  # type: ignore[arg-type]
+
+
+async def _resolve_int(
+    ctx: SharedConfigContext,
+    *,
+    env_vars: Sequence[str] = (),
+    profile_keys: Sequence[str] = (),
+) -> Resolved[int | None]:
+    """Resolve an integer value by checking providers in priority order.
+
+    :param ctx: The shared resolution context.
+    :param env_vars: Environment variable names to check, in order.
+    :param profile_keys: Config file profile keys to check, in order.
+    :returns: Resolved int value with source, or Resolved(value=UNSET) if not found.
+    """
+    result = await _resolve_str(ctx, env_vars=env_vars, profile_keys=profile_keys)
+    if result.value is UNSET:
+        return Resolved(value=UNSET, source=ConfigSource.DEFAULT)  # type: ignore[arg-type]
+    if result.value is None:
+        return Resolved(value=None, source=result.source)
+    try:
+        return Resolved(value=int(result.value), source=result.source)
+    except (ValueError, TypeError):
+        raise ConfigError(
+            f"Invalid integer value {result.value!r} for config key. "
+            "Expected a valid integer."
+        )
+
+
+def _strongest_source(*sources: ConfigSource) -> ConfigSource:
+    """Return the highest-priority source among multiple resolved values.
+
+    Priority: ENV > PROFILE > DEFAULT
+    """
+    priority = {ConfigSource.ENV: 3, ConfigSource.PROFILE: 2, ConfigSource.DEFAULT: 1}
+    return max(sources, key=lambda s: priority.get(s, 0))
+
+
+async def resolve_region(ctx: SharedConfigContext) -> Resolved[str | None]:
+    """Resolve the AWS region from environment or config file.
+
+    :param ctx: The shared resolution context.
+    :returns: Resolved region value with source.
+    """
+    return await _resolve_str(
+        ctx,
+        env_vars=("AWS_REGION", "AWS_DEFAULT_REGION"),
+        profile_keys=("region",),
+    )
+
+
+async def resolve_retry_config(
+    ctx: SharedConfigContext,
+) -> Resolved[RetryStrategyOptions]:
+    """Resolve retry configuration from environment and config file.
+
+    Combines retry_mode and max_attempts into a single RetryStrategyOptions.
+    Each sub-value is resolved independently with its own priority chain.
+
+    :param ctx: The shared resolution context.
+    :returns: Resolved RetryStrategyOptions with the strongest source.
+    """
+    mode_result = await _resolve_str(
+        ctx,
+        env_vars=("AWS_RETRY_MODE",),
+        profile_keys=("retry_mode",),
+    )
+    attempts_result = await _resolve_int(
+        ctx,
+        env_vars=("AWS_MAX_ATTEMPTS",),
+        profile_keys=("max_attempts",),
+    )
+
+    # Build kwargs for values that were resolved
+    kwargs: dict[str, Any] = {}
+    sources_found: list[ConfigSource] = []
+
+    if mode_result.value is not UNSET and mode_result.value is not None:
+        kwargs["retry_mode"] = mode_result.value
+        sources_found.append(mode_result.source)
+
+    if attempts_result.value is not UNSET and attempts_result.value is not None:
+        kwargs["max_attempts"] = attempts_result.value
+        sources_found.append(attempts_result.source)
+
+    source = (
+        _strongest_source(*sources_found) if sources_found else ConfigSource.DEFAULT
+    )
+
+    return Resolved(
+        value=RetryStrategyOptions(**kwargs),
+        source=source,
+    )
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/types.py b/packages/smithy-aws-core/src/smithy_aws_core/config/types.py
new file mode 100644
index 000000000..6ff732169
--- /dev/null
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/types.py
@@ -0,0 +1,70 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+
+class _UnsetType:
+    def __repr__(self) -> str:
+        return "UNSET"
+
+    def __bool__(self) -> bool:
+        return False
+
+
+UNSET = _UnsetType()
+
+
+class ConfigSource(str, Enum):
+    """Where a resolved config value came from."""
+
+    ENV = "env"
+    PROFILE = "profile"
+    DEFAULT = "default"
+    OVERRIDE = "override"
+
+
+@dataclass(frozen=True, kw_only=True)
+class Resolved[T]:
+    """A resolved config value paired with its provenance source."""
+
+    value: T
+    source: ConfigSource
+
+
+@dataclass(frozen=True, kw_only=True)
+class FieldSpec:
+    """Specification for a single config field.
+
+    Carries everything the resolution pipeline needs to resolve,
+    validate, and default a field.
+
+    """
+
+    default: Any = UNSET
+    """Value used if no resolver runs or all providers return unresolved."""
+
+    default_factory: Callable[[], Any] | None = None
+    """Factory for mutable defaults (lists, dicts) that need a fresh instance per config."""
+
+    resolver: Callable[..., Awaitable[Resolved[Any]]] | None = None
+    """Async function that resolves the field from env/profile/etc."""
+
+    validator: Callable[[Any], None] | None = None
+    """Function that validates the resolved value. Raises on invalid input."""
+
+    def __post_init__(self) -> None:
+        has_default = self.default is not UNSET
+        has_factory = self.default_factory is not None
+        if has_default and has_factory:
+            raise ValueError(
+                "FieldSpec: cannot set both 'default' and 'default_factory'"
+            )
+        if not has_default and not has_factory:
+            raise ValueError(
+                "FieldSpec: exactly one of 'default' or 'default_factory' must be set"
+            )
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/validators.py b/packages/smithy-aws-core/src/smithy_aws_core/config/validators.py
new file mode 100644
index 000000000..a78f5f049
--- /dev/null
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/validators.py
@@ -0,0 +1,42 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import re
+
+from smithy_aws_core.config.exceptions import ConfigError
+from smithy_core.retries import RetryStrategyOptions
+
+_REGION_PATTERN = re.compile(r"^(?![0-9]+$)(?!-)[a-zA-Z0-9-]{1,63}(?<!-)$")
+
+
+def validate_region(value: object) -> None:
+    """Validate that region is a non-empty string matching AWS region format.
+
+    Region is required and must be explicitly set via env var, config file,
+    or override.
+
+    :param value: The resolved region value.
+    :raises ConfigError: If the value is None or doesn't match the pattern.
+    """
+    if value is None:
+        raise ConfigError(
+            "Invalid value for 'region': None. Region is required and must be set."
+        )
+    if not isinstance(value, str) or not _REGION_PATTERN.match(value):
+        raise ConfigError(
+            f"Invalid value for 'region': {value!r}. "
+            "Must be a valid AWS region identifier."
+        )
+
+
+def validate_retry_strategy_options(value: object) -> None:
+    """Validate that retry_strategy_options is a RetryStrategyOptions instance.
+
+    :param value: The resolved retry strategy options value.
+    :raises ConfigError: If the value is not a RetryStrategyOptions instance.
+    """
+    if not isinstance(value, RetryStrategyOptions):
+        raise ConfigError(
+            f"Invalid value for 'retry_strategy_options': {value!r}. "
+            f"Must be RetryStrategyOptions, got {type(value).__name__}."
+        )
diff --git a/packages/smithy-aws-core/tests/unit/config/test_resolver.py b/packages/smithy-aws-core/tests/unit/config/test_resolver.py
new file mode 100644
index 000000000..b1e40512e
--- /dev/null
+++ b/packages/smithy-aws-core/tests/unit/config/test_resolver.py
@@ -0,0 +1,285 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for the config resolution pipeline.
+
+Tests AsyncAwsConfig.resolve(), resolver functions, SharedConfigContext,
+provenance tracking, and precedence behavior.
+"""
+
+from pathlib import Path
+
+import pytest
+from smithy_aws_core.config.aws_config import AsyncAwsConfig
+from smithy_aws_core.config.context import SharedConfigContext
+from smithy_aws_core.config.exceptions import ConfigError
+from smithy_aws_core.config.resolvers import (
+    resolve_region,
+    resolve_retry_config,
+)
+from smithy_aws_core.config.types import ConfigSource
+
+
+class TestResolveRegion:
+    @pytest.mark.asyncio
+    async def test_resolves_from_aws_region(self):
+        ctx = SharedConfigContext(env={"AWS_REGION": "us-west-2"})
+        result = await resolve_region(ctx)
+        assert result.value == "us-west-2"
+        assert result.source == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_resolves_from_aws_default_region(self):
+        ctx = SharedConfigContext(env={"AWS_DEFAULT_REGION": "eu-central-1"})
+        result = await resolve_region(ctx)
+        assert result.value == "eu-central-1"
+        assert result.source == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_aws_region_takes_precedence_over_default_region(self):
+        ctx = SharedConfigContext(
+            env={"AWS_REGION": "us-west-2", "AWS_DEFAULT_REGION": "eu-west-1"},
+        )
+        result = await resolve_region(ctx)
+        assert result.value == "us-west-2"
+
+    @pytest.mark.asyncio
+    async def test_resolves_from_profile_when_no_env(self, tmp_path: Path):
+        config_file = tmp_path / "config"
+        config_file.write_text("[profile default]\nregion = ap-southeast-1\n")
+
+        ctx = SharedConfigContext(
+            env={},
+            config_file_path=str(config_file),
+            credentials_file_path=str(tmp_path / "none"),
+        )
+        result = await resolve_region(ctx)
+        assert result.value == "ap-southeast-1"
+        assert result.source == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_env_takes_precedence_over_profile(self, tmp_path: Path):
+        config_file = tmp_path / "config"
+        config_file.write_text("[profile default]\nregion = eu-west-1\n")
+        ctx = SharedConfigContext(
+            env={"AWS_REGION": "us-west-2"},
+            config_file_path=str(config_file),
+            credentials_file_path=str(tmp_path / "nonexistent"),
+        )
+        result = await resolve_region(ctx)
+        assert result.value == "us-west-2"
+        assert result.source == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_empty_string_env_var_treated_as_absent(self, tmp_path: Path):
+        config_file = tmp_path / "config"
+        config_file.write_text("[profile default]\nregion = eu-west-1\n")
+        ctx = SharedConfigContext(
+            env={"AWS_REGION": ""},
+            config_file_path=str(config_file),
+            credentials_file_path=str(tmp_path / "nonexistent"),
+        )
+        result = await resolve_region(ctx)
+        assert result.value == "eu-west-1"
+        assert result.source == ConfigSource.PROFILE
+
+
+class TestResolveRetryConfig:
+    @pytest.mark.asyncio
+    async def test_resolves_both_from_env(self):
+        ctx = SharedConfigContext(
+            env={"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "10"},
+        )
+        result = await resolve_retry_config(ctx)
+        assert result.value.retry_mode == "standard"
+        assert result.value.max_attempts == 10
+        assert result.source == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_uses_defaults_when_nothing_found(self):
+        ctx = SharedConfigContext(env={})
+        result = await resolve_retry_config(ctx)
+        assert result.value.retry_mode == "standard"
+        assert result.value.max_attempts is None
+        assert result.source == ConfigSource.DEFAULT
+
+    @pytest.mark.asyncio
+    async def test_partial_resolution_uses_defaults_for_missing(self):
+        ctx = SharedConfigContext(env={"AWS_RETRY_MODE": "standard"})
+        result = await resolve_retry_config(ctx)
+        assert result.value.retry_mode == "standard"
+        assert result.value.max_attempts is None  # RetryStrategyOptions default
+        assert result.source == ConfigSource.ENV  # strongest source
+
+    @pytest.mark.asyncio
+    async def test_max_attempts_cast_to_int(self):
+        ctx = SharedConfigContext(
+            env={"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "5"}
+        )
+        result = await resolve_retry_config(ctx)
+        assert result.value.max_attempts == 5
+        assert isinstance(result.value.max_attempts, int)
+
+    @pytest.mark.asyncio
+    async def test_max_attempts_unset_when_not_found(self):
+        ctx = SharedConfigContext(env={})
+        result = await resolve_retry_config(ctx)
+        assert result.value.max_attempts is None
+
+    @pytest.mark.asyncio
+    async def test_invalid_max_attempts_raises_config_error(self):
+        ctx = SharedConfigContext(env={"AWS_MAX_ATTEMPTS": "abc"})
+        with pytest.raises(ConfigError, match="Invalid integer value"):
+            await resolve_retry_config(ctx)
+
+
+class TestAsyncAwsConfigResolve:
+    @pytest.mark.asyncio
+    async def test_resolves_region_from_env(self):
+        config = await AsyncAwsConfig.resolve(env={"AWS_REGION": "us-west-2"})
+        assert config.region == "us-west-2"
+
+    @pytest.mark.asyncio
+    async def test_resolves_retry_from_env(self):
+        config = await AsyncAwsConfig.resolve(
+            env={"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "5"},
+        )
+        assert config.retry_strategy_options is not None
+        assert config.retry_strategy_options.retry_mode == "standard"
+        assert config.retry_strategy_options.max_attempts == 5
+
+    @pytest.mark.asyncio
+    async def test_explicit_override_takes_precedence(self):
+        config = await AsyncAwsConfig.resolve(
+            env={"AWS_REGION": "us-west-2"},
+            region="eu-west-1",
+        )
+        assert config.region == "eu-west-1"
+
+    @pytest.mark.asyncio
+    async def test_default_region_raises_error(self, tmp_path: Path):
+        with pytest.raises(ConfigError, match="Region is required"):
+            await AsyncAwsConfig.resolve(
+                env={},
+                config_file_path=str(tmp_path / "no_config"),
+                credentials_file_path=str(tmp_path / "no_creds"),
+            )
+
+    @pytest.mark.asyncio
+    async def test_default_retry_strategy(self, tmp_path: Path):
+        config = await AsyncAwsConfig.resolve(
+            env={"AWS_REGION": "us-east-1"},
+            config_file_path=str(tmp_path / "no_config"),
+            credentials_file_path=str(tmp_path / "no_creds"),
+        )
+        assert config.retry_strategy_options is not None
+        assert config.retry_strategy_options.retry_mode == "standard"
+        assert config.retry_strategy_options.max_attempts is None
+
+    @pytest.mark.asyncio
+    async def test_resolves_region_from_non_default_profile(self, tmp_path: Path):
+        config_file = tmp_path / "config"
+        config_file.write_text(
+            "[profile default]\nregion = us-east-1\n"
+            "[profile work]\nregion = eu-west-1\n"
+        )
+
+        config = await AsyncAwsConfig.resolve(
+            profile="work",
+            env={},
+            config_file_path=str(config_file),
+            credentials_file_path=str(tmp_path / "none"),
+        )
+        assert config.region == "eu-west-1"
+        assert config.source_of("region") == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_invalid_override_triggers_validator(self):
+        with pytest.raises(ConfigError, match="Must be a valid AWS region"):
+            await AsyncAwsConfig.resolve(
+                env={},
+                region="bad-value!",
+            )
+
+
+class TestProvenanceTracking:
+    @pytest.mark.asyncio
+    async def test_source_of_env_resolved_field(self):
+        config = await AsyncAwsConfig.resolve(env={"AWS_REGION": "us-west-2"})
+        assert config.source_of("region") == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_source_of_profile_resolved_field(self, tmp_path: Path):
+        config_file = tmp_path / "config"
+        config_file.write_text("[profile default]\nregion = eu-west-1\n")
+
+        config = await AsyncAwsConfig.resolve(
+            env={},
+            config_file_path=str(config_file),
+            credentials_file_path=str(tmp_path / "none"),
+        )
+        assert config.source_of("region") == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_source_of_default_field(self, tmp_path: Path):
+        config = await AsyncAwsConfig.resolve(
+            env={"AWS_REGION": "us-east-1"},
+            config_file_path=str(tmp_path / "no_config"),
+            credentials_file_path=str(tmp_path / "no_creds"),
+        )
+        assert config.source_of("retry_strategy_options") == ConfigSource.DEFAULT
+
+    @pytest.mark.asyncio
+    async def test_source_of_override_field(self):
+        config = await AsyncAwsConfig.resolve(env={}, region="us-east-1")
+        assert config.source_of("region") == ConfigSource.OVERRIDE
+
+    @pytest.mark.asyncio
+    async def test_post_resolution_assignment_marks_source_as_override(self):
+        config = await AsyncAwsConfig.resolve(env={"AWS_REGION": "us-west-2"})
+        assert config.source_of("region") == ConfigSource.ENV
+        config.region = "eu-west-1"
+        assert config.source_of("region") == ConfigSource.OVERRIDE
+        assert config.region == "eu-west-1"
+
+
+class TestConstructionBlocking:
+    def test_direct_instantiation_raises_error(self):
+        with pytest.raises(ConfigError, match="cannot be constructed directly"):
+            AsyncAwsConfig()
+
+    @pytest.mark.asyncio
+    async def test_unknown_override_field_raises_error(self):
+        with pytest.raises(ConfigError, match="Unknown config field"):
+            await AsyncAwsConfig.resolve(
+                env={"AWS_REGION": "us-east-1"},
+                reigon="us-west-2",
+            )
+
+
+class TestSharedConfigContext:
+    def test_default_profile_is_default(self):
+        ctx = SharedConfigContext(env={})
+        assert ctx.profile_name == "default"
+
+    def test_profile_from_aws_profile_env(self):
+        ctx = SharedConfigContext(env={"AWS_PROFILE": "work"})
+        assert ctx.profile_name == "work"
+
+    def test_explicit_profile_overrides_env(self):
+        ctx = SharedConfigContext(profile_name="custom", env={"AWS_PROFILE": "work"})
+        assert ctx.profile_name == "custom"
+
+    @pytest.mark.asyncio
+    async def test_parsed_profiles_caches_result(self, tmp_path: Path):
+        config_file = tmp_path / "config"
+        config_file.write_text("[profile default]\nregion = us-east-1\n")
+
+        ctx = SharedConfigContext(
+            env={},
+            config_file_path=str(config_file),
+            credentials_file_path=str(tmp_path / "none"),
+        )
+
+        result1 = await ctx.parsed_profiles()
+        result2 = await ctx.parsed_profiles()
+        assert result1 is result2
diff --git a/packages/smithy-aws-core/tests/unit/config/test_validators.py b/packages/smithy-aws-core/tests/unit/config/test_validators.py
new file mode 100644
index 000000000..6a4016464
--- /dev/null
+++ b/packages/smithy-aws-core/tests/unit/config/test_validators.py
@@ -0,0 +1,93 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for config field validators."""
+
+import pytest
+from smithy_aws_core.config.exceptions import ConfigError
+from smithy_aws_core.config.types import FieldSpec
+from smithy_aws_core.config.validators import (
+    validate_region,
+    validate_retry_strategy_options,
+)
+from smithy_core.retries import RetryStrategyOptions
+
+
+class TestValidateRegion:
+    @pytest.mark.parametrize(
+        "region",
+        [
+            "us-east-1",
+            "ap-southeast-2",
+            "eu-west-1",
+            "us",
+            "us-gov-west-1",
+        ],
+    )
+    def test_valid_regions(self, region: str):
+        validate_region(region)
+
+    def test_none_raises(self):
+        with pytest.raises(ConfigError, match="Region is required"):
+            validate_region(None)
+
+    @pytest.mark.parametrize(
+        "region,reason",
+        [
+            ("", "empty string"),
+            ("-us-east-1", "starts with dash"),
+            ("us-east-1-", "ends with dash"),
+            ("12345", "all numbers"),
+            ("us-east-1!", "special characters"),
+            ("us east 1", "spaces"),
+        ],
+        ids=lambda x: x if isinstance(x, str) else "",
+    )
+    def test_invalid_regions(self, region: str, reason: str):
+        with pytest.raises(ConfigError, match="Must be a valid AWS region"):
+            validate_region(region)
+
+    def test_non_string_raises(self):
+        with pytest.raises(ConfigError, match="Must be a valid AWS region"):
+            validate_region(123)
+
+
+class TestValidateRetryStrategyOptions:
+    def test_valid_default_options(self):
+        validate_retry_strategy_options(RetryStrategyOptions())
+
+    def test_valid_options_with_values(self):
+        validate_retry_strategy_options(
+            RetryStrategyOptions(retry_mode="standard", max_attempts=5)
+        )
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            None,
+            "standard",
+            {"retry_mode": "standard"},
+            3,
+        ],
+        ids=["none", "string", "dict", "int"],
+    )
+    def test_invalid_types_raise(self, value: object):
+        with pytest.raises(ConfigError, match="Must be RetryStrategyOptions"):
+            validate_retry_strategy_options(value)
+
+
+class TestFieldSpec:
+    """Tests for FieldSpec validation constraints."""
+
+    def test_default_only_is_valid(self):
+        FieldSpec(default=None)
+
+    def test_default_factory_only_is_valid(self):
+        FieldSpec(default_factory=list)
+
+    def test_both_default_and_factory_raises(self):
+        with pytest.raises(ValueError, match="cannot set both"):
+            FieldSpec(default=None, default_factory=list)
+
+    def test_neither_default_nor_factory_raises(self):
+        with pytest.raises(ValueError, match="exactly one of"):
+            FieldSpec()

From c5d88402d91ef97ca823f12811239468bdb05d53 Mon Sep 17 00:00:00 2001
From: ubaskota <19787410+ubaskota@users.noreply.github.com>
Date: Thu, 23 Jul 2026 19:21:58 -0400
Subject: [PATCH 2/2] Add async resolver for AWS config values

---
 .../src/smithy_aws_core/config/__init__.py    |  74 +---
 .../src/smithy_aws_core/config/aws_config.py  |  19 +-
 .../src/smithy_aws_core/config/context.py     | 114 +++--
 .../src/smithy_aws_core/config/exceptions.py  |   8 -
 .../src/smithy_aws_core/config/file_parser.py |  30 +-
 .../src/smithy_aws_core/config/filesystem.py  |  50 +++
 .../src/smithy_aws_core/config/resolvers.py   |  23 +-
 .../src/smithy_aws_core/config/types.py       |   4 +-
 .../src/smithy_aws_core/config/validators.py  |  58 ++-
 .../tests/unit/config/test_config_file_io.py  |   8 +-
 .../unit/config/test_config_file_location.py  |   2 +-
 .../unit/config/test_config_file_parser.py    |  20 +-
 .../tests/unit/config/test_resolver.py        | 407 ++++++++++--------
 .../tests/unit/config/test_validators.py      |  34 +-
 .../smithy-core/src/smithy_core/exceptions.py |  12 +
 15 files changed, 497 insertions(+), 366 deletions(-)
 delete mode 100644 packages/smithy-aws-core/src/smithy_aws_core/config/exceptions.py
 create mode 100644 packages/smithy-aws-core/src/smithy_aws_core/config/filesystem.py

diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/__init__.py b/packages/smithy-aws-core/src/smithy_aws_core/config/__init__.py
index bbb4bc282..8b73473a1 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/__init__.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/__init__.py
@@ -1,76 +1,8 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-import os
-from pathlib import Path
-
-from smithy_aws_core.config.file_parser import (
-    FileType,
-    parse_config_file,
-    standardize,
-)
+from smithy_aws_core.config.context import load_config
+from smithy_aws_core.config.filesystem import DefaultFileSystem, FileSystem
 from smithy_aws_core.config.merged_config import MergedConfig
 
-_DEFAULT_CONFIG_FILE = "~/.aws/config"
-_DEFAULT_CREDENTIALS_FILE = "~/.aws/credentials"
-_CONFIG_FILE_ENV_VAR = "AWS_CONFIG_FILE"
-_CREDENTIALS_FILE_ENV_VAR = "AWS_SHARED_CREDENTIALS_FILE"
-
-
-def _resolve_config_paths(
-    config_file_path: Path | None = None,
-    credentials_file_path: Path | None = None,
-) -> tuple[Path, Path]:
-    """Resolve the final config and credentials file paths.
-
-    Resolution order for each path:
-    1. Explicit argument (if provided)
-    2. Environment variable (AWS_CONFIG_FILE / AWS_SHARED_CREDENTIALS_FILE)
-    3. Default (~/.aws/config / ~/.aws/credentials)
-
-    The ~ is expanded to the user's home directory.
-
-    :param config_file_path: Override path for config file.
-    :param credentials_file_path: Override path for credentials file.
-    :returns: Tuple of (resolved_config_path, resolved_credentials_path).
-    """
-    config_path = (
-        config_file_path
-        or Path(os.environ.get(_CONFIG_FILE_ENV_VAR, _DEFAULT_CONFIG_FILE))
-    ).expanduser()
-
-    credentials_path = (
-        credentials_file_path
-        or Path(os.environ.get(_CREDENTIALS_FILE_ENV_VAR, _DEFAULT_CREDENTIALS_FILE))
-    ).expanduser()
-
-    return config_path, credentials_path
-
-
-async def load_config(
-    config_file_path: Path | None = None,
-    credentials_file_path: Path | None = None,
-) -> MergedConfig:
-    """Load and merge AWS config and credentials files.
-
-    Parses both files, standardizes them, and returns a merged
-    MergedConfig ready for querying.
-
-    :param config_file_path: Override path for config file.
-        Defaults to AWS_CONFIG_FILE env var or ~/.aws/config.
-    :param credentials_file_path: Override path for credentials file.
-        Defaults to AWS_SHARED_CREDENTIALS_FILE env var or ~/.aws/credentials.
-    :returns: A MergedConfig with merged profiles from both files.
-    """
-
-    config_path, credentials_path = _resolve_config_paths(
-        config_file_path, credentials_file_path
-    )
-
-    raw_config = await parse_config_file(str(config_path))
-    raw_credentials = await parse_config_file(str(credentials_path))
-
-    std_config = standardize(raw_config, FileType.CONFIG)
-    std_credentials = standardize(raw_credentials, FileType.CREDENTIALS)
-
-    return MergedConfig(std_config, std_credentials)
+__all__ = ["DefaultFileSystem", "FileSystem", "MergedConfig", "load_config"]
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py b/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
index 5fbcf601b..43fbb9b7c 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
@@ -1,19 +1,20 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import Any, ClassVar, Self
 
-from smithy_aws_core.config.context import FileSystem, SharedConfigContext
-from smithy_aws_core.config.exceptions import ConfigError
+from smithy_core.exceptions import ConfigError, ConfigValidationError
+from smithy_core.retries import RetryStrategyOptions
+
+from smithy_aws_core.config.context import SharedConfigContext
+from smithy_aws_core.config.filesystem import FileSystem
 from smithy_aws_core.config.resolvers import resolve_region, resolve_retry_config
 from smithy_aws_core.config.types import UNSET, ConfigSource, FieldSpec, Resolved
 from smithy_aws_core.config.validators import (
     validate_region,
     validate_retry_strategy_options,
 )
-from smithy_core.retries import RetryStrategyOptions
 
 
 @dataclass(kw_only=True)
@@ -62,7 +63,6 @@ async def resolve(
         cls,
         *,
         profile: str | None = None,
-        env: Mapping[str, str] | None = None,
         fs: FileSystem | None = None,
         config_file_path: str | None = None,
         credentials_file_path: str | None = None,
@@ -73,7 +73,6 @@ async def resolve(
         This is the only supported way to create a config instance.
 
         :param profile: Override the active profile name.
-        :param env: Override the environment variable mapping.
         :param fs: Override the filesystem abstraction.
         :param config_file_path: Override path for config file.
         :param credentials_file_path: Override path for credentials file.
@@ -82,7 +81,6 @@ async def resolve(
         """
         ctx = SharedConfigContext(
             profile_name=profile,
-            env=env,
             fs=fs,
             config_file_path=config_file_path,
             credentials_file_path=credentials_file_path,
@@ -126,7 +124,7 @@ async def _resolve_fields(self, overrides: dict[str, Any]) -> None:
         """Run the resolution pipeline for all fields."""
         unknown = set(overrides) - set(self._FIELDS)
         if unknown:
-            raise ConfigError(
+            raise ConfigValidationError(
                 f"Unknown config field(s): {sorted(unknown)}. "
                 f"Valid fields are: {sorted(self._FIELDS)}"
             )
@@ -166,11 +164,14 @@ def _apply_default(self, field_name: str, spec: FieldSpec) -> None:
 
     def __setattr__(self, name: str, value: Any) -> None:
         """Track provenance when fields are set with plugins after construction"""
-        super().__setattr__(name, value)
         # Mark as override only if the field is in _FIELDS and was already resolved
         if (
             name in self.__class__._FIELDS
             and hasattr(self, "_sources")
             and name in self._sources
         ):
+            spec = self.__class__._FIELDS[name]
+            if spec.validator is not None:
+                spec.validator(value)
             self._sources[name] = ConfigSource.OVERRIDE
+        super().__setattr__(name, value)
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/context.py b/packages/smithy-aws-core/src/smithy_aws_core/config/context.py
index fac549038..7496cfa6c 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/context.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/context.py
@@ -1,58 +1,87 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-import asyncio
 import logging
 import os
-from collections.abc import Mapping
 from pathlib import Path
-from typing import Any, Protocol, runtime_checkable
-
-from smithy_aws_core.config import load_config
+from typing import Any
+
+from smithy_aws_core.config.file_parser import (
+    FileType,
+    parse_config_file,
+    standardize,
+)
+from smithy_aws_core.config.filesystem import DefaultFileSystem, FileSystem
 from smithy_aws_core.config.merged_config import MergedConfig
 
 logger = logging.getLogger(__name__)
 
+_DEFAULT_CONFIG_FILE = "~/.aws/config"
+_DEFAULT_CREDENTIALS_FILE = "~/.aws/credentials"
+_CONFIG_FILE_ENV_VAR = "AWS_CONFIG_FILE"
+_CREDENTIALS_FILE_ENV_VAR = "AWS_SHARED_CREDENTIALS_FILE"
 
-@runtime_checkable
-class FileSystem(Protocol):
-    """Protocol for filesystem operations.
 
-    Abstraction over file I/O so tests can provide mock implementations
-    without touching the real filesystem.
-    """
+def _resolve_config_paths(
+    config_file_path: Path | None = None,
+    credentials_file_path: Path | None = None,
+) -> tuple[Path, Path]:
+    """Resolve the final config and credentials file paths.
 
-    async def read_file(self, path: str) -> str | None:
-        """Read a file's content as UTF-8.
+    Resolution order for each path:
+    1. Explicit argument (if provided)
+    2. Environment variable (AWS_CONFIG_FILE / AWS_SHARED_CREDENTIALS_FILE)
+    3. Default (~/.aws/config / ~/.aws/credentials)
 
-        :param path: Resolved file path.
-        :returns: File content, or None if the file is inaccessible.
-        """
-        ...
+    The ~ is expanded to the user's home directory.
 
+    :param config_file_path: Override path for config file.
+    :param credentials_file_path: Override path for credentials file.
+    :returns: Tuple of (resolved_config_path, resolved_credentials_path).
+    """
+    config_path = (
+        config_file_path
+        or Path(os.environ.get(_CONFIG_FILE_ENV_VAR, _DEFAULT_CONFIG_FILE))
+    ).expanduser()
+
+    credentials_path = (
+        credentials_file_path
+        or Path(os.environ.get(_CREDENTIALS_FILE_ENV_VAR, _DEFAULT_CREDENTIALS_FILE))
+    ).expanduser()
+
+    return config_path, credentials_path
+
+
+async def load_config(
+    config_file_path: Path | None = None,
+    credentials_file_path: Path | None = None,
+    fs: FileSystem | None = None,
+) -> MergedConfig:
+    """Load and merge AWS config and credentials files.
+
+    Parses both files, standardizes them, and returns a merged
+    MergedConfig ready for querying.
+
+    :param config_file_path: Override path for config file.
+        Defaults to AWS_CONFIG_FILE env var or ~/.aws/config.
+    :param credentials_file_path: Override path for credentials file.
+        Defaults to AWS_SHARED_CREDENTIALS_FILE env var or ~/.aws/credentials.
+    :param fs: FileSystem to use for reading files.
+        Defaults to DefaultFileSystem (real disk I/O).
+    :returns: A MergedConfig with merged profiles from both files.
+    """
+    filesystem = fs or DefaultFileSystem()
+    config_path, credentials_path = _resolve_config_paths(
+        config_file_path, credentials_file_path
+    )
 
-class DefaultFileSystem:
-    """Default filesystem implementation using real disk I/O."""
-
-    async def read_file(self, path: str) -> str | None:
-        """Read a file asynchronously from disk.
+    raw_config = await parse_config_file(str(config_path), filesystem)
+    raw_credentials = await parse_config_file(str(credentials_path), filesystem)
 
-        Missing files and permission errors return None with a warning.
-        Encoding errors (invalid UTF-8) are raised to the caller.
+    std_config = standardize(raw_config, FileType.CONFIG)
+    std_credentials = standardize(raw_credentials, FileType.CREDENTIALS)
 
-        :param path: Resolved file path.
-        :returns: File content, or None if the file is inaccessible.
-        """
-        try:
-            content: str = await asyncio.to_thread(
-                Path(path).read_text, encoding="utf-8"
-            )
-            return content
-        except FileNotFoundError:
-            return None
-        except (PermissionError, OSError) as e:
-            logger.warning("Unable to read config file '%s': %s", path, e)
-            return None
+    return MergedConfig(std_config, std_credentials)
 
 
 class SharedConfigContext:
@@ -66,7 +95,6 @@ def __init__(
         self,
         *,
         profile_name: str | None = None,
-        env: Mapping[str, str] | None = None,
         fs: FileSystem | None = None,
         http_client: Any | None = None,
         config_file_path: str | Path | None = None,
@@ -76,7 +104,6 @@ def __init__(
 
         :param profile_name: The active profile to use. Defaults to
             AWS_PROFILE env var, then "default".
-        :param env: Environment variable mapping. Defaults to os.environ.
         :param fs: Filesystem abstraction for reading config files.
             Defaults to real disk I/O.
         :param http_client: HTTP client for network-based resolvers
@@ -84,7 +111,6 @@ def __init__(
         :param config_file_path: Override path for config file.
         :param credentials_file_path: Override path for credentials file.
         """
-        self._env: Mapping[str, str] = env if env is not None else os.environ
         self._fs: FileSystem = fs if fs is not None else DefaultFileSystem()
         self._http_client: Any | None = http_client
         self._profile_name: str = self._resolve_profile_name(profile_name)
@@ -96,11 +122,6 @@ def __init__(
         )
         self._cached_config_file: MergedConfig | None = None
 
-    @property
-    def env(self) -> Mapping[str, str]:
-        """The environment variable mapping."""
-        return self._env
-
     @property
     def profile_name(self) -> str:
         """The active profile name."""
@@ -126,6 +147,7 @@ async def parsed_profiles(self) -> MergedConfig:
             self._cached_config_file = await load_config(
                 config_file_path=self._config_file_path,
                 credentials_file_path=self._credentials_file_path,
+                fs=self._fs,
             )
         return self._cached_config_file
 
@@ -136,4 +158,4 @@ def _resolve_profile_name(self, explicit_profile: str | None) -> str:
         """
         if explicit_profile is not None:
             return explicit_profile
-        return self._env.get("AWS_PROFILE", "default")
+        return os.environ.get("AWS_PROFILE", "default")
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/exceptions.py b/packages/smithy-aws-core/src/smithy_aws_core/config/exceptions.py
deleted file mode 100644
index 3ad3ceb97..000000000
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/exceptions.py
+++ /dev/null
@@ -1,8 +0,0 @@
-# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
-# SPDX-License-Identifier: Apache-2.0
-
-"""Config related exceptions"""
-
-
-class ConfigParseError(Exception):
-    """Raised when a config file cannot be parsed due to invalid syntax."""
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/file_parser.py b/packages/smithy-aws-core/src/smithy_aws_core/config/file_parser.py
index 09d8b1aa2..8182ea6e1 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/file_parser.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/file_parser.py
@@ -1,14 +1,14 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-import asyncio
 import logging
 import re
 from dataclasses import dataclass, field
 from enum import Enum
-from pathlib import Path
 
-from smithy_aws_core.config.exceptions import ConfigParseError
+from smithy_core.exceptions import ConfigParseError
+
+from smithy_aws_core.config.filesystem import FileSystem
 
 logger = logging.getLogger(__name__)
 
@@ -43,16 +43,19 @@ class StandardizedOutput:
     services: SectionMap = field(default_factory=dict)  # type: ignore[assignment]
 
 
-async def parse_config_file(file_path: str) -> RawParsedSections:
+async def parse_config_file(
+    file_path: str, filesystem: FileSystem
+) -> RawParsedSections:
     """Parse an AWS config or credentials file.
 
     Reads the file asynchronously and parses it into raw sections.
 
     :param file_path: Resolved path to the file.
+    :param filesystem: FileSystem object to use.
     :returns: Raw sections dict {section_name: {key: value}}.
     :raises ConfigParseError: If the file has invalid syntax.
     """
-    content = await _read_file(file_path)
+    content = await filesystem.read_file(file_path)
     if content is None:
         return {}
     try:
@@ -111,23 +114,6 @@ def standardize(
     )
 
 
-async def _read_file(path: str) -> str | None:
-    """Read file content asynchronously.
-
-    Returns None if the file doesn't exist or can't be read due to
-    permission/OS errors. Raises on encoding errors since they likely
-    indicate a misconfigured file the user should know about.
-    """
-    try:
-        content = await asyncio.to_thread(Path(path).read_text, encoding="utf-8")
-        return content
-    except FileNotFoundError:
-        return None
-    except (PermissionError, OSError) as e:
-        logger.warning("Unable to read config file '%s': %s", path, e)
-        return None
-
-
 def _parse_content(content: str) -> RawParsedSections:
     """Parse config file content from a string into raw sections.
 
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/filesystem.py b/packages/smithy-aws-core/src/smithy_aws_core/config/filesystem.py
new file mode 100644
index 000000000..831511191
--- /dev/null
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/filesystem.py
@@ -0,0 +1,50 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+import logging
+from pathlib import Path
+from typing import Protocol, runtime_checkable
+
+logger = logging.getLogger(__name__)
+
+
+@runtime_checkable
+class FileSystem(Protocol):
+    """Protocol for filesystem operations.
+
+    Abstraction over file I/O so tests can provide mock implementations
+    without touching the real filesystem.
+    """
+
+    async def read_file(self, path: str) -> str | None:
+        """Read a file's content as UTF-8.
+
+        :param path: Resolved file path.
+        :returns: File content, or None if the file is inaccessible.
+        """
+        ...
+
+
+class DefaultFileSystem:
+    """Default filesystem implementation using real disk I/O."""
+
+    async def read_file(self, path: str) -> str | None:
+        """Read a file asynchronously from disk.
+
+        Missing files and permission errors return None with a warning.
+        Encoding errors (invalid UTF-8) are raised to the caller.
+
+        :param path: Resolved file path.
+        :returns: File content, or None if the file is inaccessible.
+        """
+        try:
+            content: str = await asyncio.to_thread(
+                Path(path).read_text, encoding="utf-8"
+            )
+            return content
+        except FileNotFoundError:
+            return None
+        except (PermissionError, OSError) as e:
+            logger.warning("Unable to read config file '%s': %s", path, e)
+            return None
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py b/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
index 388390d03..f3b21d01b 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
@@ -1,13 +1,16 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+import os
 from collections.abc import Sequence
 from typing import Any
 
+from smithy_core.exceptions import ConfigValidationError
+from smithy_core.retries import RetryStrategyOptions
+
 from smithy_aws_core.config.context import SharedConfigContext
-from smithy_aws_core.config.exceptions import ConfigError
 from smithy_aws_core.config.types import UNSET, ConfigSource, Resolved
-from smithy_core.retries import RetryStrategyOptions
+from smithy_aws_core.config.validators import validate_max_attempts, validate_retry_mode
 
 
 async def _resolve_str(
@@ -15,7 +18,7 @@ async def _resolve_str(
     *,
     env_vars: Sequence[str] = (),
     profile_keys: Sequence[str] = (),
-) -> Resolved[str | None]:
+) -> Resolved[str]:
     """Resolve a string value by checking providers in priority order.
 
     Priority: env vars (first match) > config file (profile keys, first match) > unresolved.
@@ -27,7 +30,7 @@ async def _resolve_str(
     """
     # Check environment variables first
     for var_name in env_vars:
-        value: str | None = ctx.env.get(var_name)
+        value: str | None = os.environ.get(var_name)
         if value:
             return Resolved(value=value, source=ConfigSource.ENV)
 
@@ -58,15 +61,13 @@ async def _resolve_int(
     result = await _resolve_str(ctx, env_vars=env_vars, profile_keys=profile_keys)
     if result.value is UNSET:
         return Resolved(value=UNSET, source=ConfigSource.DEFAULT)  # type: ignore[arg-type]
-    if result.value is None:
-        return Resolved(value=None, source=result.source)
     try:
         return Resolved(value=int(result.value), source=result.source)
-    except (ValueError, TypeError):
-        raise ConfigError(
+    except (ValueError, TypeError) as e:
+        raise ConfigValidationError(
             f"Invalid integer value {result.value!r} for config key. "
             "Expected a valid integer."
-        )
+        ) from e
 
 
 def _strongest_source(*sources: ConfigSource) -> ConfigSource:
@@ -117,11 +118,13 @@ async def resolve_retry_config(
     kwargs: dict[str, Any] = {}
     sources_found: list[ConfigSource] = []
 
-    if mode_result.value is not UNSET and mode_result.value is not None:
+    if mode_result.value is not UNSET:
+        validate_retry_mode(mode_result.value)
         kwargs["retry_mode"] = mode_result.value
         sources_found.append(mode_result.source)
 
     if attempts_result.value is not UNSET and attempts_result.value is not None:
+        validate_max_attempts(attempts_result.value)
         kwargs["max_attempts"] = attempts_result.value
         sources_found.append(attempts_result.source)
 
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/types.py b/packages/smithy-aws-core/src/smithy_aws_core/config/types.py
index 6ff732169..f0d3fe255 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/types.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/types.py
@@ -4,7 +4,7 @@
 
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from enum import Enum
+from enum import StrEnum
 from typing import Any
 
 
@@ -19,7 +19,7 @@ def __bool__(self) -> bool:
 UNSET = _UnsetType()
 
 
-class ConfigSource(str, Enum):
+class ConfigSource(StrEnum):
     """Where a resolved config value came from."""
 
     ENV = "env"
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/validators.py b/packages/smithy-aws-core/src/smithy_aws_core/config/validators.py
index a78f5f049..64426e979 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/validators.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/validators.py
@@ -2,9 +2,10 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import re
+from typing import get_args
 
-from smithy_aws_core.config.exceptions import ConfigError
-from smithy_core.retries import RetryStrategyOptions
+from smithy_core.exceptions import ConfigValidationError
+from smithy_core.retries import RetryStrategyOptions, RetryStrategyType
 
 _REGION_PATTERN = re.compile(r"^(?![0-9]+$)(?!-)[a-zA-Z0-9-]{1,63}(?<!-)$")
 
@@ -16,14 +17,14 @@ def validate_region(value: object) -> None:
     or override.
 
     :param value: The resolved region value.
-    :raises ConfigError: If the value is None or doesn't match the pattern.
+    :raises ConfigValidationError: If the value is None or doesn't match the pattern.
     """
     if value is None:
-        raise ConfigError(
+        raise ConfigValidationError(
             "Invalid value for 'region': None. Region is required and must be set."
         )
     if not isinstance(value, str) or not _REGION_PATTERN.match(value):
-        raise ConfigError(
+        raise ConfigValidationError(
             f"Invalid value for 'region': {value!r}. "
             "Must be a valid AWS region identifier."
         )
@@ -33,10 +34,53 @@ def validate_retry_strategy_options(value: object) -> None:
     """Validate that retry_strategy_options is a RetryStrategyOptions instance.
 
     :param value: The resolved retry strategy options value.
-    :raises ConfigError: If the value is not a RetryStrategyOptions instance.
+    :raises ConfigValidationError: If the value is not a RetryStrategyOptions instance.
     """
     if not isinstance(value, RetryStrategyOptions):
-        raise ConfigError(
+        raise ConfigValidationError(
             f"Invalid value for 'retry_strategy_options': {value!r}. "
             f"Must be RetryStrategyOptions, got {type(value).__name__}."
         )
+
+
+def validate_retry_mode(retry_mode: str):
+    """Validate retry mode.
+
+    Valid values: 'standard'
+
+    :param retry_mode: The retry mode value to validate
+    :raises: ConfigValidationError: If the retry mode is invalid
+    """
+
+    all_modes = list(get_args(RetryStrategyType))
+    if "simple" in all_modes:
+        all_modes.remove("simple")
+    valid_modes = tuple(all_modes)
+
+    if retry_mode not in valid_modes:
+        raise ConfigValidationError(
+            f"Invalid value for 'retry_mode': {retry_mode!r}. "
+            f"Must be one of {valid_modes}."
+        )
+
+
+def validate_max_attempts(
+    max_attempts: str | int,
+):
+    """Validate max_attempts
+
+    :param max_attempts: The max attempts value (string or int)
+
+    :raises ConfigValidationError: If the value is less than 1 or cannot be converted to an integer
+    """
+    try:
+        max_attempts = int(max_attempts)
+    except (ValueError, TypeError):
+        raise ConfigValidationError(
+            f"max_attempts must be a number, got {type(max_attempts).__name__}",
+        )
+
+    if max_attempts < 1:
+        raise ConfigValidationError(
+            f"max_attempts must be a positive integer, got {max_attempts}",
+        )
diff --git a/packages/smithy-aws-core/tests/unit/config/test_config_file_io.py b/packages/smithy-aws-core/tests/unit/config/test_config_file_io.py
index a8db37c2c..0b24c8327 100644
--- a/packages/smithy-aws-core/tests/unit/config/test_config_file_io.py
+++ b/packages/smithy-aws-core/tests/unit/config/test_config_file_io.py
@@ -6,7 +6,7 @@
 from unittest.mock import patch
 
 import pytest
-from smithy_aws_core.config import load_config
+from smithy_aws_core.config import DefaultFileSystem, load_config
 from smithy_aws_core.config.file_parser import parse_config_file
 
 
@@ -25,6 +25,7 @@ async def test_load_config_with_missing_files(self, tmp_path: Path):
 
     @pytest.mark.asyncio
     async def test_permission_denied_returns_empty(self, tmp_path: Path):
+        fs = DefaultFileSystem()
         restricted_file = tmp_path / "restricted_config"
         restricted_file.write_text("[profile default]\nregion = us-east-1\n")
         # Patch read_text to raise PermissionError rather than relying on
@@ -32,7 +33,7 @@ async def test_permission_denied_returns_empty(self, tmp_path: Path):
         with patch.object(
             Path, "read_text", side_effect=PermissionError("Permission denied")
         ):
-            result = await parse_config_file(str(restricted_file))
+            result = await parse_config_file(str(restricted_file), fs)
         assert result == {}
 
 
@@ -42,10 +43,11 @@ class TestEncodingErrors:
     @pytest.mark.asyncio
     async def test_bad_unicode_raises_error(self, tmp_path: Path):
         """A file with invalid UTF-8 bytes should raise UnicodeDecodeError."""
+        fs = DefaultFileSystem()
         bad_file = tmp_path / "bad_config"
         bad_file.write_bytes(b"[default]\nregion = \xff\xfe invalid")
         with pytest.raises(UnicodeDecodeError):
-            await parse_config_file(str(bad_file))
+            await parse_config_file(str(bad_file), fs)
 
 
 class TestMultiFileMerge:
diff --git a/packages/smithy-aws-core/tests/unit/config/test_config_file_location.py b/packages/smithy-aws-core/tests/unit/config/test_config_file_location.py
index 49956e9a7..68fe8d8d3 100644
--- a/packages/smithy-aws-core/tests/unit/config/test_config_file_location.py
+++ b/packages/smithy-aws-core/tests/unit/config/test_config_file_location.py
@@ -8,7 +8,7 @@
 from unittest.mock import patch
 
 import pytest
-from smithy_aws_core.config import (
+from smithy_aws_core.config.context import (
     _resolve_config_paths,  # type: ignore[reportPrivateUsage]
 )
 
diff --git a/packages/smithy-aws-core/tests/unit/config/test_config_file_parser.py b/packages/smithy-aws-core/tests/unit/config/test_config_file_parser.py
index b370bc741..239e69716 100644
--- a/packages/smithy-aws-core/tests/unit/config/test_config_file_parser.py
+++ b/packages/smithy-aws-core/tests/unit/config/test_config_file_parser.py
@@ -6,7 +6,7 @@
 from typing import cast
 
 import pytest
-from smithy_aws_core.config.exceptions import ConfigParseError
+from smithy_aws_core.config import DefaultFileSystem
 from smithy_aws_core.config.file_parser import (
     FileType,
     RawParsedSections,
@@ -14,6 +14,7 @@
     standardize,
 )
 from smithy_aws_core.config.merged_config import MergedConfig
+from smithy_core.exceptions import ConfigParseError
 
 _PARSER_TESTS_FILE = (
     Path(__file__).parent / "test-data" / "config-file-parser-tests.json"
@@ -30,10 +31,11 @@ async def _run_parse_and_standardize(
 ) -> dict[str, object]:
     """Write content to temp files, parse through public API, return merged output."""
     # Write config content to temp file
+    fs = DefaultFileSystem()
     if config_content is not None:
         config_path = tmp_path / "config"
         config_path.write_text(config_content, encoding="utf-8")
-        raw_config = await parse_config_file(str(config_path))
+        raw_config = await parse_config_file(str(config_path), fs)
     else:
         raw_config = {}
 
@@ -41,7 +43,7 @@ async def _run_parse_and_standardize(
     if credentials_content is not None:
         credentials_path = tmp_path / "credentials"
         credentials_path.write_text(credentials_content, encoding="utf-8")
-        raw_credentials = await parse_config_file(str(credentials_path))
+        raw_credentials = await parse_config_file(str(credentials_path), fs)
     else:
         raw_credentials = {}
 
@@ -144,22 +146,24 @@ def test_credentials_section_names(section_name: str, should_be_valid: bool):
 
 @pytest.mark.asyncio
 async def test_parse_error_includes_file_path(tmp_path: Path):
+    fs = DefaultFileSystem()
     bad_file = tmp_path / "config"
     bad_file.write_text("[profile p\n")  # unclosed section header
 
     with pytest.raises(ConfigParseError, match=str(bad_file)):
-        await parse_config_file(str(bad_file))
+        await parse_config_file(str(bad_file), fs)
 
 
 @pytest.mark.asyncio
 async def test_invalid_property_continuation_not_appended_to_previous(tmp_path: Path):
     """Continuation lines after an invalid property should be discarded,
     not appended to the previous valid property."""
+    fs = DefaultFileSystem()
     config_file = tmp_path / "config"
     config_file.write_text(
         "[profile p]\nregion = us-east-1\ninvalid key = ignored\n  continuation\n"
     )
-    raw = await parse_config_file(str(config_file))
+    raw = await parse_config_file(str(config_file), fs)
     result = standardize(raw, FileType.CONFIG)
     assert result.profiles["p"].properties["region"] == "us-east-1"
 
@@ -167,12 +171,13 @@ async def test_invalid_property_continuation_not_appended_to_previous(tmp_path:
 @pytest.mark.asyncio
 async def test_invalid_property_in_first_line_with_continuation_ignored(tmp_path: Path):
     """Continuation lines when the first property is invalid should be discarded"""
+    fs = DefaultFileSystem()
     config_file = tmp_path / "config"
     config_file.write_text(
         "[profile p]\ninvalid key = ignored\n  continuation\n"
         "region = us-east-1\ninvalid key = ignored\n  continuation\n"
     )
-    raw = await parse_config_file(str(config_file))
+    raw = await parse_config_file(str(config_file), fs)
     result = standardize(raw, FileType.CONFIG)
     assert result.profiles["p"].properties["region"] == "us-east-1"
 
@@ -180,13 +185,14 @@ async def test_invalid_property_in_first_line_with_continuation_ignored(tmp_path
 @pytest.mark.asyncio
 async def test_consecutive_invalid_properties_ignored(tmp_path: Path):
     """Consecutive invalid properties should be discarded"""
+    fs = DefaultFileSystem()
     config_file = tmp_path / "config"
     config_file.write_text(
         "[profile p]\nregion = us-east-1\ninvalid key = ignored\n  continuation\n"
         "invalid key = ignored\n  continuation\ninvalid key = ignored\n  continuation\n"
         "output = json\n"
     )
-    raw = await parse_config_file(str(config_file))
+    raw = await parse_config_file(str(config_file), filesystem=fs)
     result = standardize(raw, FileType.CONFIG)
     assert result.profiles["p"].properties["region"] == "us-east-1"
     assert result.profiles["p"].properties["output"] == "json"
diff --git a/packages/smithy-aws-core/tests/unit/config/test_resolver.py b/packages/smithy-aws-core/tests/unit/config/test_resolver.py
index b1e40512e..0aed9d7c2 100644
--- a/packages/smithy-aws-core/tests/unit/config/test_resolver.py
+++ b/packages/smithy-aws-core/tests/unit/config/test_resolver.py
@@ -6,240 +6,274 @@
 provenance tracking, and precedence behavior.
 """
 
-from pathlib import Path
+import os
+from unittest.mock import patch
 
 import pytest
 from smithy_aws_core.config.aws_config import AsyncAwsConfig
 from smithy_aws_core.config.context import SharedConfigContext
-from smithy_aws_core.config.exceptions import ConfigError
 from smithy_aws_core.config.resolvers import (
     resolve_region,
     resolve_retry_config,
 )
 from smithy_aws_core.config.types import ConfigSource
+from smithy_core.exceptions import ConfigError, ConfigValidationError
+
+
+class NullFileSystem:
+    async def read_file(self, path: str) -> str | None:
+        return None
+
+
+class FakeFileSystem:
+    def __init__(self, files: dict[str, str]):
+        self._files = files
+
+    async def read_file(self, path: str) -> str | None:
+        return self._files.get(path)
 
 
 class TestResolveRegion:
     @pytest.mark.asyncio
     async def test_resolves_from_aws_region(self):
-        ctx = SharedConfigContext(env={"AWS_REGION": "us-west-2"})
-        result = await resolve_region(ctx)
-        assert result.value == "us-west-2"
-        assert result.source == ConfigSource.ENV
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            ctx = SharedConfigContext()
+            result = await resolve_region(ctx)
+            assert result.value == "us-west-2"
+            assert result.source == ConfigSource.ENV
 
     @pytest.mark.asyncio
     async def test_resolves_from_aws_default_region(self):
-        ctx = SharedConfigContext(env={"AWS_DEFAULT_REGION": "eu-central-1"})
-        result = await resolve_region(ctx)
-        assert result.value == "eu-central-1"
-        assert result.source == ConfigSource.ENV
+        with patch.dict(os.environ, {"AWS_DEFAULT_REGION": "eu-central-1"}, clear=True):
+            ctx = SharedConfigContext()
+            result = await resolve_region(ctx)
+            assert result.value == "eu-central-1"
+            assert result.source == ConfigSource.ENV
 
     @pytest.mark.asyncio
     async def test_aws_region_takes_precedence_over_default_region(self):
-        ctx = SharedConfigContext(
-            env={"AWS_REGION": "us-west-2", "AWS_DEFAULT_REGION": "eu-west-1"},
-        )
-        result = await resolve_region(ctx)
-        assert result.value == "us-west-2"
+        with patch.dict(
+            os.environ,
+            {"AWS_REGION": "us-west-2", "AWS_DEFAULT_REGION": "eu-west-1"},
+            clear=True,
+        ):
+            ctx = SharedConfigContext()
+            result = await resolve_region(ctx)
+            assert result.value == "us-west-2"
 
     @pytest.mark.asyncio
-    async def test_resolves_from_profile_when_no_env(self, tmp_path: Path):
-        config_file = tmp_path / "config"
-        config_file.write_text("[profile default]\nregion = ap-southeast-1\n")
-
-        ctx = SharedConfigContext(
-            env={},
-            config_file_path=str(config_file),
-            credentials_file_path=str(tmp_path / "none"),
+    async def test_resolves_from_profile_when_no_env(self):
+        fs = FakeFileSystem(
+            {"/fake/config": "[profile default]\nregion = ap-southeast-1\n"}
         )
-        result = await resolve_region(ctx)
-        assert result.value == "ap-southeast-1"
-        assert result.source == ConfigSource.PROFILE
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            result = await resolve_region(ctx)
+            assert result.value == "ap-southeast-1"
+            assert result.source == ConfigSource.PROFILE
 
     @pytest.mark.asyncio
-    async def test_env_takes_precedence_over_profile(self, tmp_path: Path):
-        config_file = tmp_path / "config"
-        config_file.write_text("[profile default]\nregion = eu-west-1\n")
-        ctx = SharedConfigContext(
-            env={"AWS_REGION": "us-west-2"},
-            config_file_path=str(config_file),
-            credentials_file_path=str(tmp_path / "nonexistent"),
-        )
-        result = await resolve_region(ctx)
-        assert result.value == "us-west-2"
-        assert result.source == ConfigSource.ENV
+    async def test_env_takes_precedence_over_profile(self):
+        fs = FakeFileSystem({"/fake/config": "[profile default]\nregion = eu-west-1\n"})
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            result = await resolve_region(ctx)
+            assert result.value == "us-west-2"
+            assert result.source == ConfigSource.ENV
 
     @pytest.mark.asyncio
-    async def test_empty_string_env_var_treated_as_absent(self, tmp_path: Path):
-        config_file = tmp_path / "config"
-        config_file.write_text("[profile default]\nregion = eu-west-1\n")
-        ctx = SharedConfigContext(
-            env={"AWS_REGION": ""},
-            config_file_path=str(config_file),
-            credentials_file_path=str(tmp_path / "nonexistent"),
-        )
-        result = await resolve_region(ctx)
-        assert result.value == "eu-west-1"
-        assert result.source == ConfigSource.PROFILE
+    async def test_empty_string_env_var_treated_as_absent(self):
+        fs = FakeFileSystem({"/fake/config": "[profile default]\nregion = eu-west-1\n"})
+        with patch.dict(os.environ, {"AWS_REGION": ""}, clear=True):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            result = await resolve_region(ctx)
+            assert result.value == "eu-west-1"
+            assert result.source == ConfigSource.PROFILE
 
 
 class TestResolveRetryConfig:
     @pytest.mark.asyncio
     async def test_resolves_both_from_env(self):
-        ctx = SharedConfigContext(
-            env={"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "10"},
-        )
-        result = await resolve_retry_config(ctx)
-        assert result.value.retry_mode == "standard"
-        assert result.value.max_attempts == 10
-        assert result.source == ConfigSource.ENV
+        with patch.dict(
+            os.environ,
+            {"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "10"},
+            clear=True,
+        ):
+            ctx = SharedConfigContext()
+            result = await resolve_retry_config(ctx)
+            assert result.value.retry_mode == "standard"
+            assert result.value.max_attempts == 10
+            assert result.source == ConfigSource.ENV
 
     @pytest.mark.asyncio
     async def test_uses_defaults_when_nothing_found(self):
-        ctx = SharedConfigContext(env={})
-        result = await resolve_retry_config(ctx)
-        assert result.value.retry_mode == "standard"
-        assert result.value.max_attempts is None
-        assert result.source == ConfigSource.DEFAULT
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(fs=NullFileSystem())
+            result = await resolve_retry_config(ctx)
+            assert result.value.retry_mode == "standard"
+            assert result.value.max_attempts is None
+            assert result.source == ConfigSource.DEFAULT
 
     @pytest.mark.asyncio
     async def test_partial_resolution_uses_defaults_for_missing(self):
-        ctx = SharedConfigContext(env={"AWS_RETRY_MODE": "standard"})
-        result = await resolve_retry_config(ctx)
-        assert result.value.retry_mode == "standard"
-        assert result.value.max_attempts is None  # RetryStrategyOptions default
-        assert result.source == ConfigSource.ENV  # strongest source
+        with patch.dict(os.environ, {"AWS_RETRY_MODE": "standard"}, clear=True):
+            ctx = SharedConfigContext(fs=NullFileSystem())
+            result = await resolve_retry_config(ctx)
+            assert result.value.retry_mode == "standard"
+            assert result.value.max_attempts is None
+            assert result.source == ConfigSource.ENV
 
     @pytest.mark.asyncio
     async def test_max_attempts_cast_to_int(self):
-        ctx = SharedConfigContext(
-            env={"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "5"}
-        )
-        result = await resolve_retry_config(ctx)
-        assert result.value.max_attempts == 5
-        assert isinstance(result.value.max_attempts, int)
+        with patch.dict(
+            os.environ,
+            {"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "5"},
+            clear=True,
+        ):
+            ctx = SharedConfigContext()
+            result = await resolve_retry_config(ctx)
+            assert result.value.max_attempts == 5
+            assert isinstance(result.value.max_attempts, int)
 
     @pytest.mark.asyncio
     async def test_max_attempts_unset_when_not_found(self):
-        ctx = SharedConfigContext(env={})
-        result = await resolve_retry_config(ctx)
-        assert result.value.max_attempts is None
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(fs=NullFileSystem())
+            result = await resolve_retry_config(ctx)
+            assert result.value.max_attempts is None
 
     @pytest.mark.asyncio
     async def test_invalid_max_attempts_raises_config_error(self):
-        ctx = SharedConfigContext(env={"AWS_MAX_ATTEMPTS": "abc"})
-        with pytest.raises(ConfigError, match="Invalid integer value"):
-            await resolve_retry_config(ctx)
+        with patch.dict(os.environ, {"AWS_MAX_ATTEMPTS": "abc"}, clear=True):
+            ctx = SharedConfigContext(fs=NullFileSystem())
+            with pytest.raises(ConfigValidationError, match="Invalid integer value"):
+                await resolve_retry_config(ctx)
 
 
 class TestAsyncAwsConfigResolve:
     @pytest.mark.asyncio
     async def test_resolves_region_from_env(self):
-        config = await AsyncAwsConfig.resolve(env={"AWS_REGION": "us-west-2"})
-        assert config.region == "us-west-2"
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.region == "us-west-2"
 
     @pytest.mark.asyncio
     async def test_resolves_retry_from_env(self):
-        config = await AsyncAwsConfig.resolve(
-            env={"AWS_RETRY_MODE": "standard", "AWS_MAX_ATTEMPTS": "5"},
-        )
-        assert config.retry_strategy_options is not None
-        assert config.retry_strategy_options.retry_mode == "standard"
-        assert config.retry_strategy_options.max_attempts == 5
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_RETRY_MODE": "standard",
+                "AWS_MAX_ATTEMPTS": "5",
+                "AWS_REGION": "us-east-1",
+            },
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.retry_strategy_options is not None
+            assert config.retry_strategy_options.retry_mode == "standard"
+            assert config.retry_strategy_options.max_attempts == 5
 
     @pytest.mark.asyncio
     async def test_explicit_override_takes_precedence(self):
-        config = await AsyncAwsConfig.resolve(
-            env={"AWS_REGION": "us-west-2"},
-            region="eu-west-1",
-        )
-        assert config.region == "eu-west-1"
-
-    @pytest.mark.asyncio
-    async def test_default_region_raises_error(self, tmp_path: Path):
-        with pytest.raises(ConfigError, match="Region is required"):
-            await AsyncAwsConfig.resolve(
-                env={},
-                config_file_path=str(tmp_path / "no_config"),
-                credentials_file_path=str(tmp_path / "no_creds"),
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            config = await AsyncAwsConfig.resolve(
+                region="eu-west-1", fs=NullFileSystem()
             )
+            assert config.region == "eu-west-1"
 
     @pytest.mark.asyncio
-    async def test_default_retry_strategy(self, tmp_path: Path):
-        config = await AsyncAwsConfig.resolve(
-            env={"AWS_REGION": "us-east-1"},
-            config_file_path=str(tmp_path / "no_config"),
-            credentials_file_path=str(tmp_path / "no_creds"),
-        )
-        assert config.retry_strategy_options is not None
-        assert config.retry_strategy_options.retry_mode == "standard"
-        assert config.retry_strategy_options.max_attempts is None
+    async def test_default_region_raises_error(self):
+        with patch.dict(os.environ, {}, clear=True):
+            with pytest.raises(ConfigValidationError, match="Region is required"):
+                await AsyncAwsConfig.resolve(fs=NullFileSystem())
 
     @pytest.mark.asyncio
-    async def test_resolves_region_from_non_default_profile(self, tmp_path: Path):
-        config_file = tmp_path / "config"
-        config_file.write_text(
-            "[profile default]\nregion = us-east-1\n"
-            "[profile work]\nregion = eu-west-1\n"
-        )
+    async def test_default_retry_strategy(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.retry_strategy_options is not None
+            assert config.retry_strategy_options.retry_mode == "standard"
+            assert config.retry_strategy_options.max_attempts is None
 
-        config = await AsyncAwsConfig.resolve(
-            profile="work",
-            env={},
-            config_file_path=str(config_file),
-            credentials_file_path=str(tmp_path / "none"),
+    @pytest.mark.asyncio
+    async def test_resolves_region_from_non_default_profile(self):
+        fs = FakeFileSystem(
+            {
+                "/fake/config": "[profile default]\nregion = us-east-1\n"
+                "[profile work]\nregion = eu-west-1\n"
+            }
         )
-        assert config.region == "eu-west-1"
-        assert config.source_of("region") == ConfigSource.PROFILE
+        with patch.dict(os.environ, {}, clear=True):
+            config = await AsyncAwsConfig.resolve(
+                profile="work",
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            assert config.region == "eu-west-1"
+            assert config.source_of("region") == ConfigSource.PROFILE
 
     @pytest.mark.asyncio
     async def test_invalid_override_triggers_validator(self):
-        with pytest.raises(ConfigError, match="Must be a valid AWS region"):
-            await AsyncAwsConfig.resolve(
-                env={},
-                region="bad-value!",
-            )
+        with patch.dict(os.environ, {}, clear=True):
+            with pytest.raises(
+                ConfigValidationError, match="Must be a valid AWS region"
+            ):
+                await AsyncAwsConfig.resolve(region="bad-value!")
 
 
 class TestProvenanceTracking:
     @pytest.mark.asyncio
     async def test_source_of_env_resolved_field(self):
-        config = await AsyncAwsConfig.resolve(env={"AWS_REGION": "us-west-2"})
-        assert config.source_of("region") == ConfigSource.ENV
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.source_of("region") == ConfigSource.ENV
 
     @pytest.mark.asyncio
-    async def test_source_of_profile_resolved_field(self, tmp_path: Path):
-        config_file = tmp_path / "config"
-        config_file.write_text("[profile default]\nregion = eu-west-1\n")
-
-        config = await AsyncAwsConfig.resolve(
-            env={},
-            config_file_path=str(config_file),
-            credentials_file_path=str(tmp_path / "none"),
-        )
-        assert config.source_of("region") == ConfigSource.PROFILE
+    async def test_source_of_profile_resolved_field(self):
+        fs = FakeFileSystem({"/fake/config": "[profile default]\nregion = eu-west-1\n"})
+        with patch.dict(os.environ, {}, clear=True):
+            config = await AsyncAwsConfig.resolve(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            assert config.source_of("region") == ConfigSource.PROFILE
 
     @pytest.mark.asyncio
-    async def test_source_of_default_field(self, tmp_path: Path):
-        config = await AsyncAwsConfig.resolve(
-            env={"AWS_REGION": "us-east-1"},
-            config_file_path=str(tmp_path / "no_config"),
-            credentials_file_path=str(tmp_path / "no_creds"),
-        )
-        assert config.source_of("retry_strategy_options") == ConfigSource.DEFAULT
+    async def test_source_of_default_field(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.source_of("retry_strategy_options") == ConfigSource.DEFAULT
 
     @pytest.mark.asyncio
     async def test_source_of_override_field(self):
-        config = await AsyncAwsConfig.resolve(env={}, region="us-east-1")
-        assert config.source_of("region") == ConfigSource.OVERRIDE
+        with patch.dict(os.environ, {}, clear=True):
+            config = await AsyncAwsConfig.resolve(
+                region="us-east-1", fs=NullFileSystem()
+            )
+            assert config.source_of("region") == ConfigSource.OVERRIDE
 
     @pytest.mark.asyncio
     async def test_post_resolution_assignment_marks_source_as_override(self):
-        config = await AsyncAwsConfig.resolve(env={"AWS_REGION": "us-west-2"})
-        assert config.source_of("region") == ConfigSource.ENV
-        config.region = "eu-west-1"
-        assert config.source_of("region") == ConfigSource.OVERRIDE
-        assert config.region == "eu-west-1"
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.source_of("region") == ConfigSource.ENV
+            config.region = "eu-west-1"
+            assert config.source_of("region") == ConfigSource.OVERRIDE
+            assert config.region == "eu-west-1"
 
 
 class TestConstructionBlocking:
@@ -249,37 +283,60 @@ def test_direct_instantiation_raises_error(self):
 
     @pytest.mark.asyncio
     async def test_unknown_override_field_raises_error(self):
-        with pytest.raises(ConfigError, match="Unknown config field"):
-            await AsyncAwsConfig.resolve(
-                env={"AWS_REGION": "us-east-1"},
-                reigon="us-west-2",
-            )
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            with pytest.raises(ConfigValidationError, match="Unknown config field"):
+                await AsyncAwsConfig.resolve(reigon="us-west-2")
+
+    @pytest.mark.parametrize(
+        "field_name,invalid_value,match",
+        [
+            ("region", "bad-value!", "Must be a valid AWS region"),
+            ("region", None, "Region is required"),
+            (
+                "retry_strategy_options",
+                "not-a-strategy",
+                "Must be RetryStrategyOptions",
+            ),
+        ],
+    )
+    @pytest.mark.asyncio
+    async def test_setattr_validates_during_override(
+        self,
+        field_name: str,
+        invalid_value: str | None,
+        match: str,
+    ):
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            with pytest.raises(ConfigValidationError, match=match):
+                setattr(config, field_name, invalid_value)
 
 
 class TestSharedConfigContext:
     def test_default_profile_is_default(self):
-        ctx = SharedConfigContext(env={})
-        assert ctx.profile_name == "default"
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext()
+            assert ctx.profile_name == "default"
 
     def test_profile_from_aws_profile_env(self):
-        ctx = SharedConfigContext(env={"AWS_PROFILE": "work"})
-        assert ctx.profile_name == "work"
+        with patch.dict(os.environ, {"AWS_PROFILE": "work"}, clear=True):
+            ctx = SharedConfigContext()
+            assert ctx.profile_name == "work"
 
     def test_explicit_profile_overrides_env(self):
-        ctx = SharedConfigContext(profile_name="custom", env={"AWS_PROFILE": "work"})
-        assert ctx.profile_name == "custom"
+        with patch.dict(os.environ, {"AWS_PROFILE": "work"}, clear=True):
+            ctx = SharedConfigContext(profile_name="custom")
+            assert ctx.profile_name == "custom"
 
     @pytest.mark.asyncio
-    async def test_parsed_profiles_caches_result(self, tmp_path: Path):
-        config_file = tmp_path / "config"
-        config_file.write_text("[profile default]\nregion = us-east-1\n")
-
-        ctx = SharedConfigContext(
-            env={},
-            config_file_path=str(config_file),
-            credentials_file_path=str(tmp_path / "none"),
-        )
-
-        result1 = await ctx.parsed_profiles()
-        result2 = await ctx.parsed_profiles()
-        assert result1 is result2
+    async def test_parsed_profiles_caches_result(self):
+        fs = FakeFileSystem({"/fake/config": "[profile default]\nregion = us-east-1\n"})
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            result1 = await ctx.parsed_profiles()
+            result2 = await ctx.parsed_profiles()
+            assert result1 is result2
diff --git a/packages/smithy-aws-core/tests/unit/config/test_validators.py b/packages/smithy-aws-core/tests/unit/config/test_validators.py
index 6a4016464..4dfdf1d1d 100644
--- a/packages/smithy-aws-core/tests/unit/config/test_validators.py
+++ b/packages/smithy-aws-core/tests/unit/config/test_validators.py
@@ -3,12 +3,14 @@
 """Tests for config field validators."""
 
 import pytest
-from smithy_aws_core.config.exceptions import ConfigError
 from smithy_aws_core.config.types import FieldSpec
 from smithy_aws_core.config.validators import (
+    validate_max_attempts,
     validate_region,
+    validate_retry_mode,
     validate_retry_strategy_options,
 )
+from smithy_core.exceptions import ConfigValidationError
 from smithy_core.retries import RetryStrategyOptions
 
 
@@ -27,7 +29,7 @@ def test_valid_regions(self, region: str):
         validate_region(region)
 
     def test_none_raises(self):
-        with pytest.raises(ConfigError, match="Region is required"):
+        with pytest.raises(ConfigValidationError, match="Region is required"):
             validate_region(None)
 
     @pytest.mark.parametrize(
@@ -43,11 +45,11 @@ def test_none_raises(self):
         ids=lambda x: x if isinstance(x, str) else "",
     )
     def test_invalid_regions(self, region: str, reason: str):
-        with pytest.raises(ConfigError, match="Must be a valid AWS region"):
+        with pytest.raises(ConfigValidationError, match="Must be a valid AWS region"):
             validate_region(region)
 
     def test_non_string_raises(self):
-        with pytest.raises(ConfigError, match="Must be a valid AWS region"):
+        with pytest.raises(ConfigValidationError, match="Must be a valid AWS region"):
             validate_region(123)
 
 
@@ -71,7 +73,7 @@ def test_valid_options_with_values(self):
         ids=["none", "string", "dict", "int"],
     )
     def test_invalid_types_raise(self, value: object):
-        with pytest.raises(ConfigError, match="Must be RetryStrategyOptions"):
+        with pytest.raises(ConfigValidationError, match="Must be RetryStrategyOptions"):
             validate_retry_strategy_options(value)
 
 
@@ -91,3 +93,25 @@ def test_both_default_and_factory_raises(self):
     def test_neither_default_nor_factory_raises(self):
         with pytest.raises(ValueError, match="exactly one of"):
             FieldSpec()
+
+
+class TestValidateRetryMode:
+    @pytest.mark.parametrize("mode", ["standard"])
+    def test_valid_modes(self, mode: str):
+        validate_retry_mode(mode)
+
+    @pytest.mark.parametrize("mode", ["fake-mode", "", "STANDARD"])
+    def test_invalid_modes(self, mode: str):
+        with pytest.raises(ConfigValidationError, match="retry_mode"):
+            validate_retry_mode(mode)
+
+
+class TestValidateMaxAttempts:
+    @pytest.mark.parametrize("value", [1, 3, 10, 100])
+    def test_valid_values(self, value: int):
+        validate_max_attempts(value)
+
+    @pytest.mark.parametrize("value", [0, -1, -100])
+    def test_invalid_values(self, value: int):
+        with pytest.raises(ConfigValidationError, match="max_attempts"):
+            validate_max_attempts(value)
diff --git a/packages/smithy-core/src/smithy_core/exceptions.py b/packages/smithy-core/src/smithy_core/exceptions.py
index 0a99976f9..9faf08ad4 100644
--- a/packages/smithy-core/src/smithy_core/exceptions.py
+++ b/packages/smithy-core/src/smithy_core/exceptions.py
@@ -116,3 +116,15 @@ class UnsupportedStreamError(SmithyError):
 
 class EndpointResolutionError(SmithyError):
     """Exception type for all exceptions raised by endpoint resolution."""
+
+
+class ConfigError(SmithyError):
+    """Raised when a config value cannot be constructed directly"""
+
+
+class ConfigParseError(ConfigError):
+    """Raised when a config file cannot be parsed due to invalid syntax."""
+
+
+class ConfigValidationError(ConfigError):
+    """Raised when a config value cannot be validated"""