Coverage for sources/agentsmgr/commands/base.py: 22%

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 __ 

32 

33 

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

35 

36 

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

38 [ __.cabc.Callable[ 

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

40 __.cabc.Callable[ 

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

42]: 

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

44 

45 Provides clean separation between business logic and error handling: 

46 

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

48 business logic while the decorator handles all error presentation 

49 concerns. 

50 

51 **Responsibilities**: 

52 

53 - Intercepts Omnierror exceptions from command execution 

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

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

56 

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

58 decorator handles presentation and process termination. This 

59 separation ensures commands remain testable and focused. 

60 

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

62 checks in individual command execute methods serve type narrowing 

63 purposes and must be retained for proper type checking. 

64 ''' 

65 def decorator( 

66 function: __.cabc.Callable[ 

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

68 ) -> __.cabc.Callable[ 

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

70 ]: 

71 @__.funct.wraps( function ) 

72 async def wrapper( 

73 self: __.typx.Any, 

74 auxdata: __.typx.Any, 

75 *posargs: __.typx.Any, 

76 **nomargs: __.typx.Any, 

77 ) -> None: 

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

79 except __.Omnierror as exception: 

80 if isinstance( auxdata, __.Globals ): 

81 await __.render_and_print_result( 

82 exception, auxdata.display, auxdata.exits ) 

83 else: 

84 for line in exception.render_as_markdown( ): 

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

86 raise SystemExit( 1 ) from None 

87 return wrapper 

88 return decorator 

89 

90 

91async def retrieve_configuration( 

92 target: __.Path 

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

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

95 

96 Unified configuration loading used by multiple command 

97 implementations. Reads from standard Copier answers location 

98 and validates required fields. 

99 ''' 

100 answers_file = ( 

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

102 if not answers_file.exists( ): 

103 raise __.ConfigurationAbsence( target ) 

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

105 except ( OSError, IOError ) as exception: 

106 raise __.ConfigurationAbsence( ) from exception 

107 try: 

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

109 _yaml.safe_load( content ) ) 

110 except _yaml.YAMLError as exception: 

111 raise __.ConfigurationInvalidity( exception ) from exception 

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

113 raise __.ConfigurationInvalidity( ) 

114 await validate_configuration( configuration ) 

115 return configuration 

116 

117 

118async def validate_configuration( 

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

120) -> None: 

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

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

123 raise __.ConfigurationInvalidity( ) 

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

125 raise __.ConfigurationInvalidity( ) 

126 

127 

128def retrieve_data_location( source_spec: str ) -> __.Path: 

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

130 

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

132 pluggable source handlers. Uses registered handlers to resolve 

133 various URL schemes to local filesystem paths. 

134 ''' 

135 return __.resolve_source_location( source_spec ) 

136 

137 

138def retrieve_variant_answers_file( 

139 auxdata: __.Globals, variant: str 

140) -> __.Path: 

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

142 

143 Validates file existence and raises ConfigurationAbsence if not 

144 found. 

145 ''' 

146 data_directory = auxdata.provide_data_location( ) 

147 project_root = data_directory.parent 

148 answers_file = ( 

149 project_root / 'tests' / 'data' / 'profiles' 

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

151 if not answers_file.exists( ): 

152 raise __.ConfigurationAbsence( ) 

153 return answers_file