Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return make_property(getter)
def __call__(self):
new = "'caplog.{0}' property".format(self._prop_name)
if self._prop_name == 'records':
new += ' (or caplog.clear())'
self._warn_compat(old="'caplog.{0}()' syntax".format(self._prop_name),
new=new)
return self._naked_value # to let legacy clients modify the object
class CallableList(CallablePropertyMixin, list):
pass
class CallableStr(CallablePropertyMixin, py.builtin.text):
pass
class CompatLogCaptureFixture(LogCaptureFixture):
"""Backward compatibility with pytest-capturelog."""
def _warn_compat(self, old, new):
self._item.warn(code='L1',
message=("{0} is deprecated, use {1} instead"
.format(old, new)))
@CallableStr.compat_property
def text(self):
return super(CompatLogCaptureFixture, self).text
@CallableList.compat_property
import difflib
import pytest
import py
from xdist.slavemanage import NodeManager
queue = py.builtin._tryimport('queue', 'Queue')
class EachScheduling:
def __init__(self, numnodes, log=None):
self.numnodes = numnodes
self.node2collection = {}
self.node2pending = {}
if log is None:
self.log = py.log.Producer("eachsched")
else:
self.log = log.eachsched
self.collection_is_completed = False
def hasnodes(self):
return bool(self.node2pending)
except ImportError: # pragma: no cover
# Only available in Python 3.4+ or as a backport
enum = None
import _pytest
import _pytest._pluggy as pluggy
cutdir2 = py.path.local(_pytest.__file__).dirpath()
cutdir1 = py.path.local(pluggy.__file__.rstrip("oc"))
NoneType = type(None)
NOTSET = object()
isfunction = inspect.isfunction
isclass = inspect.isclass
callable = py.builtin.callable
# used to work around a python2 exception info leak
exc_clear = getattr(sys, 'exc_clear', lambda: None)
# The type of re.compile objects is not exposed in Python.
REGEX_TYPE = type(re.compile(''))
_PY3 = sys.version_info > (3, 0)
_PY2 = not _PY3
if hasattr(inspect, 'signature'):
def _format_args(func):
return str(inspect.signature(func))
else:
def _format_args(func):
return inspect.formatargspec(*inspect.getargspec(func))
def findsource(obj):
try:
sourcelines, lineno = py.std.inspect.findsource(obj)
except py.builtin._sysex:
raise
except:
return None, None
source = Source()
source.lines = [line.rstrip() for line in sourcelines]
return source, lineno
def addfinalizer(self, finalizer, colitem):
""" attach a finalizer to the given colitem.
if colitem is None, this will add a finalizer that
is called at the end of teardown_all().
"""
assert colitem and not isinstance(colitem, tuple)
assert py.builtin.callable(finalizer)
#assert colitem in self.stack # some unit tests don't setup stack :/
self._finalizers.setdefault(colitem, []).append(finalizer)
import py
import execnet
from py.impl.test.dist.txnode import TXNode
queue = py.builtin._tryimport("queue", "Queue")
Queue = queue.Queue
class EventQueue:
def __init__(self, registry, queue=None):
if queue is None:
queue = Queue()
self.queue = queue
registry.register(self)
def geteventargs(self, eventname, timeout=2.0):
events = []
while 1:
try:
eventcall = self.queue.get(timeout=timeout)
except queue.Empty:
#print "node channel", self.node.channel
def rsync(self, source, notify=None, verbose=False, ignores=None):
""" perform rsync to all remote hosts.
"""
rsync = HostRSync(source, verbose=verbose, ignores=ignores)
seen = py.builtin.set()
gateways = []
for gateway in self.group:
spec = gateway.spec
if spec.popen and not spec.chdir:
# XXX this assumes that sources are python-packages
# and that adding the basedir does not hurt
gateway.remote_exec("""
import sys ; sys.path.insert(0, %r)
""" % os.path.dirname(str(source))).waitclose()
continue
if spec not in seen:
def finished():
if notify:
notify("rsyncrootready", spec, source)
rsync.add_target_host(gateway, finished=finished)
seen.add(spec)
def addfinalizer(self, finalizer, colitem):
""" attach a finalizer to the given colitem.
if colitem is None, this will add a finalizer that
is called at the end of teardown_all().
"""
assert colitem and not isinstance(colitem, tuple)
assert py.builtin.callable(finalizer)
#assert colitem in self.stack # some unit tests don't setup stack :/
self._finalizers.setdefault(colitem, []).append(finalizer)
def sysexec(self, *argv, **popen_opts):
""" return stdout text from executing a system child process,
where the 'self' path points to executable.
The process is directly invoked and not through a system shell.
"""
from subprocess import Popen, PIPE
argv = map_as_list(str, argv)
popen_opts['stdout'] = popen_opts['stderr'] = PIPE
proc = Popen([str(self)] + argv, **popen_opts)
stdout, stderr = proc.communicate()
ret = proc.wait()
if py.builtin._isbytes(stdout):
stdout = py.builtin._totext(stdout, sys.getdefaultencoding())
if ret != 0:
if py.builtin._isbytes(stderr):
stderr = py.builtin._totext(stderr, sys.getdefaultencoding())
raise py.process.cmdexec.Error(ret, ret, str(self),
stdout, stderr,)
return stdout
def patch_builtins(assertion=True, compile=True):
""" put compile and AssertionError builtins to Python's builtins. """
if assertion:
from py._code import assertion
l = oldbuiltins.setdefault('AssertionError', [])
l.append(py.builtin.builtins.AssertionError)
py.builtin.builtins.AssertionError = assertion.AssertionError
if compile:
l = oldbuiltins.setdefault('compile', [])
l.append(py.builtin.builtins.compile)
py.builtin.builtins.compile = py.code.compile