Coverage for sources/copiertv/cli.py: 69%

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''' Command-line interface. ''' 

22 

23 

24from . import __ 

25from . import configuration as _configuration 

26from . import engine as _engine 

27from . import exceptions as _exceptions 

28from . import state as _state 

29 

30 

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

32 [ __.cabc.Callable[ 

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

34 __.cabc.Callable[ 

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

36]: 

37 ''' Decorator that catches Omnierror for CLI display. ''' 

38 def decorator( 

39 function: __.cabc.Callable[ 

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

41 ) -> __.cabc.Callable[ 

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

43 ]: 

44 @__.functools.wraps( function ) 

45 async def wrapper( 

46 *args: __.typx.Any, **kwargs: __.typx.Any 

47 ) -> None: 

48 try: await function( *args, **kwargs ) 

49 except _exceptions.Omnierror as exc: 

50 renderer = getattr( 

51 exc, 'render_as_markdown', None ) 

52 if renderer: 52 ↛ 55line 52 didn't jump to line 55 because the condition on line 52 was always true

53 lines = renderer( ) 

54 print( '\n'.join( lines ) ) 

55 raise SystemExit( 1 ) from exc 

56 return wrapper 

57 return decorator 

58 

59 

60async def _survey( config: _configuration.Configuration ) -> None: 

61 ''' Lists discovered template variants. ''' 

62 answers_dir = config.answers_directory 

63 if __.is_absent( answers_dir ): 

64 from .exceptions import ConfigurationInvalidity 

65 raise ConfigurationInvalidity( 'answers directory' ) # noqa: TRY003 

66 for variant in _engine.survey_variants( answers_dir ): 

67 print( variant ) 

68 

69 

70async def _validate( 

71 variant: str, 

72 config: _configuration.Configuration, 

73) -> None: 

74 ''' Validates a template variant. ''' 

75 result = _engine.validate_variant( variant, config ) 

76 lines = result.render_as_markdown( ) 

77 print( '\n'.join( lines ) ) 

78 

79 

80class _SurveyCommand( __.appcore_cli.Command ): 

81 ''' Surveys available template configuration variants. ''' 

82 

83 @intercept_errors( ) 

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

85 if not isinstance( auxdata, _state.Globals ): 

86 raise _exceptions.ConfigurationInvalidity( ) 

87 await _survey( auxdata.copiertv_configuration ) 

88 

89 

90class _ValidateCommand( __.appcore_cli.Command ): 

91 ''' Validates template against configuration variant. ''' 

92 

93 variant: __.typx.Annotated[ 

94 str, 

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

96 __.tyro.conf.Positional, 

97 ] 

98 preserve: __.typx.Annotated[ 

99 bool, 

100 __.tyro.conf.arg( 

101 help = 'Keep temporary files for inspection.', 

102 prefix_name = False ), 

103 ] = False 

104 

105 @intercept_errors( ) 

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

107 if not isinstance( auxdata, _state.Globals ): 

108 raise _exceptions.ConfigurationInvalidity( ) 

109 cli_config = _configuration.Configuration( 

110 preserve = self.preserve ) 

111 config = _configuration.merge_configurations( 

112 auxdata.copiertv_configuration, cli_config ) 

113 await _validate( self.variant, config ) 

114 

115 

116class _Application( __.appcore_cli.Application ): 

117 ''' Copiertv CLI application. ''' 

118 

119 command: __.typx.Union[ 

120 __.typx.Annotated[ 

121 _SurveyCommand, 

122 __.tyro.conf.subcommand( 

123 'survey', prefix_name = False ), 

124 ], 

125 __.typx.Annotated[ 

126 _ValidateCommand, 

127 __.tyro.conf.subcommand( 

128 'validate', prefix_name = False ), 

129 ], 

130 ] = __.dcls.field( default_factory = _SurveyCommand ) 

131 

132 async def execute( self, auxdata: __.Globals ) -> None: 

133 ''' Dispatches to the selected command. ''' 

134 await self.command( auxdata ) 

135 

136 async def prepare( 

137 self, exits: __.ctxl.AsyncExitStack 

138 ) -> _state.Globals: 

139 ''' Prepares copiertv-specific global state. ''' 

140 auxdata_base = await super( ).prepare( exits ) 

141 config = _configuration.acquire_configuration( 

142 auxdata_base.configuration ) 

143 return _state.Globals( 

144 copiertv_configuration = config, 

145 **{ 

146 field.name: getattr( auxdata_base, field.name ) 

147 for field in __.dcls.fields( auxdata_base ) 

148 if not field.name.startswith( '_' ) } ) 

149 

150 

151def execute( ) -> None: 

152 ''' Entrypoint for CLI execution. ''' 

153 config = ( 

154 __.tyro.conf.EnumChoicesFromValues, 

155 __.tyro.conf.HelptextFromCommentsOff, 

156 ) 

157 try: __.asyncio.run( __.tyro.cli( _Application, config = config )( ) ) 

158 except SystemExit: raise 

159 except BaseException: raise SystemExit( 1 ) from None