Coverage for sources/agentsmgr/commands/userdata.py: 9%

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 __ 

32 

33 

34def _is_json_dict( 

35 value: __.typx.Any 

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

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

38 return isinstance( value, dict ) 

39 

40 

41def populate_globals( 

42 data_location: __.Path, 

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

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

45 simulate: bool = False, 

46) -> tuple[ int, int ]: 

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

48 

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

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

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

52 (preserving user values). 

53 

54 Returns tuple of (files_attempted, files_updated) counts. 

55 ''' 

56 globals_directory = data_location / 'globals' 

57 if not globals_directory.exists( ): 

58 return ( 0, 0 ) 

59 files_attempted = 0 

60 files_updated = 0 

61 for coder in coders: 

62 coder_globals = globals_directory / coder 

63 if not coder_globals.exists( ): 

64 continue 

65 try: renderer = __.RENDERERS[ coder ] 

66 except KeyError as exception: 

67 raise __.CoderAbsence( coder ) from exception 

68 per_user_directory = renderer.resolve_base_directory( 

69 mode = 'per-user', 

70 target = __.Path.cwd( ), 

71 configuration = application_configuration, 

72 environment = __.os.environ, 

73 ) 

74 for global_file in coder_globals.iterdir( ): 

75 if not global_file.is_file( ): 

76 continue 

77 files_attempted += 1 

78 target_file = per_user_directory / global_file.name 

79 if _is_settings_file( global_file, coder ): 

80 updated = _merge_settings_file( 

81 global_file, target_file, simulate ) 

82 else: 

83 updated = _copy_file_directly( 

84 global_file, target_file, simulate ) 

85 if updated: 

86 files_updated += 1 

87 return ( files_attempted, files_updated ) 

88 

89 

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

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

92 

93 Settings files have coder-specific names and contain JSON 

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

95 files are directly copied, replacing any existing version. 

96 ''' 

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

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

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

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

101 } 

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

103 

104 

105def _copy_file_directly( 

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

107) -> bool: 

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

109 

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

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

112 ''' 

113 if simulate: 

114 return True 

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

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

117 except ( OSError, IOError ) as exception: 

118 raise __.GlobalsPopulationFailure( source, target ) from exception 

119 return True 

120 

121 

122def _merge_settings_file( 

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

124) -> bool: 

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

126 

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

128 merge adding missing keys from template while preserving all user 

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

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

131 ''' 

132 template = _load_json_file( source, target ) 

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

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

135 else { } ) 

136 merged = _deep_merge_settings( user_settings, template ) 

137 if simulate: 

138 return True 

139 _write_merged_settings( target, merged ) 

140 return True 

141 

142 

143def _load_json_file( 

144 filepath: __.Path, target_context: __.Path 

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

146 ''' Loads JSON file with error handling. 

147 

148 Raises GlobalsPopulationFailure with source context on any error. 

149 ''' 

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

151 except ( OSError, IOError ) as exception: 

152 raise __.GlobalsPopulationFailure( 

153 filepath, target_context ) from exception 

154 try: 

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

156 except ValueError as exception: 

157 raise __.GlobalsPopulationFailure( 

158 filepath, target_context ) from exception 

159 if not _is_json_dict( loaded ): 

160 raise __.GlobalsPopulationFailure( filepath, target_context ) 

161 return loaded 

162 

163 

164def _write_merged_settings( 

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

166) -> None: 

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

168 

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

170 writing merged result. 

171 ''' 

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

173 if target.exists( ): 

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

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

176 except ( OSError, IOError ) as exception: 

177 raise __.GlobalsPopulationFailure( 

178 target, target ) from exception 

179 try: 

180 target.write_text( 

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

182 except ( OSError, IOError ) as exception: 

183 raise __.GlobalsPopulationFailure( target, target ) from exception 

184 

185 

186def _deep_merge_settings( 

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

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

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

190 

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

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

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

194 wins (user preferences preserved). 

195 ''' 

196 result = target.copy( ) 

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

198 if key not in result: 

199 result[ key ] = source_value 

200 elif ( 

201 _is_json_dict( result[ key ] ) 

202 and _is_json_dict( source_value ) 

203 ): 

204 result[ key ] = _deep_merge_settings( 

205 result[ key ], source_value ) 

206 return result