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-12 08:09 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

try:
    from urllib.parse import urlparse
except ImportError:
    from urlparse import urlparse

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 parse_elasticsearch_storage(string, default_index="benchmark", default_doctype="benchmark"):
    storage_url = urlparse(string)
    hosts = ["{scheme}://{netloc}".format(scheme=storage_url.scheme, netloc=netloc) for netloc in storage_url.netloc.split(',')]
    index = default_index
    doctype = default_doctype
    if storage_url.path:
        splitted = storage_url.path.strip("/").split("/")
        index = splitted[0]
        if len(splitted) >= 2:
            doctype = splitted[1]
    return hosts, index, doctype


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

ionelmc / pytest-benchmark

Pull Request — master (#58)

get_project_name() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like