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

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 sources as _sources 

35 

36 

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

38 

39 

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

41 [ __.cabc.Callable[ 

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

43 __.cabc.Callable[ 

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

45]: 

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

47 

48 Provides clean separation between business logic and error handling: 

49 

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

51 business logic while the decorator handles all error presentation 

52 concerns. 

53 

54 **Responsibilities**: 

55 

56 - Intercepts Omnierror exceptions from command execution 

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

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

59 

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

61 decorator handles presentation and process termination. This 

62 separation ensures commands remain testable and focused. 

63 

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

65 checks in individual command execute methods serve type narrowing 

66 purposes and must be retained for proper type checking. 

67 ''' 

68 def decorator( 

69 function: __.cabc.Callable[ 

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

71 ) -> __.cabc.Callable[ 

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

73 ]: 

74 @__.funct.wraps( function ) 

75 async def wrapper( 

76 self: __.typx.Any, 

77 auxdata: __.typx.Any, 

78 *posargs: __.typx.Any, 

79 **nomargs: __.typx.Any, 

80 ) -> None: 

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

82 except _exceptions.Omnierror as exception: 

83 if isinstance( auxdata, _core.Globals ): 

84 await _core.render_and_print_result( 

85 exception, auxdata.display, auxdata.exits ) 

86 else: 

87 for line in exception.render_as_markdown( ): 

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

89 raise SystemExit( 1 ) from None 

90 return wrapper 

91 return decorator 

92 

93 

94async def retrieve_configuration( 

95 target: __.Path, 

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

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

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

99 

100 Unified configuration loading used by multiple command 

101 implementations. Reads from standard Copier answers location 

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

103 ''' 

104 if profile is not None: 

105 answers_file = profile 

106 else: 

107 answers_file = ( 

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

109 if not answers_file.exists( ): 

110 raise _exceptions.ConfigurationAbsence( target ) 

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

112 except ( OSError, IOError ) as exception: 

113 raise _exceptions.ConfigurationAbsence( ) from exception 

114 try: 

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

116 _yaml.safe_load( content ) ) 

117 except _yaml.YAMLError as exception: 

118 raise _exceptions.ConfigurationInvalidity( exception ) from exception 

119 if not isinstance( configuration, __.cabc.Mapping ): 

120 raise _exceptions.ConfigurationInvalidity( ) 

121 await validate_configuration( configuration ) 

122 return configuration 

123 

124 

125async def validate_configuration( 

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

127) -> None: 

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

129 if not configuration.get( 'coders' ): 

130 raise _exceptions.ConfigurationInvalidity( ) 

131 if not configuration.get( 'languages' ): 

132 raise _exceptions.ConfigurationInvalidity( ) 

133 

134 

135def retrieve_data_location( 

136 source_spec: str, 

137 tag_prefix: __.typx.Annotated[ 

138 __.Absential[ str ], 

139 __.ddoc.Doc( 

140 "Prefix for filtering version tags when no explicit ref " 

141 "is specified. Only tags starting with this prefix will be " 

142 "considered, and the prefix will be stripped before version " 

143 "parsing." ), 

144 ] = __.absent, 

145) -> __.Path: 

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

147 

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

149 pluggable source handlers. Uses registered handlers to resolve 

150 various URL schemes to local filesystem paths. 

151 ''' 

152 return _sources.resolve_source_location( source_spec, tag_prefix ) 

153 

154 

155def retrieve_variant_answers_file( 

156 auxdata: _core.Globals, variant: str 

157) -> __.Path: 

158 ''' Retrieves path to variant answers file in test data directory. 

159 

160 Validates file existence and raises ConfigurationAbsence if not 

161 found. 

162 ''' 

163 data_directory = auxdata.provide_data_location( ) 

164 project_root = data_directory.parent 

165 answers_file = ( 

166 project_root / 'tests' / 'data' / 'profiles' 

167 / f"answers-{variant}.yaml" ) 

168 if not answers_file.exists( ): 

169 raise _exceptions.ConfigurationAbsence( ) 

170 return answers_file