Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def runtest(self):
call = py.io.StdCapture.call
found_errors, out, err = call(
check_file,
self.fspath,
self.flake8ignore,
self.maxlength,
self.maxcomplexity,
self.showshource,
self.statistics)
if found_errors:
raise Flake8Error(out, err)
# update mtime only if test passed
# otherwise failures would not be re-run next time
if hasattr(self.config, "_flake8mtimes"):
self.config._flake8mtimes[str(self.fspath)] = (self._flake8mtime,
self.flake8ignore)
def pickletest(mod):
f1 = py.io.TextIO()
f2 = py.io.TextIO()
pickler1 = mod.Pickler(f1)
unpickler1 = mod.Unpickler(f2)
pickler2 = mod.Pickler(f2)
unpickler2 = mod.Unpickler(f1)
#pickler1.memo = unpickler1.memo = {}
#pickler2.memo = unpickler2.memo = {}
d = {}
pickler1.dump(d)
f1.seek(0)
d_other = unpickler2.load()
def _notin_text(term, text):
index = text.find(term)
head = text[:index]
tail = text[index+len(term):]
correct_text = head + tail
diff = _diff_text(correct_text, text)
newdiff = ['%s is contained here:' % py.io.saferepr(term, maxsize=42)]
for line in diff:
if line.startswith('Skipping'):
continue
if line.startswith('- '):
continue
if line.startswith('+ '):
newdiff.append(' ' + line[2:])
else:
newdiff.append(line)
return newdiff
def test_stdouterrin_setnull():
cap = py.io.StdCaptureFD()
from py.__.execnet.register import stdouterrin_setnull
stdouterrin_setnull()
import os
os.write(1, "hello")
if os.name == "nt":
os.write(2, "world")
os.read(0, 1)
out, err = cap.reset()
assert not out
assert not err
assert link.href.endswith("/pkg1-2.6.tgz")
# we check here that the upload of docs without version was
# automatically tied to the newest release metadata
link = vv.get_link(rel="doczip")
history_log = link.log
assert len(history_log) == 2
assert history_log[0]['what'] == 'upload'
assert history_log[0]['who'] == 'user1'
assert history_log[0]['dst'] == 'user1/dev'
assert history_log[1]['what'] == 'push'
assert history_log[1]['who'] == 'user1'
assert history_log[1]['src'] == 'user1/dev'
assert history_log[1]['dst'] == 'user2/prod'
assert link.href.endswith("/pkg1-2.6.doc.zip")
r = testapp.get(link.href)
archive = Archive(py.io.BytesIO(r.body))
assert 'index.html' in archive.namelist()
# reconfigure inheritance and see if get shadowing information
mapp.modify_index("user1/dev", indexconfig=dict(bases=("/user2/prod",)))
vv = get_view_version_links(testapp, "/user1/dev", "pkg1", "2.6", proj=proj)
link = vv.get_link(rel="releasefile")
assert link.href.endswith("/pkg1-2.6.tgz")
shadows = vv.shadowed()
assert len(shadows) == 1, vv.versiondata
vv = shadows[0]
link = vv.get_link(rel="releasefile")
assert link.href.endswith("/pkg1-2.6.tgz")
def _saferepr(self, obj):
return py.io.saferepr(obj)
def test_stdouterrin_setnull():
cap = py.io.StdCaptureFD()
from py.__.execnet.gateway import stdouterrin_setnull
stdouterrin_setnull()
import os
os.write(1, "hello".encode('ascii'))
if os.name == "nt":
os.write(2, "world")
os.read(0, 1)
out, err = cap.reset()
assert not out
assert not err
def pretty_print(self):
tw = py.io.TerminalWriter()
if self.layer_name == self.DATA_LAYER:
tw.write('DATA')
return
tw.write('Layer %s:' % self.layer_name.upper() + os.linesep, yellow=True, bold=True)
for field_line in self._get_all_field_lines():
if ':' in field_line:
field_name, field_line = field_line.split(':', 1)
tw.write(field_name + ':', green=True, bold=True)
tw.write(field_line, bold=True)
def __init__(self, name, justification=20):
self.justification = justification
self.name = name
self.op_code = None
self.bar = progress(self.name, _translate(None), 10, just=self.justification)
res, out, self._err = py.io.StdCaptureFD.call(self.bar.start)
if not config.path_nodeinfo.exists():
fatal("The path '%s' contains no devpi-server data, use devpi-init to initialize." % config.serverdir)
if args.init or args.import_:
sdir = config.serverdir
if not (sdir.exists() and len(sdir.listdir()) >= 2):
set_state_version(config, DATABASE_VERSION)
xom = xom_from_config(config)
init_default_indexes(xom)
if args.start or args.stop or args.log or args.status:
xprocdir = config.serverdir.join(".xproc")
from devpi_server.bgserver import BackgroundServer
tw = py.io.TerminalWriter()
bgserver = BackgroundServer(tw, xprocdir)
if args.start:
return bgserver.start(args, argv[1:])
elif args.stop:
return bgserver.stop()
elif args.log:
return bgserver.log()
elif args.status:
if bgserver.info.isrunning():
bgserver.line("server is running with pid %s" %
bgserver.info.pid)
else:
bgserver.line("no server is running")
return
if args.passwd: