Coverage for src/qdrant_loader/cli/commands/config.py: 40%

1from __future__ import annotations 

2 

3import json 

4from pathlib import Path 

5 

6from click.exceptions import ClickException 

7 

8 

9def run_show_config( 

10 workspace: Path | None, 

11 config: Path | None, 

12 env: Path | None, 

13 log_level: str, 

14) -> str: 

15 """Load configuration and return a JSON string representation. 

16 

17 This function is a command helper used by the CLI wrapper to display configuration. 

18 It performs validation, optional workspace setup, logging setup, and settings loading. 

19 """ 

20 try: 

21 from qdrant_loader.cli.config_loader import ( 

22 load_config_with_workspace as _load_config_with_workspace, 

23 ) 

24 from qdrant_loader.cli.config_loader import ( 

25 setup_workspace as _setup_workspace_impl, 

26 ) 

27 from qdrant_loader.config import get_settings 

28 from qdrant_loader.config.workspace import validate_workspace_flags 

29 from qdrant_loader.utils.logging import LoggingConfig 

30 

31 # Validate flag combinations 

32 validate_workspace_flags(workspace, config, env) 

33 

34 # Setup workspace if provided 

35 workspace_config = None 

36 if workspace: 

37 workspace_config = _setup_workspace_impl(workspace) 

38 

39 # Setup/reconfigure logging once with workspace support 

40 log_file = ( 

41 str(workspace_config.logs_path) if workspace_config else "qdrant-loader.log" 

42 ) 

43 if getattr(LoggingConfig, "reconfigure", None): # Core supports reconfigure 

44 if getattr(LoggingConfig, "_initialized", False): # type: ignore[attr-defined] 

45 LoggingConfig.reconfigure(file=log_file) # type: ignore[attr-defined] 

46 else: 

47 LoggingConfig.setup(level=log_level, format="console", file=log_file) 

48 else: 

49 # Compatibility path when running with an older core: clear root handlers 

50 # to avoid duplicates, then perform a full setup once. 

51 import logging as _py_logging 

52 

53 _py_logging.getLogger().handlers = [] 

54 LoggingConfig.setup(level=log_level, format="console", file=log_file) 

55 

56 # Emit a single set of workspace-related info logs (no duplicates) 

57 if workspace_config: 

58 logger = LoggingConfig.get_logger(__name__) 

59 logger.info( 

60 "Using workspace", workspace=str(workspace_config.workspace_path) 

61 ) 

62 if getattr(workspace_config, "env_path", None): 

63 logger.info( 

64 "Environment file found", env_path=str(workspace_config.env_path) 

65 ) 

66 if getattr(workspace_config, "config_path", None): 

67 logger.info( 

68 "Config file found", config_path=str(workspace_config.config_path) 

69 ) 

70 

71 # Load configuration (skip validation to avoid directory prompts) 

72 _load_config_with_workspace(workspace_config, config, env, skip_validation=True) 

73 settings = get_settings() 

74 if settings is None: 

75 raise ClickException("Settings not available") 

76 

77 # Redact sensitive values before dumping to JSON 

78 def _redact_secrets(value): 

79 """Recursively redact sensitive keys within nested structures. 

80 

81 Keys are compared case-insensitively and redacted if their name contains 

82 any common secret-like token. 

83 """ 

84 sensitive_tokens = { 

85 "password", 

86 "secret", 

87 "api_key", 

88 "token", 

89 "key", 

90 "credentials", 

91 "access_token", 

92 "private_key", 

93 "client_secret", 

94 } 

95 

96 mask = "****" 

97 

98 if isinstance(value, dict): 

99 redacted: dict = {} 

100 for k, v in value.items(): 

101 key_lc = str(k).lower() 

102 if any(token in key_lc for token in sensitive_tokens): 

103 redacted[k] = mask 

104 else: 

105 redacted[k] = _redact_secrets(v) 

106 return redacted 

107 if isinstance(value, list): 

108 return [_redact_secrets(item) for item in value] 

109 # Primitive or unknown types are returned as-is 

110 return value 

111 

112 config_dict = settings.model_dump(mode="json") 

113 safe_config = _redact_secrets(config_dict) 

114 return json.dumps(safe_config, indent=2, ensure_ascii=False) 

115 except ClickException: 

116 raise 

117 except Exception as e: 

118 raise ClickException(f"Failed to display configuration: {str(e)!s}") from e