From 1a31c6ed2eecee3f846a7863398e94fc5e978053 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Fri, 30 Jan 2026 03:41:40 +0000 Subject: [PATCH] Initial commit: Add OpenAPI Mock Server project --- .src/openapi_mock/core/config.py | 204 +++++++++++++++++++++++++++++++ 1 file changed, 204 insertions(+) create mode 100644 .src/openapi_mock/core/config.py diff --git a/.src/openapi_mock/core/config.py b/.src/openapi_mock/core/config.py new file mode 100644 index 0000000..1c70cd9 --- /dev/null +++ b/.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