Coverage for sources/agentsmgr/cmdbase.py: 70%

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''' Shared infrastructure for command implementations. 

22 

23 This module provides common utilities used across multiple command 

24 implementations, including error handling, configuration management, 

25 and data location resolution. 

26''' 

27 

28 

29import yaml as _yaml 

30 

31from . import __ 

32from . import core as _core 

33from . import exceptions as _exceptions 

34from . import nomina as _nomina 

35from . import sources as _sources 

36 

37 

38CoderConfiguration: __.typx.TypeAlias = __.cabc.Mapping[ str, __.typx.Any ] 

39 

40 

41def intercept_errors( ) -> __.cabc.Callable[ 

42 [ __.cabc.Callable[ 

43 ..., __.cabc.Coroutine[ __.typx.Any, __.typx.Any, None ] ] ], 

44 __.cabc.Callable[ 

45 ..., __.cabc.Coroutine[ __.typx.Any, __.typx.Any, None ] ] 

46]: 

47 ''' Decorator for CLI command handlers to intercept and render errors. 

48 

49 Provides clean separation between business logic and error handling: 

50 

51 **Purpose**: Enables command implementations to focus purely on 

52 business logic while the decorator handles all error presentation 

53 concerns. 

54 

55 **Responsibilities**: 

56 

57 - Intercepts Omnierror exceptions from command execution 

58 - Renders errors in appropriate format (markdown for CLI) 

59 - Ensures proper exit code handling (SystemExit with code 1) 

60 

61 **Pattern**: Commands implement business logic and raise exceptions; 

62 decorator handles presentation and process termination. This 

63 separation ensures commands remain testable and focused. 

64 

65 **Type narrowing note**: The isinstance(auxdata, _core.Globals) 

66 checks in individual command execute methods serve type narrowing 

67 purposes and must be retained for proper type checking. 

68 ''' 

69 def decorator( 

70 function: __.cabc.Callable[ 

71 ..., __.cabc.Coroutine[ __.typx.Any, __.typx.Any, None ] ] 

72 ) -> __.cabc.Callable[ 

73 ..., __.cabc.Coroutine[ __.typx.Any, __.typx.Any, None ] 

74 ]: 

75 @__.funct.wraps( function ) 

76 async def wrapper( 

77 self: __.typx.Any, 

78 auxdata: __.typx.Any, 

79 *posargs: __.typx.Any, 

80 **nomargs: __.typx.Any, 

81 ) -> None: 

82 try: return await function( self, auxdata, *posargs, **nomargs ) 

83 except _exceptions.Omnierror as exception: 

84 if isinstance( auxdata, _core.Globals ): 84 ↛ 88line 84 didn't jump to line 88 because the condition on line 84 was always true

85 await _core.render_and_print_result( 

86 exception, auxdata.display, auxdata.exits ) 

87 else: 

88 for line in exception.render_as_markdown( ): 

89 print( line, file = __.sys.stderr ) 

90 raise SystemExit( 1 ) from None 

91 return wrapper 

92 return decorator 

93 

94 

95async def retrieve_configuration( 

96 target: __.Path, 

97 profile: __.typx.Optional[ __.Path ] = None, 

98) -> __.cabc.Mapping[ str, __.typx.Any ]: 

99 ''' Loads and validates configuration from Copier answers file. 

100 

101 Unified configuration loading used by multiple command 

102 implementations. Reads from standard Copier answers location 

103 (or specified profile path) and validates required fields. 

104 ''' 

105 if profile is not None: 

106 answers_file = profile 

107 else: 

108 answers_file = ( 

109 target / ".auxiliary/configuration/copier-answers--agents.yaml" ) 

110 if not answers_file.exists( ): 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 raise _exceptions.ConfigurationAbsence( target ) 

112 try: content = answers_file.read_text( encoding = 'utf-8' ) 

113 except ( OSError, IOError ) as exception: 

114 raise _exceptions.ConfigurationAbsence( ) from exception 

115 try: 

116 configuration: __.cabc.Mapping[ str, __.typx.Any ] = ( 

117 _yaml.safe_load( content ) ) 

118 except _yaml.YAMLError as exception: 

119 raise _exceptions.ConfigurationInvalidity( exception ) from exception 

120 if not isinstance( configuration, __.cabc.Mapping ): 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 raise _exceptions.ConfigurationInvalidity( ) 

122 await validate_configuration( configuration ) 

123 return configuration 

124 

125 

126async def validate_configuration( 

127 configuration: __.cabc.Mapping[ str, __.typx.Any ] 

128) -> None: 

129 ''' Validates required configuration fields are present and non-empty. ''' 

130 if not configuration.get( 'coders' ): 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true

131 raise _exceptions.ConfigurationInvalidity( ) 

132 if not configuration.get( 'languages' ): 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true

133 raise _exceptions.ConfigurationInvalidity( ) 

134 

135 

136def retrieve_data_location( 

137 source_spec: str, 

138 tag_prefix: _nomina.TagPrefixArgument = __.absent, 

139) -> __.Path: 

140 ''' Resolves data source specification to local filesystem path. 

141 

142 Supports local paths, Git repositories, and remote sources through 

143 pluggable source handlers. Uses registered handlers to resolve 

144 various URL schemes to local filesystem paths. 

145 ''' 

146 return _sources.resolve_source_location( source_spec, tag_prefix ) 

147 

148 

149def validate_data_source_structure( 

150 location: __.Path, 

151 required_directories: __.cabc.Sequence[ str ], 

152) -> None: 

153 ''' Validates data source contains required directory structure. 

154 

155 Checks that all required directories exist at the data source 

156 location. Raises DataSourceInvalidity with list of missing 

157 directories if validation fails. 

158 ''' 

159 missing = tuple( 

160 directory for directory in required_directories 

161 if not ( location / directory ).is_dir( ) ) 

162 if missing: 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true

163 raise _exceptions.DataSourceInvalidity( location, missing )