Coverage for sources / agentsmgr / maintenance / template.py: 29%

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''' Commands for Copier template survey and validation. ''' 

22 

23 

24from . import __ 

25 

26 

27_scribe = __.provide_scribe( __name__ ) 

28 

29 

30class SurveyCommand( __.appcore_cli.Command ): 

31 ''' Surveys available template configuration variants. ''' 

32 

33 @__.cmdbase.intercept_errors( ) 

34 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride] 

35 if not isinstance( auxdata, __.Globals ): # pragma: no cover 

36 raise __.ContextInvalidity 

37 stream = await auxdata.display.provide_stream( auxdata.exits ) 

38 for variant in survey_variants( auxdata ): 

39 print( variant, file = stream ) 

40 

41 

42class ValidateCommand( __.appcore_cli.Command ): 

43 ''' Validates Copier template using configuration variant answers. ''' 

44 

45 variant: __.typx.Annotated[ 

46 str, 

47 __.typx.Doc( ''' Configuration variant to validate. ''' ), 

48 __.tyro.conf.Positional, 

49 ] 

50 preserve: __.typx.Annotated[ 

51 bool, 

52 __.tyro.conf.arg( 

53 help = "Keep temporary files for inspection.", 

54 prefix_name = False ), 

55 ] = False 

56 

57 @__.cmdbase.intercept_errors( ) 

58 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride] 

59 if not isinstance( auxdata, __.Globals ): # pragma: no cover 

60 raise __.ContextInvalidity 

61 _scribe.info( f"Validating Copier template for {self.variant}" ) 

62 repository_directory = _provide_repository_directory( ) 

63 answers_file = __.cmdbase.retrieve_variant_answers_file( 

64 auxdata, self.variant ) 

65 try: temporary_directory = __.Path( __.tempfile.mkdtemp( 

66 prefix = f"agents-template-{self.variant}-" ) ) 

67 except ( OSError, IOError ) as exception: 

68 raise __.FileOperationFailure( 

69 __.Path( __.tempfile.gettempdir( ) ), 

70 "create directory" ) from exception 

71 _scribe.debug( f"Created temporary directory: {temporary_directory}" ) 

72 try: 

73 project_directory = temporary_directory / self.variant 

74 commands = _provide_validation_commands( 

75 repository_directory, project_directory ) 

76 _copy_template( 

77 answers_file, 

78 project_directory, 

79 repository_directory, 

80 ) 

81 _validate_variant_project( commands, repository_directory ) 

82 result = __.ValidationResult( 

83 variant = self.variant, 

84 temporary_directory = temporary_directory, 

85 items_attempted = len( commands ) + 1, 

86 items_generated = len( commands ) + 1, 

87 preserved = self.preserve, 

88 ) 

89 finally: 

90 if not self.preserve: 

91 _scribe.debug( 

92 f"Cleaning up temporary directory: {temporary_directory}" ) 

93 with __.ctxl.suppress( OSError, IOError ): 

94 __.shutil.rmtree( temporary_directory ) 

95 await __.render_and_print_result( 

96 result, auxdata.display, auxdata.exits ) 

97 

98 

99class CommandDispatcher( __.appcore_cli.Command ): 

100 ''' Dispatches maintainer commands for Copier template workflows. ''' 

101 

102 command: __.typx.Union[ 

103 __.typx.Annotated[ 

104 SurveyCommand, 

105 __.tyro.conf.subcommand( 'survey', prefix_name = False ), 

106 ], 

107 __.typx.Annotated[ 

108 ValidateCommand, 

109 __.tyro.conf.subcommand( 'validate', prefix_name = False ), 

110 ], 

111 ] = __.dcls.field( default_factory = SurveyCommand ) 

112 

113 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride] 

114 await self.command( auxdata ) 

115 

116 

117def survey_variants( auxdata: __.Globals ) -> tuple[ str, ... ]: 

118 ''' Surveys available template configuration variants. ''' 

119 return __.cmdbase.survey_variant_names( auxdata ) 

120 

121 

122def _copy_template( 

123 answers_file: __.Path, 

124 project_directory: __.Path, 

125 repository_directory: __.Path, 

126) -> None: 

127 ''' Copies template to temporary project directory using answers file. ''' 

128 command = ( 

129 'copier', 'copy', 

130 '--data-file', str( answers_file ), 

131 '--defaults', 

132 '--overwrite', 

133 '--vcs-ref', 'HEAD', 

134 '.', str( project_directory ), 

135 ) 

136 _run_checked_command( command, repository_directory ) 

137 

138 

139def _provide_repository_directory( ) -> __.Path: 

140 ''' Provides repository directory for template-copy operations. ''' 

141 repository_directory = __.Path.cwd( ) 

142 if not ( repository_directory / 'copier.yaml' ).is_file( ): 

143 raise __.ConfigurationAbsence( repository_directory ) 

144 if not ( repository_directory / 'template' ).is_dir( ): 

145 raise __.ConfigurationAbsence( repository_directory ) 

146 return repository_directory 

147 

148 

149def _provide_validation_commands( 

150 repository_directory: __.Path, 

151 project_directory: __.Path, 

152) -> tuple[ tuple[ str, ... ], ... ]: 

153 ''' Provides validation commands for generated template project. ''' 

154 # Keep template validation narrowly focused on Copier rendering. 

155 # Content generation is validated via `agentsmgr-maintain content`. 

156 return ( 

157 ( 

158 'hatch', '--env', 'develop', 'run', 

159 'agentsmgr', 'detect', 

160 '--source', str( project_directory ), 

161 ), 

162 ) 

163 

164 

165def _run_checked_command( 

166 command: tuple[ str, ... ], cwd: __.Path 

167) -> None: 

168 ''' Runs command and converts failures into configuration errors. ''' 

169 try: __.subprocess.run( command, cwd = cwd, check = True ) 

170 except FileNotFoundError as exception: 

171 raise __.ConfigurationInvalidity( exception ) from exception 

172 except __.subprocess.CalledProcessError as exception: 

173 raise __.ConfigurationInvalidity( exception ) from exception 

174 

175 

176def _validate_variant_project( 

177 commands: tuple[ tuple[ str, ... ], ... ], 

178 repository_directory: __.Path, 

179) -> None: 

180 ''' Validates generated project with configured command sequence. ''' 

181 for command in commands: 

182 _scribe.debug( f"Running validation command: {' '.join( command )}" ) 

183 _run_checked_command( command, repository_directory )