Coverage for quibble/util.py: 88%

1# Copyright 2017-2018, Antoine "hashar" Musso 

2# Copyright 2017, Tyler Cipriani 

3# Copyright 2017-2018, Wikimedia Foundation Inc. 

4# 

5# Licensed under the Apache License, Version 2.0 (the "License"); 

6# you may not use this file except in compliance with the License. 

7# You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, software 

12# distributed under the License is distributed on an "AS IS" BASIS, 

13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

14# See the License for the specific language governing permissions and 

15# limitations under the License. 

16 

17import contextlib 

18import json 

19import logging 

20import os 

21import urllib.request 

22from shutil import copyfile 

23import sys 

24import threading 

25import time 

26 

27log = logging.getLogger(__name__) 

28 

29 

30def copylog(src, dest): 

31 log.info('Copying %s to %s', src, dest) 

32 copyfile(src, dest) 

33 

34 

35def isCoreOrVendor(project): 

36 """ 

37 project: a gerrit repository name 

38 

39 Returns boolean, whether the repository is mediawiki/core or 

40 mediawiki/vendor. 

41 """ 

42 return project == 'mediawiki/core' or project == 'mediawiki/vendor' 

43 

44 

45def isExtOrSkin(project): 

46 """ 

47 project: a gerrit repository name 

48 

49 Returns boolean, whether the repository is a MediaWiki extension or skin. 

50 """ 

51 return project.startswith(('mediawiki/extensions/', 'mediawiki/skins/')) 

52 

53 

54def move_item_to_head(dependencies, project): 

55 repos = list(dependencies) 

56 repos.insert(0, repos.pop(repos.index(project))) 

57 return repos 

58 

59 

60@contextlib.contextmanager 

61def _redirect_stream(source, sink): 

62 """Redirects at the OS level, so that the new pipes are inherited by 

63 subprocesses. 

64 

65 Can't reuse contextlib redirectors here because they only affect the Python 

66 `std*` globals. 

67 """ 

68 old_source_fileno = os.dup(source.fileno()) 

69 os.dup2(sink.fileno(), source.fileno()) 

70 

71 yield 

72 source.flush() 

73 

74 os.dup2(old_source_fileno, source.fileno()) 

75 

76 

77class BytesStreamHandler(logging.StreamHandler): 

78 """ 

79 Logging handling converting received strings to bytes 

80 

81 This is used when redirecting logging to TemporaryFile() which by default 

82 is a binary file. The write() expects a byte like object. 

83 

84 Invalid unicode characters are replaced by backslashreplace. 

85 

86 With Python 3.8 we can remove it and use: 

87 

88 TemporaryFile(errors='backslashreplace') 

89 """ 

90 

91 def __init__(self, stream): 

92 logging.StreamHandler.__init__(self, stream) 

93 

94 def emit(self, record): 

95 msg = self.format(record) 

96 self.stream.write( 

97 bytes( 

98 msg + self.terminator, 

99 encoding='utf-8', 

100 errors='backslashreplace', 

101 ) 

102 ) 

103 self.stream.flush() 

104 

105 

106@contextlib.contextmanager 

107def _redirect_logging(sink): 

108 """Redirect logging to a single stream, and reconnect when finished.""" 

109 log_handler = BytesStreamHandler(sink) 

110 

111 logger = logging.getLogger() 

112 old_handlers = logger.handlers 

113 for handler in old_handlers: 

114 logger.removeHandler(handler) 

115 logger.addHandler(log_handler) 

116 

117 yield 

118 log_handler.flush() 

119 

120 logger.removeHandler(log_handler) 

121 for handler in old_handlers: 

122 logger.addHandler(handler) 

123 log_handler.close() 

124 

125 

126@contextlib.contextmanager 

127def redirect_all_streams(sink): 

128 """Redirect stdout, stderr, and logging to a single stream.""" 

129 with _redirect_logging(sink), _redirect_stream( 

130 sys.stdout, sink 

131 ), _redirect_stream(sys.stderr, sink): 

132 yield 

133 

134 

135class ProgressReporter: 

136 """Report job progress at regular intervals, wraps an iterable and tracks 

137 how many items have been served from it. 

138 

139 Inspired by tqdm. 

140 """ 

141 

142 def __init__(self, *, iterable, desc, sleep_interval, total): 

143 self.iterable = iterable 

