40 return ISOLATED_MODULES[mod] |
48 return ISOLATED_MODULES[mod] |
41 |
49 |
42 os = isolate_module(os) |
50 os = isolate_module(os) |
43 |
51 |
44 |
52 |
|
53 class SysModuleSaver: |
|
54 """Saves the contents of sys.modules, and removes new modules later.""" |
|
55 def __init__(self): |
|
56 self.old_modules = set(sys.modules) |
|
57 |
|
58 def restore(self): |
|
59 """Remove any modules imported since this object started.""" |
|
60 new_modules = set(sys.modules) - self.old_modules |
|
61 for m in new_modules: |
|
62 del sys.modules[m] |
|
63 |
|
64 |
|
65 @contextlib.contextmanager |
|
66 def sys_modules_saved(): |
|
67 """A context manager to remove any modules imported during a block.""" |
|
68 saver = SysModuleSaver() |
|
69 try: |
|
70 yield |
|
71 finally: |
|
72 saver.restore() |
|
73 |
|
74 |
|
75 def import_third_party(modname): |
|
76 """Import a third-party module we need, but might not be installed. |
|
77 |
|
78 This also cleans out the module after the import, so that coverage won't |
|
79 appear to have imported it. This lets the third party use coverage for |
|
80 their own tests. |
|
81 |
|
82 Arguments: |
|
83 modname (str): the name of the module to import. |
|
84 |
|
85 Returns: |
|
86 The imported module, or None if the module couldn't be imported. |
|
87 |
|
88 """ |
|
89 with sys_modules_saved(): |
|
90 try: |
|
91 return importlib.import_module(modname) |
|
92 except ImportError: |
|
93 return None |
|
94 |
|
95 |
45 def dummy_decorator_with_args(*args_unused, **kwargs_unused): |
96 def dummy_decorator_with_args(*args_unused, **kwargs_unused): |
46 """Dummy no-op implementation of a decorator with arguments.""" |
97 """Dummy no-op implementation of a decorator with arguments.""" |
47 def _decorator(func): |
98 def _decorator(func): |
48 return func |
99 return func |
49 return _decorator |
100 return _decorator |
50 |
101 |
51 |
102 |
52 # Environment COVERAGE_NO_CONTRACTS=1 can turn off contracts while debugging |
|
53 # tests to remove noise from stack traces. |
|
54 # $set_env.py: COVERAGE_NO_CONTRACTS - Disable PyContracts to simplify stack traces. |
|
55 USE_CONTRACTS = env.TESTING and not bool(int(os.environ.get("COVERAGE_NO_CONTRACTS", 0))) |
|
56 |
|
57 # Use PyContracts for assertion testing on parameters and returns, but only if |
103 # Use PyContracts for assertion testing on parameters and returns, but only if |
58 # we are running our own test suite. |
104 # we are running our own test suite. |
59 if USE_CONTRACTS: |
105 if env.USE_CONTRACTS: |
60 from contracts import contract # pylint: disable=unused-import |
106 from contracts import contract # pylint: disable=unused-import |
61 from contracts import new_contract as raw_new_contract |
107 from contracts import new_contract as raw_new_contract |
62 |
108 |
63 def new_contract(*args, **kwargs): |
109 def new_contract(*args, **kwargs): |
64 """A proxy for contracts.new_contract that doesn't mind happening twice.""" |
110 """A proxy for contracts.new_contract that doesn't mind happening twice.""" |
69 # PyContracts doesn't like redefining contracts. It's OK. |
115 # PyContracts doesn't like redefining contracts. It's OK. |
70 pass |
116 pass |
71 |
117 |
72 # Define contract words that PyContract doesn't have. |
118 # Define contract words that PyContract doesn't have. |
73 new_contract('bytes', lambda v: isinstance(v, bytes)) |
119 new_contract('bytes', lambda v: isinstance(v, bytes)) |
74 if env.PY3: |
120 new_contract('unicode', lambda v: isinstance(v, str)) |
75 new_contract('unicode', lambda v: isinstance(v, unicode_class)) |
|
76 |
121 |
77 def one_of(argnames): |
122 def one_of(argnames): |
78 """Ensure that only one of the argnames is non-None.""" |
123 """Ensure that only one of the argnames is non-None.""" |
79 def _decorator(func): |
124 def _decorator(func): |
80 argnameset = {name.strip() for name in argnames.split(",")} |
125 argnameset = {name.strip() for name in argnames.split(",")} |
154 def ensure_dir(directory): |
199 def ensure_dir(directory): |
155 """Make sure the directory exists. |
200 """Make sure the directory exists. |
156 |
201 |
157 If `directory` is None or empty, do nothing. |
202 If `directory` is None or empty, do nothing. |
158 """ |
203 """ |
159 if directory and not os.path.isdir(directory): |
204 if directory: |
160 os.makedirs(directory) |
205 os.makedirs(directory, exist_ok=True) |
161 |
206 |
162 |
207 |
163 def ensure_dir_for_file(path): |
208 def ensure_dir_for_file(path): |
164 """Make sure the directory for the path exists.""" |
209 """Make sure the directory for the path exists.""" |
165 ensure_dir(os.path.dirname(path)) |
210 ensure_dir(os.path.dirname(path)) |
195 dice = random.Random(os.urandom(8)).randint(0, 999999) |
240 dice = random.Random(os.urandom(8)).randint(0, 999999) |
196 suffix = "%s.%s.%06d" % (socket.gethostname(), os.getpid(), dice) |
241 suffix = "%s.%s.%06d" % (socket.gethostname(), os.getpid(), dice) |
197 return suffix |
242 return suffix |
198 |
243 |
199 |
244 |
200 class Hasher(object): |
245 class Hasher: |
201 """Hashes Python data into md5.""" |
246 """Hashes Python data for fingerprinting.""" |
202 def __init__(self): |
247 def __init__(self): |
203 self.md5 = hashlib.md5() |
248 self.hash = hashlib.new("sha3_256") |
204 |
249 |
205 def update(self, v): |
250 def update(self, v): |
206 """Add `v` to the hash, recursively if needed.""" |
251 """Add `v` to the hash, recursively if needed.""" |
207 self.md5.update(to_bytes(str(type(v)))) |
252 self.hash.update(str(type(v)).encode("utf-8")) |
208 if isinstance(v, unicode_class): |
253 if isinstance(v, str): |
209 self.md5.update(v.encode('utf8')) |
254 self.hash.update(v.encode("utf-8")) |
210 elif isinstance(v, bytes): |
255 elif isinstance(v, bytes): |
211 self.md5.update(v) |
256 self.hash.update(v) |
212 elif v is None: |
257 elif v is None: |
213 pass |
258 pass |
214 elif isinstance(v, (int, float)): |
259 elif isinstance(v, (int, float)): |
215 self.md5.update(to_bytes(str(v))) |
260 self.hash.update(str(v).encode("utf-8")) |
216 elif isinstance(v, (tuple, list)): |
261 elif isinstance(v, (tuple, list)): |
217 for e in v: |
262 for e in v: |
218 self.update(e) |
263 self.update(e) |
219 elif isinstance(v, dict): |
264 elif isinstance(v, dict): |
220 keys = v.keys() |
265 keys = v.keys() |
228 a = getattr(v, k) |
273 a = getattr(v, k) |
229 if inspect.isroutine(a): |
274 if inspect.isroutine(a): |
230 continue |
275 continue |
231 self.update(k) |
276 self.update(k) |
232 self.update(a) |
277 self.update(a) |
233 self.md5.update(b'.') |
278 self.hash.update(b'.') |
234 |
279 |
235 def hexdigest(self): |
280 def hexdigest(self): |
236 """Retrieve the hex digest of the hash.""" |
281 """Retrieve the hex digest of the hash.""" |
237 return self.md5.hexdigest() |
282 return self.hash.hexdigest()[:32] |
238 |
283 |
239 |
284 |
240 def _needs_to_implement(that, func_name): |
285 def _needs_to_implement(that, func_name): |
241 """Helper to raise NotImplementedError in interface stubs.""" |
286 """Helper to raise NotImplementedError in interface stubs.""" |
242 if hasattr(that, "_coverage_plugin_name"): |
287 if hasattr(that, "_coverage_plugin_name"): |
243 thing = "Plugin" |
288 thing = "Plugin" |
244 name = that._coverage_plugin_name |
289 name = that._coverage_plugin_name |
245 else: |
290 else: |
246 thing = "Class" |
291 thing = "Class" |
247 klass = that.__class__ |
292 klass = that.__class__ |
248 name = "{klass.__module__}.{klass.__name__}".format(klass=klass) |
293 name = f"{klass.__module__}.{klass.__name__}" |
249 |
294 |
250 raise NotImplementedError( |
295 raise NotImplementedError( |
251 "{thing} {name!r} needs to implement {func_name}()".format( |
296 f"{thing} {name!r} needs to implement {func_name}()" |
252 thing=thing, name=name, func_name=func_name |
|
253 ) |
|
254 ) |
297 ) |
255 |
298 |
256 |
299 |
257 class DefaultValue(object): |
300 class DefaultValue: |
258 """A sentinel object to use for unusual default-value needs. |
301 """A sentinel object to use for unusual default-value needs. |
259 |
302 |
260 Construct with a string that will be used as the repr, for display in help |
303 Construct with a string that will be used as the repr, for display in help |
261 and Sphinx output. |
304 and Sphinx output. |
262 |
305 |
297 )? # maybe. |
340 )? # maybe. |
298 } |
341 } |
299 ) |
342 ) |
300 """ |
343 """ |
301 |
344 |
|
345 dollar_groups = ('dollar', 'word1', 'word2') |
|
346 |
302 def dollar_replace(match): |
347 def dollar_replace(match): |
303 """Called for each $replacement.""" |
348 """Called for each $replacement.""" |
304 # Only one of the groups will have matched, just get its text. |
349 # Only one of the groups will have matched, just get its text. |
305 word = next(g for g in match.group('dollar', 'word1', 'word2') if g) |
350 word = next(g for g in match.group(*dollar_groups) if g) # pragma: always breaks |
306 if word == "$": |
351 if word == "$": |
307 return "$" |
352 return "$" |
308 elif word in variables: |
353 elif word in variables: |
309 return variables[word] |
354 return variables[word] |
310 elif match.group('strict'): |
355 elif match.group('strict'): |
311 msg = "Variable {} is undefined: {!r}".format(word, text) |
356 msg = f"Variable {word} is undefined: {text!r}" |
312 raise CoverageException(msg) |
357 raise CoverageException(msg) |
313 else: |
358 else: |
314 return match.group('defval') |
359 return match.group('defval') |
315 |
360 |
316 text = re.sub(dollar_pattern, dollar_replace, text) |
361 text = re.sub(dollar_pattern, dollar_replace, text) |
317 return text |
362 return text |
318 |
363 |
319 |
364 |
320 class BaseCoverageException(Exception): |
365 def format_local_datetime(dt): |
321 """The base of all Coverage exceptions.""" |
366 """Return a string with local timezone representing the date. |
322 pass |
367 """ |
323 |
368 return dt.astimezone().strftime('%Y-%m-%d %H:%M %z') |
324 |
369 |
325 class CoverageException(BaseCoverageException): |
370 |
326 """An exception raised by a coverage.py function.""" |
371 def import_local_file(modname, modfile=None): |
327 pass |
372 """Import a local file as a module. |
328 |
373 |
329 |
374 Opens a file in the current directory named `modname`.py, imports it |
330 class NoSource(CoverageException): |
375 as `modname`, and returns the module object. `modfile` is the file to |
331 """We couldn't find the source for a module.""" |
376 import if it isn't in the current directory. |
332 pass |
377 |
333 |
378 """ |
334 |
379 if modfile is None: |
335 class NoCode(NoSource): |
380 modfile = modname + '.py' |
336 """We couldn't find any code at all.""" |
381 spec = importlib.util.spec_from_file_location(modname, modfile) |
337 pass |
382 mod = importlib.util.module_from_spec(spec) |
338 |
383 sys.modules[modname] = mod |
339 |
384 spec.loader.exec_module(mod) |
340 class NotPython(CoverageException): |
385 |
341 """A source file turned out not to be parsable Python.""" |
386 return mod |
342 pass |
387 |
343 |
388 |
344 |
389 def human_key(s): |
345 class ExceptionDuringRun(CoverageException): |
390 """Turn a string into a list of string and number chunks. |
346 """An exception happened while running customer code. |
391 "z23a" -> ["z", 23, "a"] |
347 |
392 """ |
348 Construct it with three arguments, the values from `sys.exc_info`. |
393 def tryint(s): |
349 |
394 """If `s` is a number, return an int, else `s` unchanged.""" |
350 """ |
395 try: |
351 pass |
396 return int(s) |
352 |
397 except ValueError: |
353 |
398 return s |
354 class StopEverything(BaseCoverageException): |
399 |
355 """An exception that means everything should stop. |
400 return [tryint(c) for c in re.split(r"(\d+)", s)] |
356 |
401 |
357 The CoverageTest class converts these to SkipTest, so that when running |
402 def human_sorted(strings): |
358 tests, raising this exception will automatically skip the test. |
403 """Sort the given iterable of strings the way that humans expect. |
359 |
404 |
360 """ |
405 Numeric components in the strings are sorted as numbers. |
361 pass |
406 |
|
407 Returns the sorted list. |
|
408 |
|
409 """ |
|
410 return sorted(strings, key=human_key) |
|
411 |
|
412 def human_sorted_items(items, reverse=False): |
|
413 """Sort the (string, value) items the way humans expect. |
|
414 |
|
415 Returns the sorted list of items. |
|
416 """ |
|
417 return sorted(items, key=lambda pair: (human_key(pair[0]), pair[1]), reverse=reverse) |