1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
|
# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt
"""Miscellaneous stuff for coverage.py."""
import errno
import hashlib
import importlib.util
import inspect
import locale
import os
import os.path
import random
import re
import socket
import sys
import types
from coverage import env
ISOLATED_MODULES = {}
def isolate_module(mod):
"""Copy a module so that we are isolated from aggressive mocking.
If a test suite mocks os.path.exists (for example), and then we need to use
it during the test, everything will get tangled up if we use their mock.
Making a copy of the module when we import it will isolate coverage.py from
those complications.
"""
if mod not in ISOLATED_MODULES:
new_mod = types.ModuleType(mod.__name__)
ISOLATED_MODULES[mod] = new_mod
for name in dir(mod):
value = getattr(mod, name)
if isinstance(value, types.ModuleType):
value = isolate_module(value)
setattr(new_mod, name, value)
return ISOLATED_MODULES[mod]
os = isolate_module(os)
def dummy_decorator_with_args(*args_unused, **kwargs_unused):
"""Dummy no-op implementation of a decorator with arguments."""
def _decorator(func):
return func
return _decorator
# Environment COVERAGE_NO_CONTRACTS=1 can turn off contracts while debugging
# tests to remove noise from stack traces.
# $set_env.py: COVERAGE_NO_CONTRACTS - Disable PyContracts to simplify stack traces.
USE_CONTRACTS = env.TESTING and not bool(int(os.environ.get("COVERAGE_NO_CONTRACTS", 0)))
# Use PyContracts for assertion testing on parameters and returns, but only if
# we are running our own test suite.
if USE_CONTRACTS:
from contracts import contract # pylint: disable=unused-import
from contracts import new_contract as raw_new_contract
def new_contract(*args, **kwargs):
"""A proxy for contracts.new_contract that doesn't mind happening twice."""
try:
raw_new_contract(*args, **kwargs)
except ValueError:
# During meta-coverage, this module is imported twice, and
# PyContracts doesn't like redefining contracts. It's OK.
pass
# Define contract words that PyContract doesn't have.
new_contract('bytes', lambda v: isinstance(v, bytes))
new_contract('unicode', lambda v: isinstance(v, str))
def one_of(argnames):
"""Ensure that only one of the argnames is non-None."""
def _decorator(func):
argnameset = {name.strip() for name in argnames.split(",")}
def _wrapper(*args, **kwargs):
vals = [kwargs.get(name) for name in argnameset]
assert sum(val is not None for val in vals) == 1
return func(*args, **kwargs)
return _wrapper
return _decorator
else: # pragma: not testing
# We aren't using real PyContracts, so just define our decorators as
# stunt-double no-ops.
contract = dummy_decorator_with_args
one_of = dummy_decorator_with_args
def new_contract(*args_unused, **kwargs_unused):
"""Dummy no-op implementation of `new_contract`."""
pass
def nice_pair(pair):
"""Make a nice string representation of a pair of numbers.
If the numbers are equal, just return the number, otherwise return the pair
with a dash between them, indicating the range.
"""
start, end = pair
if start == end:
return "%d" % start
else:
return "%d-%d" % (start, end)
def expensive(fn):
"""A decorator to indicate that a method shouldn't be called more than once.
Normally, this does nothing. During testing, this raises an exception if
called more than once.
"""
if env.TESTING:
attr = "_once_" + fn.__name__
def _wrapper(self):
if hasattr(self, attr):
raise AssertionError("Shouldn't have called %s more than once" % fn.__name__)
setattr(self, attr, True)
return fn(self)
return _wrapper
else:
return fn # pragma: not testing
def bool_or_none(b):
"""Return bool(b), but preserve None."""
if b is None:
return None
else:
return bool(b)
def join_regex(regexes):
"""Combine a list of regexes into one that matches any of them."""
return "|".join("(?:%s)" % r for r in regexes)
def file_be_gone(path):
"""Remove a file, and don't get annoyed if it doesn't exist."""
try:
os.remove(path)
except OSError as e:
if e.errno != errno.ENOENT:
raise
def ensure_dir(directory):
"""Make sure the directory exists.
If `directory` is None or empty, do nothing.
"""
if directory and not os.path.isdir(directory):
os.makedirs(directory)
def ensure_dir_for_file(path):
"""Make sure the directory for the path exists."""
ensure_dir(os.path.dirname(path))
def output_encoding(outfile=None):
"""Determine the encoding to use for output written to `outfile` or stdout."""
if outfile is None:
outfile = sys.stdout
encoding = (
getattr(outfile, "encoding", None) or
getattr(sys.__stdout__, "encoding", None) or
locale.getpreferredencoding()
)
return encoding
def filename_suffix(suffix):
"""Compute a filename suffix for a data file.
If `suffix` is a string or None, simply return it. If `suffix` is True,
then build a suffix incorporating the hostname, process id, and a random
number.
Returns a string or None.
"""
if suffix is True:
# If data_suffix was a simple true value, then make a suffix with
# plenty of distinguishing information. We do this here in
# `save()` at the last minute so that the pid will be correct even
# if the process forks.
dice = random.Random(os.urandom(8)).randint(0, 999999)
suffix = "%s.%s.%06d" % (socket.gethostname(), os.getpid(), dice)
return suffix
class Hasher(object):
"""Hashes Python data into md5."""
def __init__(self):
self.md5 = hashlib.md5()
def update(self, v):
"""Add `v` to the hash, recursively if needed."""
self.md5.update(str(type(v)).encode("utf8"))
if isinstance(v, str):
self.md5.update(v.encode('utf8'))
elif isinstance(v, bytes):
self.md5.update(v)
elif v is None:
pass
elif isinstance(v, (int, float)):
self.md5.update(str(v).encode("utf8"))
elif isinstance(v, (tuple, list)):
for e in v:
self.update(e)
elif isinstance(v, dict):
keys = v.keys()
for k in sorted(keys):
self.update(k)
self.update(v[k])
else:
for k in dir(v):
if k.startswith('__'):
continue
a = getattr(v, k)
if inspect.isroutine(a):
continue
self.update(k)
self.update(a)
self.md5.update(b'.')
def hexdigest(self):
"""Retrieve the hex digest of the hash."""
return self.md5.hexdigest()
def _needs_to_implement(that, func_name):
"""Helper to raise NotImplementedError in interface stubs."""
if hasattr(that, "_coverage_plugin_name"):
thing = "Plugin"
name = that._coverage_plugin_name
else:
thing = "Class"
klass = that.__class__
name = "{klass.__module__}.{klass.__name__}".format(klass=klass)
raise NotImplementedError(
"{thing} {name!r} needs to implement {func_name}()".format(
thing=thing, name=name, func_name=func_name
)
)
class DefaultValue(object):
"""A sentinel object to use for unusual default-value needs.
Construct with a string that will be used as the repr, for display in help
and Sphinx output.
"""
def __init__(self, display_as):
self.display_as = display_as
def __repr__(self):
return self.display_as
def substitute_variables(text, variables):
"""Substitute ``${VAR}`` variables in `text` with their values.
Variables in the text can take a number of shell-inspired forms::
$VAR
${VAR}
${VAR?} strict: an error if VAR isn't defined.
${VAR-missing} defaulted: "missing" if VAR isn't defined.
$$ just a dollar sign.
`variables` is a dictionary of variable values.
Returns the resulting text with values substituted.
"""
dollar_pattern = r"""(?x) # Use extended regex syntax
\$ # A dollar sign,
(?: # then
(?P<dollar>\$) | # a dollar sign, or
(?P<word1>\w+) | # a plain word, or
{ # a {-wrapped
(?P<word2>\w+) # word,
(?:
(?P<strict>\?) | # with a strict marker
-(?P<defval>[^}]*) # or a default value
)? # maybe.
}
)
"""
def dollar_replace(match):
"""Called for each $replacement."""
# Only one of the groups will have matched, just get its text.
word = next(g for g in match.group('dollar', 'word1', 'word2') if g)
if word == "$":
return "$"
elif word in variables:
return variables[word]
elif match.group('strict'):
msg = "Variable {} is undefined: {!r}".format(word, text)
raise CoverageException(msg)
else:
return match.group('defval')
text = re.sub(dollar_pattern, dollar_replace, text)
return text
def format_local_datetime(dt):
"""Return a string with local timezone representing the date.
"""
return dt.astimezone().strftime('%Y-%m-%d %H:%M %z')
def import_local_file(modname, modfile=None):
"""Import a local file as a module.
Opens a file in the current directory named `modname`.py, imports it
as `modname`, and returns the module object. `modfile` is the file to
import if it isn't in the current directory.
"""
if modfile is None:
modfile = modname + '.py'
spec = importlib.util.spec_from_file_location(modname, modfile)
mod = importlib.util.module_from_spec(spec)
sys.modules[modname] = mod
spec.loader.exec_module(mod)
return mod
class BaseCoverageException(Exception):
"""The base of all Coverage exceptions."""
pass
class CoverageException(BaseCoverageException):
"""An exception raised by a coverage.py function."""
pass
class NoSource(CoverageException):
"""We couldn't find the source for a module."""
pass
class NoCode(NoSource):
"""We couldn't find any code at all."""
pass
class NotPython(CoverageException):
"""A source file turned out not to be parsable Python."""
pass
class ExceptionDuringRun(CoverageException):
"""An exception happened while running customer code.
Construct it with three arguments, the values from `sys.exc_info`.
"""
pass
class StopEverything(BaseCoverageException):
"""An exception that means everything should stop.
The CoverageTest class converts these to SkipTest, so that when running
tests, raising this exception will automatically skip the test.
"""
pass
|