NameWrapper.__str__() - Code Metrics - Inspection of "Add support for pygal 2.2 and test all the pygal v..." - ionelmc/pytest-benchmark - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7b504c...b6f606 )

by Ionel Cristian

created 2016-05-26 23:01 UTC

NameWrapper.str() A

↳ Parent: NameWrapper

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
c	0
b	0
f	0
dl	0
loc	4
rs	10

from __future__ import division
from __future__ import print_function

import argparse
import genericpath
import json
import ntpath
import os
import platform
import re
import subprocess
import sys
import types
from datetime import datetime
from decimal import Decimal
from functools import partial

from .compat import PY3

try:
    from subprocess import check_output
except ImportError:
    def check_output(*popenargs, **kwargs):
        if 'stdout' in kwargs:
            raise ValueError('stdout argument not allowed, it will be overridden.')
        process = subprocess.Popen(stdout=subprocess.PIPE, *popenargs, **kwargs)
        output, unused_err = process.communicate()
        retcode = process.poll()
        if retcode:
            cmd = kwargs.get("args")
            if cmd is None:
                cmd = popenargs[0]
            raise subprocess.CalledProcessError(retcode, cmd)
        return output

TIME_UNITS = {
    "": "Seconds",
    "m": "Miliseconds (ms)",
    "u": "Microseconds (us)",
    "n": "Nanoseconds (ns)"
}
ALLOWED_COLUMNS = ["min", "max", "mean", "stddev", "median", "iqr", "outliers", "rounds", "iterations"]


class SecondsDecimal(Decimal):
    def __float__(self):
        return float(super(SecondsDecimal, self).__str__())

    def __str__(self):
        return "{0}s".format(format_time(float(super(SecondsDecimal, self).__str__())))

    @property
    def as_string(self):
        return super(SecondsDecimal, self).__str__()


class NameWrapper(object):
    def __init__(self, target):
        self.target = target

    def __str__(self):
        name = self.target.__module__ + "." if hasattr(self.target, '__module__') else ""
        name += self.target.__name__ if hasattr(self.target, '__name__') else repr(self.target)
        return name

    def __repr__(self):
        return "NameWrapper(%s)" % repr(self.target)


def get_tag():
    info = get_commit_info()
    return '%s_%s%s' % (info['id'], get_current_time(), '_uncommitted-changes' if info['dirty'] else '')


def get_machine_id():
    return "%s-%s-%s-%s" % (
        platform.system(),
        platform.python_implementation(),
        ".".join(platform.python_version_tuple()[:2]),
        platform.architecture()[0]
    )


def get_commit_info():
    dirty = False
    commit = 'unversioned'
    try:
        if os.path.exists('.git'):
            desc = check_output('git describe --dirty --always --long --abbrev=40'.split(),
                                universal_newlines=True).strip()
            desc = desc.split('-')
            if desc[-1].strip() == 'dirty':
                dirty = True
                desc.pop()
            commit = desc[-1].strip('g')
        elif os.path.exists('.hg'):
            desc = check_output('hg id --id --debug'.split(), universal_newlines=True).strip()
            if desc[-1] == '+':
                dirty = True
            commit = desc.strip('+')
        return {
            'id': commit,
            'dirty': dirty
        }
    except Exception as exc:
        return {
            'id': 'unknown',
            'dirty': dirty,
            'error': repr(exc),
        }


def get_current_time():
    return datetime.now().strftime("%Y%m%d_%H%M%S")


def first_or_value(obj, value):
    if obj:
        value, = obj

    return value


def short_filename(path, machine_id=None):
    parts = []
    last = len(path.parts) - 1
    for pos, part in enumerate(path.parts):
        if not pos and part == machine_id:
            continue
        if pos == last:
            part = part.rsplit('.', 1)[0]
            # if len(part) > 16:
            #     part = "%.13s..." % part
        parts.append(part)
    return '/'.join(parts)


