diff --git a/src/sap_cloud_sdk/core/secret_resolver/__init__.py b/src/sap_cloud_sdk/core/secret_resolver/__init__.py index 79cf79ff..c151654f 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/__init__.py +++ b/src/sap_cloud_sdk/core/secret_resolver/__init__.py @@ -1,26 +1,64 @@ """ -Secret resolver: load configuration/secrets from mounted files or environment variables +Secret resolver: load configuration/secrets from mounted files or environment variables. -Usage: - from dataclasses import dataclass, field - from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var +Built-in resolvers and chain builder:: - @dataclass - class MyConfig: - username: str = field(metadata={"secret": "username"}) - password: str = field(metadata={"secret": "password"}) - endpoint: str = "http://localhost" + from sap_cloud_sdk.core.secret_resolver import ( + MountResolver, + EnvVarResolver, + ChainedResolver, + ) + + # Build a chain explicitly + resolver = ChainedResolver([MountResolver(), EnvVarResolver()]) + resolver.resolve("destination", "default", binding) + +Legacy function-based API (still supported):: + + from sap_cloud_sdk.core.secret_resolver import read_from_mount_and_fallback_to_env_var - cfg = MyConfig() read_from_mount_and_fallback_to_env_var( base_volume_mount="/etc/secrets/appfnd", base_var_name="CLOUD_SDK_CFG", - module="objectstore", + module="destination", instance="default", - target=cfg + target=binding, ) """ -from .resolver import read_from_mount_and_fallback_to_env_var, resolve_base_mount +from sap_cloud_sdk.core.secret_resolver.resolver import ( + read_from_mount_and_fallback_to_env_var, +) +from sap_cloud_sdk.core.secret_resolver._resolvers import ( + Resolver, + ChainedResolver, +) + +from sap_cloud_sdk.core.secret_resolver.mount_resolver import ( + MountResolver, + resolve_base_mount, +) +from sap_cloud_sdk.core.secret_resolver.env_resolver import EnvVarResolver + +from sap_cloud_sdk.core.secret_resolver.sdk_config import ( + SdkConfig, + configure, + get_sdk_config, + get_resolver, +) -__all__ = ["read_from_mount_and_fallback_to_env_var", "resolve_base_mount"] +__all__ = [ + # Class-based API + "Resolver", + "MountResolver", + "EnvVarResolver", + "ChainedResolver", + # Global configuration + "SdkConfig", + "configure", + "get_sdk_config", + "get_resolver", + # Legacy function-based API + "read_from_mount_and_fallback_to_env_var", + "resolve_base_mount", +] diff --git a/src/sap_cloud_sdk/core/secret_resolver/_mapping.py b/src/sap_cloud_sdk/core/secret_resolver/_mapping.py new file mode 100644 index 00000000..f5316bed --- /dev/null +++ b/src/sap_cloud_sdk/core/secret_resolver/_mapping.py @@ -0,0 +1,32 @@ +"""Utilities for mapping dataclass fields to secret store keys.""" + +from typing import Any, Dict, Tuple +from dataclasses import fields, is_dataclass + + +def _get_field_map(target: Any) -> dict[str, tuple[str, type]]: + """ + Build a mapping from secret key -> (attribute_name, attribute_type) for a dataclass instance. + + Priority: + 1. Use field.metadata["secret"] if present as the key + 2. Fallback to the lowercase dataclass field name + Only string-typed fields are supported. + """ + if not is_dataclass(target) or isinstance(target, type): + raise TypeError("target must be a dataclass instance") + + mapping: Dict[str, Tuple[str, type]] = {} + for f in fields(target): + # Only support string fields for secrets (consistent with Go SDK) + # Allow plain 'str' annotations; reject others to keep behavior predictable + if f.type is not str: + raise TypeError( + f"target field '{f.name}' is not a string (only str fields are supported)" + ) + key = f.metadata.get("secret") if hasattr(f, "metadata") else None + if key and isinstance(key, str) and key.strip(): + mapping[key] = (f.name, f.type) + else: + mapping[f.name.lower()] = (f.name, f.type) + return mapping diff --git a/src/sap_cloud_sdk/core/secret_resolver/_resolvers.py b/src/sap_cloud_sdk/core/secret_resolver/_resolvers.py new file mode 100644 index 00000000..e74b6cfe --- /dev/null +++ b/src/sap_cloud_sdk/core/secret_resolver/_resolvers.py @@ -0,0 +1,92 @@ +"""BindingResolver protocol and built-in implementations. + +This module defines the core extensibility contract for secret resolution. +Each resolver encapsulates one binding source. Compose them into an ordered +chain via :class:`ChainedResolver` — the first resolver that succeeds wins. + +Protocol contract:: + + resolver.resolve(module, instance, target) + +- On success: populates ``target`` in-place, returns ``None`` +- On failure: raises any exception; the chain tries the next resolver +""" + +from __future__ import annotations + +from dataclasses import fields, is_dataclass +from typing import Any, Protocol, runtime_checkable + + +@runtime_checkable +class Resolver(Protocol): + """Contract for a single binding resolution strategy. + + A ``BindingResolver`` reads credentials from one source and populates + ``target`` in-place. Implementations raise on failure so that a + :class:`ChainedResolver` can try the next strategy. + + Any object implementing ``resolve`` with this signature satisfies the + protocol — no inheritance required. + """ + + def resolve(self, module: str, instance: str, target: Any) -> None: + """Populate ``target`` with credentials for ``module``/``instance``. + + Args: + module: Service module name (e.g. ``"destination"``). + instance: Instance identifier (e.g. ``"default"``). + target: Dataclass instance whose ``str`` fields will be set. + + Raises: + Any exception on failure; the caller determines how to handle it. + """ + ... + + +class ChainedResolver: + """Tries each resolver in order; returns on the first success. + + Collects failure messages from each resolver and raises a + :class:`RuntimeError` with an aggregated report when all resolvers fail. + + Args: + resolvers: Ordered list of :class:`BindingResolver` implementations to try. + base_var_name: Used only for the error guidance message. + """ + + def __init__( + self, + resolvers: list[Resolver], + base_var_name: str = "CLOUD_SDK_CFG", + ) -> None: + if not resolvers: + raise ValueError("resolvers list must not be empty") + self._resolvers = resolvers + self._base_var_name = base_var_name + + def resolve(self, module: str, instance: str, target: Any) -> None: + """Try each resolver in order; raise on total failure.""" + if not is_dataclass(target) or isinstance(target, type): + raise TypeError("target must be a dataclass instance") + for f in fields(target): + if f.type is not str and f.type != "str": + raise TypeError( + f"target field {f.name!r} is not a string (only str fields are supported)" + ) + + errors: list[str] = [] + for resolver in self._resolvers: + try: + resolver.resolve(module, instance, target) + return + except Exception as e: + label = type(resolver).__name__ + errors.append(f"{label} failed: {e}") + + raise RuntimeError( + f"module={module!r} instance={instance!r} failed to read secrets from all resolvers: " + f"{errors}. " + "Options: mount secrets under the service binding path, set environment variables " + f"like {self._base_var_name}_{module}_{instance}_ (uppercased), or set VCAP_SERVICES." + ) diff --git a/src/sap_cloud_sdk/core/secret_resolver/constants.py b/src/sap_cloud_sdk/core/secret_resolver/constants.py index 7d66cf40..165dbc79 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/constants.py +++ b/src/sap_cloud_sdk/core/secret_resolver/constants.py @@ -3,3 +3,5 @@ """ BASE_MOUNT_PATH = "/etc/secrets/appfnd" +BASE_VAR_NAME = "CLOUD_SDK_CFG" +SERVICE_BINDING_ROOT = "SERVICE_BINDING_ROOT" diff --git a/src/sap_cloud_sdk/core/secret_resolver/env_resolver.py b/src/sap_cloud_sdk/core/secret_resolver/env_resolver.py new file mode 100644 index 00000000..4e42f0c8 --- /dev/null +++ b/src/sap_cloud_sdk/core/secret_resolver/env_resolver.py @@ -0,0 +1,47 @@ +"""Resolver that reads service binding secrets from environment variables.""" + +import os +from typing import Any + +from sap_cloud_sdk.core.secret_resolver._mapping import _get_field_map +from sap_cloud_sdk.core.secret_resolver.constants import BASE_VAR_NAME + + +class EnvVarResolver: + """Resolves bindings from environment variables. + + Reads variables named ``{base_var_name}_{module}_{instance}_{field_key}`` + (uppercased, hyphens in module/instance replaced with underscores). + + Args: + base_var_name: Env var name prefix. Defaults to ``"CLOUD_SDK_CFG"``. + """ + + def __init__(self, base_var_name: str = BASE_VAR_NAME) -> None: + self._base_var_name = base_var_name + + def resolve(self, module: str, instance: str, target: Any) -> None: + """Load secrets from environment variables.""" + normalized_module = module.replace("-", "_") + normalized_instance = instance.replace("-", "_") + _load_from_env( + self._base_var_name, normalized_module, normalized_instance, target + ) + + +def _load_from_env(base_var_name: str, module: str, instance: str, target: Any) -> None: + """ + Load secrets from environment variables with names: + {base_var_name}_{module}_{instance}_{field_key} (uppercased) + instance names have '-' replaced with '_' for env var compatibility. + """ + field_map = _get_field_map(target) + prefix = f"{base_var_name}_{module}_{instance}".upper() + + for key, (attr_name, _) in field_map.items(): + var_name = f"{prefix}_{key}".upper() + value = os.environ.get(var_name) + if value is None: + # Align with Go: error if env var not found + raise KeyError(f"env var not found: {var_name}") + setattr(target, attr_name, value) diff --git a/src/sap_cloud_sdk/core/secret_resolver/mount_resolver.py b/src/sap_cloud_sdk/core/secret_resolver/mount_resolver.py new file mode 100644 index 00000000..f24dde78 --- /dev/null +++ b/src/sap_cloud_sdk/core/secret_resolver/mount_resolver.py @@ -0,0 +1,93 @@ +"""Resolver that reads service binding secrets from a mounted volume path.""" + +import os +from typing import Any + +from sap_cloud_sdk.core.secret_resolver._mapping import _get_field_map +from sap_cloud_sdk.core.secret_resolver.constants import ( + BASE_MOUNT_PATH, + SERVICE_BINDING_ROOT, +) + + +class MountResolver: + """Resolves bindings from a mounted volume path. + + Reads secret files at ``{base_volume_mount}/{module}/{instance}/{field_key}``. + Respects the ``SERVICE_BINDING_ROOT`` environment variable (servicebinding.io + spec) as an override for ``base_volume_mount``. + + Args: + base_volume_mount: Base path for mounted secrets. Defaults to + ``/etc/secrets/appfnd``. + """ + + def __init__(self, base_volume_mount: str = BASE_MOUNT_PATH) -> None: + self._base_volume_mount = base_volume_mount + + def resolve(self, module: str, instance: str, target: Any) -> None: + """Load secrets from the mounted volume path.""" + effective_base = resolve_base_mount(self._base_volume_mount) + _load_from_mount(effective_base, module, instance, target) + + +def resolve_base_mount(base_volume_mount: str = BASE_MOUNT_PATH) -> str: + """Resolve the base mount path for service binding discovery. + + Checks the ``SERVICE_BINDING_ROOT`` environment variable first (as defined + by the `servicebinding.io `_ + specification). Falls back to ``base_volume_mount`` when the env var is + absent. + + Args: + base_volume_mount: Default base path used when ``SERVICE_BINDING_ROOT`` + is not set. Defaults to ``/etc/secrets/appfnd``. + + Returns: + The effective base path for secret mount resolution. + """ + return os.environ.get(SERVICE_BINDING_ROOT, base_volume_mount) + + +def _load_from_mount( + base_volume_mount: str, module: str, instance: str, target: Any +) -> None: + """ + Load secrets from files at: + {base_volume_mount}/{module}/{instance}/{field_key} + + Sets string attributes directly on the dataclass instance. + """ + secret_dir = os.path.join(base_volume_mount, module, instance) + _validate_path(secret_dir) + + field_map = _get_field_map(target) + for key, (attr_name, _) in field_map.items(): + file_path = os.path.join(secret_dir, key) + try: + # Read entire file content as text; do not strip newlines to match Go behavior + with open(file_path, "r", encoding="utf-8") as f: + content = f.read() + except FileNotFoundError as e: + # Align with Go: surface precise file error + raise FileNotFoundError( + f"failed to read secret file {file_path}: {e}" + ) from e + except OSError as e: + raise OSError(f"failed to read secret file {file_path}: {e}") from e + + # Set target field (string only) + setattr(target, attr_name, content) + + +def _validate_path(path: str) -> None: + """Validate that the given path exists and is a directory.""" + try: + _st = os.stat(path) + except FileNotFoundError as e: + raise FileNotFoundError(f"path does not exist: {path}") from e + except OSError as e: + raise OSError(f"cannot access path {path}: {e}") from e + # If exists, ensure it's a directory + if not os.path.isdir(path): + raise NotADirectoryError(f"path is not a directory: {path}") diff --git a/src/sap_cloud_sdk/core/secret_resolver/resolver.py b/src/sap_cloud_sdk/core/secret_resolver/resolver.py index c76bbc74..a5433f6c 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/resolver.py +++ b/src/sap_cloud_sdk/core/secret_resolver/resolver.py @@ -1,136 +1,14 @@ """Core secret resolver implementation.""" from __future__ import annotations - import os -from dataclasses import fields, is_dataclass -from typing import Any, Dict, Tuple -from .constants import BASE_MOUNT_PATH - - -def resolve_base_mount(base_volume_mount: str = BASE_MOUNT_PATH) -> str: - """Resolve the base mount path for service binding discovery. - - Checks the ``SERVICE_BINDING_ROOT`` environment variable first (as defined - by the `servicebinding.io `_ - specification). Falls back to ``base_volume_mount`` when the env var is - absent. - - Args: - base_volume_mount: Default base path used when ``SERVICE_BINDING_ROOT`` - is not set. Defaults to ``/etc/secrets/appfnd``. - - Returns: - The effective base path for secret mount resolution. - """ - return os.environ.get("SERVICE_BINDING_ROOT", base_volume_mount) - - -def _validate_inputs(module: str, instance: str) -> None: - """Validate module and instance inputs.""" - if not isinstance(module, str) or not module.strip(): - raise ValueError("module name cannot be empty") - if not isinstance(instance, str) or not instance.strip(): - raise ValueError("instance name cannot be empty") +from typing import Any - -def _validate_path(path: str) -> None: - """Validate that the given path exists and is a directory.""" - try: - _st = os.stat(path) - except FileNotFoundError as e: - raise FileNotFoundError(f"path does not exist: {path}") from e - except OSError as e: - raise OSError(f"cannot access path {path}: {e}") from e - # If exists, ensure it's a directory - if not os.path.isdir(path): - raise NotADirectoryError(f"path is not a directory: {path}") - - -def _get_field_map(target: Any) -> Dict[str, Tuple[str, type]]: - """ - Build a mapping from secret key -> (attribute_name, attribute_type) for a dataclass instance. - - Priority: - 1. Use field.metadata["secret"] if present as the key - 2. Fallback to the lowercase dataclass field name - Only string-typed fields are supported. - """ - if not is_dataclass(target) or isinstance(target, type): - raise TypeError("target must be a dataclass instance") - - mapping: Dict[str, Tuple[str, type]] = {} - for f in fields(target): - # Only support string fields for secrets (consistent with Go SDK) - # Allow plain 'str' annotations; reject others to keep behavior predictable - if f.type is not str: - raise TypeError( - f"target field '{f.name}' is not a string (only str fields are supported)" - ) - key = f.metadata.get("secret") if hasattr(f, "metadata") else None - if key and isinstance(key, str) and key.strip(): - mapping[key] = (f.name, f.type) - else: - mapping[f.name.lower()] = (f.name, f.type) - return mapping - - -def _load_from_path(secret_dir: str, target: Any) -> None: - """ - Load secrets from files directly in secret_dir into target dataclass. - - Reads each field key as a file name under secret_dir. Used for both the - servicebinding.io flat layout ($ROOT//) and the legacy - three-level layout via :func:`_load_from_mount`. - """ - _validate_path(secret_dir) - - field_map = _get_field_map(target) - for key, (attr_name, _) in field_map.items(): - file_path = os.path.join(secret_dir, key) - try: - # Read entire file content as text; do not strip newlines to match Go behavior - with open(file_path, "r", encoding="utf-8") as f: - content = f.read() - except FileNotFoundError as e: - # Align with Go: surface precise file error - raise FileNotFoundError( - f"failed to read secret file {file_path}: {e}" - ) from e - except OSError as e: - raise OSError(f"failed to read secret file {file_path}: {e}") from e - - # Set target field (string only) - setattr(target, attr_name, content) - - -def _load_from_mount( - base_volume_mount: str, module: str, instance: str, target: Any -) -> None: - """ - Load secrets from files at: - {base_volume_mount}/{module}/{instance}/{field_key} - """ - secret_dir = os.path.join(base_volume_mount, module, instance) - _load_from_path(secret_dir, target) - - -def _load_from_env(base_var_name: str, module: str, instance: str, target: Any) -> None: - """ - Load secrets from environment variables with names: - {base_var_name}_{module}_{instance}_{field_key} (uppercased) - instance names have '-' replaced with '_' for env var compatibility. - """ - field_map = _get_field_map(target) - prefix = f"{base_var_name}_{module}_{instance}".upper() - - for key, (attr_name, _) in field_map.items(): - var_name = f"{prefix}_{key}".upper() - value = os.environ.get(var_name) - if value is None: - # Align with Go: error if env var not found - raise KeyError(f"env var not found: {var_name}") - setattr(target, attr_name, value) +from sap_cloud_sdk.core.secret_resolver.mount_resolver import ( + resolve_base_mount, + _load_from_mount, +) +from sap_cloud_sdk.core.secret_resolver.env_resolver import _load_from_env def read_from_mount_and_fallback_to_env_var( @@ -142,21 +20,17 @@ def read_from_mount_and_fallback_to_env_var( ) -> None: """ Load secrets for a given module and instance into the provided dataclass instance `target`. - - Fallback order when ``SERVICE_BINDING_ROOT`` is set: - 1. Flat path: {SERVICE_BINDING_ROOT}/{module}/{field_key} (servicebinding.io spec) - 2. Full path: {SERVICE_BINDING_ROOT}/{module}/{instance}/{field_key} (legacy convention) - 3. Environment variables: {base_var_name}_{module}_{instance}_{field_key} (uppercased) - - Fallback order when ``SERVICE_BINDING_ROOT`` is **not** set: - 1. Full path: {base_volume_mount}/{module}/{instance}/{field_key} + Fallback order: + 1. Mounted volume path: {base_volume_mount}/{module}/{instance}/{field_key} + (``SERVICE_BINDING_ROOT`` env var overrides ``base_volume_mount`` — see + :func:`resolve_base_mount`) 2. Environment variables: {base_var_name}_{module}_{instance}_{field_key} (uppercased) Raises: ValueError: If inputs are invalid or target is not a dataclass instance FileNotFoundError / NotADirectoryError / OSError: If mount path issues occur KeyError: If environment variables are missing on fallback - RuntimeError: If all strategies fail (aggregated error) + RuntimeError: If both mount and env var loading fail (aggregated error) """ _validate_inputs(module, instance) @@ -165,15 +39,6 @@ def read_from_mount_and_fallback_to_env_var( normalized_module = module.replace("-", "_") normalized_instance = instance.replace("-", "_") - # servicebinding.io: when SERVICE_BINDING_ROOT is explicitly set, try the flat path - # $ROOT// before the legacy $ROOT/// path. - if os.environ.get("SERVICE_BINDING_ROOT") is not None: - try: - _load_from_path(os.path.join(resolved_base_path, module), target) - return - except Exception as e: - errors.append(f"mount failed: {e};") - try: _load_from_mount(resolved_base_path, module, instance, target) return @@ -202,3 +67,11 @@ def read_from_mount_and_fallback_to_env_var( raise RuntimeError( f"module={module} instance={instance} failed to read secrets: {errors} {guidance}" ) + + +def _validate_inputs(module: str, instance: str) -> None: + """Validate module and instance inputs.""" + if not isinstance(module, str) or not module.strip(): + raise ValueError("module name cannot be empty") + if not isinstance(instance, str) or not instance.strip(): + raise ValueError("instance name cannot be empty") diff --git a/src/sap_cloud_sdk/core/secret_resolver/sdk_config.py b/src/sap_cloud_sdk/core/secret_resolver/sdk_config.py new file mode 100644 index 00000000..223a2cf6 --- /dev/null +++ b/src/sap_cloud_sdk/core/secret_resolver/sdk_config.py @@ -0,0 +1,94 @@ +"""Process-wide SDK configuration. + +Provides :class:`SdkConfig` and :func:`configure` so an application can set a +custom binding resolver chain once at startup, and every ``create_client()`` call +across all modules will use it automatically. + +Usage:: + + from sap_cloud_sdk.core.secret_resolver import ( + configure, SdkConfig, ChainedResolver, EnvVarResolver, + ) + from sap_cloud_sdk.core.secrets_resolver_extended import VcapResolver + + # Cloud Foundry: VCAP first, env vars as fallback + configure(SdkConfig( + resolver=ChainedResolver([VcapResolver(), EnvVarResolver()]) + )) + +If no configuration is set, all modules fall back to the default chain: +``ChainedResolver([MountResolver(), EnvVarResolver()])``, which is identical +to the previous ``read_from_mount_and_fallback_to_env_var()`` behaviour. +""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass +from typing import Optional + +from sap_cloud_sdk.core.secret_resolver._resolvers import ( + Resolver, + ChainedResolver, +) + +from sap_cloud_sdk.core.secret_resolver.env_resolver import EnvVarResolver +from sap_cloud_sdk.core.secret_resolver.mount_resolver import MountResolver + +_lock = threading.Lock() +_sdk_config: Optional[SdkConfig] = None + + +@dataclass +class SdkConfig: + """Process-wide SDK configuration. + + Attributes: + resolver: The :class:`BindingResolver` that all ``create_client()`` calls + use when no explicit config is passed. ``None`` means each module uses + the default ``ChainedResolver([MountResolver(), EnvVarResolver()])`` chain. + """ + + resolver: Optional[Resolver] = None + + +def configure(config: SdkConfig) -> None: + """Set the process-wide SDK configuration. + + Thread-safe. Replaces any previously set configuration. Call once at + application startup before any ``create_client()`` is invoked. + + Args: + config: :class:`SdkConfig` instance to install. + """ + global _sdk_config + with _lock: + _sdk_config = config + + +def get_sdk_config() -> Optional[SdkConfig]: + """Return the current process-wide SDK configuration, or ``None`` if unset.""" + return _sdk_config + + +def _reset_sdk_config() -> None: + """Reset the global configuration to the unset state. + + Intended for test teardown only. Not part of the public API. + """ + global _sdk_config + with _lock: + _sdk_config = None + + +def get_resolver() -> Resolver: + """Return the resolver all modules should use for binding resolution. + + Returns the custom resolver set via :func:`configure` if one has been + installed; otherwise returns a fresh default + ``ChainedResolver([MountResolver(), EnvVarResolver()])``. + """ + cfg = get_sdk_config() + if cfg is not None and cfg.resolver is not None: + return cfg.resolver + return ChainedResolver([MountResolver(), EnvVarResolver()]) diff --git a/src/sap_cloud_sdk/core/secret_resolver/user-guide.md b/src/sap_cloud_sdk/core/secret_resolver/user-guide.md index aeea94b6..61c588b7 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/user-guide.md +++ b/src/sap_cloud_sdk/core/secret_resolver/user-guide.md @@ -1,256 +1,179 @@ -# Secret Resolver User Guide +# Secret Resolver — User Guide -This module provides secure credential management by loading secrets from mounted volumes (Kubernetes-style) with fallback to environment variables. It supports type-safe configuration using dataclasses and follows Cloud patterns for secret resolution. - -The Secret Resolver is designed to work seamlessly in both Kubernetes environments with mounted secrets and with environment variables. - -## Import - -```python -from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var -``` +The `secret_resolver` module loads service-binding credentials (secrets) into +dataclass instances from two sources: **mounted volume files** and **environment +variables**. Resolvers are composable and the default chain can be replaced +process-wide for custom environments (e.g. Cloud Foundry VCAP). --- -## Getting Started +## Concepts -The Secret Resolver loads configuration into dataclass objects using a hierarchical approach: +### Target dataclass -1. **First**: Try to read from mounted volume paths (Kubernetes secrets) -2. **Fallback**: Use environment variables if mounted secrets are not available +All resolvers populate a **dataclass instance** whose fields are all `str`. +Each field maps to one secret key. By default the key is the lowercase field +name; you can override it with `field(metadata={"secret": "custom-key"})`. ```python -from dataclasses import dataclass -from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var +from dataclasses import dataclass, field @dataclass -class DatabaseConfig: - host: str = "" - port: str = "" - username: str = "" - password: str = "" - -# Load configuration -config = DatabaseConfig() -read_from_mount_and_fallback_to_env_var( - base_volume_mount="/etc/secrets", # Base mount path - base_var_name="DB", # Environment variable prefix - module="database", # Module/service name - instance="primary", # Instance name - target=config # Target dataclass instance -) - -print(f"Database: {config.username}@{config.host}:{config.port}") +class DestinationBinding: + clientid: str = "" + clientsecret: str = "" + url: str = "" + # Override the lookup key to "token_service_url" + token_url: str = field(default="", metadata={"secret": "token_service_url"}) ``` +### module / instance + +Every resolver call takes a `module` (service category, e.g. `"destination"`) +and an `instance` (service instance name, e.g. `"default"`). These are used to +build the lookup path or variable name prefix. Hyphens in both are normalised +to underscores where required. + --- -## Configuration Patterns +## Resolver types -### Mount Path Structure +### `MountResolver` -The Secret Resolver expects mounted secrets to follow this hierarchy: +Reads secret files at: ``` -/etc/secrets/appfnd -└── / - └── / - ├── host - ├── port - ├── username - └── password +{base_volume_mount}/{module}/{instance}/{field_key} ``` -### Base Path Resolution -By default, the resolver looks for secrets under `/etc/secrets/appfnd`. You can override this by setting the `SERVICE_BINDING_ROOT` environment variable, which follows the [servicebinding.io](https://servicebinding.io) specification used across SAP SDKs and Kubernetes-native tooling. - -When `SERVICE_BINDING_ROOT` is set, it takes precedence over the default `/etc/secrets/appfnd` path: +```python +from sap_cloud_sdk.core.secret_resolver import MountResolver -```bash -export SERVICE_BINDING_ROOT=/bindings +resolver = MountResolver() # defaults to /etc/secrets/appfnd +resolver = MountResolver("/custom/mount/path") # explicit base path ``` -With this set, the resolver looks for secrets at `$SERVICE_BINDING_ROOT///` instead of `/etc/secrets/appfnd///`. +### `EnvVarResolver` + +Reads environment variables named: -Example for the above configuration: ``` -/etc/secrets/appfnd -└── database/ - └── primary/ - ├── host # Contains: "db.example.com" - ├── port # Contains: "5432" - ├── username # Contains: "app_user" - └── password # Contains: "secret123" +{BASE_VAR_NAME}_{MODULE}_{INSTANCE}_{FIELD_KEY} (all uppercased) ``` -### Environment Variable Fallback +Example: `CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTID` -If mounted secrets are not available, the resolver falls back to environment variables using this pattern: +```python +from sap_cloud_sdk.core.secret_resolver import EnvVarResolver +resolver = EnvVarResolver() # base prefix: CLOUD_SDK_CFG +resolver = EnvVarResolver("MY_APP_SECRETS") # custom prefix ``` -___ -``` -- INSTANCE_NAME has `'-'` replaced with `'_'` for compatibility with system environment variable naming rules. +### `ChainedResolver` -For the example above: -```bash -export DB_DATABASE_PRIMARY_HOST="db.example.com" -export DB_DATABASE_PRIMARY_PORT="5432" -export DB_DATABASE_PRIMARY_USERNAME="app_user" -export DB_DATABASE_PRIMARY_PASSWORD="secret123" -``` +Tries each resolver in order and returns on the first success. Raises +`RuntimeError` with an aggregated report when all resolvers fail. ---- +```python +from sap_cloud_sdk.core.secret_resolver import ChainedResolver, MountResolver, EnvVarResolver -## Usage Examples +resolver = ChainedResolver([MountResolver(), EnvVarResolver()]) +resolver.resolve("destination", "default", binding) +``` -### ObjectStore Configuration +### Custom resolver -This is how the ObjectStore module uses the Secret Resolver internally: +Any object with a `resolve(module, instance, target)` method satisfies the +`Resolver` protocol — no inheritance needed. ```python -from dataclasses import dataclass -from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var - -@dataclass -class ObjectStoreConfig: - access_key_id: str = "" - secret_access_key: str = "" - bucket: str = "" - host: str = "" - -# Load ObjectStore credentials -config = ObjectStoreConfig() -read_from_mount_and_fallback_to_env_var( - base_volume_mount="/etc/secrets", - base_var_name="OBJECTSTORE", - module="objectstore", - instance="credentials", - target=config -) +class VaultResolver: + def resolve(self, module: str, instance: str, target: object) -> None: + # fetch from HashiCorp Vault, populate target fields + ... ``` -**Mounted secrets structure:** -``` -/etc/secrets/appfnd -└── objectstore/ - └── credentials/ - ├── access_key_id - ├── secret_access_key - ├── bucket - └── host -``` +--- -**Environment variable fallback:** -```bash -export OBJECTSTORE_OBJECTSTORE_CREDENTIALS_ACCESS_KEY_ID="AKIA..." -export OBJECTSTORE_OBJECTSTORE_CREDENTIALS_SECRET_ACCESS_KEY="secret" -export OBJECTSTORE_OBJECTSTORE_CREDENTIALS_BUCKET="my-bucket" -export OBJECTSTORE_OBJECTSTORE_CREDENTIALS_HOST="s3.amazonaws.com" -``` +## Process-wide configuration -### Database Configuration with Multiple Instances +Call `configure()` once at application startup to install a custom resolver +chain. All `create_client()` calls across every module will use it. ```python -from dataclasses import dataclass -from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var +from sap_cloud_sdk.core.secret_resolver import configure, SdkConfig, ChainedResolver, EnvVarResolver -@dataclass -class DatabaseConfig: - host: str = "" - port: str = "5432" # Default value - database: str = "" - username: str = "" - password: str = "" - ssl_mode: str = "require" # Default value - -# Load primary database config -primary_db = DatabaseConfig() -read_from_mount_and_fallback_to_env_var( - base_volume_mount="/etc/secrets", - base_var_name="APP", - module="database", - instance="primary", - target=primary_db -) +configure(SdkConfig( + resolver=ChainedResolver([VaultResolver(), EnvVarResolver()]) +)) +``` -# Load read replica config -replica_db = DatabaseConfig() -read_from_mount_and_fallback_to_env_var( - base_volume_mount="/etc/secrets", - base_var_name="APP", - module="database", - instance="replica", - target=replica_db -) +If `configure()` is never called, each module falls back to the default chain: +`ChainedResolver([MountResolver(), EnvVarResolver()])`. -print(f"Primary DB: {primary_db.username}@{primary_db.host}") -print(f"Replica DB: {replica_db.username}@{replica_db.host}") -``` +### Configuration API + +| Function | Description | +|---|---| +| `configure(config)` | Install a process-wide `SdkConfig`. Thread-safe. | +| `get_sdk_config()` | Return the current `SdkConfig`, or `None` if unset. | +| `get_resolver()` | Return the active resolver (custom or default chain). | +| `reset_sdk_config()` | Reset to unset state. Intended for test teardown only. | + +--- -### API Configuration +## Putting it together ```python -from dataclasses import dataclass -from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var +from dataclasses import dataclass, field +from sap_cloud_sdk.core.secret_resolver import ChainedResolver, MountResolver, EnvVarResolver @dataclass -class ApiConfig: - base_url: str = "" - api_key: str = "" - timeout: str = "30" - retries: str = "3" - -# Load external API configuration -api_config = ApiConfig() -read_from_mount_and_fallback_to_env_var( - base_volume_mount="/etc/secrets", - base_var_name="EXTERNAL", - module="payment", - instance="stripe", - target=api_config -) +class DestinationBinding: + clientid: str = "" + clientsecret: str = "" + url: str = "" + +binding = DestinationBinding() -# Convert string values to appropriate types -timeout_seconds = int(api_config.timeout) -max_retries = int(api_config.retries) +resolver = ChainedResolver([MountResolver(), EnvVarResolver()]) +resolver.resolve("destination", "my-instance", binding) + +print(binding.clientid) ``` --- -## Error Handling +## Legacy API -The Secret Resolver handles missing secrets gracefully by leaving default values unchanged: +The function-based API from earlier SDK versions is still supported: ```python -from dataclasses import dataclass -from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var - -@dataclass -class ServiceConfig: - api_key: str = "" - timeout: str = "30" # Default value - retries: str = "3" # Default value - debug: str = "false" # Default value +from sap_cloud_sdk.core.secret_resolver import read_from_mount_and_fallback_to_env_var -config = ServiceConfig() - -# This won't raise an error if secrets are missing read_from_mount_and_fallback_to_env_var( - base_volume_mount="/etc/secrets", - base_var_name="API", - module="external", - instance="service", - target=config + base_volume_mount="/etc/secrets/appfnd", + base_var_name="CLOUD_SDK_CFG", + module="destination", + instance="default", + target=binding, ) - -# Check if required values were loaded -if not config.api_key: - raise ValueError("API key is required but not found in secrets or environment") - -print(f"Loaded config: timeout={config.timeout}, retries={config.retries}") ``` +Prefer the class-based API for new code — it is more composable and supports +process-wide configuration. + --- + +## Error handling + +| Situation | Exception raised | +|---|---| +| Target is not a dataclass instance | `TypeError` | +| A target field is not `str` | `TypeError` | +| Mount directory does not exist | `FileNotFoundError` | +| Mount path is not a directory | `NotADirectoryError` | +| Expected env var is absent | `KeyError` | +| All resolvers in a chain fail | `RuntimeError` (aggregated message with guidance) | diff --git a/tests/core/unit/secret_resolver/unit/test_secret_resolver.py b/tests/core/unit/secret_resolver/unit/test_secret_resolver.py index f02445ff..baa2c257 100644 --- a/tests/core/unit/secret_resolver/unit/test_secret_resolver.py +++ b/tests/core/unit/secret_resolver/unit/test_secret_resolver.py @@ -2,10 +2,22 @@ import os from dataclasses import dataclass, field -from unittest.mock import patch, mock_open +from unittest.mock import patch, mock_open, MagicMock import pytest -from sap_cloud_sdk.core.secret_resolver import read_from_mount_and_fallback_to_env_var +from sap_cloud_sdk.core.secret_resolver import ( + read_from_mount_and_fallback_to_env_var, + MountResolver, + EnvVarResolver, + ChainedResolver, + Resolver, + SdkConfig, + configure, + get_sdk_config, + get_resolver, + resolve_base_mount, +) +from sap_cloud_sdk.core.secret_resolver.sdk_config import _reset_sdk_config @dataclass @@ -20,7 +32,11 @@ class NonStringConfig: count: int = 0 -class TestSecretResolver: +# --------------------------------------------------------------------------- +# Legacy function-based API +# --------------------------------------------------------------------------- + +class TestLegacyResolver: def test_validate_inputs_empty_module(self): config = SampleConfig() @@ -48,9 +64,8 @@ def test_load_from_mount_success(self, mock_file, mock_stat, mock_isdir): mock_file.side_effect = [ mock_open(read_data="test_user").return_value, mock_open(read_data="test_pass").return_value, - mock_open(read_data="test_endpoint").return_value + mock_open(read_data="test_endpoint").return_value, ] - config = SampleConfig() read_from_mount_and_fallback_to_env_var("/secrets", "VAR", "module", "instance", config) @@ -82,7 +97,7 @@ def test_validate_path_not_directory(self, mock_stat, mock_isdir): @patch.dict(os.environ, { "VAR_MODULE_INSTANCE_USER": "env_user", "VAR_MODULE_INSTANCE_PASSWORD": "env_pass", - "VAR_MODULE_INSTANCE_ENDPOINT": "env_endpoint" + "VAR_MODULE_INSTANCE_ENDPOINT": "env_endpoint", }) def test_load_from_env_success(self): config = SampleConfig() @@ -109,9 +124,8 @@ def test_mount_success_no_env_fallback(self, mock_file, mock_stat, mock_isdir): mock_file.side_effect = [ mock_open(read_data="mount_user").return_value, mock_open(read_data="mount_pass").return_value, - mock_open(read_data="mount_endpoint").return_value + mock_open(read_data="mount_endpoint").return_value, ] - config = SampleConfig() read_from_mount_and_fallback_to_env_var("/secrets", "VAR", "module", "instance", config) @@ -132,9 +146,8 @@ def test_preserves_newlines(self, mock_file, mock_stat, mock_isdir): mock_file.side_effect = [ mock_open(read_data="user\nwith\nnewlines").return_value, mock_open(read_data="pass").return_value, - mock_open(read_data="endpoint").return_value + mock_open(read_data="endpoint").return_value, ] - config = SampleConfig() read_from_mount_and_fallback_to_env_var("/secrets", "VAR", "module", "instance", config) @@ -160,9 +173,8 @@ def test_metadata_secret_priority(self, mock_file, mock_stat, mock_isdir): mock_file.side_effect = [ mock_open(read_data="metadata_user").return_value, mock_open(read_data="field_pass").return_value, - mock_open(read_data="field_endpoint").return_value + mock_open(read_data="field_endpoint").return_value, ] - config = SampleConfig() read_from_mount_and_fallback_to_env_var("/secrets", "VAR", "module", "instance", config) @@ -174,7 +186,6 @@ def test_metadata_secret_priority(self, mock_file, mock_stat, mock_isdir): "VAR_MODULE_MY_INSTANCE_ENDPOINT": "env_endpoint_hyphen", }) def test_env_instance_name_hyphen_normalization(self): - # Given instance name with hyphen, the resolver should replace '-' with '_' config = SampleConfig() with patch('os.path.isdir', return_value=False), \ patch('os.stat', side_effect=FileNotFoundError()): @@ -199,8 +210,8 @@ def test_service_binding_root_overrides_base_mount(self, mock_file, mock_stat, m config = SampleConfig() read_from_mount_and_fallback_to_env_var("/etc/secrets/appfnd", "VAR", "module", "instance", config) first_call_path = mock_file.call_args_list[0][0][0] - # With SERVICE_BINDING_ROOT set, flat path is tried first: $ROOT// - assert first_call_path == "/custom/root/module/user" + # SERVICE_BINDING_ROOT overrides the base; path is still $ROOT/// + assert first_call_path == "/custom/root/module/instance/user" @patch.dict(os.environ, {}, clear=True) @patch('os.path.isdir', return_value=True) @@ -215,43 +226,281 @@ def test_default_base_mount_used_when_no_service_binding_root(self, mock_file, m config = SampleConfig() read_from_mount_and_fallback_to_env_var("/etc/secrets/appfnd", "VAR", "module", "instance", config) first_call_path = mock_file.call_args_list[0][0][0] - # Without SERVICE_BINDING_ROOT, only the legacy $ROOT/// path is tried assert first_call_path == "/etc/secrets/appfnd/module/instance/user" - @patch.dict(os.environ, {"SERVICE_BINDING_ROOT": "/bindings"}) + +# --------------------------------------------------------------------------- +# resolve_base_mount +# --------------------------------------------------------------------------- + +class TestResolveBaseMount: + + @patch.dict(os.environ, {"SERVICE_BINDING_ROOT": "/from/env"}) + def test_returns_env_var_when_set(self): + assert resolve_base_mount("/default") == "/from/env" + + @patch.dict(os.environ, {}, clear=True) + def test_returns_default_when_env_var_absent(self): + assert resolve_base_mount("/my/default") == "/my/default" + + @patch.dict(os.environ, {}, clear=True) + def test_returns_module_constant_by_default(self): + from sap_cloud_sdk.core.secret_resolver.constants import BASE_MOUNT_PATH + assert resolve_base_mount() == BASE_MOUNT_PATH + + +# --------------------------------------------------------------------------- +# MountResolver +# --------------------------------------------------------------------------- + +class TestMountResolver: + @patch('os.path.isdir', return_value=True) @patch('os.stat') @patch('builtins.open', new_callable=mock_open) - def test_service_binding_root_flat_path_success(self, mock_file, mock_stat, mock_isdir): + def test_resolve_success(self, mock_file, mock_stat, mock_isdir): mock_file.side_effect = [ - mock_open(read_data="flat_user").return_value, - mock_open(read_data="flat_pass").return_value, - mock_open(read_data="flat_endpoint").return_value, + mock_open(read_data="u").return_value, + mock_open(read_data="p").return_value, + mock_open(read_data="e").return_value, ] config = SampleConfig() - read_from_mount_and_fallback_to_env_var("/etc/secrets/appfnd", "VAR", "module", "instance", config) + MountResolver(base_volume_mount="/secrets").resolve("mod", "inst", config) + + assert config.username == "u" + assert config.password == "p" + assert config.endpoint == "e" + + @patch('os.stat', side_effect=FileNotFoundError("not found")) + def test_resolve_raises_when_dir_missing(self, mock_stat): + config = SampleConfig() + with pytest.raises(FileNotFoundError): + MountResolver(base_volume_mount="/noexist").resolve("mod", "inst", config) + + @patch('os.path.isdir', return_value=False) + @patch('os.stat') + def test_resolve_raises_when_path_not_directory(self, mock_stat, mock_isdir): + config = SampleConfig() + with pytest.raises(NotADirectoryError): + MountResolver(base_volume_mount="/file").resolve("mod", "inst", config) + + @patch.dict(os.environ, {"SERVICE_BINDING_ROOT": "/override"}) + @patch('os.path.isdir', return_value=True) + @patch('os.stat') + @patch('builtins.open', new_callable=mock_open) + def test_service_binding_root_overrides_constructor_path(self, mock_file, mock_stat, mock_isdir): + mock_file.side_effect = [ + mock_open(read_data="x").return_value, + mock_open(read_data="y").return_value, + mock_open(read_data="z").return_value, + ] + config = SampleConfig() + MountResolver(base_volume_mount="/original").resolve("mod", "inst", config) first_call_path = mock_file.call_args_list[0][0][0] - # Flat path $ROOT// is tried first - assert first_call_path == "/bindings/module/user" - assert config.username == "flat_user" + assert first_call_path.startswith("/override/") - @patch.dict(os.environ, {"SERVICE_BINDING_ROOT": "/bindings"}) @patch('os.path.isdir', return_value=True) @patch('os.stat') - def test_service_binding_root_flat_fails_falls_back_to_module_instance(self, mock_stat, mock_isdir): - # Flat path: directory exists but field files are not there (old AppFND structure) - # Full path: files are present under // - flat_not_found = FileNotFoundError("flat file missing") - mock_file_calls = [ - flat_not_found, # flat: /user → not found - mock_open(read_data="legacy_user").return_value, # full: //user - mock_open(read_data="legacy_pass").return_value, # full: //password - mock_open(read_data="legacy_ep").return_value, # full: //endpoint + @patch('builtins.open', new_callable=mock_open) + def test_uses_default_base_mount_path(self, mock_file, mock_stat, mock_isdir): + from sap_cloud_sdk.core.secret_resolver.constants import BASE_MOUNT_PATH + mock_file.side_effect = [ + mock_open(read_data="u").return_value, + mock_open(read_data="p").return_value, + mock_open(read_data="e").return_value, ] - with patch('builtins.open', side_effect=mock_file_calls): - config = SampleConfig() - read_from_mount_and_fallback_to_env_var("/bindings", "VAR", "module", "instance", config) + config = SampleConfig() + with patch.dict(os.environ, {}, clear=True): + MountResolver().resolve("mod", "inst", config) + first_call_path = mock_file.call_args_list[0][0][0] + assert first_call_path.startswith(BASE_MOUNT_PATH) + + +# --------------------------------------------------------------------------- +# EnvVarResolver +# --------------------------------------------------------------------------- + +class TestEnvVarResolver: + + @patch.dict(os.environ, { + "CLOUD_SDK_CFG_MOD_INST_USER": "eu", + "CLOUD_SDK_CFG_MOD_INST_PASSWORD": "ep", + "CLOUD_SDK_CFG_MOD_INST_ENDPOINT": "ee", + }) + def test_resolve_success(self): + config = SampleConfig() + EnvVarResolver().resolve("mod", "inst", config) + + assert config.username == "eu" + assert config.password == "ep" + assert config.endpoint == "ee" + + @patch.dict(os.environ, { + "MYPREFIX_MOD_INST_USER": "cu", + "MYPREFIX_MOD_INST_PASSWORD": "cp", + "MYPREFIX_MOD_INST_ENDPOINT": "ce", + }) + def test_custom_base_var_name(self): + config = SampleConfig() + EnvVarResolver(base_var_name="MYPREFIX").resolve("mod", "inst", config) + + assert config.username == "cu" + + @patch.dict(os.environ, { + "CLOUD_SDK_CFG_MOD_MY_INST_USER": "hu", + "CLOUD_SDK_CFG_MOD_MY_INST_PASSWORD": "hp", + "CLOUD_SDK_CFG_MOD_MY_INST_ENDPOINT": "he", + }) + def test_hyphen_normalization_in_instance(self): + config = SampleConfig() + EnvVarResolver().resolve("mod", "my-inst", config) + + assert config.username == "hu" + + @patch.dict(os.environ, { + "CLOUD_SDK_CFG_MY_MOD_INST_USER": "mu", + "CLOUD_SDK_CFG_MY_MOD_INST_PASSWORD": "mp", + "CLOUD_SDK_CFG_MY_MOD_INST_ENDPOINT": "me", + }) + def test_hyphen_normalization_in_module(self): + config = SampleConfig() + EnvVarResolver().resolve("my-mod", "inst", config) + + assert config.username == "mu" + + @patch.dict(os.environ, {}, clear=True) + def test_resolve_raises_when_var_missing(self): + config = SampleConfig() + with pytest.raises(KeyError, match="env var not found"): + EnvVarResolver().resolve("mod", "inst", config) + + +# --------------------------------------------------------------------------- +# ChainedResolver +# --------------------------------------------------------------------------- + +class TestChainedResolver: + + def test_empty_resolvers_raises(self): + with pytest.raises(ValueError, match="resolvers list must not be empty"): + ChainedResolver([]) + + def test_non_dataclass_target_raises(self): + resolver = ChainedResolver([MountResolver()]) + with pytest.raises(TypeError, match="target must be a dataclass instance"): + resolver.resolve("mod", "inst", "not_a_dataclass") + + def test_non_string_field_raises(self): + resolver = ChainedResolver([MountResolver()]) + config = NonStringConfig() + with pytest.raises(TypeError, match="is not a string"): + resolver.resolve("mod", "inst", config) + + def test_first_resolver_succeeds(self): + first = MagicMock(spec=Resolver) + second = MagicMock(spec=Resolver) + chain = ChainedResolver([first, second]) + config = SampleConfig() + + chain.resolve("mod", "inst", config) + + first.resolve.assert_called_once_with("mod", "inst", config) + second.resolve.assert_not_called() + + def test_falls_back_to_second_on_first_failure(self): + first = MagicMock(spec=Resolver) + first.resolve.side_effect = RuntimeError("first failed") + second = MagicMock(spec=Resolver) + chain = ChainedResolver([first, second]) + config = SampleConfig() + + chain.resolve("mod", "inst", config) + + first.resolve.assert_called_once() + second.resolve.assert_called_once_with("mod", "inst", config) + + def test_raises_runtime_error_when_all_resolvers_fail(self): + first = MagicMock(spec=Resolver) + first.resolve.side_effect = RuntimeError("first failed") + second = MagicMock(spec=Resolver) + second.resolve.side_effect = KeyError("second failed") + chain = ChainedResolver([first, second]) + config = SampleConfig() + + with pytest.raises(RuntimeError, match="failed to read secrets from all resolvers"): + chain.resolve("mod", "inst", config) + + def test_aggregated_error_contains_all_failures(self): + first = MagicMock(spec=Resolver) + first.resolve.side_effect = RuntimeError("mount error") + second = MagicMock(spec=Resolver) + second.resolve.side_effect = KeyError("env error") + chain = ChainedResolver([first, second]) + config = SampleConfig() + + with pytest.raises(RuntimeError) as exc_info: + chain.resolve("mod", "inst", config) + + msg = str(exc_info.value) + assert "mount error" in msg + assert "env error" in msg + + def test_custom_base_var_name_appears_in_error(self): + resolver = MagicMock(spec=Resolver) + resolver.resolve.side_effect = RuntimeError("fail") + chain = ChainedResolver([resolver], base_var_name="MY_APP_CFG") + config = SampleConfig() + + with pytest.raises(RuntimeError, match="MY_APP_CFG"): + chain.resolve("mod", "inst", config) + + +# --------------------------------------------------------------------------- +# SdkConfig / configure / get_resolver / reset_sdk_config +# --------------------------------------------------------------------------- + +class TestSdkConfig: + + def teardown_method(self): + _reset_sdk_config() + + def test_get_sdk_config_is_none_by_default(self): + assert get_sdk_config() is None + + def test_configure_sets_global_config(self): + cfg = SdkConfig() + configure(cfg) + assert get_sdk_config() is cfg + + def test_reset_clears_config(self): + configure(SdkConfig()) + _reset_sdk_config() + assert get_sdk_config() is None + + def test_configure_replaces_previous_config(self): + first = SdkConfig() + second = SdkConfig() + configure(first) + configure(second) + assert get_sdk_config() is second + + def test_get_resolver_returns_default_chain_when_no_config(self): + resolver = get_resolver() + assert isinstance(resolver, ChainedResolver) + + def test_get_resolver_returns_default_chain_when_config_has_no_resolver(self): + configure(SdkConfig(resolver=None)) + resolver = get_resolver() + assert isinstance(resolver, ChainedResolver) + + def test_get_resolver_returns_custom_resolver_when_configured(self): + custom = MagicMock(spec=Resolver) + configure(SdkConfig(resolver=custom)) + assert get_resolver() is custom - assert config.username == "legacy_user" - assert config.password == "legacy_pass" - assert config.endpoint == "legacy_ep" + def test_get_resolver_default_chain_contains_mount_and_env(self): + resolver = get_resolver() + assert isinstance(resolver, ChainedResolver) + # Verify it contains MountResolver and EnvVarResolver by attempting resolution + # (the internal _resolvers list is private, so we check behaviour via the protocol) + assert hasattr(resolver, "resolve")