144 self.desc = desc 

145 self.completed = 0 

146 self.total = total 

147 self.start_time = time.time() 

148 self.sleep_interval = sleep_interval 

149 

150 self.monitor = _RepeatingTimer(self.sleep_interval, self._refresh) 

151 

152 def __iter__(self): 

153 self.monitor.start() 

154 

155 for obj in self.iterable: 

156 yield obj 

157 

158 self.completed += 1 

159 

160 self.monitor.cancel() 

161 

162 def _refresh(self): 

163 elapsed = int(time.time() - self.start_time) 

164 log.debug( 

165 "Waiting for %s: %ss elapsed, %s/%s completed", 

166 self.desc, 

167 elapsed, 

168 self.completed, 

169 self.total, 

170 ) 

171 

172 

173class _RepeatingTimer(threading.Timer): 

174 def __init__(self, *args, **kwargs): 

175 # This is a daemon thread so that it can be immediately killed if the 

176 # program crashes before fully consuming the iterator. Otherwise, the 

177 # `self.finished` Event might never receive the flag set by 

178 # `self.monitor.cancel` above. 

179 super(_RepeatingTimer, self).__init__(*args, **kwargs) 

180 self.daemon = True 

181 

182 def run(self): 

183 while not self.finished.wait(self.interval): 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true

184 self.function(*self.args, **self.kwargs) 

185 

186 

187class FetchInfo: 

188 url = None 

189 project = None 

190 branch = None 

191 ref = None 

192 

193 @staticmethod 

194 def change(change, patchset=None): 

195 fetchinfo = FetchInfo() 

196 

197 url = 'https://gerrit.wikimedia.org/r/changes/?q=change:%s' % change 

198 if patchset: 

199 url += '&o=CURRENT_REVISION' 

200 else: 

201 url += '&o=ALL_REVISIONS' 

202 

203 change_info = FetchInfo.fetch(url) 

204 

205 fetchinfo.project = change_info['project'] 

206 fetchinfo.branch = change_info['branch'] 

207 

208 def get_url_ref(rev): 

209 fetch = rev['fetch']['anonymous http'] 

210 # url, reference 

211 return (fetch['url'][: -len(fetchinfo.project)], fetch['ref']) 

212 

213 if patchset is None: 

214 current = change_info['current_revision'] 

215 (fetchinfo.url, fetchinfo.ref) = get_url_ref( 

216 change_info['revisions'][current] 

217 ) 

218 return fetchinfo 

219 else: 

220 for sha1, rev_data in change_info['revisions'].items(): 220 ↛ 224line 220 didn't jump to line 224 because the loop on line 220 didn't complete

221 if rev_data["_number"] == patchset: 221 ↛ 220line 221 didn't jump to line 220 because the condition on line 221 was always true

222 (fetchinfo.url, fetchinfo.ref) = get_url_ref(rev_data) 

223 return fetchinfo 

224 raise Exception( 

225 "Could not find metadata for %s,%s" % (change, patchset) 

226 ) 

227 

228 @staticmethod 

229 def fetch(url): 

230 with urllib.request.urlopen(url) as f: 

231 f.readline() # consumes "')]}'\n" 

232 changes = json.loads(f.readline().decode('utf-8')) 

233 if len(changes) != 1: 

234 raise Exception("Got multiple changes from %s" % url) 

235 return changes[0] 

236 

237 def asZuulEnv(self): 

238 return { 

239 'ZUUL_URL': self.url, 

240 'ZUUL_PROJECT': self.project, 

241 'ZUUL_BRANCH': self.branch, 

242 'ZUUL_REF': self.ref, 

243 } 

244 

245 

246# The strtobool code has been copied from Python which removed it with v3.12. 

247# There is most probably NO reason to touch this code. 

248def strtobool(val): 

249 """Convert a string representation of truth to true (1) or false (0). 

250 

251 True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values 

252 are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if 

253 'val' is anything else. 

254 """ 

255 val = val.lower() 

256 if val in ('y', 'yes', 't', 'true', 'on', '1'): 

257 return 1 

258 elif val in ('n', 'no', 'f', 'false', 'off', '0'): 258 ↛ 261line 258 didn't jump to line 261 because the condition on line 258 was always true

259 return 0 

260 else: 

261 raise ValueError("invalid truth value %r" % (val,))