def load_timer(string):
    if "." not in string:
        raise argparse.ArgumentTypeError("Value for --benchmark-timer must be in dotted form. Eg: 'module.attr'.")
    mod, attr = string.rsplit(".", 1)
    if mod == 'pep418':
        if PY3:
            import time
            return NameWrapper(getattr(time, attr))
        else:
            from . import pep418
            return NameWrapper(getattr(pep418, attr))
    else:
        __import__(mod)
        mod = sys.modules[mod]
        return NameWrapper(getattr(mod, attr))


class RegressionCheck(object):
    def __init__(self, field, threshold):
        self.field = field
        self.threshold = threshold

    def fails(self, current, compared):
        val = self.compute(current, compared)
        if val > self.threshold:
            return "Field %r has failed %s: %.9f > %.9f" % (
                self.field, self.__class__.__name__, val, self.threshold
            )


class PercentageRegressionCheck(RegressionCheck):
    def compute(self, current, compared):
        val = compared[self.field]
        if not val:
            return float("inf")
        return current[self.field] / val * 100 - 100


class DifferenceRegressionCheck(RegressionCheck):
    def compute(self, current, compared):
        return current[self.field] - compared[self.field]


def parse_compare_fail(string,
                       rex=re.compile('^(?P<field>min|max|mean|median|stddev|iqr):'
                                      '((?P<percentage>[0-9]?[0-9])%|(?P<difference>[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?))$')):
    m = rex.match(string)
    if m:
        g = m.groupdict()
        if g['percentage']:
            return PercentageRegressionCheck(g['field'], int(g['percentage']))
        elif g['difference']:
            return DifferenceRegressionCheck(g['field'], float(g['difference']))

    raise argparse.ArgumentTypeError("Could not parse value: %r." % string)


def parse_warmup(string):
    string = string.lower().strip()
    if string == "auto":
        return platform.python_implementation() == "PyPy"
    elif string in ["off", "false", "no"]:
        return False
    elif string in ["on", "true", "yes", ""]:
        return True
    else:
        raise argparse.ArgumentTypeError("Could not parse value: %r." % string)


def name_formatter_short(bench):
    name = bench["name"]
    if bench["source"]:
        name = "%s (%.4s)" % (name, os.path.split(bench["source"])[-1])
    if name.startswith("test_"):
        name = name[5:]
    return name


def name_formatter_normal(bench):
    name = bench["name"]
    if bench["source"]:
        parts = bench["source"].split('/')
        parts[-1] = parts[-1][:12]
        name = "%s (%s)" % (name, '/'.join(parts))
    return name


def name_formatter_long(bench):
    if bench["source"]:
        return "%(fullname)s (%(source)s)" % bench
    else:
        return bench["fullname"]


NAME_FORMATTERS = {
    "short": name_formatter_short,
    "normal": name_formatter_normal,
    "long": name_formatter_long,
}


def parse_name_format(string):
    string = string.lower().strip()
    if string in NAME_FORMATTERS:
        return string
    else:
        raise argparse.ArgumentTypeError("Could not parse value: %r." % string)


def parse_timer(string):
    return str(load_timer(string))


def parse_sort(string):
    string = string.lower().strip()
    if string not in ("min", "max", "mean", "stddev", "name", "fullname"):
        raise argparse.ArgumentTypeError(
            "Unacceptable value: %r. "
            "Value for --benchmark-sort must be one of: 'min', 'max', 'mean', "
            "'stddev', 'name', 'fullname'." % string)
    return string


def parse_columns(string):
    columns = [str.strip(s) for s in string.lower().split(',')]
    invalid = set(columns) - set(ALLOWED_COLUMNS)
    if invalid:
        # there are extra items in columns!
        msg = "Invalid column name(s): %s. " % ', '.join(invalid)
        msg += "The only valid column names are: %s" % ', '.join(ALLOWED_COLUMNS)
        raise argparse.ArgumentTypeError(msg)
    return columns


