Initial commit: Add OpenAPI Mock Server project

.src/openapi_mock/core/config.py

@@ -0,0 +1,204 @@
+"""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