Coverage for sources/agentsmgr/operations.py: 9%

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''' Core operations for content generation and directory population. 

22 

23 This module provides functions for orchestrating content generation, 

24 including directory population and file writing operations with 

25 simulation support. 

26''' 

27 

28 

29from . import __ 

30from . import exceptions as _exceptions 

31from . import generator as _generator 

32 

33 

34def populate_directory( 

35 generator: _generator.ContentGenerator, 

36 target: __.Path, 

37 simulate: bool = False 

38) -> tuple[ int, int ]: 

39 ''' Generates all content items to target directory. 

40 

41 Orchestrates content generation for all coders and item types 

42 configured in generator. Returns tuple of (items_attempted, 

43 items_written). 

44 ''' 

45 items_attempted = 0 

46 items_written = 0 

47 for coder_name in generator.configuration[ 'coders' ]: 

48 for item_type in ( 'commands', 'agents' ): 

49 attempted, written = produce_coder_item_type( 

50 generator, coder_name, item_type, target, simulate ) 

51 items_attempted += attempted 

52 items_written += written 

53 return ( items_attempted, items_written ) 

54 

55 

56def produce_coder_item_type( 

57 generator: _generator.ContentGenerator, 

58 coder: str, 

59 item_type: str, 

60 target: __.Path, 

61 simulate: bool 

62) -> tuple[ int, int ]: 

63 ''' Produces items of specific type for a coder. 

64 

65 Generates all items (commands or agents) for specified coder by 

66 iterating through configuration files. Returns tuple of 

67 (items_attempted, items_written). 

68 ''' 

69 items_attempted = 0 

70 items_written = 0 

71 if generator.mode == 'nowhere': 

72 return ( items_attempted, items_written ) 

73 configuration_directory = ( 

74 generator.location / 'configurations' / item_type ) 

75 if not configuration_directory.exists( ): 

76 return ( items_attempted, items_written ) 

77 for configuration_file in configuration_directory.glob( '*.toml' ): 

78 items_attempted += 1 

79 result = generator.render_single_item( 

80 item_type, configuration_file.stem, coder, target ) 

81 if update_content( result.content, result.location, simulate ): 

82 items_written += 1 

83 return ( items_attempted, items_written ) 

84 

85 

86def update_content( 

87 content: str, location: __.Path, simulate: bool = False 

88) -> bool: 

89 ''' Updates content file, creating directories as needed. 

90 

91 Writes content to specified location, creating parent directories 

92 if necessary. In simulation mode, no actual writing occurs. 

93 Returns True if file was written, False if simulated. 

94 ''' 

95 if simulate: return False 

96 try: location.parent.mkdir( parents = True, exist_ok = True ) 

97 except ( OSError, IOError ) as exception: 

98 raise _exceptions.FileOperationFailure( 

99 location.parent, "create directory" ) from exception 

100 try: location.write_text( content, encoding = 'utf-8' ) 

101 except ( OSError, IOError ) as exception: 

102 raise _exceptions.FileOperationFailure( 

103 location, "update content" ) from exception 

104 return True 

105 

106 

107def update_git_exclude( 

108 target: __.Path, 

109 symlinks: __.cabc.Sequence[ str ], 

110 simulate: bool = False 

111) -> int: 

112 ''' Updates .git/info/exclude with symlink names if not already present. 

113 

114 Adds symlink names to git exclude file to prevent accidental 

115 commits of generated symlinks. Processes file line-by-line to 

116 preserve existing content and avoid duplicates. 

117 

118 Returns count of symlink names added to exclude file. 

119 ''' 

120 if simulate or not symlinks: return 0 

121 exclude_file = target / '.git' / 'info' / 'exclude' 

122 if not exclude_file.exists( ): return 0 

123 try: content = exclude_file.read_text( encoding = 'utf-8' ) 

124 except ( OSError, IOError ) as exception: 

125 raise _exceptions.FileOperationFailure( 

126 exclude_file, "read git exclude file" ) from exception 

127 existing_lines = content.splitlines( ) 

128 existing_patterns = frozenset( existing_lines ) 

129 additions = [ 

130 symlink for symlink in symlinks 

131 if symlink not in existing_patterns 

132 ] 

133 if not additions: return 0 

134 new_content_lines = existing_lines.copy( ) 

135 if new_content_lines and not new_content_lines[ -1 ].strip( ): 

136 new_content_lines.extend( additions ) 

137 else: 

138 new_content_lines.append( '' ) 

139 new_content_lines.extend( additions ) 

140 new_content = '\n'.join( new_content_lines ) 

141 if not new_content.endswith( '\n' ): new_content += '\n' 

142 try: exclude_file.write_text( new_content, encoding = 'utf-8' ) 

143 except ( OSError, IOError ) as exception: 

144 raise _exceptions.FileOperationFailure( 

145 exclude_file, "update git exclude file" ) from exception 

146 return len( additions )