def parse_rounds(string):
    try:
        value = int(string)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(exc)
    else:
        if value < 1:
            raise argparse.ArgumentTypeError("Value for --benchmark-rounds must be at least 1.")
        return value


def parse_seconds(string):
    try:
        return SecondsDecimal(string).as_string
    except Exception as exc:
        raise argparse.ArgumentTypeError("Invalid decimal value %r: %r" % (string, exc))


def parse_save(string):
    if not string:
        raise argparse.ArgumentTypeError("Can't be empty.")
    illegal = ''.join(c for c in r"\/:*?<>|" if c in string)
    if illegal:
        raise argparse.ArgumentTypeError("Must not contain any of these characters: /:*?<>|\\ (it has %r)" % illegal)
    return string


def time_unit(value):
    if value < 1e-6:
        return "n", 1e9
    elif value < 1e-3:
        return "u", 1e6
    elif value < 1:
        return "m", 1e3
    else:
        return "", 1.


def format_time(value):
    unit, adjustment = time_unit(value)
    return "{0:.2f}{1:s}".format(value * adjustment, unit)


class cached_property(object):
    def __init__(self, func):
        self.__doc__ = getattr(func, '__doc__')
        self.func = func

    def __get__(self, obj, cls):
        if obj is None:
            return self
        value = obj.__dict__[self.func.__name__] = self.func(obj)
        return value


def funcname(f):
    try:
        if isinstance(f, partial):
            return f.func.__name__
        else:
            return f.__name__
    except AttributeError:
        return str(f)


