openapi-mock-server/.src/openapi_mock/core/config.py

"""Configuration file support for OpenAPI Mock Server."""

import os
from pathlib import Path
from typing import Any, Dict, List, Optional

import yaml


class ConfigError(Exception):
    """Base exception for configuration errors."""
    pass


class ConfigFileNotFoundError(ConfigError):
    """Raised when config file is not found."""
    pass


class ConfigParseError(ConfigError):
    """Raised when config file has invalid YAML."""
    pass


DEFAULT_CONFIG_FILES = [
    ".openapi-mock.yaml",
    ".openapi-mock.yml",
    "mock-config.yaml",
    "mock-config.yml",
]


def find_config_file(config_dir: Optional[str] = None) -> Optional[Path]:
    """Find a configuration file in the given or current directory.

    Args:
        config_dir: Directory to search for config file.

    Returns:
        Path to config file or None if not found.
    """
    search_dir = Path(config_dir) if config_dir else Path.cwd()

    for config_name in DEFAULT_CONFIG_FILES:
        config_path = search_dir / config_name
        if config_path.exists():
            return config_path

    return None


def load_config(config_path: Optional[str] = None) -> Dict[str, Any]:
    """Load configuration from a YAML file.

    Args:
        config_path: Path to config file. If None, searches default locations.

    Returns:
        Configuration dictionary.

    Raises:
        ConfigFileNotFoundError: If config file not found.
        ConfigParseError: If config file has invalid YAML.
    """
    if config_path is None:
        config_path = find_config_file()
        if config_path is None:
            return {}
    else:
        config_path = Path(config_path)
        if not config_path.exists():
            raise ConfigFileNotFoundError(f"Config file not found: {config_path}")

    try:
        with open(config_path, "r", encoding="utf-8") as f:
            config = yaml.safe_load(f)
    except yaml.YAMLError as e:
        raise ConfigParseError(f"Invalid YAML in config: {e}")
    except IOError as e:
        raise ConfigError(f"Error reading config: {e}")

    return config or {}


def apply_env_overrides(config: Dict[str, Any]) -> Dict[str, Any]:
    """Apply environment variable overrides to configuration.

    Args:
        config: Base configuration dictionary.

    Returns:
        Configuration with environment overrides applied.
    """
    result = config.copy()

    if "OPENAPI_MOCK_PORT" in os.environ:
        try:
            result["port"] = int(os.environ["OPENAPI_MOCK_PORT"])
        except ValueError:
            pass

    if "OPENAPI_MOCK_HOST" in os.environ:
        result["host"] = os.environ["OPENAPI_MOCK_HOST"]

    if "OPENAPI_MOCK_DELAY" in os.environ:
        delay_str = os.environ["OPENAPI_MOCK_DELAY"]
        try:
            if "," in delay_str:
                parts = delay_str.split(",")
                result["delay"] = (float(parts[0]), float(parts[1]))
            else:
                result["delay"] = (float(delay_str), float(delay_str))
        except ValueError:
            pass

    if "OPENAPI_MOCK_AUTH" in os.environ:
        result["auth"] = os.environ["OPENAPI_MOCK_AUTH"]

    return result


def get_cli_overrides(
    host: Optional[str] = None,
    port: Optional[int] = None,
    delay: Optional[tuple] = None,
    auth: Optional[str] = None,
    watch: Optional[bool] = None,
    reload: Optional[bool] = None
) -> Dict[str, Any]:
    """Get CLI overrides as a configuration dictionary.

    Args:
        host: Host override.
        port: Port override.
        delay: Delay override.
        auth: Auth type override.
        watch: Watch flag override.
        reload: Reload flag override.

    Returns:
        Configuration dictionary with non-None values.
    """
    config: Dict[str, Any] = {}

    if host is not None:
        config["host"] = host
    if port is not None:
        config["port"] = port
    if delay is not None:
        config["delay"] = delay
    if auth is not None:
        config["auth"] = auth
    if watch is not None:
        config["watch"] = watch
    if reload is not None:
        config["reload"] = reload

    return config


def merge_configs(
    base_config: Dict[str, Any],
    override_config: Dict[str, Any]
) -> Dict[str, Any]:
    """Merge two configuration dictionaries, with override taking precedence.

    Args:
        base_config: Base configuration.
        override_config: Override configuration.

    Returns:
        Merged configuration.
    """
    result = base_config.copy()

    for key, value in override_config.items():
        if isinstance(value, dict) and key in result and isinstance(result[key], dict):
            result[key] = merge_configs(result[key], value)
        else:
            result[key] = value

    return result


def get_server_config(
    config_path: Optional[str] = None,
    cli_overrides: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """Get complete server configuration.

    Args:
        config_path: Path to config file.
        cli_overrides: CLI argument overrides.

    Returns:
        Complete server configuration.
    """
    config = load_config(config_path)
    config = apply_env_overrides(config)

    if cli_overrides:
        config = merge_configs(config, cli_overrides)

    return config