Coverage for sources / agentsmgr / userdata.py: 9%
102 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-01 15:37 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
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#============================================================================#
21''' Global settings management for coder configurations.
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'''
29import json as _json
31from . import __
32from . import exceptions as _exceptions
33from . import renderers as _renderers
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 )
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.
51 Surveys defaults/user/configurations directory for coder-specific
52 files and populates them to per-user locations. Handles two types
53 of files: direct copy for non-settings files and merge for settings
54 files (preserving user values).
56 Returns tuple of (files_attempted, files_updated) counts.
57 '''
58 globals_directory = data_location / 'user' / 'configurations'
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 )
92def _is_settings_file( file: __.Path, coder: str ) -> bool:
93 ''' Determines whether file is a settings file requiring merge logic.
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, ( ) )
107def _copy_file_directly(
108 source: __.Path, target: __.Path, simulate: bool
109) -> bool:
110 ''' Copies file directly from source to target location.
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
126def _merge_settings_file(
127 source: __.Path, target: __.Path, simulate: bool
128) -> bool:
129 ''' Merges JSON settings file preserving user values.
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
147def _load_json_file(
148 filepath: __.Path, target_context: __.Path
149) -> dict[ str, __.typx.Any ]:
150 ''' Loads JSON file with error handling.
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
168def _write_merged_settings(
169 target: __.Path, merged: dict[ str, __.typx.Any ]
170) -> None:
171 ''' Writes merged settings with backup of existing file.
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
192def populate_user_wrappers(
193 data_location: __.Path,
194 simulate: bool = False,
195) -> tuple[ int, int ]:
196 ''' Installs wrapper scripts to user bin directory.
198 Copies wrapper scripts from data source to ~/.local/bin,
199 making them executable. Returns tuple of (files_attempted,
200 files_installed) counts.
201 '''
202 wrappers_dir = data_location / 'user' / 'executables'
203 user_bin = __.Path.home( ) / '.local' / 'bin'
204 if not wrappers_dir.exists( ):
205 return ( 0, 0 )
206 files_attempted = 0
207 files_installed = 0
208 for script in wrappers_dir.iterdir( ):
209 if not script.is_file( ):
210 continue
211 files_attempted += 1
212 target = user_bin / script.name
213 if not simulate:
214 user_bin.mkdir( parents = True, exist_ok = True )
215 try: __.shutil.copy2( script, target )
216 except ( OSError, IOError ) as exception:
217 raise _exceptions.GlobalsPopulationFailure(
218 script, target
219 ) from exception
220 try: target.chmod( target.stat( ).st_mode | 0o111 )
221 except ( OSError, IOError ) as exception:
222 raise _exceptions.GlobalsPopulationFailure(
223 script, target
224 ) from exception
225 files_installed += 1
226 return ( files_attempted, files_installed )
229def _deep_merge_settings(
230 target: dict[ str, __.typx.Any ], source: dict[ str, __.typx.Any ]
231) -> dict[ str, __.typx.Any ]:
232 ''' Recursively merges source into target preserving target values.
234 Implements additive merge: adds keys from source that are missing
235 in target. When both contain same key with dict values, recursively
236 merges nested dicts. For conflicting scalar values, target value
237 wins (user preferences preserved).
238 '''
239 result = target.copy( )
240 for key, source_value in source.items( ):
241 if key not in result:
242 result[ key ] = source_value
243 elif (
244 _is_json_dict( result[ key ] )
245 and _is_json_dict( source_value )
246 ):
247 result[ key ] = _deep_merge_settings(
248 result[ key ], source_value )
249 return result