Coverage for sources/agentsmgr/userdata.py: 11%

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

4#============================================================================# 

5# # 

6# Licensed under the Apache License, Version 2.0 (the "License"); # 

7# you may not use this file except in compliance with the License. # 

8# You may obtain a copy of the License at # 

9# # 

10# http://www.apache.org/licenses/LICENSE-2.0 # 

11# # 

12# Unless required by applicable law or agreed to in writing, software # 

13# distributed under the License is distributed on an "AS IS" BASIS, # 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

15# See the License for the specific language governing permissions and # 

16# limitations under the License. # 

17# # 

18#============================================================================# 

19 

20 

21''' Global settings management for coder configurations. 

22 

23 Provides functionality for populating per-user global settings files, 

24 including direct file copying and JSON settings merging with user 

25 preservation semantics. 

26''' 

27 

28 

29import json as _json 

30 

31from . import __ 

32from . import exceptions as _exceptions 

33from . import renderers as _renderers 

34 

35 

36def _is_json_dict( 

37 value: __.typx.Any 

38) -> __.typx.TypeGuard[ dict[ str, __.typx.Any ] ]: 

39 ''' Type guard for JSON dictionary values. ''' 

40 return isinstance( value, dict ) 

41 

42 

43def populate_globals( 

44 data_location: __.Path, 

45 coders: __.cabc.Sequence[ str ], 

46 application_configuration: __.cabc.Mapping[ str, __.typx.Any ], 

47 simulate: bool = False, 

48) -> tuple[ int, int ]: 

49 ''' Populates per-user global files for configured coders. 

50 

51 Surveys defaults/globals directory for coder-specific files and 

52 populates them to per-user locations. Handles two types of files: 

53 direct copy for non-settings files and merge for settings files 

54 (preserving user values). 

55 

56 Returns tuple of (files_attempted, files_updated) counts. 

57 ''' 

58 globals_directory = data_location / 'globals' 

59 if not globals_directory.exists( ): 

60 return ( 0, 0 ) 

61 files_attempted = 0 

62 files_updated = 0 

63 for coder in coders: 

64 coder_globals = globals_directory / coder 

65 if not coder_globals.exists( ): 

66 continue 

67 try: renderer = _renderers.RENDERERS[ coder ] 

68 except KeyError as exception: 

69 raise _exceptions.CoderAbsence( coder ) from exception 

70 per_user_directory = renderer.resolve_base_directory( 

71 mode = 'per-user', 

72 target = __.Path.cwd( ), 

73 configuration = application_configuration, 

74 environment = __.os.environ, 

75 ) 

76 for global_file in coder_globals.iterdir( ): 

77 if not global_file.is_file( ): 

78 continue 

79 files_attempted += 1 

80 target_file = per_user_directory / global_file.name 

81 if _is_settings_file( global_file, coder ): 

82 updated = _merge_settings_file( 

83 global_file, target_file, simulate ) 

84 else: 

85 updated = _copy_file_directly( 

86 global_file, target_file, simulate ) 

87 if updated: 

88 files_updated += 1 

89 return ( files_attempted, files_updated ) 

90 

91 

92def _is_settings_file( file: __.Path, coder: str ) -> bool: 

93 ''' Determines whether file is a settings file requiring merge logic. 

94 

95 Settings files have coder-specific names and contain JSON 

96 configuration that should be merged rather than replaced. Non-settings 

97 files are directly copied, replacing any existing version. 

98 ''' 

99 settings_names: dict[ str, tuple[ str, ... ] ] = { 

100 'claude': ( 'settings.json', ), 

101 'opencode': ( 'opencode.json', 'opencode.jsonc' ), 

102 'codex': ( 'config.json', ), 

103 } 

104 return file.name in settings_names.get( coder, ( ) ) 

105 

106 

107def _copy_file_directly( 

108 source: __.Path, target: __.Path, simulate: bool 

109) -> bool: 

