Coverage for sources / agentsmgr / maintenance / content.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 validating generated content in temporary directories. ''' 

22 

23 

24import yaml as _yaml 

25 

26from . import __ 

27 

28 

29_scribe = __.provide_scribe( __name__ ) 

30 

31 

32class ValidateCommand( __.appcore_cli.Command ): 

33 ''' Validates content generation in temporary directory. ''' 

34 

35 variant: __.typx.Annotated[ 

36 str, 

37 __.typx.Doc( ''' Configuration variant to test. ''' ), 

38 __.tyro.conf.Positional, 

39 ] = 'default' 

40 preserve: __.typx.Annotated[ 

41 bool, 

42 __.tyro.conf.arg( 

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

44 prefix_name = False ), 

45 ] = False 

46 

47 @__.cmdbase.intercept_errors( ) 

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

49 ''' Validates content generation and displays result. ''' 

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

51 raise __.ContextInvalidity 

52 _scribe.info( f"Validating content generation for {self.variant}" ) 

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

54 prefix = f"agents-validate-{self.variant}-" ) ) 

55 except ( OSError, IOError ) as exception: 

56 raise __.FileOperationFailure( 

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

58 "create directory" ) from exception 

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

60 try: 

61 configuration = self._produce_test_configuration( auxdata ) 

62 location = __.cmdbase.retrieve_data_location( "defaults" ) 

63 generator = __.generator.ContentGenerator( 

64 location = location, configuration = configuration ) 

65 items_attempted, items_generated = ( 

66 __.operations.populate_directory( 

67 generator, temporary_directory, simulate = False ) ) 

68 _scribe.info( 

69 f"Generated {items_generated}/{items_attempted} items" ) 

70 finally: 

71 if not self.preserve: 

72 _scribe.debug( 

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

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

75 __.shutil.rmtree( temporary_directory ) 

76 result = __.ValidationResult( 

77 variant = self.variant, 

78 temporary_directory = temporary_directory, 

79 items_attempted = items_attempted, 

80 items_generated = items_generated, 

81 preserved = self.preserve, 

82 ) 

83 await __.render_and_print_result( 

84 result, auxdata.display, auxdata.exits ) 

85 

86 def _produce_test_configuration( 

87 self, auxdata: __.Globals 

88 ) -> __.cmdbase.CoderConfiguration: 

89 ''' Produces test configuration for specified variant. 

90 

91 Creates configuration by loading variant answers file from 

92 data directory. 

93 ''' 

94 answers_file = __.cmdbase.retrieve_variant_answers_file( 

95 auxdata, self.variant ) 

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

97 except ( OSError, IOError ) as exception: 

98 raise __.ConfigurationAbsence( ) from exception 

99 try: 

100 configuration: __.cmdbase.CoderConfiguration = ( 

101 _yaml.safe_load( content ) ) 

102 except _yaml.YAMLError as exception: 

103 raise __.ConfigurationInvalidity( exception ) from exception 

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

105 raise __.ConfigurationInvalidity( ) 

106 return __.immut.Dictionary( configuration ) 

107 

108 

109class SurveyCommand( __.appcore_cli.Command ): 

110 ''' Surveys available configuration variants for content validation. ''' 

111 

112 @__.cmdbase.intercept_errors( ) 

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

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

115 raise __.ContextInvalidity 

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

117 for variant in survey_variants( auxdata ): 

118 print( variant, file = stream ) 

119 

120 

121class CommandDispatcher( __.appcore_cli.Command ): 

122 ''' Dispatches maintainer commands for generated content validation. ''' 

123 

124 command: __.typx.Union[ 

125 __.typx.Annotated[ 

126 SurveyCommand, 

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

128 ], 

129 __.typx.Annotated[ 

130 ValidateCommand, 

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

132 ], 

133 ] = __.dcls.field( default_factory = ValidateCommand ) 

134 

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

136 await self.command( auxdata ) 

137 

138 

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

140 ''' Surveys available configuration variants. ''' 

141 return __.cmdbase.survey_variant_names( auxdata )