Coverage for sources / mimeogram / apply.py: 89%

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''' Application of mimeograms. ''' 

22# TODO? Use BSD sysexits. 

23 

24 

25from . import __ 

26from . import exceptions as _exceptions 

27from . import interfaces as _interfaces 

28from . import parts as _parts 

29from . import updaters as _updaters 

30 

31 

32_scribe = __.produce_scribe( __name__ ) 

33 

34 

35class Command( 

36 _interfaces.CliCommand, 

37 decorators = ( __.standard_tyro_class, ), 

38): 

39 ''' Applies mimeogram to filesystem locations. ''' 

40 

41 source: __.typx.Annotated[ 

42 str, 

43 __.typx.Doc( 

44 ''' Source file for mimeogram. 

45 

46 Defaults to stdin if '--clip' not specified. 

47 ''' ), 

48 ] = '-' 

49 clip: __.typx.Annotated[ 

50 __.tyro.conf.DisallowNone[ bool | None ], 

51 __.typx.Doc( 

52 ''' Read mimeogram from clipboard instead of file or stdin. ''' ), 

53 __.tyro.conf.arg( aliases = ( '--clipboard', '--from-clipboard' ) ), 

54 ] = None 

55 mode: __.typx.Annotated[ 

56 __.typx.Optional[ _updaters.ReviewModes ], 

57 __.typx.Doc( 

58 ''' Controls how changes are reviewed. 

59 

60 'silent': Apply without review. 

61 'partitive': Review each change interactively. 

62 

63 Partitive, if not specified and on a terminal. 

64 Silent, if not specified and not on a terminal. 

65 ''' ), 

66 __.tyro.conf.arg( aliases = ( '--review-mode', ) ), 

67 ] = None 

68 base: __.typx.Annotated[ 

69 __.typx.Optional[ __.Path ], 

70 __.typx.Doc( 

71 ''' Base directory for relative locations. 

72 

73 Defaults to current working directory. 

74 ''' ), 

75 __.tyro.conf.arg( aliases = ( '--base-directory', ) ), 

76 ] = None 

77 force: __.typx.Annotated[ 

78 __.tyro.conf.DisallowNone[ bool | None ], 

79 __.typx.Doc( '''Override protected path checks.''' ), 

80 ] = None 

81 

82 async def __call__( 

83 self, auxdata: __.appcore.state.Globals 

84 ) -> None: 

85 ''' Executes command to apply mimeogram. ''' 

86 await apply( auxdata, self ) 

87 

88 def provide_configuration_edits( 

89 self, 

90 ) -> __.appcore.dictedits.Edits: 

91 ''' Provides edits against configuration from options. ''' 

92 edits: list[ __.appcore.dictedits.Edit ] = [ ] 

93 if None is not self.clip: 

94 edits.append( __.appcore.dictedits.SimpleEdit( # pyright: ignore 

95 address = ( 'apply', 'from-clipboard' ), value = self.clip ) ) 

96 if None is not self.force: 

97 edits.append( __.appcore.dictedits.SimpleEdit( # pyright: ignore 

98 address = ( 'update-parts', 'disable-protections' ), 

99 value = self.force ) ) 

100 return tuple( edits ) 

101 

102 

103class ContentAcquirer( 

104 __.immut.DataclassProtocol, __.typx.Protocol, 

105 decorators = ( __.typx.runtime_checkable, ), 

106): 

107 ''' Acquires content for apply command. ''' 

108 

109 @__.abc.abstractmethod 

110 def stdin_is_tty( self ) -> bool: 

111 ''' Checks if input is from a terminal. ''' 

112 raise NotImplementedError 

113 

114 @__.abc.abstractmethod 

115 async def acquire_clipboard( self ) -> str: 

116 ''' Acquires content from clipboard. ''' 

117 raise NotImplementedError 

118 

119 @__.abc.abstractmethod 

120 async def acquire_file( self, path: str | __.Path ) -> str: 

121 ''' Acquires content from file. ''' 

122 raise NotImplementedError 

123 

124 @__.abc.abstractmethod 

125 async def acquire_stdin( self ) -> str: 

126 ''' Acquires content from standard input. ''' 