def clonefunc(f):
    """Deep clone the given function to create a new one.

    By default, the PyPy JIT specializes the assembler based on f.__code__:
    clonefunc makes sure that you will get a new function with a **different**
    __code__, so that PyPy will produce independent assembler. This is useful
    e.g. for benchmarks and microbenchmarks, so you can make sure to compare
    apples to apples.

    Use it with caution: if abused, this might easily produce an explosion of
    produced assembler.

    from: https://bitbucket.org/antocuni/pypytools/src/tip/pypytools/util.py?at=default
    """

    # first of all, we clone the code object
    try:
        co = f.__code__
        if PY3:
            co2 = types.CodeType(co.co_argcount, co.co_kwonlyargcount,
                                 co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
                                 co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
                                 co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
        else:
            co2 = types.CodeType(co.co_argcount, co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
                                 co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
                                 co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
        #
        # then, we clone the function itself, using the new co2
        return types.FunctionType(co2, f.__globals__, f.__name__, f.__defaults__, f.__closure__)
    except AttributeError:
        return f


def format_dict(obj):
    return "{%s}" % ", ".join("%s: %s" % (k, json.dumps(v)) for k, v in sorted(obj.items()))


class SafeJSONEncoder(json.JSONEncoder):
    def default(self, o):
        return "UNSERIALIZABLE[%r]" % o


def safe_dumps(obj, **kwargs):
    return json.dumps(obj, cls=SafeJSONEncoder, **kwargs)


def report_progress(iterable, terminal_reporter, format_string, **kwargs):
    total = len(iterable)

    def progress_reporting_wrapper():
        for pos, item in enumerate(iterable):
            string = format_string.format(pos=pos + 1, total=total, value=item, **kwargs)
            terminal_reporter.rewrite(string, black=True, bold=True)
            yield string, item
    return progress_reporting_wrapper()


def report_noprogress(iterable, *args, **kwargs):
    for pos, item in enumerate(iterable):
        yield "", item


def slugify(name):
    for c in "\/:*?<>| ":
        name = name.replace(c, '_').replace('__', '_')
    return name


def commonpath(paths):
    """Given a sequence of path names, returns the longest common sub-path."""

    if not paths:
        raise ValueError('commonpath() arg is an empty sequence')

    if isinstance(paths[0], bytes):
        sep = b'\\'
        altsep = b'/'
        curdir = b'.'
    else:
        sep = '\\'
        altsep = '/'
        curdir = '.'

    try:
        drivesplits = [ntpath.splitdrive(p.replace(altsep, sep).lower()) for p in paths]
        split_paths = [p.split(sep) for d, p in drivesplits]

        try:
            isabs, = set(p[:1] == sep for d, p in drivesplits)
        except ValueError:
            raise ValueError("Can't mix absolute and relative paths")

        # Check that all drive letters or UNC paths match. The check is made only
        # now otherwise type errors for mixing strings and bytes would not be
        # caught.
        if len(set(d for d, p in drivesplits)) != 1:
            raise ValueError("Paths don't have the same drive")

        drive, path = ntpath.splitdrive(paths[0].replace(altsep, sep))
        common = path.split(sep)
        common = [c for c in common if c and c != curdir]

        split_paths = [[c for c in s if c and c != curdir] for s in split_paths]
        s1 = min(split_paths)
        s2 = max(split_paths)
        for i, c in enumerate(s1):
            if c != s2[i]:
                common = common[:i]
                break
        else:
            common = common[:len(s1)]

        prefix = drive + sep if isabs else drive
        return prefix + sep.join(common)
    except (TypeError, AttributeError):
        genericpath._check_arg_types('commonpath', *paths)
        raise


1			from __future__ import division
2			from __future__ import print_function
3
4			import argparse
5			import genericpath
6			import json
7			import ntpath
8			import os
9			import platform
10			import re
11			import subprocess
12			import sys
13			import types
14			from datetime import datetime
15			from decimal import Decimal
16			from functools import partial
17
18			from .compat import PY3
19
20			try:
21			from subprocess import check_output
22			except ImportError:
23			def check_output(popenargs, *kwargs):
24			if 'stdout' in kwargs:
25			raise ValueError('stdout argument not allowed, it will be overridden.')
26			process = subprocess.Popen(stdout=subprocess.PIPE, popenargs, *kwargs)
27			output, unused_err = process.communicate()
28			retcode = process.poll()
29			if retcode:
30			cmd = kwargs.get("args")
31			if cmd is None:
32			cmd = popenargs[0]
33			raise subprocess.CalledProcessError(retcode, cmd)
34			return output
35
36			TIME_UNITS = {
37			"": "Seconds",
38			"m": "Miliseconds (ms)",
39			"u": "Microseconds (us)",
40			"n": "Nanoseconds (ns)"
41			}
42			ALLOWED_COLUMNS = ["min", "max", "mean", "stddev", "median", "iqr", "outliers", "rounds", "iterations"]
43
44
45			class SecondsDecimal(Decimal):
46			def __float__(self):
47			return float(super(SecondsDecimal, self).__str__())
48
49			def __str__(self):
50			return "{0}s".format(format_time(float(super(SecondsDecimal, self).__str__())))
51
52			@property
53			def as_string(self):
54			return super(SecondsDecimal, self).__str__()
55
56
57			class NameWrapper(object):
58			def __init__(self, target):
59			self.target = target
60
61			def __str__(self):
62			name = self.target.__module__ + "." if hasattr(self.target, '__module__') else ""
63			name += self.target.__name__ if hasattr(self.target, '__name__') else repr(self.target)
64			return name
65
66			def __repr__(self):
67			return "NameWrapper(%s)" % repr(self.target)
68
69
70			def get_tag():
71			info = get_commit_info()
72			return '%s_%s%s' % (info['id'], get_current_time(), '_uncommitted-changes' if info['dirty'] else '')
73
74
75			def get_machine_id():
76			return "%s-%s-%s-%s" % (
77			platform.system(),
78			platform.python_implementation(),
79			".".join(platform.python_version_tuple()[:2]),
80			platform.architecture()[0]
81			)
82
83
84			def get_commit_info():
85			dirty = False
86			commit = 'unversioned'
87			try:
88			if os.path.exists('.git'):
89			desc = check_output('git describe --dirty --always --long --abbrev=40'.split(),
90			universal_newlines=True).strip()
91			desc = desc.split('-')
92			if desc[-1].strip() == 'dirty':
93			dirty = True
94			desc.pop()
95			commit = desc[-1].strip('g')
96			elif os.path.exists('.hg'):
97			desc = check_output('hg id --id --debug'.split(), universal_newlines=True).strip()
98			if desc[-1] == '+':
99			dirty = True
100			commit = desc.strip('+')
101			return {
102			'id': commit,
103			'dirty': dirty
104			}
105			except Exception as exc:
106			return {
107			'id': 'unknown',
108			'dirty': dirty,
109			'error': repr(exc),
110			}
111
112
113			def get_current_time():
114			return datetime.now().strftime("%Y%m%d_%H%M%S")
115
116
117			def first_or_value(obj, value):
118			if obj:
119			value, = obj
120
121			return value
122
123
124			def short_filename(path, machine_id=None):
125			parts = []
126			last = len(path.parts) - 1
127			for pos, part in enumerate(path.parts):
128			if not pos and part == machine_id:
129			continue
130			if pos == last:
131			part = part.rsplit('.', 1)[0]
132			# if len(part) > 16:
133			# part = "%.13s..." % part
134			parts.append(part)
135			return '/'.join(parts)
136
137
138			def load_timer(string):
139			if "." not in string:
140			raise argparse.ArgumentTypeError("Value for --benchmark-timer must be in dotted form. Eg: 'module.attr'.")
141			mod, attr = string.rsplit(".", 1)
142			if mod == 'pep418':
143			if PY3:
144			import time
145			return NameWrapper(getattr(time, attr))
146			else:
147			from . import pep418
148			return NameWrapper(getattr(pep418, attr))
149			else:
150			__import__(mod)
151			mod = sys.modules[mod]
152			return NameWrapper(getattr(mod, attr))
153
154
155			class RegressionCheck(object):
156			def __init__(self, field, threshold):
157			self.field = field
158			self.threshold = threshold
159
160			def fails(self, current, compared):
161			val = self.compute(current, compared)
162			if val > self.threshold:
163			return "Field %r has failed %s: %.9f > %.9f" % (
164			self.field, self.__class__.__name__, val, self.threshold
165			)
166
167
168			class PercentageRegressionCheck(RegressionCheck):
169			def compute(self, current, compared):
170			val = compared[self.field]
171			if not val:
172			return float("inf")
173			return current[self.field] / val * 100 - 100
174
175
176			class DifferenceRegressionCheck(RegressionCheck):
177			def compute(self, current, compared):
178			return current[self.field] - compared[self.field]
179
180
181			def parse_compare_fail(string,
182			rex=re.compile('^(?P<field>min\|max\|mean\|median\|stddev\|iqr):'
183			'((?P<percentage>[0-9]?[0-9])%\|(?P<difference>[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?))$')):
184			m = rex.match(string)
185			if m:
186			g = m.groupdict()
187			if g['percentage']:
188			return PercentageRegressionCheck(g['field'], int(g['percentage']))
189			elif g['difference']:
190			return DifferenceRegressionCheck(g['field'], float(g['difference']))
191
192			raise argparse.ArgumentTypeError("Could not parse value: %r." % string)
193
194
195			def parse_warmup(string):
196			string = string.lower().strip()
197			if string == "auto":
198			return platform.python_implementation() == "PyPy"
199			elif string in ["off", "false", "no"]:
200			return False
201			elif string in ["on", "true", "yes", ""]:
202			return True
203			else:
204			raise argparse.ArgumentTypeError("Could not parse value: %r." % string)
205
206
207			def name_formatter_short(bench):
208			name = bench["name"]
209			if bench["source"]:
210			name = "%s (%.4s)" % (name, os.path.split(bench["source"])[-1])
211			if name.startswith("test_"):
212			name = name[5:]
213			return name
214
215
216			def name_formatter_normal(bench):
217			name = bench["name"]
218			if bench["source"]:
219			parts = bench["source"].split('/')
220			parts[-1] = parts[-1][:12]
221			name = "%s (%s)" % (name, '/'.join(parts))
222			return name
223
224
225			def name_formatter_long(bench):
226			if bench["source"]:
227			return "%(fullname)s (%(source)s)" % bench
228			else:
229			return bench["fullname"]
230
231
232			NAME_FORMATTERS = {
233			"short": name_formatter_short,
234			"normal": name_formatter_normal,
235			"long": name_formatter_long,
236			}
237
238
239			def parse_name_format(string):
240			string = string.lower().strip()
241			if string in NAME_FORMATTERS:
242			return string
243			else:
244			raise argparse.ArgumentTypeError("Could not parse value: %r." % string)
245
246
247			def parse_timer(string):
248			return str(load_timer(string))
249
250
251			def parse_sort(string):
252			string = string.lower().strip()
253			if string not in ("min", "max", "mean", "stddev", "name", "fullname"):
254			raise argparse.ArgumentTypeError(
255			"Unacceptable value: %r. "
256			"Value for --benchmark-sort must be one of: 'min', 'max', 'mean', "
257			"'stddev', 'name', 'fullname'." % string)
258			return string
259
260
261			def parse_columns(string):
262			columns = [str.strip(s) for s in string.lower().split(',')]
263			invalid = set(columns) - set(ALLOWED_COLUMNS)
264			if invalid:
265			# there are extra items in columns!
266			msg = "Invalid column name(s): %s. " % ', '.join(invalid)
267			msg += "The only valid column names are: %s" % ', '.join(ALLOWED_COLUMNS)
268			raise argparse.ArgumentTypeError(msg)
269			return columns
270
271
272			def parse_rounds(string):
273			try:
274			value = int(string)
275			except ValueError as exc:
276			raise argparse.ArgumentTypeError(exc)
277			else:
278			if value < 1:
279			raise argparse.ArgumentTypeError("Value for --benchmark-rounds must be at least 1.")
280			return value
281
282
283			def parse_seconds(string):
284			try:
285			return SecondsDecimal(string).as_string
286			except Exception as exc:
287			raise argparse.ArgumentTypeError("Invalid decimal value %r: %r" % (string, exc))
288
289
290			def parse_save(string):
291			if not string:
292			raise argparse.ArgumentTypeError("Can't be empty.")
293			illegal = ''.join(c for c in r"\/:*?<>\|" if c in string)
294			if illegal:
295			raise argparse.ArgumentTypeError("Must not contain any of these characters: /:*?<>\|\\ (it has %r)" % illegal)
296			return string
297
298
299			def time_unit(value):
300			if value < 1e-6:
301			return "n", 1e9
302			elif value < 1e-3:
303			return "u", 1e6
304			elif value < 1:
305			return "m", 1e3
306			else:
307			return "", 1.
308
309
310			def format_time(value):
311			unit, adjustment = time_unit(value)
312			return "{0:.2f}{1:s}".format(value * adjustment, unit)
313
314
315			class cached_property(object):
316			def __init__(self, func):
317			self.__doc__ = getattr(func, '__doc__')
318			self.func = func
319
320			def __get__(self, obj, cls):
321			if obj is None:
322			return self
323			value = obj.__dict__[self.func.__name__] = self.func(obj)
324			return value
325
326
327			def funcname(f):
328			try:
329			if isinstance(f, partial):
330			return f.func.__name__
331			else:
332			return f.__name__
333			except AttributeError:
334			return str(f)
335
336
337			def clonefunc(f):
338			"""Deep clone the given function to create a new one.
339
340			By default, the PyPy JIT specializes the assembler based on f.__code__:
341			clonefunc makes sure that you will get a new function with a different
342			__code__, so that PyPy will produce independent assembler. This is useful
343			e.g. for benchmarks and microbenchmarks, so you can make sure to compare
344			apples to apples.
345
346			Use it with caution: if abused, this might easily produce an explosion of
347			produced assembler.
348
349			from: https://bitbucket.org/antocuni/pypytools/src/tip/pypytools/util.py?at=default
350			"""
351
352			# first of all, we clone the code object
353			try:
354			co = f.__code__
355			if PY3:
356			co2 = types.CodeType(co.co_argcount, co.co_kwonlyargcount,
357			co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
358			co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
359			co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
360			else:
361			co2 = types.CodeType(co.co_argcount, co.co_nlocals, co.co_stacksize, co.co_flags, co.co_code,
362			co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
363			co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars)
364			#
365			# then, we clone the function itself, using the new co2
366			return types.FunctionType(co2, f.__globals__, f.__name__, f.__defaults__, f.__closure__)
367			except AttributeError:
368			return f
369
370
371			def format_dict(obj):
372			return "{%s}" % ", ".join("%s: %s" % (k, json.dumps(v)) for k, v in sorted(obj.items()))
373
374
375			class SafeJSONEncoder(json.JSONEncoder):
376			def default(self, o):
377			return "UNSERIALIZABLE[%r]" % o
378
379
380			def safe_dumps(obj, **kwargs):
381			return json.dumps(obj, cls=SafeJSONEncoder, **kwargs)
382
383
384			def report_progress(iterable, terminal_reporter, format_string, **kwargs):
385			total = len(iterable)
386
387			def progress_reporting_wrapper():
388			for pos, item in enumerate(iterable):
389			string = format_string.format(pos=pos + 1, total=total, value=item, **kwargs)
390			terminal_reporter.rewrite(string, black=True, bold=True)
391			yield string, item
392			return progress_reporting_wrapper()
393
394
395			def report_noprogress(iterable, args, *kwargs):
396			for pos, item in enumerate(iterable):
397			yield "", item
398
399
400			def slugify(name):
401			for c in "\/:*?<>\| ":
402			name = name.replace(c, '_').replace('__', '_')
403			return name
404
405
406			def commonpath(paths):
407			"""Given a sequence of path names, returns the longest common sub-path."""
408
409			if not paths:
410			raise ValueError('commonpath() arg is an empty sequence')
411
412			if isinstance(paths[0], bytes):
413			sep = b'\\'
414			altsep = b'/'
415			curdir = b'.'
416			else:
417			sep = '\\'
418			altsep = '/'
419			curdir = '.'
420
421			try:
422			drivesplits = [ntpath.splitdrive(p.replace(altsep, sep).lower()) for p in paths]
423			split_paths = [p.split(sep) for d, p in drivesplits]
424
425			try:
426			isabs, = set(p[:1] == sep for d, p in drivesplits)
427			except ValueError:
428			raise ValueError("Can't mix absolute and relative paths")
429
430			# Check that all drive letters or UNC paths match. The check is made only
431			# now otherwise type errors for mixing strings and bytes would not be
432			# caught.
433			if len(set(d for d, p in drivesplits)) != 1:
434			raise ValueError("Paths don't have the same drive")
435
436			drive, path = ntpath.splitdrive(paths[0].replace(altsep, sep))
437			common = path.split(sep)
438			common = [c for c in common if c and c != curdir]
439
440			split_paths = [[c for c in s if c and c != curdir] for s in split_paths]
441			s1 = min(split_paths)
442			s2 = max(split_paths)
443			for i, c in enumerate(s1):
444			if c != s2[i]:
445			common = common[:i]
446			break
447			else:
448			common = common[:len(s1)]
449
450			prefix = drive + sep if isabs else drive
451			return prefix + sep.join(common)
452			except (TypeError, AttributeError):
453			genericpath._check_arg_types('commonpath', *paths)
454			raise
455

ionelmc / pytest-benchmark

Push — master ( 7b504c...b6f606 )

NameWrapper.__str__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

NameWrapper.str() A