diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/__init__.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/__init__.py index 73eadc2bd..eb774a9b3 100644 --- a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/__init__.py +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/__init__.py @@ -2,7 +2,38 @@ # Licensed under the MIT License. from ._load_configuration import load_configuration_from_env +from .configuration import Configuration, _Configurable +from ._configuration_protocol import _ConfigurationProtocol, _DictConfigurable +from ._resolve import _resolve +from ._utils import ( + _MissingType, + _MISSING, + _get, + _is_missing, + _path as path, + _xpath as xpath, + _set, + _try_set, +) +from ._coercion import coerce_bool, coerce_int, coerce_float, _get_str __all__ = [ "load_configuration_from_env", + "Configuration", + "_ConfigurationProtocol", + "_MissingType", + "_MISSING", + "_is_missing", + "xpath", + "path", + "_get", + "_set", + "_try_set", + "_resolve", + "_Configurable", + "_DictConfigurable", + "coerce_bool", + "coerce_int", + "coerce_float", + "_get_str", ] diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_coercion.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_coercion.py index bdcea6201..1ba7f9c64 100644 --- a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_coercion.py +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_coercion.py @@ -13,13 +13,34 @@ from __future__ import annotations -from typing import Any +from typing import ( + Any, + Mapping, + overload, +) + +from ._utils import ( + _MISSING, + _MissingType, + _get, + _is_missing, +) _TRUTHY = ("1", "true") _FALSY = ("0", "false") -def coerce_bool(value: Any, default: bool | None = None, name: str = "value") -> bool: +@overload +def _handle_bool( + value: Any, default: None = None, name: str = "value" +) -> bool | None: ... +@overload +def _handle_bool( + value: Any, default: bool | _MissingType = _MISSING, name: str = "value" +) -> bool: ... +def _handle_bool( + value: Any, default: bool | None | _MissingType = None, name: str = "value" +) -> bool | None: """Coerce a config value (possibly an env string) to a bool. Environment variables always arrive as strings, so ``bool("false")`` would @@ -27,31 +48,59 @@ def coerce_bool(value: Any, default: bool | None = None, name: str = "value") -> are ``0/false`` (case-insensitive). An empty/whitespace-only string or ``None`` is treated as *unset* (not as an - explicit ``false``). For unset and unrecognized values: if ``default`` is - provided it is returned, otherwise a ``ValueError`` is raised naming the - offending setting -- mirroring :func:`coerce_int` and :func:`coerce_float`. + explicit ``false``). Unset values return the default, or raise if no default + was supplied. Unrecognized nonempty values always raise ``ValueError``. + Errors identify the setting without including its input value. """ if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) - if value is not None: + if value is not None and not _is_missing(value): text = str(value).strip().lower() if text in _TRUTHY: return True if text in _FALSY: return False if text: - if default is not None: - return default - raise ValueError(f"Invalid boolean for {name}: {value!r}") - # Unset (None or empty/whitespace-only string). - if default is not None: - return default - raise ValueError(f"Invalid boolean for {name}: {value!r}") + raise ValueError(f"Invalid boolean for {name}") + if _is_missing(default): + raise ValueError(f"Missing boolean for {name}") + return default + + +def coerce_bool(value: Any, default: bool | None = None, name: str = "value") -> bool: + """Coerce a config value (possibly an env string) to a bool. + + Environment variables always arrive as strings, so ``bool("false")`` would + be ``True``. Recognized truthy spellings are ``1/true`` and falsy spellings + are ``0/false`` (case-insensitive). + + An empty/whitespace-only string or ``None`` is treated as *unset* (not as an + explicit ``false``). Unset and unrecognized values return the default, or + raise ``ValueError`` if no non-None default was supplied. + Errors identify the setting without including its input value. + """ + new_default = default if default is not None else _MISSING + try: + return _handle_bool(value, new_default, name) + except ValueError: + if default is not None: + return default + raise -def coerce_int(value: Any, default: int, name: str = "value") -> int: +@overload +def _handle_int( + value: Any, default: int | _MissingType = _MISSING, name: str = "value" +) -> int: ... +@overload +def _handle_int( + value: Any, default: None = None, name: str = "value" +) -> int | None: ... +def _handle_int( + value: Any, default: int | None | _MissingType = None, name: str = "value" +) -> int | None: """Coerce a config value (possibly an env string) to an int. Empty/``None`` values fall back to ``default``. Bools are treated as unset @@ -59,41 +108,174 @@ def coerce_int(value: Any, default: int, name: str = "value") -> int: as ``"5"`` or ``"5.0"`` are accepted; anything else raises ``ValueError`` naming the offending setting. """ - if value is None: + if value is None or _is_missing(value): + if _is_missing(default): + raise ValueError(f"Missing integer for {name}") return default if isinstance(value, bool): - return default + if _is_missing(default): + raise ValueError(f"Missing integer for {name}") + return default # TODO -> should this raise? if isinstance(value, int): return value text = str(value).strip() if not text: + if _is_missing(default): + raise ValueError(f"Missing integer for {name}") return default try: return int(text, 10) except ValueError: try: return int(float(text)) - except ValueError as ex: - raise ValueError(f"Invalid integer for {name}: {value!r}") from ex + except ValueError: + raise ValueError(f"Invalid integer for {name}") from None + + +def coerce_int(value: Any, default: int | None = None, name: str = "value") -> int: + """Coerce a config value (possibly an env string) to an int. + + Empty/``None`` values fall back to ``default``. Bools are treated as unset + (a stray ``True``/``False`` is not a meaningful count). Numeric strings such + as ``"5"`` or ``"5.0"`` are accepted; anything else raises ``ValueError`` + naming the offending setting. + """ + new_default = default if default is not None else _MISSING + return _handle_int(value, new_default, name) -def coerce_float(value: Any, default: float, name: str = "value") -> float: +@overload +def _handle_float( + value: Any, default: float | _MissingType, name: str = "value" +) -> float: ... +@overload +def _handle_float( + value: Any, default: None = None, name: str = "value" +) -> float | None: ... +def _handle_float( + value: Any, default: float | None | _MissingType = None, name: str = "value" +) -> float | None: """Coerce a config value (possibly an env string) to a float. Empty/``None`` values fall back to ``default``. Bools are treated as unset. Anything that is not parseable as a float raises ``ValueError`` naming the offending setting. """ - if value is None: + if value is None or _is_missing(value): + if _is_missing(default): + raise ValueError(f"Missing number for {name}") return default if isinstance(value, bool): - return default + if _is_missing(default): + raise ValueError(f"Missing number for {name}") + return default # TODO -> should this also raise? if isinstance(value, (int, float)): return float(value) text = str(value).strip() if not text: + if _is_missing(default): + raise ValueError(f"Missing number for {name}") return default try: return float(text) - except ValueError as ex: - raise ValueError(f"Invalid number for {name}: {value!r}") from ex + except ValueError: + raise ValueError(f"Invalid number for {name}") from None + + +def coerce_float( + value: Any, default: float | None = None, name: str = "value" +) -> float: + """Coerce a config value (possibly an env string) to a float. + + Empty/``None`` values fall back to ``default``. Bools are treated as unset. + Anything that is not parseable as a float raises ``ValueError`` naming the + offending setting. + """ + new_default = default if default is not None else _MISSING + return _handle_float(value, new_default, name) + + +def _check_key_error(value: Any, default: Any, key: str) -> None: + if _is_missing(value) and _is_missing(default): + raise KeyError(key) + + +@overload +def _handle_str( + value: Any, default: None = None, name: str = "value" +) -> str | None: ... +@overload +def _handle_str( + value: Any, default: str | _MissingType = _MISSING, name: str = "value" +) -> str: ... +def _handle_str( + value: Any, default: str | None | _MissingType = None, name: str = "value" +) -> str | None: + if value is None or _is_missing(value): + if _is_missing(default): + raise ValueError(f"Missing string for {name}") + return default + if isinstance(value, str): + return value + return str(value) + + +def _handle_str_list( + value: Any, default: list[str] | None = None, name: str = "value" +) -> list[str]: + if not isinstance(value, list): + raise ValueError(f"{name} must be a list.") + values = [_handle_str(val) for val in value] + return [val for val in values if val is not None] + + +@overload +def _get_bool(mapping: Mapping, key: str, default: bool | _MissingType) -> bool: ... +@overload +def _get_bool(mapping: Mapping, key: str, default: None) -> bool | None: ... +def _get_bool( + mapping: Mapping, key: str, default: bool | None | _MissingType = _MISSING +) -> bool | None: + val = _get(mapping, key, default=_MISSING) + _check_key_error(val, default, key) + return _handle_bool(val, default, key) + + +@overload +def _get_int(mapping: Mapping, key: str, default: int | _MissingType) -> int: ... +@overload +def _get_int(mapping: Mapping, key: str, default: None) -> int | None: ... +def _get_int( + mapping: Mapping, key: str, default: int | None | _MissingType = _MISSING +) -> int | None: + val = _get(mapping, key, default=_MISSING) + _check_key_error(val, default, key) + return _handle_int(val, default, key) + + +@overload +def _get_float(mapping: Mapping, *key: str, default: float | _MissingType) -> float: ... +@overload +def _get_float(mapping: Mapping, *key: str, default: None) -> float | None: ... +def _get_float( + mapping: Mapping, *key: str, default: float | None | _MissingType = _MISSING +) -> float | None: + if not key: + raise ValueError("At least one key must be provided.") + val = _get(mapping, *key, default=_MISSING) + _check_key_error(val, default, key[0]) + return _handle_float(val, default, key[0]) + + +@overload +def _get_str(mapping: Mapping, *key: str, default: str | _MissingType) -> str: ... +@overload +def _get_str(mapping: Mapping, *key: str, default: None) -> str | None: ... +def _get_str( + mapping: Mapping, *key: str, default: str | None | _MissingType = _MISSING +) -> str | None: + if not key: + raise ValueError("At least one key must be provided.") + val = _get(mapping, *key, default=_MISSING) + _check_key_error(val, default, key[0]) + return _handle_str(val, default, key[0]) diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_configuration_protocol.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_configuration_protocol.py new file mode 100644 index 000000000..e77c80602 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_configuration_protocol.py @@ -0,0 +1,35 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from abc import ABC, abstractmethod +from typing import ( + Mapping, + Protocol, + runtime_checkable, +) +from typing_extensions import Self + + +@runtime_checkable +class _ConfigurationProtocol(Protocol): + + @classmethod + def from_mapping(cls, _mapping: Mapping) -> Self: ... + + +class _DictConfigurableProtocol(_ConfigurationProtocol, Protocol): + + def __init__(self, data: dict, *args, **kwargs) -> None: ... + + @classmethod + def from_mapping(cls, mapping: Mapping) -> Self: ... + + +class _DictConfigurable(ABC): + + @abstractmethod + def __init__(self, data: dict, *args, **kwargs) -> None: ... + + @classmethod + def from_mapping(cls, mapping: Mapping) -> Self: + return cls(dict(mapping)) diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_configure_logging.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_configure_logging.py deleted file mode 100644 index 37c64527a..000000000 --- a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_configure_logging.py +++ /dev/null @@ -1,92 +0,0 @@ -# Copyright (c) Microsoft Corporation. All rights reserved. -# Licensed under the MIT License. - -import logging - -from ..errors import configuration_errors - -# in Python 3.11, we can move to using -# logging.getLevelNamesMapping() -_NAME_TO_LEVEL = { - "CRITICAL": logging.CRITICAL, - "FATAL": logging.FATAL, - "ERROR": logging.ERROR, - "WARN": logging.WARNING, - "WARNING": logging.WARNING, - "INFO": logging.INFO, - "INFORMATION": logging.INFO, # .NET parity - "DEBUG": logging.DEBUG, - "NOTSET": logging.NOTSET, -} - - -class ColorFormatter(logging.Formatter): - """Logging formatter that can add ANSI color based on severity.""" - - MAGENTA = "\033[1;35m" - RED = "\033[1;31m" - YELLOW = "\033[1;33m" - GREEN = "\033[1;32m" - BLUE = "\033[1;34m" - RESET = "\033[0m" - - FMT = ( - "%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)" - ) - - def __init__( - self, - use_colors: bool = True, - ) -> None: - super().__init__(fmt=self.FMT) - self._plain_formatter = logging.Formatter(fmt=self.FMT) - if not use_colors: - self._formatters: dict[int, logging.Formatter] = {} - return - self._formatters = { - logging.CRITICAL: logging.Formatter( - f"{self.MAGENTA}{self.FMT}{self.RESET}" - ), - logging.ERROR: logging.Formatter(f"{self.RED}{self.FMT}{self.RESET}"), - logging.WARNING: logging.Formatter(f"{self.YELLOW}{self.FMT}{self.RESET}"), - logging.INFO: logging.Formatter(f"{self.GREEN}{self.FMT}{self.RESET}"), - logging.DEBUG: logging.Formatter(f"{self.BLUE}{self.FMT}{self.RESET}"), - } - - def format(self, record: logging.LogRecord) -> str: - formatter = self._formatters.get(record.levelno) - return (formatter or self._plain_formatter).format(record) - - -def _configure_logging(logging_config: dict): - """Configures logging based on the provided logging configuration dictionary. - - :param logging_config: A dictionary containing logging configuration. - :raises ValueError: If an invalid log level is provided in the configuration. - """ - - log_levels = logging_config.get("LOGLEVEL", {}) - - console_handler = logging.StreamHandler() - console_handler.setFormatter( - ColorFormatter(use_colors=console_handler.stream.isatty()) - ) - - for key in log_levels: - level_name = log_levels[key].upper() - level = _NAME_TO_LEVEL.get(level_name) - if level is None: - raise ValueError( - configuration_errors.InvalidLoggingConfiguration.format(key, level_name) - ) - - namespace = key.lower() - if namespace == "default": - logger = logging.getLogger() - else: - logger = logging.getLogger(namespace) - - logger.propagate = False # Prevent log messages from being propagated - logger.handlers.clear() # Remove existing handlers to prevent duplicates - logger.addHandler(console_handler) - logger.setLevel(level) diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_constants.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_constants.py new file mode 100644 index 000000000..8ea66aba0 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_constants.py @@ -0,0 +1,5 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +_PATH_SEPARATOR = "::" +_PATH_ANY = "*" diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_load_configuration.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_load_configuration.py index 8c851fc85..754732c4f 100644 --- a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_load_configuration.py +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_load_configuration.py @@ -3,7 +3,7 @@ from typing import Any, Mapping -from ._configure_logging import _configure_logging +from ._logging import _configure_logging def load_configuration_from_env(env_vars: Mapping[str, Any]) -> dict: diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/__init__.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/__init__.py new file mode 100644 index 000000000..25d95b305 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/__init__.py @@ -0,0 +1,12 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from ._color_formatter import ColorFormatter +from ._load import _configure_logging +from ._logging_configuration import _LoggingConfiguration + +__all__ = [ + "ColorFormatter", + "_configure_logging", + "_LoggingConfiguration", +] diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_color_formatter.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_color_formatter.py new file mode 100644 index 000000000..9dcee2773 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_color_formatter.py @@ -0,0 +1,42 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import logging + + +class ColorFormatter(logging.Formatter): + """Logging formatter that can add ANSI color based on severity.""" + + MAGENTA = "\033[1;35m" + RED = "\033[1;31m" + YELLOW = "\033[1;33m" + GREEN = "\033[1;32m" + BLUE = "\033[1;34m" + RESET = "\033[0m" + + FMT = ( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)" + ) + + def __init__( + self, + use_colors: bool = True, + ) -> None: + super().__init__(fmt=self.FMT) + self._plain_formatter = logging.Formatter(fmt=self.FMT) + if not use_colors: + self._formatters: dict[int, logging.Formatter] = {} + return + self._formatters = { + logging.CRITICAL: logging.Formatter( + f"{self.MAGENTA}{self.FMT}{self.RESET}" + ), + logging.ERROR: logging.Formatter(f"{self.RED}{self.FMT}{self.RESET}"), + logging.WARNING: logging.Formatter(f"{self.YELLOW}{self.FMT}{self.RESET}"), + logging.INFO: logging.Formatter(f"{self.GREEN}{self.FMT}{self.RESET}"), + logging.DEBUG: logging.Formatter(f"{self.BLUE}{self.FMT}{self.RESET}"), + } + + def format(self, record: logging.LogRecord) -> str: + formatter = self._formatters.get(record.levelno) + return (formatter or self._plain_formatter).format(record) diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_constants.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_constants.py new file mode 100644 index 000000000..9b96888e4 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_constants.py @@ -0,0 +1,20 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import logging + +# in Python 3.11, we can move to using +# logging.getLevelNamesMapping() +_NAME_TO_LEVEL = { + "CRITICAL": logging.CRITICAL, + "FATAL": logging.FATAL, + "ERROR": logging.ERROR, + "WARN": logging.WARNING, + "WARNING": logging.WARNING, + "INFO": logging.INFO, + "INFORMATION": logging.INFO, # .NET parity + "DEBUG": logging.DEBUG, + "NOTSET": logging.NOTSET, +} + +_DEFAULT_NAMESPACE = "default" diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_load.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_load.py new file mode 100644 index 000000000..4f468f023 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_load.py @@ -0,0 +1,38 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import logging +from typing import Mapping + +from ..configuration import Configuration +from ._logging_configuration import _LoggingConfiguration +from ._color_formatter import ColorFormatter +from ._constants import _DEFAULT_NAMESPACE + + +def _configure_logging(_config: Mapping | Configuration | _LoggingConfiguration): + """Configures logging based on the provided logging configuration dictionary. + + :param config: A Configuration object or _LoggingConfiguration instance containing logging configuration. + :raises ValueError: If an invalid log level is provided in the configuration. + """ + + if isinstance(_config, Mapping): + config = _LoggingConfiguration.from_mapping(_config) + else: + config = _LoggingConfiguration.from_configuration(_config) + + console_handler = logging.StreamHandler() + console_handler.setFormatter( + ColorFormatter(use_colors=console_handler.stream.isatty()) + ) + + for entry in config.log_levels: + if entry.namespace == _DEFAULT_NAMESPACE: + logger = logging.getLogger() + else: + logger = logging.getLogger(entry.namespace) + logger.propagate = False # Prevent log messages from being propagated + logger.handlers.clear() # Remove existing handlers to prevent duplicates + logger.addHandler(console_handler) + logger.setLevel(entry.level) diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_logging_configuration.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_logging_configuration.py new file mode 100644 index 000000000..106aecbc7 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_logging/_logging_configuration.py @@ -0,0 +1,57 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Mapping + +from microsoft_agents.activity.errors import configuration_errors + +from ..configuration import Configuration, _Configurable +from .._utils import _xpath +from ._constants import _NAME_TO_LEVEL + + +@dataclass(frozen=True) +class _LogLevelEntry: + """Represents a logging level entry with a namespace and level.""" + + namespace: str + level: int + + def __init__(self, namespace: str, level: str) -> None: + level_upper = level.upper() + if level_upper not in _NAME_TO_LEVEL: + raise ValueError( + configuration_errors.InvalidLoggingConfiguration.format( + namespace, level + ) + ) + object.__setattr__(self, "namespace", namespace.lower()) + object.__setattr__(self, "level", _NAME_TO_LEVEL[level_upper]) + + +@dataclass(frozen=True) +class _LoggingConfiguration(_Configurable): + """Configuration class for logging levels.""" + + log_levels: list[_LogLevelEntry] + + @classmethod + def _from_configuration(cls, config: Configuration) -> _LoggingConfiguration: + + raw_log_levels = config.get(_xpath("log_level"), default={}) + if not isinstance(raw_log_levels, Mapping): + raise ValueError( + configuration_errors.InvalidLoggingConfiguration.format( + "log_levels", raw_log_levels + ) + ) + log_levels: list[_LogLevelEntry] = [ + _LogLevelEntry(namespace, level) + for namespace, level in raw_log_levels.items() + ] + return cls(log_levels) diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_resolve.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_resolve.py new file mode 100644 index 000000000..d6396a758 --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_resolve.py @@ -0,0 +1,96 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from typing import Any, Callable, Mapping + +from ._utils import _MISSING, _get, _parse_env_var_name, _path, _try_set + + +def _expand_paths( + source: Mapping, path_transformer: Callable[[str], str] +) -> dict[str, Any]: + expanded: dict[str, Any] = {} + + def copy_value(value: Any) -> Any: + if isinstance(value, Mapping): + return _expand_paths(value, path_transformer) + if isinstance(value, list): + return [copy_value(item) for item in value] + return value + + def populate(mapping: Mapping, parts: list[str]) -> None: + for key, value in mapping.items(): + if not isinstance(key, (str, int)): + raise TypeError("Configuration keys must be strings or integers") + path = parts + [str(key)] + if isinstance(value, Mapping) and value: + populate(value, path) + continue + name = path_transformer(_path(*path)) + existing = _get(expanded, name, default=_MISSING) + if isinstance(value, Mapping) and isinstance(existing, Mapping): + continue + if not _try_set(expanded, name, copy_value(value)): + raise ValueError(f"Conflicting configuration path {name}") + + populate(source, []) + return expanded + + +def _normalize( + source: Mapping, *, path_transformer: Callable[[str], str] | None = None +) -> dict[str, Any]: + """Expand paths and copy containers, normalizing indexed mappings to lists.""" + transform = path_transformer or _parse_env_var_name + expanded = _expand_paths(source, transform) + + def convert(value: Any) -> Any: + if isinstance(value, dict): + converted = {key: convert(item) for key, item in value.items()} + if converted and all(key.isdecimal() for key in converted): + indices = sorted(converted, key=int) + if len({int(key) for key in indices}) != len(indices): + raise ValueError("Duplicate configuration list indices") + return [converted[key] for key in indices] + return converted + if isinstance(value, list): + return [convert(item) for item in value] + return value + + return {key: convert(value) for key, value in expanded.items()} + + +def _merge( + defaults: dict[str, Any], + primary: dict[str, Any], + *, + _base_path: str = "", +) -> dict[str, Any]: + """Merge owned mappings in place; primary lists replace default lists.""" + for key, value in primary.items(): + if key not in defaults: + defaults[key] = value + continue + previous = defaults[key] + name = _path(_base_path, key) if _base_path else key + if isinstance(previous, dict) and isinstance(value, dict): + _merge(previous, value, _base_path=name) + elif isinstance(previous, dict) or isinstance(value, dict): + raise TypeError(f"Conflicting configuration structure at {name}") + else: + defaults[key] = value + return defaults + + +def _resolve( + primary: Mapping, + defaults: Mapping | None = None, + *, + path_transformer: Callable[[str], str] | None = None, +) -> dict[str, Any]: + """Normalize each source independently, then merge with primary precedence.""" + base = _normalize( + defaults if defaults is not None else {}, path_transformer=path_transformer + ) + overrides = _normalize(primary, path_transformer=path_transformer) + return _merge(base, overrides) diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_utils.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_utils.py new file mode 100644 index 000000000..a0b55b74c --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/_utils.py @@ -0,0 +1,366 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import functools + +from typing import Any, Callable, Mapping, MutableMapping, TypeVar +from typing_extensions import TypeIs + +from ._constants import _PATH_SEPARATOR, _PATH_ANY + +T = TypeVar("T") + + +class _MissingType: + """A special class used to represent missing values in the configuration.""" + + def __bool__(self): + return False + + +_MISSING = _MissingType() # Singleton instance representing a missing value + + +def _is_missing(val: Any) -> TypeIs[_MissingType]: + """Check if a value is the special _MissingType instance. + + :param val: The value to check. + :return: True if the value is an instance of _MissingType, False otherwise. + """ + return isinstance(val, _MissingType) + + +def _path(*args: str) -> str: + """Construct a ``::``-separated key path from the given arguments. + + :param args: The individual parts of the key path. + :return: The ``::``-separated key path. + """ + return _PATH_SEPARATOR.join(args) + + +def _split(key: str) -> list[str]: + return key.split(_PATH_SEPARATOR) + + +def _get_with_path( + mapping: Mapping, + *keys: str, + default: Any = None, + skip_none: bool = False, + none_is_missing: bool = False, +) -> tuple[Any, list[str]]: + """Return the first matched value and its relative source path segments.""" + + final_path: list[str] = [] + none_found: bool = False + for key in keys: + path = _split(key) + current = mapping + break_from_inner: bool = False + path_candidate = [] + for part in path: + if isinstance(current, Mapping) and part in current: + path_candidate.append(part) + current = current[part] + else: + break_from_inner = True + break + if not break_from_inner: + + if current is None: + none_found = True + if not final_path: + final_path = path_candidate + if skip_none: + continue + + return current, path_candidate + + if none_found and not none_is_missing: + return None, final_path + return default, final_path + + +def _get( + mapping: Mapping, + *keys: str, + default: Any = None, + skip_none: bool = False, + none_is_missing: bool = False, +) -> Any: + """Retrieve a value using alternative ``::``-separated key paths. + + :param mapping: The mapping to retrieve the value from. + :param keys: The individual key paths to try. + :param default: The default value to return if the key path is not found. + :param skip_none: Whether to skip None values while traversing the mapping. + :return: The retrieved value or the default value if not found. + """ + value, _ = _get_with_path( + mapping, + *keys, + default=default, + skip_none=skip_none, + none_is_missing=none_is_missing, + ) + return value + + +def _collect_helper( + current: Any, + path_left: list[str], + collection: dict[str, Any], + current_path: list[str], +) -> None: + + if not path_left: + collection[_path(*current_path)] = current + return + + if isinstance(current, Mapping): + + part = path_left[0] + for key in current: + if part == _PATH_ANY or part == key: + _collect_helper( + current[key], path_left[1:], collection, current_path + [key] + ) + if part != _PATH_ANY: + break + if isinstance(current, list): + part = path_left[0] + for idx, item in enumerate(current): + if part == _PATH_ANY or part == str(idx): + _collect_helper( + item, path_left[1:], collection, current_path + [str(idx)] + ) + if part != _PATH_ANY: + break + + +def _collect_with_paths(mapping: Mapping, *keys: str) -> dict[str, Any]: + collection: dict[str, Any] = {} + for key in keys: + _collect_helper(mapping, _split(key), collection, []) + return collection + + +def _collect(mapping: Mapping, *keys: str) -> list[Any]: + return list(_collect_with_paths(mapping, *keys).values()) + + +def _try_set( + mapping: MutableMapping, key: str, value: Any, *, overwrite: bool = False +) -> bool: + """Set a value in a nested mapping using a ``::``-separated key path. + + :param mapping: The mapping to set the value in. + :param key: The ``::``-separated key path. + :param value: The value to set. + :param overwrite: Whether to overwrite the existing value if the key already exists. + """ + path = _split(key) + current = mapping + for part in path[:-1]: + if isinstance(current, MutableMapping): + if part in current: + current = current[part] + else: + current[part] = {} + current = current[part] + else: + raise TypeError( + f"Cannot set value for key path {key} in non-mutable mapping or list object" + ) + + if isinstance(current, MutableMapping): + if overwrite or path[-1] not in current: + current[path[-1]] = value + return True + return False + else: + raise TypeError(f"Cannot set value for key path {key} in non-mutable mapping") + + +def _set(mapping: MutableMapping, key: str, value: Any) -> None: + """Set a value using a ``::``-separated path, overwriting any existing value. + + :param mapping: The mapping to set the value in. + :param key: The ``::``-separated key path. + :param value: The value to set. + """ + _try_set(mapping, key, value, overwrite=True) + + +def _parse_env_var_name(name: str) -> str: + """Replace environment-variable ``__`` separators with ``::``. + + :param name: The environment variable name. + :return: The ``::``-separated key path, preserving case. + """ + return name.replace("__", _PATH_SEPARATOR) + + +@functools.lru_cache(maxsize=10) +def _snake_to_camel(name: str) -> str: + """Convert a snake_case name to PascalCase. + + :param name: The snake_case name. + :return: The PascalCase name. + """ + camel = "" + j = 0 + for word in name.split("_"): + if not word.strip(): + continue + + if j == 0: + camel += word + else: + camel += word.capitalize() + + j += 1 + + return camel + + +@functools.lru_cache(maxsize=10) +def _xcase_helper( + snake_case_name: str, + *, + upper_snake: bool = True, + lower_snake: bool = False, + camel: bool = False, + pascal: bool = False, + all_caps: bool = True, +) -> set[str]: + """Generate a list of name variants for a given snake_case name. + + :param snake_case_name: The snake_case name. + :param upper_snake: Include upper snake case variant. + :param lower_snake: Include lower snake case variant. + :param camel: Include camel case variant. + :param pascal: Include pascal case variant. + :param all_caps: Include all caps variant. + :return: A list of name variants. + """ + variants = set() + + if not snake_case_name.strip(): + return variants + + if upper_snake: + variants.add(snake_case_name.upper()) + if lower_snake: + variants.add(snake_case_name.lower()) + + if camel or pascal or all_caps: + camel_case = _snake_to_camel(snake_case_name) + if camel_case: # a little silly, but "_" becomes "" in the above call. + if camel: + variants.add(camel_case) + if pascal: + variants.add(camel_case[0].capitalize() + camel_case[1:]) + if all_caps: + variants.add(camel_case.upper()) + + return variants + + +def _xcase( + *snake_case_names: str, + upper_snake: bool = True, + lower_snake: bool = False, + camel: bool = False, + pascal: bool = False, + all_caps: bool = True, +) -> list[str]: + """Generate a list of name variants for the given snake_case names. + + :param snake_case_names: The snake_case names. + :param upper_snake: Include upper snake case variant. + :param lower_snake: Include lower snake case variant. + :param camel: Include camel case variant. + :param pascal: Include pascal case variant. + :param all_caps: Include all caps variant. + :return: A list of name variants. + """ + variants = set() + for name in snake_case_names: + variants.update( + _xcase_helper( + name, + upper_snake=upper_snake, + lower_snake=lower_snake, + camel=camel, + pascal=pascal, + all_caps=all_caps, + ) + ) + return list(variants) + + +@functools.lru_cache(maxsize=10) +def _xpath( + *path_parts: str, + upper_snake: bool = True, + lower_snake: bool = True, + camel: bool = False, + pascal: bool = False, + all_caps: bool = True, +) -> list[str]: + """Generate consistently cased variants of a configuration path. + + :param path_parts: Snake_case path components, optionally containing ``::``. + :param upper_snake: Include upper snake case variant. + :param lower_snake: Include lower snake case variant. + :param camel: Include camel case variant. + :param pascal: Include pascal case variant. + :param all_caps: Include all caps variant. + :return: A list of name variants. + """ + + if not path_parts: + return [] + + components = _split(_path(*path_parts)) + styles = { + "upper_snake": upper_snake, + "lower_snake": lower_snake, + "camel": camel, + "pascal": pascal, + "all_caps": all_caps, + } + paths = set() + for style, enabled in styles.items(): + if not enabled: + continue + options = {name: name == style for name in styles} + variants = [_xcase(component, **options) for component in components] + if all(variants): + paths.add(_path(*(variant[0] for variant in variants))) + + return sorted(paths) + + +def str_in(base: str, *sub: str) -> bool: + """Check if all substrings are present in the base string. + + :param base: The base string. + :param sub: Substrings to check for. + :return: True if all substrings are present in the base string, False otherwise. + """ + return any(s in base for s in sub) + + +def _resolve_keys_arg(*keys: str | list[str]) -> list[str]: + resolved_keys = [] + for key in keys: + if isinstance(key, str): + resolved_keys.append(key) + elif isinstance(key, list): + resolved_keys.extend(key) + else: + raise ValueError(f"Invalid key type: {type(key)}") + return resolved_keys diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/config/configuration.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/configuration.py new file mode 100644 index 000000000..ee3b25f9c --- /dev/null +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/config/configuration.py @@ -0,0 +1,431 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import ( + Any, + Callable, + Iterable, + Mapping, + TypeVar, + overload, +) +from typing_extensions import Self + +from ._configuration_protocol import _ConfigurationProtocol +from ._constants import _PATH_SEPARATOR +from ._resolve import _resolve +from ._coercion import ( + _handle_bool, + _handle_int, + _handle_float, + _handle_str, + _handle_str_list, +) +from ._utils import ( + _MissingType, + _MISSING, + _get, + _get_with_path, + _collect, + _collect_with_paths, + _is_missing, + _resolve_keys_arg, +) + +T = TypeVar("T") +_ConfigurationProtocolT = TypeVar( + "_ConfigurationProtocolT", bound=_ConfigurationProtocol +) + + +class Configuration(dict, _ConfigurationProtocol): + """Dictionary-backed configuration with path lookup and typed conversion. + + Paths use ``::`` separators. Alternative paths are tried in the supplied + order. Sections carry the matched source path for diagnostics. + Construction makes a shallow copy; nested mutable values can remain shared. + """ + + def __init__( + self, _original: Mapping | None = None, _base_path: str | None = None, **kwargs + ): + """Initialize a Configuration instance. + + :param _original: The original mapping to initialize the configuration from. + :param _base_path: The base path for resolving nested configuration keys. + """ + if _original is not None and kwargs: + raise ValueError("Cannot provide both '_original' and keyword arguments.") + super().__init__(_original or kwargs) + self._base_path = _base_path or "" + + @classmethod + def from_mapping( + cls, _mapping: Mapping | None = None, _base_path: str | None = None, **kwargs + ) -> Configuration: + """Create configuration, using keyword arguments as fallback values. + + An existing Configuration is reused unless defaults or a different + diagnostic base path are supplied. + """ + if ( + isinstance(_mapping, Configuration) + and not kwargs + and (_base_path is None or _base_path == _mapping._base_path) + ): + return _mapping + if kwargs: + if not _mapping: + resolved = kwargs + else: + resolved = _resolve(_mapping, kwargs, path_transformer=lambda x: x) + else: + resolved = _mapping + return cls(resolved, _base_path=_base_path) + + @classmethod + def from_config(self, base: Configuration | None, **kwargs) -> Configuration: + """Create a new Configuration instance by forking the current configuration with additional key-value pairs.""" + if base is None: + return Configuration(**kwargs) + new_config = base.copy() + return Configuration.from_mapping( + new_config, _base_path=base._base_path, **kwargs + ) + + def _resolve_path(self, key: str) -> str: + """Resolve a configuration path by combining the base path and the key.""" + return f"{self._base_path}{_PATH_SEPARATOR}{key}" if self._base_path else key + + def _get_with_path( + self, *_keys: str | list[str], default: Any = _MISSING, skip_none: bool = False + ) -> tuple[Any, list[str]]: + """Return a value and its matched path segments, relative to this section.""" + keys = _resolve_keys_arg(*_keys) + res, path = _get_with_path(self, *keys, default=_MISSING, skip_none=skip_none) + if _is_missing(res): + if _is_missing(default): + raise KeyError(self._resolve_path(keys[0])) + return default, path + return res, path + + def _get( + self, *_keys: str | list[str], default: Any = _MISSING, skip_none: bool = False + ) -> Any: + """Return the first matched value, raising if absent and no default is given.""" + val, _ = self._get_with_path(*_keys, default=default, skip_none=skip_none) + return val + + def get( + self, key: str | list[str], default: Any = None, skip_none: bool = False + ) -> Any: + """Retrieve a value from the configuration by key, returning the default if not found.""" + keys = _resolve_keys_arg(key) + try: + return self._get(*keys, default=default, skip_none=skip_none) + except KeyError: + return default + + def has(self, *keys: str | list[str]) -> bool: + """Check if the configuration contains the specified keys. + + :param keys: The keys to check. + :return: True if the keys exist in the configuration, False otherwise. + """ + resolved_keys = _resolve_keys_arg(*keys) + try: + self._get(*resolved_keys) + return True + except KeyError: + return False + + @overload + def get_section( + self, + *section: str | list[str], + default: Configuration | _MissingType = _MISSING, + ) -> Configuration: ... + @overload + def get_section( + self, *section: str | list[str], default: None + ) -> Configuration | None: ... + def get_section( + self, + *section: str | list[str], + default: Configuration | None | _MissingType = _MISSING, + ) -> Configuration | None: + """Retrieve a section of the configuration, returning it as a Configuration object. + + :param section: Alternative section paths, tried in order. + :param default: The default value to return if the section is not found. + :return: The retrieved section as a Configuration object, or the default value if not found. + :raises KeyError: If the section is not found and no default is provided. + :raises TypeError: If the section is found but is not a mapping. + """ + keys = _resolve_keys_arg(*section) + val, path = self._get_with_path(*keys, default=default) + if not path: + return val + base_path = self._resolve_path(_PATH_SEPARATOR.join(path)) + + if isinstance(val, Mapping): + return Configuration.from_mapping(val, _base_path=base_path) + if val is None and not _is_missing(default): + return default + raise TypeError(f"Section {base_path} is not a mapping") + + def get_sections(self, *section: str) -> list[Configuration]: + """Retrieve all subsections of the specified section as Configuration objects. + + :param section: The names of the nested sections to retrieve. + :return: A list of Configuration objects representing the subsections. + :raises TypeError: If the section is not a mapping. + """ + map = _collect_with_paths(self, *section) + + if not map: + return [] + + for base_path, v in map.items(): + if not isinstance(v, Mapping): + raise TypeError( + f"Section {self._resolve_path(base_path)} is not a mapping" + ) + + return [ + Configuration.from_mapping(v, _base_path=self._resolve_path(base_path)) + for base_path, v in map.items() + ] + + @overload + def get_bool( + self, *_keys: str | list[str], default: bool | _MissingType = _MISSING + ) -> bool: ... + @overload + def get_bool( + self, *_keys: str | list[str], default: None = None + ) -> bool | None: ... + def get_bool( + self, *_keys: str | list[str], default: bool | None | _MissingType = _MISSING + ) -> bool | None: + """Retrieve a boolean value from the configuration. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved boolean value or the default value if not found. + """ + keys = _resolve_keys_arg(*_keys) + val, path = self._get_with_path(*keys, default=default, skip_none=True) + name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0]) + return _handle_bool(val, default, name) + + def get_booln( + self, *_keys: str | list[str], default: bool | None = None + ) -> bool | None: + """Retrieve a boolean value from the configuration, allowing None as default. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved boolean value or the default value if not found. + """ + return self.get_bool(*_keys, default=default) + + @overload + def get_int(self, *_keys: str | list[str], default: None = None) -> int | None: ... + @overload + def get_int( + self, *_keys: str | list[str], default: int | _MissingType = _MISSING + ) -> int: ... + def get_int( + self, *_keys: str | list[str], default: int | None | _MissingType = _MISSING + ) -> int | None: + """Retrieve an integer value from the configuration. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved integer value or the default value if not found. + """ + keys = _resolve_keys_arg(*_keys) + val, path = self._get_with_path(*keys, default=default, skip_none=True) + name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0]) + return _handle_int(val, default, name) + + def get_intn( + self, *_keys: str | list[str], default: int | None = None + ) -> int | None: + """Retrieve an integer value from the configuration, allowing None as default. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved integer value or the default value if not found. + """ + return self.get_int(*_keys, default=default) + + @overload + def get_float( + self, *_keys: str | list[str], default: None = None + ) -> float | None: ... + @overload + def get_float( + self, *_keys: str | list[str], default: float | _MissingType = _MISSING + ) -> float: ... + def get_float( + self, *_keys: str | list[str], default: float | None | _MissingType = _MISSING + ) -> float | None: + """Retrieve a float value from the configuration. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved float value or the default value if not found. + """ + keys = _resolve_keys_arg(*_keys) + val, path = self._get_with_path(*keys, default=default, skip_none=True) + name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0]) + return _handle_float(val, default, name) + + def get_floatn( + self, *_keys: str | list[str], default: float | None = None + ) -> float | None: + """Retrieve a float value from the configuration, allowing None as default. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved float value or the default value if not found. + """ + return self.get_float(*_keys, default=default) + + @overload + def get_str(self, *_keys: str | list[str], default: None) -> str | None: ... + @overload + def get_str( + self, *_keys: str | list[str], default: str | _MissingType = _MISSING + ) -> str: ... + def get_str( + self, *_keys: str | list[str], default: str | None | _MissingType = _MISSING + ) -> str | None: + """Retrieve a string value from the configuration. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved string value or the default value if not found. + """ + keys = _resolve_keys_arg(*_keys) + val, path = self._get_with_path(*keys, default=default, skip_none=True) + name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0]) + return _handle_str(val, default, name) + + def get_strn(self, *_keys: str | list[str], default: str | None = None): + """Retrieve a string, allowing None as the default. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved string or the default value if not found. + """ + return self.get_str(*_keys, default=default) + + def get_str_list( + self, *_keys: str | list[str], default: list[str] | None = None + ) -> list[str]: + """Retrieve a list of strings from the configuration. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved list of strings or the default value if not found. + """ + keys = _resolve_keys_arg(*_keys) + val, path = self._get_with_path(*keys, default=default, skip_none=True) + if val is None: + if default is None: + return [] + return default + if isinstance(val, list): + name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0]) + return _handle_str_list(val, default, name) + elif isinstance(val, Mapping): + raise ValueError( + f"Invalid list value for {self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0])}" + ) + elif not isinstance(val, Mapping): + name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0]) + before_split = _handle_str(val, None, name) + if before_split is None: + return default + if "," in before_split: + return [s.strip() for s in before_split.split(",")] + return [s.strip() for s in before_split.strip().split()] + + def get_items( + self, *_keys: str | list[str], default: dict | None = None + ) -> Iterable[tuple[Any, Any]]: + """Retrieve items from a dictionary in the configuration. + + :param keys: The keys of the value to retrieve. + :param default: The default value to return if the key is not found. + :return: The retrieved items as a list of key-value tuples or the default value if not found. + """ + keys = _resolve_keys_arg(*_keys) + base = _get(self, *keys, default=default) + if isinstance(base, Mapping): + return base.items() + elif isinstance(base, list): + return enumerate(base) + return [] + + def collect(self, *_keys: str | list[str]) -> list[Any]: + keys = _resolve_keys_arg(*_keys) + return _collect(self, *keys) + + def collect_str(self, *_keys: str | list[str]) -> list[str]: + keys = _resolve_keys_arg(*_keys) + vals = _collect(self, *keys) + vals = [_handle_str(v, None, keys[0]) for v in vals if v is not None] + return [v for v in vals if v is not None] + + def get_scopes(self, *_keys: str | list[str]) -> list[str]: + """Retrieve a list of scopes from the configuration. + + :param keys: The keys of the scopes value to retrieve. + :return: The scopes as a list, or an empty list if absent. + """ + keys = _resolve_keys_arg(*_keys) + val, path = self._get_with_path(*keys, default=None, skip_none=True) + if val is None: + return [] + if isinstance(val, list): + return val + if isinstance(val, str): + return Configuration.format_scopes(val) + name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0]) + raise ValueError(f"Invalid scopes value for {name}") + + @staticmethod + def format_scopes(scopes: str | list[str] | None) -> list[str]: + """Format scopes into a list of strings. If scopes is a string, it will be split by whitespace. If scopes is None, an empty list is returned.""" + if scopes is None: + return [] + if isinstance(scopes, str): + lst = scopes.strip().split() + return [s for s in lst if s] + return scopes + + +class _Configurable(ABC): + + @classmethod + def from_mapping(cls, _mapping: Mapping | None = None, **kwargs) -> Self: + return cls.from_configuration(Configuration.from_mapping(_mapping, **kwargs)) + + @classmethod + @abstractmethod + def _from_configuration(cls, config: Configuration) -> Self: ... + + @classmethod + def from_configuration(cls, config: Configuration | Self) -> Self: + if isinstance(config, cls): + return config + if isinstance(config, Configuration): + return cls._from_configuration(config) + raise TypeError(f"Cannot create {cls.__name__} from {type(config).__name__}") diff --git a/pytest.ini b/pytest.ini index b26925549..7e79e6e61 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,5 +1,6 @@ [pytest] # Pytest configuration for Microsoft Agents for Python +asyncio_default_fixture_loop_scope = function # Treat all warnings as errors by default # This ensures that any code generating warnings will fail tests, diff --git a/tests/activity/config/test_configurable.py b/tests/activity/config/test_configurable.py new file mode 100644 index 000000000..8d84fc318 --- /dev/null +++ b/tests/activity/config/test_configurable.py @@ -0,0 +1,60 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from dataclasses import dataclass +from types import MappingProxyType + +import pytest + +from microsoft_agents.activity.config import Configuration, _Configurable + + +@dataclass(frozen=True) +class ApplicationSettings(_Configurable): + enabled: bool + count: int + + @classmethod + def _from_configuration(cls, config: Configuration): + return cls( + enabled=config.get_bool("enabled", default=False), + count=config.get_int("count", default=0), + ) + + +def test_binds_a_configuration_section(): + root = Configuration({"app": {"enabled": "true", "count": "4"}}) + + settings = ApplicationSettings.from_configuration(root.get_section("app")) + + assert settings == ApplicationSettings(enabled=True, count=4) + + +def test_from_mapping_accepts_read_only_mappings_and_defaults(): + source = MappingProxyType({"enabled": "false"}) + + settings = ApplicationSettings.from_mapping(source, enabled=True, count=3) + + assert settings == ApplicationSettings(enabled=False, count=3) + assert dict(source) == {"enabled": "false"} + + +def test_from_configuration_reuses_bound_instances(): + settings = ApplicationSettings(enabled=True, count=2) + + assert ApplicationSettings.from_configuration(settings) is settings + + +@pytest.mark.parametrize("value", [{}, None, "invalid", 3]) +def test_from_configuration_rejects_unbound_values(value): + with pytest.raises(TypeError, match="ApplicationSettings"): + ApplicationSettings.from_configuration(value) + + +def test_binding_preserves_section_paths_in_conversion_errors(): + root = Configuration({"app": {"count": "synthetic-secret"}}) + + with pytest.raises(ValueError, match="app::count") as error: + ApplicationSettings.from_configuration(root.get_section("app")) + + assert "synthetic-secret" not in str(error.value) diff --git a/tests/activity/config/test_configuration.py b/tests/activity/config/test_configuration.py new file mode 100644 index 000000000..50523edce --- /dev/null +++ b/tests/activity/config/test_configuration.py @@ -0,0 +1,325 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import traceback + +import pytest + +from microsoft_agents.activity.config import Configuration, path, xpath + + +def test_section_uses_matched_alias_and_nested_path(): + config = Configuration( + {"Connections": {"Service": {"SETTINGS": {"Enabled": True}}}} + ) + + connection = config.get_section(["CONNECTIONS", "Connections"]) + settings = connection.get_section(["Service::settings", "Service::SETTINGS"]) + + assert connection._base_path == "Connections" + assert settings._base_path == "Connections::Service::SETTINGS" + assert settings.get_bool("Enabled") is True + + +def test_existing_configuration_section_is_rebased_without_modifying_original(): + original = Configuration({"Enabled": True}, _base_path="Previous") + config = Configuration({"Settings": original}, _base_path="Service") + + section = config.get_section("Settings") + + assert section == original + assert section is not original + assert section._base_path == "Service::Settings" + assert original._base_path == "Previous" + + +def test_from_mapping_reuses_configuration_when_base_path_is_unchanged(): + original = Configuration({"Enabled": True}, _base_path="Service") + + assert Configuration.from_mapping(original) is original + assert Configuration.from_mapping(original, _base_path="Service") is original + + +def test_from_mapping_can_rebase_to_root(): + original = Configuration({"Enabled": True}, _base_path="Service") + + rebased = Configuration.from_mapping(original, _base_path="") + + assert rebased._base_path == "" + assert original._base_path == "Service" + + +def test_missing_section_preserves_default_configuration_identity(): + config = Configuration(_base_path="Service") + default = Configuration({"Enabled": True}, _base_path="Fallback") + + assert config.get_section("Missing", default=default) is default + assert default._base_path == "Fallback" + assert config.get_section("Missing", default=None) is None + + +@pytest.mark.parametrize("getter", ["get_section", "get_int"]) +def test_missing_required_value_reports_full_path(getter): + config = Configuration({"Settings": {}}).get_section("Settings") + + with pytest.raises(KeyError) as error: + getattr(config, getter)("Missing") + + assert error.value.args == ("Settings::Missing",) + + +def test_non_mapping_section_reports_matched_alias_without_its_value(): + config = Configuration({"Settings": {"TOKEN": "synthetic-secret"}}) + settings = config.get_section("Settings") + + with pytest.raises(TypeError) as error: + settings.get_section("token", "TOKEN") + + assert str(error.value) == "Section Settings::TOKEN is not a mapping" + + +@pytest.mark.parametrize( + "getter, value", + [ + ("get_bool", "synthetic-secret"), + ("get_int", "synthetic-secret"), + ("get_float", "synthetic-secret"), + ("get_str_list", {"token": "synthetic-secret"}), + ("get_scopes", {"token": "synthetic-secret"}), + ], +) +def test_conversion_errors_report_matched_full_path_without_input(getter, value): + config = Configuration({"Settings": {"VALUE": value}}) + settings = config.get_section("Settings") + + with pytest.raises(ValueError) as error: + getattr(settings, getter)("value", "VALUE") + + assert "Settings::VALUE" in str(error.value) + assert "Settings::value" not in str(error.value) + assert "synthetic-secret" not in "".join( + traceback.format_exception_only(type(error.value), error.value) + ) + assert error.value.__cause__ is None + if error.value.__context__ is not None: + assert error.value.__suppress_context__ + + +@pytest.mark.parametrize( + "getter, value, expected", + [ + ("get_bool", "false", False), + ("get_int", "42", 42), + ("get_float", "1.5", 1.5), + ("get_str", 42, "42"), + ("get_str_list", ["read", "write"], ["read", "write"]), + ("get_scopes", "read write", ["read", "write"]), + ], +) +def test_diagnostic_paths_preserve_successful_conversion(getter, value, expected): + settings = Configuration({"Settings": {"VALUE": value}}).get_section("Settings") + + assert getattr(settings, getter)("value", "VALUE") == expected + + +@pytest.mark.parametrize( + "getter, default", + [ + ("get_bool", False), + ("get_int", 5), + ("get_float", 1.5), + ("get_str", "fallback"), + ("get_str_list", ["fallback"]), + ], +) +def test_conversion_preserves_missing_defaults(getter, default): + settings = Configuration(_base_path="Settings") + + assert getattr(settings, getter)("Missing", default=default) == default + + +@pytest.mark.parametrize( + "value, expected", + [ + (["read", "write"], ["read", "write"]), + ("read write", ["read", "write"]), + (" read\twrite\n", ["read", "write"]), + ("read,write", ["read", "write"]), + (" read, write ", ["read", "write"]), + ("", []), + (" ", []), + (None, []), + ], +) +def test_string_lists_bind_supported_formats(value, expected): + config = Configuration({"VALUES": value}) + + assert config.get_str_list("VALUES") == expected + + +def test_string_list_rejects_unresolved_indexed_mapping_even_with_default(): + config = Configuration({"VALUES": {"0": "read"}}, _base_path="Settings") + + with pytest.raises(ValueError, match="Settings::VALUES"): + config.get_str_list("VALUES", default=["fallback"]) + + +@pytest.mark.parametrize("getter", ["get_bool", "get_booln"]) +@pytest.mark.parametrize("default", [True, False, None]) +@pytest.mark.parametrize("value", ["yes", "on", "no", "off", "maybe"]) +def test_boolean_getters_reject_unrecognized_values_with_defaults( + getter, default, value +): + settings = Configuration({"Enabled": value}, _base_path="Settings") + + with pytest.raises(ValueError, match="Invalid boolean for Settings::Enabled"): + getattr(settings, getter)("Enabled", default=default) + + +def test_null_value_preserves_matched_path_when_skipped(): + config = Configuration({"Settings": {"VALUE": None}}) + settings = config.get_section("Settings") + + value, matched_path = settings._get_with_path( + "value", "VALUE", default=None, skip_none=True + ) + + assert value is None + assert matched_path == ["VALUE"] + assert settings.get_int("value", "VALUE", default=5) == 5 + + +def test_collected_sections_rebase_existing_configurations(): + original = Configuration({"Enabled": True}, _base_path="Previous") + config = Configuration({"Entries": {"Service": original}}, _base_path="Root") + + sections = config.get_sections(path("Entries", "*")) + + assert len(sections) == 1 + assert sections[0]._base_path == "Root::Entries::Service" + assert original._base_path == "Previous" + + +def test_non_mapping_collected_section_reports_full_path(): + config = Configuration( + {"Entries": {"Service": "synthetic-secret"}}, _base_path="Root" + ) + + with pytest.raises(TypeError) as error: + config.get_sections(path("Entries", "*")) + + assert str(error.value) == "Section Root::Entries::Service is not a mapping" + + +def test_configuration_is_a_dictionary_with_positional_default_lookup(): + config = Configuration({"Enabled": False, "Optional": None}) + fallback = object() + + assert isinstance(config, dict) + assert config.get("Enabled", fallback) is False + assert config.get("Optional", fallback) is None + assert config.get("Missing", fallback) is fallback + assert config.get(["Missing", "Enabled"], fallback) is False + + +def test_from_mapping_merges_defaults_without_mutating_inputs(): + primary = {"Settings": {"Enabled": False}} + defaults = {"Settings": {"Enabled": True, "Count": 3}} + + config = Configuration.from_mapping(primary, **defaults) + config["Settings"]["Count"] = 4 + + assert config.get_bool("Settings::Enabled") is False + assert primary == {"Settings": {"Enabled": False}} + assert defaults == {"Settings": {"Enabled": True, "Count": 3}} + + +def test_from_config_forks_top_level_values_and_preserves_base_path(): + original = Configuration({"Enabled": False}, _base_path="Service") + + fork = Configuration.from_config(original, Enabled=True, Count=3) + fork["Enabled"] = True + + assert fork is not original + assert fork._base_path == "Service" + assert fork["Count"] == 3 + assert original["Enabled"] is False + assert "Count" not in original + + +def test_from_config_preserves_primary_values_over_fallbacks(): + original = Configuration({"Enabled": False}) + + assert Configuration.from_config(original, Enabled=True)["Enabled"] is False + + +def test_constructor_makes_a_shallow_copy(): + source = {"Values": ["original"]} + config = Configuration(source) + + source["New"] = True + source["Values"].append("shared") + + assert "New" not in config + assert config["Values"] == ["original", "shared"] + + +@pytest.mark.parametrize( + "entries", + [ + [("client_id", "lower"), ("CLIENT_ID", "upper"), ("CLIENTID", "compact")], + [("CLIENTID", "compact"), ("CLIENT_ID", "upper"), ("client_id", "lower")], + ], +) +def test_alias_precedence_does_not_depend_on_insertion_order(entries): + assert Configuration(dict(entries)).get_str(xpath("client_id")) == "compact" + + +def test_xpath_components_form_one_nested_path(): + config = Configuration( + {"AGENTAPPLICATION": {"USERAUTHORIZATION": {"Enabled": True}}} + ) + + section = config.get_section(xpath("agent_application", "user_authorization")) + + assert section.get_bool("Enabled") is True + assert section._base_path == "AGENTAPPLICATION::USERAUTHORIZATION" + + +@pytest.mark.parametrize("values", [["first", "second"], {"0": "first", "1": "second"}]) +def test_collect_str_selects_children_of_lists_and_indexed_mappings(values): + config = Configuration({"audiences": values}) + + assert config.collect_str("audiences::*") == ["first", "second"] + + +def test_collect_does_not_implicitly_flatten_selected_containers(): + values = ["first", "second"] + config = Configuration({"audiences": values}) + + assert config.collect("audiences") == [values] + assert config.collect_str("audiences") == [str(values)] + assert config.collect_str("audiences::*") == values + + +def test_collect_deduplicates_overlapping_paths_not_equal_values(): + config = Configuration({"Values": ["same", "same"]}) + + assert config.collect_str("Values::*", "Values::0") == ["same", "same"] + + +@pytest.mark.parametrize( + "value, expected", + [ + (None, []), + ("", []), + (" ", []), + ([], []), + ("read\twrite\n", ["read", "write"]), + (["read", "write"], ["read", "write"]), + ], +) +def test_scopes_bind_supported_list_and_whitespace_formats(value, expected): + config = Configuration({"SCOPES": value}) + + assert config.get_scopes("SCOPES") == expected diff --git a/tests/activity/config/test_logging_configuration.py b/tests/activity/config/test_logging_configuration.py new file mode 100644 index 000000000..3931cf85c --- /dev/null +++ b/tests/activity/config/test_logging_configuration.py @@ -0,0 +1,34 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import logging + +import pytest + +from microsoft_agents.activity.config._logging._logging_configuration import ( + _LoggingConfiguration, +) + + +def test_missing_logging_configuration_has_no_level_entries(): + assert _LoggingConfiguration.from_mapping({}).log_levels == [] + + +@pytest.mark.parametrize("key", ["log_level", "LOG_LEVEL", "LOGLEVEL"]) +def test_logging_level_aliases_bind_namespace_and_level(key): + settings = _LoggingConfiguration.from_mapping({key: {"My.Namespace": "debug"}}) + + assert len(settings.log_levels) == 1 + assert settings.log_levels[0].namespace == "my.namespace" + assert settings.log_levels[0].level == logging.DEBUG + + +@pytest.mark.parametrize("value", ["DEBUG", [], 1]) +def test_non_mapping_log_levels_are_rejected(value): + with pytest.raises(ValueError): + _LoggingConfiguration.from_mapping({"LOGLEVEL": value}) + + +def test_unrecognized_log_level_is_rejected(): + with pytest.raises(ValueError): + _LoggingConfiguration.from_mapping({"LOGLEVEL": {"default": "not-a-level"}}) diff --git a/tests/activity/config/test_resolve.py b/tests/activity/config/test_resolve.py new file mode 100644 index 000000000..02d50e5c8 --- /dev/null +++ b/tests/activity/config/test_resolve.py @@ -0,0 +1,119 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from copy import deepcopy +from types import MappingProxyType + +import pytest + +from microsoft_agents.activity.config import Configuration +from microsoft_agents.activity.config._resolve import _normalize, _resolve + + +@pytest.mark.parametrize( + "indices", + [ + {"10": "ten", "2": "two", "0": "zero"}, + {10: "ten", 2: "two", 0: "zero"}, + ], +) +def test_normalize_sorts_indexed_mappings_numerically(indices): + source = {"Values": MappingProxyType(indices), "Empty": {}, "List": []} + + assert _normalize(MappingProxyType(source)) == { + "Values": ["zero", "two", "ten"], + "Empty": {}, + "List": [], + } + + +def test_normalize_rejects_duplicate_list_indices(): + with pytest.raises(ValueError, match="Duplicate configuration list indices"): + _normalize({"Values": {"0": "first", "00": "second"}}) + + +@pytest.mark.parametrize( + "primary", + [ + {"Scopes": ["admin"]}, + {"Scopes": {"0": "admin"}}, + {"Scopes__0": "admin"}, + ], +) +@pytest.mark.parametrize( + "defaults", + [ + {"Scopes": ["read", "write"]}, + {"Scopes__0": "read", "Scopes__1": "write"}, + ], +) +def test_resolve_replaces_whole_lists_regardless_of_representation(primary, defaults): + original_primary = deepcopy(primary) + original_defaults = deepcopy(defaults) + + assert _resolve(primary, defaults) == {"Scopes": ["admin"]} + assert primary == original_primary + assert defaults == original_defaults + + +def test_resolve_replaces_list_entries_without_inheriting_nested_defaults(): + defaults = { + "App": {"Entries": [{"Name": "default", "Scopes": ["read"]}], "Keep": 1} + } + primary = {"App": {"Entries": [{"Name": "custom"}]}} + + assert _resolve(primary, defaults) == { + "App": {"Entries": [{"Name": "custom"}], "Keep": 1} + } + + +def test_resolve_empty_primary_list_clears_indexed_defaults(): + assert _resolve({"Scopes": []}, {"Scopes__0": "read", "Scopes__1": "write"}) == { + "Scopes": [] + } + + +@pytest.mark.parametrize("as_defaults", [True, False]) +@pytest.mark.parametrize( + "entries", + [ + [{"Scopes": {"0": "read"}}], + [{"Groups": [{"Scopes": {"0": "read"}}]}], + [[{"Scopes": {"0": "read"}}]], + ], +) +def test_configuration_resolution_copies_all_nested_containers(as_defaults, entries): + source = {"Entries": entries} + original = deepcopy(source) + if as_defaults: + result = Configuration.from_mapping({"Unrelated": True}, **source) + else: + result = Configuration.from_mapping(source, Unrelated=True) + + assert source == original + assert result["Entries"] is not entries + if isinstance(entries[0], list): + scopes = result["Entries"][0][0]["Scopes"] + elif "Groups" in entries[0]: + scopes = result["Entries"][0]["Groups"][0]["Scopes"] + else: + scopes = result["Entries"][0]["Scopes"] + assert scopes == ["read"] + scopes.append("write") + assert source == original + + +def test_normalize_copies_read_only_mappings_inside_nested_lists(): + scopes = {"0": "read"} + source = { + "Entries": [ + [MappingProxyType({"Scopes": MappingProxyType(scopes)})], + [], + ] + } + + result = _normalize(source) + + assert result == {"Entries": [[{"Scopes": ["read"]}], []]} + result["Entries"][0][0]["Scopes"].append("write") + assert scopes == {"0": "read"} diff --git a/tests/activity/config/test_utils.py b/tests/activity/config/test_utils.py new file mode 100644 index 000000000..bf61a3915 --- /dev/null +++ b/tests/activity/config/test_utils.py @@ -0,0 +1,388 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from copy import deepcopy +from types import MappingProxyType + +import pytest + +from microsoft_agents.activity.config._resolve import _resolve +from microsoft_agents.activity.config._utils import ( + _MISSING, + _get, + _is_missing, + _parse_env_var_name, + _path, + _set, + _try_set, + _collect, + _xpath, +) + + +@pytest.mark.parametrize("value", [None, False, 0, "", {}, []]) +def test_is_missing_distinguishes_unset_from_present_values(value): + assert _is_missing(_MISSING) + assert not _is_missing(value) + + +def test_path_joins_segments(): + assert ( + _path("Connections", "Service", "Settings") == "Connections::Service::Settings" + ) + assert _path("Connections") == "Connections" + assert _path() == "" + + +def test_parse_env_var_name_preserves_single_underscores_and_case(): + assert ( + _parse_env_var_name("Connections__SERVICE_CONNECTION__CLIENT_ID") + == "Connections::SERVICE_CONNECTION::CLIENT_ID" + ) + + +@pytest.mark.parametrize("value", [None, False, 0, "", [], {"C": "value"}]) +def test_get_returns_raw_values(value): + mapping = MappingProxyType({"A": MappingProxyType({"B": value})}) + + assert _get(mapping, "A::B", default=_MISSING) is value + + +@pytest.mark.parametrize("path", ["Missing", "A::Missing", "A::B::C"]) +def test_get_returns_default_for_missing_or_non_mapping_path(path): + assert _get({"A": {"B": "scalar"}}, path, default=_MISSING) is _MISSING + assert _get({"A": {"B": "scalar"}}, path) is None + + +def test_try_set_creates_nested_mappings(): + mapping = {} + + assert _try_set(mapping, "A::B::C", "value") is True + assert mapping == {"A": {"B": {"C": "value"}}} + + +@pytest.mark.parametrize("existing", [None, False, 0, "", {}, []]) +def test_try_set_does_not_overwrite_existing_values_by_default(existing): + mapping = {"A": {"B": existing}} + + assert _try_set(mapping, "A::B", "replacement") is False + assert mapping["A"]["B"] is existing + + +def test_try_set_can_explicitly_overwrite(): + mapping = {"A": {"B": "old"}} + + assert _try_set(mapping, "A::B", "new", overwrite=True) is True + assert mapping == {"A": {"B": "new"}} + + +@pytest.mark.parametrize("path", ["A::B", "A::B::C"]) +def test_try_set_rejects_traversal_through_scalar(path): + mapping = {"A": "scalar"} + + with pytest.raises(TypeError, match=path): + _try_set(mapping, path, "value") + + assert mapping == {"A": "scalar"} + + +def test_try_set_rejects_read_only_mapping(): + mapping = {"A": MappingProxyType({"B": "old"})} + + with pytest.raises(TypeError, match="A::B"): + _try_set(mapping, "A::B", "new") + + assert mapping["A"]["B"] == "old" + + +def test_set_overwrites_existing_leaf_and_preserves_siblings(): + mapping = {"A": {"B": "old", "C": "keep"}} + + _set(mapping, "A::B", "new") + + assert mapping == {"A": {"B": "new", "C": "keep"}} + + +@pytest.mark.parametrize( + "primary", + [ + {"App": {"Enabled": "new"}}, + {"App__Enabled": "new"}, + {"App::Enabled": "new"}, + ], +) +def test_resolve_merges_equivalent_path_representations(primary): + defaults = {"App": {"Enabled": "old", "Name": "keep"}} + + assert _resolve(primary, defaults) == {"App": {"Enabled": "new", "Name": "keep"}} + + +@pytest.mark.parametrize("value", [None, False, 0, "", "false", "123"]) +def test_resolve_preserves_explicit_primary_values_without_coercion(value): + assert _resolve({"A__B": value}, {"A": {"B": "default"}}) == {"A": {"B": value}} + + +@pytest.mark.parametrize("source", ["primary", "defaults"]) +def test_resolve_preserves_empty_lists_from_either_source(source): + mapping = {"Empty": [], "Nested": {"Empty": []}, "Keep": "value"} + primary, defaults = (mapping, {}) if source == "primary" else ({}, mapping) + original = deepcopy(mapping) + + result = _resolve(primary, defaults) + + assert result == mapping + assert result["Empty"] is not mapping["Empty"] + assert result["Nested"]["Empty"] is not mapping["Nested"]["Empty"] + assert mapping == original + + +@pytest.mark.parametrize("empty_is_primary", [True, False]) +@pytest.mark.parametrize("value", ["fallback", ["read"]]) +def test_resolve_empty_lists_follow_primary_precedence(empty_is_primary, value): + empty = {"A": []} + populated = {"A": value} + primary, defaults = (empty, populated) if empty_is_primary else (populated, empty) + original_primary = deepcopy(primary) + original_defaults = deepcopy(defaults) + + assert _resolve(primary, defaults) == primary + assert primary == original_primary + assert defaults == original_defaults + + +def test_resolve_preserves_case_in_both_sources(): + assert _resolve( + {"app__client_id": "primary"}, {"App": {"CLIENT_ID": "default"}} + ) == { + "app": {"client_id": "primary"}, + "App": {"CLIENT_ID": "default"}, + } + + +def test_resolve_accepts_custom_path_transformer(): + assert _resolve( + {"App__Name": "new"}, + {"App": {"Name": "old", "Enabled": True}}, + path_transformer=lambda path: path.replace("__", "::"), + ) == {"App": {"Name": "new", "Enabled": True}} + + +@pytest.mark.parametrize("primary_first", [True, False]) +def test_resolve_rejects_structural_conflicts_between_sources(primary_first): + scalar = {"A": "scalar"} + subtree = {"A__B": "child"} + primary, defaults = (scalar, subtree) if primary_first else (subtree, scalar) + + with pytest.raises((TypeError, ValueError), match="A"): + _resolve(primary, defaults) + + +@pytest.mark.parametrize("source", ["primary", "defaults"]) +@pytest.mark.parametrize("parent_first", [True, False]) +def test_resolve_rejects_conflicting_paths_regardless_of_order(source, parent_first): + entries = [("A", "scalar"), ("A__B", "child")] + if not parent_first: + entries.reverse() + mapping = dict(entries) + primary, defaults = (mapping, {}) if source == "primary" else ({}, mapping) + + with pytest.raises((TypeError, ValueError), match="A"): + _resolve(primary, defaults) + + +def test_resolve_accepts_read_only_sources_without_modifying_inputs(): + primary = {"A__B": "new"} + defaults = {"A": {"B": "old", "C": "keep"}} + original_primary = deepcopy(primary) + original_defaults = deepcopy(defaults) + + result = _resolve( + MappingProxyType(primary), + MappingProxyType({"A": MappingProxyType(defaults["A"])}), + ) + + assert result == {"A": {"B": "new", "C": "keep"}} + assert primary == original_primary + assert defaults == original_defaults + result["A"]["B"] = "changed" + assert defaults == original_defaults + + +def test_resolve_empty_sources(): + assert _resolve({}, {}) == {} + + +@pytest.mark.parametrize("source", ["primary", "defaults"]) +def test_resolve_preserves_empty_sections_without_sharing_them(source): + mapping = {"A": {}, "B": {"C": {}}} + primary, defaults = (mapping, {}) if source == "primary" else ({}, mapping) + + result = _resolve(primary, defaults) + + assert result == {"A": {}, "B": {"C": {}}} + assert result["A"] is not mapping["A"] + assert result["B"]["C"] is not mapping["B"]["C"] + result["A"]["New"] = 1 + result["B"]["C"]["New"] = 2 + assert mapping == {"A": {}, "B": {"C": {}}} + + +@pytest.mark.parametrize("empty_is_primary", [True, False]) +@pytest.mark.parametrize( + "empty, populated", + [ + ({"A": {}}, {"A": {"B": 1}}), + ({"A": {}}, {"A": {"B": {}}}), + ({"A": {"B": {}}}, {"A": {"B": {"C": 1}, "D": 2}}), + ], +) +def test_resolve_merges_empty_and_populated_sections_without_mutating_sources( + empty_is_primary, empty, populated +): + primary, defaults = (empty, populated) if empty_is_primary else (populated, empty) + original_primary = deepcopy(primary) + original_defaults = deepcopy(defaults) + + result = _resolve(primary, defaults) + + assert result == populated + assert primary == original_primary + assert defaults == original_defaults + result["A"]["New"] = 3 + assert primary == original_primary + assert defaults == original_defaults + + +def test_resolve_merges_read_only_empty_section(): + defaults = MappingProxyType({"A": MappingProxyType({})}) + + assert _resolve({"A__B": 1}, defaults) == {"A": {"B": 1}} + assert defaults["A"] == {} + + +@pytest.mark.parametrize("source", ["primary", "defaults"]) +@pytest.mark.parametrize("parent_first", [True, False]) +def test_resolve_accepts_empty_parent_and_child_paths_in_either_order( + source, parent_first +): + entries = [("A", {}), ("A__B", 1)] + if not parent_first: + entries.reverse() + mapping = dict(entries) + primary, defaults = (mapping, {}) if source == "primary" else ({}, mapping) + + assert _resolve(primary, defaults) == {"A": {"B": 1}} + assert mapping["A"] == {} + + +@pytest.mark.parametrize("empty_is_primary", [True, False]) +@pytest.mark.parametrize("scalar", [None, False, 0, "", "scalar", []]) +def test_resolve_rejects_non_mapping_parent_of_empty_section(empty_is_primary, scalar): + empty = {"A": {"B": {}}} + non_mapping = {"A": scalar} + primary, defaults = ( + (empty, non_mapping) if empty_is_primary else (non_mapping, empty) + ) + + with pytest.raises((TypeError, ValueError), match="A"): + _resolve(primary, defaults) + + +def test_resolve_merges_identical_empty_sections(): + primary = {"A": {}} + defaults = {"A": {}} + + result = _resolve(primary, defaults) + + assert result == {"A": {}} + assert result["A"] is not primary["A"] + assert result["A"] is not defaults["A"] + + +def test_get_tries_alternative_paths_in_order(): + assert _get({"A": 1, "B": 2}, "Missing", "B", "A") == 2 + + +def test_get_can_skip_null_candidates(): + assert _get({"A": None, "B": False}, "A", "B", skip_none=True) is False + + +@pytest.mark.parametrize( + "mapping, path, expected", + [ + ({"First": 1, "Second": 2}, "Second", [2]), + ({"Entries": ["first", "second"]}, "Entries::1", ["second"]), + ({"Entries": {"A": 1, "B": 2}}, "Entries::*", [1, 2]), + ({"Entries": [1, 2]}, "Entries::*", [1, 2]), + ({"Entries": [1, 2]}, "Missing::*", []), + ], +) +def test_collect_selects_named_and_wildcard_paths(mapping, path, expected): + assert _collect(mapping, path) == expected + + +def test_xpath_has_stable_alias_precedence(): + assert _xpath("client_id") == ["CLIENTID", "CLIENT_ID", "client_id"] + + +def test_xpath_joins_path_components(): + assert _xpath("agent_application", "user_authorization") == [ + "AGENTAPPLICATION::USERAUTHORIZATION", + "AGENT_APPLICATION::USER_AUTHORIZATION", + "agent_application::user_authorization", + ] + + +def test_xpath_three_components_deduplicates_equivalent_styles(): + assert _xpath("a", "b", "c") == ["A::B::C", "a::b::c"] + + +@pytest.mark.parametrize( + "style, expected", + [ + ("upper_snake", "AGENT_APPLICATION::USER_AUTHORIZATION::CLIENT_ID"), + ("lower_snake", "agent_application::user_authorization::client_id"), + ("camel", "agentApplication::userAuthorization::clientId"), + ("pascal", "AgentApplication::UserAuthorization::ClientId"), + ("all_caps", "AGENTAPPLICATION::USERAUTHORIZATION::CLIENTID"), + ], +) +def test_xpath_applies_only_the_enabled_style_to_every_component(style, expected): + options = { + "upper_snake": False, + "lower_snake": False, + "all_caps": False, + style: True, + } + + assert _xpath( + "agent_application", "user_authorization", "client_id", **options + ) == [expected] + + +@pytest.mark.parametrize( + "parts", + [ + ("agent_application", "user_authorization"), + ("agent_application::user_authorization",), + ], +) +def test_xpath_sorts_complete_paths_without_mixing_enabled_styles(parts): + assert _xpath(*parts, camel=True, pascal=True) == [ + "AGENTAPPLICATION::USERAUTHORIZATION", + "AGENT_APPLICATION::USER_AUTHORIZATION", + "AgentApplication::UserAuthorization", + "agentApplication::userAuthorization", + "agent_application::user_authorization", + ] + + +@pytest.mark.parametrize("parts", [(), ("",), (" ",), ("a", "")]) +def test_xpath_empty_components_produce_no_aliases(parts): + assert _xpath(*parts) == [] + + +def test_xpath_all_styles_disabled_produces_no_aliases(): + assert ( + _xpath("client_id", upper_snake=False, lower_snake=False, all_caps=False) == [] + ) diff --git a/tests/activity/test_channel_id.py b/tests/activity/test_channel_id.py index 04747699d..78adb52a9 100644 --- a/tests/activity/test_channel_id.py +++ b/tests/activity/test_channel_id.py @@ -75,72 +75,3 @@ def test_init_errors(self): ChannelId("", channel=" ", sub_channel="support") with pytest.raises(Exception): ChannelId(channel="a:b", sub_channel="support") - - -class TestChannelIdIncludes: - @pytest.mark.parametrize( - ("filter_id", "candidate", "expected"), - [ - # Exact channel - ("msteams", "msteams", True), - ("msteams", "webchat", False), - # Parent filter includes every sub-channel - ("msteams", "msteams:copilot", True), - ("msteams", "msteams:meeting", True), - # Specific sub-channel filter is exact - ("msteams:copilot", "msteams:copilot", True), - ("msteams:copilot", "msteams:meeting", False), - ("msteams:copilot", "msteams", False), - ("msteams:copilot", "slack:copilot", False), - # Wildcard sub-channel behaves like the parent filter - ("msteams:*", "msteams", True), - ("msteams:*", "msteams:copilot", True), - ("msteams:*", "slack:copilot", False), - # Wildcard channel - ("*", "msteams", True), - ("*", "msteams:copilot", True), - ("*:copilot", "slack:copilot", True), - ("*:copilot", "msteams:copilot", True), - ("*:copilot", "msteams:meeting", False), - ("*:copilot", "msteams", False), - ("*:*", "slack:anything", True), - # Case-sensitive - ("msteams", "MSTeams", False), - ("msteams:copilot", "msteams:COPILOT", False), - # Whitespace is normalized by ChannelId parsing - ("msteams", " msteams:copilot ", True), - ], - ) - def test_includes(self, filter_id, candidate, expected): - assert ChannelId(filter_id).includes(candidate) is expected - - @pytest.mark.parametrize( - ("filter_id", "candidate", "expected"), - [ - ("msteams", ChannelId("msteams:copilot"), True), - ("msteams:copilot", ChannelId(channel="msteams"), False), - ("msteams", ChannelId(channel="msteams", sub_channel="copilot"), True), - ], - ) - def test_includes_accepts_channel_id_candidate( - self, filter_id, candidate, expected - ): - assert ChannelId(filter_id).includes(candidate) is expected - - @pytest.mark.parametrize("candidate", [None, "", " ", ":copilot"]) - def test_includes_rejects_missing_or_invalid_candidate(self, candidate): - assert ChannelId("msteams").includes(candidate) is False - assert ChannelId("*").includes(candidate) is False - - def test_includes_is_directional(self): - parent = ChannelId("msteams") - child = ChannelId("msteams:copilot") - - assert parent.includes(child) - assert not child.includes(parent) - - def test_includes_does_not_change_equality(self): - parent = ChannelId("msteams") - - assert parent.includes("msteams:copilot") - assert parent != "msteams:copilot" diff --git a/tests/activity/test_config_coercion.py b/tests/activity/test_config_coercion.py index 154516c6d..a8a763ba9 100644 --- a/tests/activity/test_config_coercion.py +++ b/tests/activity/test_config_coercion.py @@ -57,6 +57,16 @@ def test_unrecognized_requires_default(self): assert coerce_bool("maybe", default=True) is True assert coerce_bool("maybe", default=False) is False + @pytest.mark.parametrize("value, expected", [("true", True), ("false", False)]) + def test_recognized_value_overrides_default(self, value, expected): + assert coerce_bool(value, default=not expected) is expected + + def test_unrecognized_with_none_default_reports_name_without_value(self): + with pytest.raises(ValueError) as error: + coerce_bool("synthetic-secret", default=None, name="Settings::Enabled") + + assert str(error.value) == "Invalid boolean for Settings::Enabled" + class TestCoerceInt: def test_none_returns_default(self):