110 ''' Copies file directly from source to target location. 

111 

112 Creates target directory if needed. Returns True if file was 

113 updated (or would be updated in simulation mode). 

114 ''' 

115 if simulate: 

116 return True 

117 target.parent.mkdir( parents = True, exist_ok = True ) 

118 try: __.shutil.copy2( source, target ) 

119 except ( OSError, IOError ) as exception: 

120 raise _exceptions.GlobalsPopulationFailure( 

121 source, target 

122 ) from exception 

123 return True 

124 

125 

126def _merge_settings_file( 

127 source: __.Path, target: __.Path, simulate: bool 

128) -> bool: 

129 ''' Merges JSON settings file preserving user values. 

130 

131 Loads both source template and target user settings, performs deep 

132 merge adding missing keys from template while preserving all user 

133 values. Creates backup before writing merged result. Returns True 

134 if file was updated (or would be updated in simulation mode). 

135 ''' 

136 template = _load_json_file( source, target ) 

137 user_settings: dict[ str, __.typx.Any ] = ( 

138 _load_json_file( target, target ) if target.exists( ) 

139 else { } ) 

140 merged = _deep_merge_settings( user_settings, template ) 

141 if simulate: 

142 return True 

143 _write_merged_settings( target, merged ) 

144 return True 

145 

146 

147def _load_json_file( 

148 filepath: __.Path, target_context: __.Path 

149) -> dict[ str, __.typx.Any ]: 

150 ''' Loads JSON file with error handling. 

151 

152 Raises GlobalsPopulationFailure with source context on any error. 

153 ''' 

154 try: content = filepath.read_text( encoding = 'utf-8' ) 

155 except ( OSError, IOError ) as exception: 

156 raise _exceptions.GlobalsPopulationFailure( 

157 filepath, target_context ) from exception 

158 try: 

159 loaded: __.typx.Any = _json.loads( content ) 

160 except ValueError as exception: 

161 raise _exceptions.GlobalsPopulationFailure( 

162 filepath, target_context ) from exception 

163 if not _is_json_dict( loaded ): 

164 raise _exceptions.GlobalsPopulationFailure( filepath, target_context ) 

165 return loaded 

166 

167 

168def _write_merged_settings( 

169 target: __.Path, merged: dict[ str, __.typx.Any ] 

170) -> None: 

171 ''' Writes merged settings with backup of existing file. 

172 

173 Creates target directory if needed. Backs up existing file before 

174 writing merged result. 

175 ''' 

176 target.parent.mkdir( parents = True, exist_ok = True ) 

177 if target.exists( ): 

178 backup_path = target.with_suffix( '.json.backup' ) 

179 try: __.shutil.copy2( target, backup_path ) 

180 except ( OSError, IOError ) as exception: 

181 raise _exceptions.GlobalsPopulationFailure( 

182 target, target ) from exception 

183 try: 

184 target.write_text( 

185 _json.dumps( merged, indent = 2 ), encoding = 'utf-8' ) 

186 except ( OSError, IOError ) as exception: 

187 raise _exceptions.GlobalsPopulationFailure( 

188 target, target 

189 ) from exception 

190 

191 

192def _deep_merge_settings( 

193 target: dict[ str, __.typx.Any ], source: dict[ str, __.typx.Any ] 

194) -> dict[ str, __.typx.Any ]: 

195 ''' Recursively merges source into target preserving target values. 

196 

197 Implements additive merge: adds keys from source that are missing 

198 in target. When both contain same key with dict values, recursively 

199 merges nested dicts. For conflicting scalar values, target value 

200 wins (user preferences preserved). 

201 ''' 

202 result = target.copy( ) 

203 for key, source_value in source.items( ): 

204 if key not in result: 

205 result[ key ] = source_value 

206 elif ( 

207 _is_json_dict( result[ key ] ) 

208 and _is_json_dict( source_value ) 

209 ): 

210 result[ key ] = _deep_merge_settings( 

211 result[ key ], source_value ) 

212 return result