127 raise NotImplementedError 

128 

129 

130class StandardContentAcquirer( ContentAcquirer ): 

131 ''' Standard implementation of content acquisition. ''' 

132 

133 def stdin_is_tty( self ) -> bool: 

134 return __.sys.stdin.isatty( ) 

135 

136 async def acquire_clipboard( self ) -> str: 

137 from . import clipboard 

138 return clipboard.copy_from_clipboard( ) 

139 

140 async def acquire_file( self, path: str | __.Path ) -> str: 

141 return await __.appcore.io.acquire_text_file_async( path ) 

142 

143 async def acquire_stdin( self ) -> str: 

144 return __.sys.stdin.read( ) 

145 

146 

147async def apply( 

148 auxdata: __.appcore.state.Globals, 

149 command: Command, 

150 *, 

151 acquirer: __.Absential[ ContentAcquirer ] = __.absent, 

152 parser: __.Absential[ 

153 __.cabc.Callable[ [ str ], __.cabc.Sequence[ _parts.Part ] ] 

154 ] = __.absent, 

155 updater: __.Absential[ 

156 __.cabc.Callable[ 

157 [ __.appcore.state.Globals, 

158 __.cabc.Sequence[ _parts.Part ], 

159 _updaters.ReviewModes ], 

160 __.cabc.Coroutine[ None, None, None ] 

161 ] 

162 ] = __.absent, 

163) -> __.typx.Never: 

164 ''' Applies mimeogram. ''' 

165 if __.is_absent( acquirer ): 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true

166 acquirer = StandardContentAcquirer( ) 

167 if __.is_absent( parser ): 

168 from .parsers import parse as parser 

169 if __.is_absent( updater ): 

170 from .updaters import update as updater 

171 review_mode = _determine_review_mode( command, acquirer ) 

172 with _exceptions.report_exceptions( 

173 _scribe, "Could not acquire mimeogram to apply." 

174 ): mgtext = await _acquire( auxdata, command, acquirer ) 

175 if not mgtext: 

176 _scribe.error( "Cannot apply empty mimeogram." ) 

177 raise SystemExit( 1 ) 

178 with _exceptions.report_exceptions( 

179 _scribe, "Could not parse mimeogram." 

180 ): 

181 parts = parser( mgtext ) 

182 nomargs: dict[ str, __.typx.Any ] = { } 

183 if command.base: nomargs[ 'base' ] = command.base 

184 with _exceptions.report_exceptions( 

185 _scribe, "Could not apply mimeogram." 

186 ): 

187 await updater( auxdata, parts, review_mode, **nomargs ) 

188 # TODO: If all parts ignored or inapplicable, then do not mention success. 

189 _scribe.info( "Successfully applied mimeogram" ) 

190 raise SystemExit( 0 ) 

191 

192 

193async def _acquire( 

194 auxdata: __.appcore.state.Globals, 

195 cmd: Command, 

196 acquirer: ContentAcquirer, 

197) -> str: 

198 ''' Acquires content to parse from clipboard, file, or stdin. ''' 

199 options = auxdata.configuration.get( 'apply', { } ) 

200 if options.get( 'from-clipboard', False ): 

201 content = await acquirer.acquire_clipboard( ) 

202 if not content: 

203 _scribe.error( "Clipboard is empty." ) 

204 raise SystemExit( 1 ) 

205 _scribe.debug( 

206 "Read {} characters from clipboard.".format( len( content ) ) ) 

207 return content 

208 match cmd.source: 

209 case '-': return await acquirer.acquire_stdin( ) 

210 case _: return await acquirer.acquire_file( cmd.source ) 

211 

212 

213def _determine_review_mode( 

214 command: Command, acquirer: ContentAcquirer 

215) -> _updaters.ReviewModes: 

216 on_tty = acquirer.stdin_is_tty( ) 

217 if command.mode is None: 

218 if on_tty: return _updaters.ReviewModes.Partitive 

219 return _updaters.ReviewModes.Silent 

220 if not on_tty and command.mode is not _updaters.ReviewModes.Silent: 

221 _scribe.error( "Cannot use an interactive mode without terminal." ) 

222 raise SystemExit( 1 ) 

223 return command.mode