Coverage for src / local_deep_research / web / warning_checks / __init__.py: 88%

1"""Warning calculation for the settings UI. 

2 

3Thin orchestrator: reads settings from a single DB session, 

4delegates to pure check functions in hardware.py and context.py. 

5""" 

6 

7from typing import List 

8 

9from flask import session 

10from loguru import logger 

11 

12from ...database.session_context import get_user_db_session 

13from ...utilities.db_utils import get_settings_manager 

14from .context import ( 

15 check_context_below_history, 

16 check_context_truncation_history, 

17) 

18from .backup import ( 

19 check_backup_disabled, 

20 check_backup_healthy, 

21 check_no_backups_exist, 

22) 

23from .hardware import ( 

24 LOCAL_PROVIDERS, 

25 check_high_context, 

26 check_legacy_server_config, 

27 check_model_mismatch, 

28) 

29 

30 

31def _safe_check(check_fn, *args, **kwargs): 

32 """Run a single warning check, returning None on failure.""" 

33 try: 

34 return check_fn(*args, **kwargs) 

35 except Exception: 

36 name = getattr(check_fn, "__name__", repr(check_fn)) 

37 logger.exception(f"Warning check {name} failed") 

38 return None 

39 

40 

41def calculate_warnings() -> List[dict]: 

42 """Calculate current warning conditions based on settings. 

43 

44 Uses a single DB session for all setting reads and history queries. 

45 """ 

46 warnings: List[dict] = [] 

47 

48 try: 

49 username = session.get("username") 

50 with get_user_db_session(username) as db_session: 

51 if not db_session: 

52 return [] 

53 

54 settings_manager = get_settings_manager(db_session, username) 

55 

56 # Read all needed settings in one session 

57 provider = settings_manager.get_setting( 

58 "llm.provider", "ollama" 

59 ).lower() 

60 local_context = settings_manager.get_setting( 

61 "llm.local_context_window_size", 8192 

62 ) 

63 current_model = settings_manager.get_setting("llm.model", "") 

64 dismiss_high_context = settings_manager.get_setting( 

65 "app.warnings.dismiss_high_context", False 

66 ) 

67 dismiss_model_mismatch = settings_manager.get_setting( 

68 "app.warnings.dismiss_model_mismatch", False 

69 ) 

70 dismiss_context_warning = settings_manager.get_setting( 

71 "app.warnings.dismiss_context_reduced", False 

72 ) 

73 dismiss_legacy_config = settings_manager.get_setting( 

74 "app.warnings.dismiss_legacy_config", False 

75 ) 

76 backup_enabled = settings_manager.get_setting( 

77 "backup.enabled", True 

78 ) 

79 dismiss_backup_disabled = settings_manager.get_setting( 

80 "app.warnings.dismiss_backup_disabled", False 

81 ) 

82 dismiss_no_backups = settings_manager.get_setting( 

83 "app.warnings.dismiss_no_backups", False 

84 ) 

85 

86 logger.debug(f"Starting warning calculation - provider={provider}") 

87 

88 is_local = provider in LOCAL_PROVIDERS 

89 

90 # --- Hardware / settings checks (pure functions) --- 

91 w = _safe_check( 

92 check_high_context, 

93 provider, 

94 local_context, 

95 dismiss_high_context, 

96 ) 

97 if w: 

98 warnings.append(w) 

99 

100 w = _safe_check( 

101 check_model_mismatch, 

102 provider, 

103 current_model, 

104 local_context, 

105 dismiss_model_mismatch, 

106 ) 

107 if w: 

108 warnings.append(w) 

109 

110 w = _safe_check(check_legacy_server_config, dismiss_legacy_config) 

111 if w: 

112 warnings.append(w) 

113 

114 # --- Backup checks --- 

115 w = _safe_check( 

116 check_backup_disabled, backup_enabled, dismiss_backup_disabled 

117 ) 

118 if w: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 warnings.append(w) 

120 

121 # Check backup file status (lightweight filesystem glob) 

122 dismiss_backup_info = settings_manager.get_setting( 

123 "app.warnings.dismiss_backup_info", False 

124 ) 

125 try: 

126 from ...config.paths import get_user_backup_directory 

127 from ...utilities.formatting import human_size 

128 

129 username = session.get("username") 

130 if username: 130 ↛ 163line 130 didn't jump to line 163 because the condition on line 130 was always true

131 backup_dir = get_user_backup_directory(username) 

132 total_size = 0 

133 backup_count = 0 

134 for f in backup_dir.glob("ldr_backup_*.db"): 134 ↛ 135line 134 didn't jump to line 135 because the loop on line 134 never started

135 try: 

136 total_size += f.stat().st_size 

137 backup_count += 1 

138 except FileNotFoundError: 

139 continue 

140 

141 w = _safe_check( 

142 check_no_backups_exist, 

143 backup_enabled, 

144 backup_count, 

145 dismiss_no_backups, 

146 ) 

147 if w: 

148 warnings.append(w) 

149 

150 w = _safe_check( 

151 check_backup_healthy, 

152 backup_enabled, 

153 backup_count, 

154 human_size(total_size), 

155 dismiss_backup_info, 

156 ) 

157 if w: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true

158 warnings.append(w) 

159 except Exception: 

160 logger.debug("Backup status check skipped") 

161 

162 # --- History-based checks (need DB queries) --- 

163 if is_local and not dismiss_context_warning: 

164 w = _safe_check( 

165 check_context_below_history, db_session, local_context 

166 ) 

167 if w: 

168 warnings.append(w) 

169 

170 w = _safe_check( 

171 check_context_truncation_history, db_session, local_context 

172 ) 

173 if w: 

174 warnings.append(w) 

175 

176 except Exception: 

177 logger.exception("Error calculating warnings") 

178 

179 return warnings