get_project_name() - Code Metrics - Inspection of "Elasticsearch report backend" - ionelmc/pytest-benchmark - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#58)

unknown

created 2016-09-09 08:36 UTC

get_project_name() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
c	1
b	0
f	0
dl	0
loc	18
rs	8.5454

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()
    parts = []
    if info['project']:
        parts.append(info['project'])
    parts.append(info['id'])
    parts.append(get_current_time())
    if info['dirty']:
        parts.append("uncommited-changes")
    return "_".join(parts)


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_project_name():
    if os.path.exists('.git'):
        try:
            project_address = check_output("git config --local remote.origin.url".split())
            project_name = re.findall(r'/([^/]*)\.git', project_address.decode())[0]
            return project_name
        except (IndexError, subprocess.CalledProcessError):
            return os.path.basename(os.getcwd())
    elif os.path.exists('.hg'):
        try:
            project_address = check_output("hg path default".split())
            project_address = project_address.decode()
            project_name = project_address.split("/")[-1]
            return project_name.strip()
        except (IndexError, subprocess.CalledProcessError):
            return os.path.basename(os.getcwd())
    else:
        return os.path.basename(os.getcwd())


def get_commit_info():
    dirty = False
    commit = 'unversioned'
    project_name = get_project_name()
    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,
            'project': project_name,
        }
    except Exception as exc:
        return {
            'id': 'unknown',
            'dirty': dirty,
            'error': repr(exc),
            'project': project_name,
        }


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

ionelmc / pytest-benchmark

Pull Request — master (#58)

get_project_name() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like