fix py/io classes and tests to pass 3.1
introduce py.builtin._totext helper to make a 2k=unicode / 3k=str object, allow a string as data --HG-- branch : trunk
This commit is contained in:
holger krekel
parent
d75f7b2dd7
commit
1dafcc6b37
|
|
@ -148,6 +148,7 @@ initpkg(__name__,
|
|||
'builtin._reraise' : ('./builtin/builtin31.py', '_reraise'),
|
||||
'builtin.exec_' : ('./builtin/builtin31.py', 'exec_'),
|
||||
'builtin.basestring' : ('./builtin/builtin31.py', 'basestring'),
|
||||
'builtin._totext' : ('./builtin/builtin31.py', '_totext'),
|
||||
'builtin.builtins' : ('./builtin/builtin31.py', 'builtins'),
|
||||
|
||||
# gateways into remote contexts
|
||||
|
|
|
|||
|
|
@ -4,8 +4,13 @@ if sys.version_info >= (3, 0):
|
|||
exec ("print_ = print ; exec_=exec")
|
||||
import builtins
|
||||
basestring = str
|
||||
def _totext(obj, encoding):
|
||||
if isinstance(obj, str):
|
||||
obj = obj.encode(encoding)
|
||||
return str(obj, encoding)
|
||||
|
||||
else:
|
||||
_totext = unicode
|
||||
basestring = basestring
|
||||
import __builtin__ as builtins
|
||||
def print_(*args, **kwargs):
|
||||
|
|
|
|||
|
|
@ -84,6 +84,9 @@ def test_print_simple():
|
|||
s = f.getvalue()
|
||||
assert s == "xyzabc"
|
||||
|
||||
def test_totext():
|
||||
py.builtin._totext("hello", "UTF-8")
|
||||
|
||||
def test_reraise():
|
||||
from py.builtin import _reraise
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -8,12 +8,14 @@ try:
|
|||
except ImportError:
|
||||
from StringIO import StringIO
|
||||
|
||||
class TextIO(StringIO):
|
||||
if sys.version_info < (3,0):
|
||||
if sys.version_info < (3,0):
|
||||
class TextIO(StringIO):
|
||||
def write(self, data):
|
||||
if not isinstance(data, unicode):
|
||||
data = unicode(data, getattr(self, '_encoding', 'UTF-8'))
|
||||
StringIO.write(self, data)
|
||||
else:
|
||||
TextIO = StringIO
|
||||
|
||||
try:
|
||||
from io import BytesIO
|
||||
|
|
@ -28,9 +30,16 @@ class FDCapture:
|
|||
""" Capture IO to/from a given os-level filedescriptor. """
|
||||
|
||||
def __init__(self, targetfd, tmpfile=None):
|
||||
""" save targetfd descriptor, and open a new
|
||||
temporary file there. If no tmpfile is
|
||||
specified a tempfile.Tempfile() will be opened
|
||||
in text mode.
|
||||
"""
|
||||
self.targetfd = targetfd
|
||||
if tmpfile is None:
|
||||
tmpfile = self.maketmpfile()
|
||||
f = tempfile.TemporaryFile('wb+')
|
||||
tmpfile = dupfile(f, encoding="UTF-8")
|
||||
f.close()
|
||||
self.tmpfile = tmpfile
|
||||
self._savefd = os.dup(targetfd)
|
||||
os.dup2(self.tmpfile.fileno(), targetfd)
|
||||
|
|
@ -59,26 +68,18 @@ class FDCapture:
|
|||
self.tmpfile.seek(0)
|
||||
return self.tmpfile
|
||||
|
||||
def maketmpfile(self):
|
||||
""" create a temporary file
|
||||
"""
|
||||
f = tempfile.TemporaryFile()
|
||||
newf = dupfile(f)
|
||||
f.close()
|
||||
return newf
|
||||
|
||||
def writeorg(self, str):
|
||||
def writeorg(self, data):
|
||||
""" write a string to the original file descriptor
|
||||
"""
|
||||
tempfp = tempfile.TemporaryFile()
|
||||
try:
|
||||
os.dup2(self._savefd, tempfp.fileno())
|
||||
tempfp.write(str)
|
||||
tempfp.write(data)
|
||||
finally:
|
||||
tempfp.close()
|
||||
|
||||
|
||||
def dupfile(f, mode=None, buffering=0, raising=False):
|
||||
def dupfile(f, mode=None, buffering=0, raising=False, encoding=None):
|
||||
""" return a new open file object that's a duplicate of f
|
||||
|
||||
mode is duplicated if not given, 'buffering' controls
|
||||
|
|
@ -95,6 +96,11 @@ def dupfile(f, mode=None, buffering=0, raising=False):
|
|||
return f
|
||||
newfd = os.dup(fd)
|
||||
mode = mode and mode or f.mode
|
||||
if encoding is not None and sys.version_info >= (3,0):
|
||||
mode = mode.replace("b", "")
|
||||
buffering = True
|
||||
return os.fdopen(newfd, mode, buffering, encoding, closefd=False)
|
||||
else:
|
||||
return os.fdopen(newfd, mode, buffering)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -139,16 +139,18 @@ class TerminalWriter(object):
|
|||
Black=40, Red=41, Green=42, Yellow=43,
|
||||
Blue=44, Purple=45, Cyan=46, White=47,
|
||||
bold=1, light=2, blink=5, invert=7)
|
||||
_encoding = "utf-8"
|
||||
|
||||
def __init__(self, file=None, stringio=False):
|
||||
# XXX deprecate stringio argument
|
||||
def __init__(self, file=None, stringio=False, encoding=None):
|
||||
self.encoding = encoding
|
||||
|
||||
if file is None:
|
||||
if stringio:
|
||||
self.stringio = file = py.io.TextIO()
|
||||
else:
|
||||
file = py.std.sys.stdout
|
||||
elif hasattr(file, '__call__'):
|
||||
file = WriteFile(file)
|
||||
file = WriteFile(file, encoding=encoding)
|
||||
self._file = file
|
||||
self.fullwidth = get_terminal_width()
|
||||
self.hasmarkup = should_do_markup(file)
|
||||
|
|
@ -202,8 +204,9 @@ class TerminalWriter(object):
|
|||
self._file.flush()
|
||||
|
||||
def _getbytestring(self, s):
|
||||
if sys.version_info < (3,0) and isinstance(s, unicode):
|
||||
return s.encode(self._encoding)
|
||||
# XXX review this and the whole logic
|
||||
if self.encoding and sys.version_info < (3,0) and isinstance(s, unicode):
|
||||
return s.encode(self.encoding)
|
||||
elif not isinstance(s, str):
|
||||
return str(s)
|
||||
return s
|
||||
|
|
@ -246,8 +249,15 @@ if sys.platform == 'win32':
|
|||
TerminalWriter = Win32ConsoleWriter
|
||||
|
||||
class WriteFile(object):
|
||||
def __init__(self, writemethod):
|
||||
self.write = writemethod
|
||||
def __init__(self, writemethod, encoding=None):
|
||||
self.encoding = encoding
|
||||
self._writemethod = writemethod
|
||||
|
||||
def write(self, data):
|
||||
if self.encoding:
|
||||
data = data.encode(self.encoding)
|
||||
self._writemethod(data)
|
||||
|
||||
def flush(self):
|
||||
return
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,34 @@
|
|||
import os, sys
|
||||
import py
|
||||
|
||||
from py.builtin import print_
|
||||
|
||||
if sys.version_info >= (3,0):
|
||||
def tobytes(obj):
|
||||
if isinstance(obj, str):
|
||||
obj = obj.encode('UTF-8')
|
||||
assert isinstance(obj, bytes)
|
||||
return obj
|
||||
def totext(obj):
|
||||
if isinstance(obj, bytes):
|
||||
obj = str(obj, 'UTF-8')
|
||||
assert isinstance(obj, str)
|
||||
return obj
|
||||
else:
|
||||
def tobytes(obj):
|
||||
if isinstance(obj, unicode):
|
||||
obj = obj.encode('UTF-8')
|
||||
assert isinstance(obj, str)
|
||||
return obj
|
||||
def totext(obj):
|
||||
if isinstance(obj, str):
|
||||
obj = unicode(obj, 'UTF-8')
|
||||
assert isinstance(obj, unicode)
|
||||
return obj
|
||||
|
||||
def oswritebytes(fd, obj):
|
||||
os.write(fd, tobytes(obj))
|
||||
|
||||
class TestTextIO:
|
||||
def test_text(self):
|
||||
f = py.io.TextIO()
|
||||
|
|
@ -11,18 +39,23 @@ class TestTextIO:
|
|||
|
||||
def test_unicode_and_str_mixture(self):
|
||||
f = py.io.TextIO()
|
||||
f.write(u"\u00f6")
|
||||
f.write(str("hello"))
|
||||
if sys.version_info >= (3,0):
|
||||
f.write("\u00f6")
|
||||
py.test.skip("3k IO beahviour?")
|
||||
f.write(bytes("hello", 'UTF-8'))
|
||||
else:
|
||||
f.write(unicode("\u00f6", 'UTF-8'))
|
||||
f.write("hello") # bytes
|
||||
s = f.getvalue()
|
||||
f.close()
|
||||
assert isinstance(s, unicode)
|
||||
|
||||
def test_bytes_io():
|
||||
f = py.io.BytesIO()
|
||||
f.write("hello")
|
||||
py.test.raises(TypeError, "f.write(u'hello')")
|
||||
f.write(tobytes("hello"))
|
||||
py.test.raises(TypeError, "f.write(totext('hello'))")
|
||||
s = f.getvalue()
|
||||
assert s == "hello"
|
||||
assert s == tobytes("hello")
|
||||
|
||||
def test_dontreadfrominput():
|
||||
from py.__.io.capture import DontReadFromInput
|
||||
|
|
@ -33,30 +66,36 @@ def test_dontreadfrominput():
|
|||
py.test.raises(IOError, iter, f)
|
||||
py.test.raises(ValueError, f.fileno)
|
||||
|
||||
def test_dupfile():
|
||||
somefile = py.std.os.tmpfile()
|
||||
def pytest_funcarg__tmpfile(request):
|
||||
testdir = request.getfuncargvalue("testdir")
|
||||
f = testdir.makepyfile("").open('wb+')
|
||||
request.addfinalizer(f.close)
|
||||
return f
|
||||
|
||||
def test_dupfile(tmpfile):
|
||||
somefile = tmpfile
|
||||
flist = []
|
||||
for i in range(5):
|
||||
nf = py.io.dupfile(somefile)
|
||||
assert nf != somefile
|
||||
assert nf.fileno() != somefile.fileno()
|
||||
assert nf not in flist
|
||||
print >>nf, i,
|
||||
print_(i, end="", file=nf)
|
||||
flist.append(nf)
|
||||
for i in range(5):
|
||||
f = flist[i]
|
||||
f.close()
|
||||
somefile.seek(0)
|
||||
s = somefile.read()
|
||||
assert s.startswith("01234")
|
||||
assert "01234" in repr(s)
|
||||
somefile.close()
|
||||
|
||||
class TestFDCapture:
|
||||
def test_basic(self):
|
||||
tmpfile = py.std.os.tmpfile()
|
||||
def test_stdout(self, tmpfile):
|
||||
fd = tmpfile.fileno()
|
||||
cap = py.io.FDCapture(fd)
|
||||
os.write(fd, "hello")
|
||||
data = tobytes("hello")
|
||||
os.write(fd, data)
|
||||
f = cap.done()
|
||||
s = f.read()
|
||||
assert s == "hello"
|
||||
|
|
@ -64,48 +103,34 @@ class TestFDCapture:
|
|||
def test_stderr(self):
|
||||
cap = py.io.FDCapture(2)
|
||||
cap.setasfile('stderr')
|
||||
print >>sys.stderr, "hello"
|
||||
print_("hello", file=sys.stderr)
|
||||
f = cap.done()
|
||||
s = f.read()
|
||||
assert s == "hello\n"
|
||||
|
||||
def test_stdin(self):
|
||||
f = os.tmpfile()
|
||||
print >>f, "3"
|
||||
f.seek(0)
|
||||
cap = py.io.FDCapture(0, tmpfile=f)
|
||||
def test_stdin(self, tmpfile):
|
||||
tmpfile.write(tobytes("3"))
|
||||
tmpfile.seek(0)
|
||||
cap = py.io.FDCapture(0, tmpfile=tmpfile)
|
||||
# check with os.read() directly instead of raw_input(), because
|
||||
# sys.stdin itself may be redirected (as py.test now does by default)
|
||||
x = os.read(0, 100).strip()
|
||||
f = cap.done()
|
||||
assert x == "3"
|
||||
assert x == tobytes("3")
|
||||
|
||||
def test_writeorg(self):
|
||||
tmppath = py.test.ensuretemp('test_writeorg').ensure('stderr',
|
||||
file=True)
|
||||
tmpfp = tmppath.open('w+b')
|
||||
def test_writeorg(self, tmpfile):
|
||||
data1, data2 = tobytes("foo"), tobytes("bar")
|
||||
try:
|
||||
cap = py.io.FDCapture(tmpfp.fileno())
|
||||
print >>tmpfp, 'foo'
|
||||
cap.writeorg('bar\n')
|
||||
cap = py.io.FDCapture(tmpfile.fileno())
|
||||
tmpfile.write(data1)
|
||||
cap.writeorg(data2)
|
||||
finally:
|
||||
tmpfp.close()
|
||||
tmpfile.close()
|
||||
f = cap.done()
|
||||
scap = f.read()
|
||||
assert scap == 'foo\n'
|
||||
stmp = tmppath.read()
|
||||
assert stmp == "bar\n"
|
||||
|
||||
def test_writeorg_wrongtype(self):
|
||||
tmppath = py.test.ensuretemp('test_writeorg').ensure('stdout',
|
||||
file=True)
|
||||
tmpfp = tmppath.open('r')
|
||||
try:
|
||||
cap = py.io.FDCapture(tmpfp.fileno())
|
||||
py.test.raises(IOError, "cap.writeorg('bar\\n')")
|
||||
finally:
|
||||
tmpfp.close()
|
||||
f = cap.done()
|
||||
assert scap == totext(data1)
|
||||
stmp = open(tmpfile.name, 'rb').read()
|
||||
assert stmp == data2
|
||||
|
||||
|
||||
class TestStdCapture:
|
||||
|
|
@ -114,16 +139,18 @@ class TestStdCapture:
|
|||
|
||||
def test_capturing_done_simple(self):
|
||||
cap = self.getcapture()
|
||||
print "hello world"
|
||||
print >>sys.stderr, "hello error"
|
||||
sys.stdout.write("hello")
|
||||
sys.stderr.write("world")
|
||||
outfile, errfile = cap.done()
|
||||
assert outfile.read() == "hello world\n"
|
||||
assert errfile.read() == "hello error\n"
|
||||
s = outfile.read()
|
||||
assert s == "hello"
|
||||
s = errfile.read()
|
||||
assert s == "world"
|
||||
|
||||
def test_capturing_reset_simple(self):
|
||||
cap = self.getcapture()
|
||||
print "hello world"
|
||||
print >>sys.stderr, "hello error"
|
||||
print("hello world")
|
||||
sys.stderr.write("hello error\n")
|
||||
out, err = cap.reset()
|
||||
assert out == "hello world\n"
|
||||
assert err == "hello error\n"
|
||||
|
|
@ -131,28 +158,28 @@ class TestStdCapture:
|
|||
def test_capturing_readouterr(self):
|
||||
cap = self.getcapture()
|
||||
try:
|
||||
print "hello world"
|
||||
print >>sys.stderr, "hello error"
|
||||
print ("hello world")
|
||||
sys.stderr.write("hello error\n")
|
||||
out, err = cap.readouterr()
|
||||
assert out == "hello world\n"
|
||||
assert err == "hello error\n"
|
||||
print >>sys.stderr, "error2"
|
||||
sys.stderr.write("error2")
|
||||
finally:
|
||||
out, err = cap.reset()
|
||||
assert err == "error2\n"
|
||||
assert err == "error2"
|
||||
|
||||
def test_capturing_mixed(self):
|
||||
cap = self.getcapture(mixed=True)
|
||||
print "hello",
|
||||
print >>sys.stderr, "world",
|
||||
print >>sys.stdout, ".",
|
||||
sys.stdout.write("hello ")
|
||||
sys.stderr.write("world")
|
||||
sys.stdout.write(".")
|
||||
out, err = cap.reset()
|
||||
assert out.strip() == "hello world ."
|
||||
assert out.strip() == "hello world."
|
||||
assert not err
|
||||
|
||||
def test_capturing_twice_error(self):
|
||||
cap = self.getcapture()
|
||||
print "hello"
|
||||
print ("hello")
|
||||
cap.reset()
|
||||
py.test.raises(Exception, "cap.reset()")
|
||||
|
||||
|
|
@ -160,12 +187,12 @@ class TestStdCapture:
|
|||
oldout = sys.stdout
|
||||
olderr = sys.stderr
|
||||
cap = self.getcapture()
|
||||
print "hello",
|
||||
print >>sys.stderr, "world",
|
||||
sys.stdout.write("hello")
|
||||
sys.stderr.write("world")
|
||||
sys.stdout = py.io.TextIO()
|
||||
sys.stderr = py.io.TextIO()
|
||||
print "not seen"
|
||||
print >>sys.stderr, "not seen"
|
||||
print ("not seen" )
|
||||
sys.stderr.write("not seen\n")
|
||||
out, err = cap.reset()
|
||||
assert out == "hello"
|
||||
assert err == "world"
|
||||
|
|
@ -174,9 +201,9 @@ class TestStdCapture:
|
|||
|
||||
def test_capturing_error_recursive(self):
|
||||
cap1 = self.getcapture()
|
||||
print "cap1"
|
||||
print ("cap1")
|
||||
cap2 = self.getcapture()
|
||||
print "cap2"
|
||||
print ("cap2")
|
||||
out2, err2 = cap2.reset()
|
||||
py.test.raises(Exception, "cap2.reset()")
|
||||
out1, err1 = cap1.reset()
|
||||
|
|
@ -185,18 +212,18 @@ class TestStdCapture:
|
|||
|
||||
def test_just_out_capture(self):
|
||||
cap = self.getcapture(out=True, err=False)
|
||||
print >>sys.stdout, "hello"
|
||||
print >>sys.stderr, "world"
|
||||
sys.stdout.write("hello")
|
||||
sys.stderr.write("world")
|
||||
out, err = cap.reset()
|
||||
assert out == "hello\n"
|
||||
assert out == "hello"
|
||||
assert not err
|
||||
|
||||
def test_just_err_capture(self):
|
||||
cap = self.getcapture(out=False, err=True)
|
||||
print >>sys.stdout, "hello"
|
||||
print >>sys.stderr, "world"
|
||||
sys.stdout.write("hello")
|
||||
sys.stderr.write("world")
|
||||
out, err = cap.reset()
|
||||
assert err == "world\n"
|
||||
assert err == "world"
|
||||
assert not out
|
||||
|
||||
def test_stdin_restored(self):
|
||||
|
|
@ -208,9 +235,9 @@ class TestStdCapture:
|
|||
assert sys.stdin is old
|
||||
|
||||
def test_stdin_nulled_by_default(self):
|
||||
print "XXX this test may well hang instead of crashing"
|
||||
print "XXX which indicates an error in the underlying capturing"
|
||||
print "XXX mechanisms"
|
||||
print ("XXX this test may well hang instead of crashing")
|
||||
print ("XXX which indicates an error in the underlying capturing")
|
||||
print ("XXX mechanisms" )
|
||||
cap = self.getcapture()
|
||||
py.test.raises(IOError, "sys.stdin.read()")
|
||||
out, err = cap.reset()
|
||||
|
|
@ -218,15 +245,15 @@ class TestStdCapture:
|
|||
def test_suspend_resume(self):
|
||||
cap = self.getcapture(out=True, err=False, in_=False)
|
||||
try:
|
||||
print "hello"
|
||||
print ("hello")
|
||||
sys.stderr.write("error\n")
|
||||
out, err = cap.suspend()
|
||||
assert out == "hello\n"
|
||||
assert not err
|
||||
print "in between"
|
||||
print ("in between")
|
||||
sys.stderr.write("in between\n")
|
||||
cap.resume()
|
||||
print "after"
|
||||
print ("after")
|
||||
sys.stderr.write("error_after\n")
|
||||
finally:
|
||||
out, err = cap.reset()
|
||||
|
|
@ -239,20 +266,22 @@ class TestStdCaptureFD(TestStdCapture):
|
|||
|
||||
def test_intermingling(self):
|
||||
cap = self.getcapture()
|
||||
os.write(1, "1")
|
||||
print >>sys.stdout, 2,
|
||||
os.write(1, "3")
|
||||
os.write(2, "a")
|
||||
print >>sys.stderr, "b",
|
||||
os.write(2, "c")
|
||||
oswritebytes(1, "1")
|
||||
sys.stdout.write(str(2))
|
||||
sys.stdout.flush()
|
||||
oswritebytes(1, "3")
|
||||
oswritebytes(2, "a")
|
||||
sys.stderr.write("b")
|
||||
sys.stderr.flush()
|
||||
oswritebytes(2, "c")
|
||||
out, err = cap.reset()
|
||||
assert out == "123"
|
||||
assert err == "abc"
|
||||
|
||||
def test_callcapture(self):
|
||||
def func(x, y):
|
||||
print x
|
||||
print >>py.std.sys.stderr, y
|
||||
print (x)
|
||||
py.std.sys.stderr.write(str(y))
|
||||
return 42
|
||||
|
||||
res, out, err = py.io.StdCaptureFD.call(func, 3, y=4)
|
||||
|
|
@ -264,10 +293,10 @@ def test_capture_no_sys():
|
|||
capsys = py.io.StdCapture()
|
||||
try:
|
||||
cap = py.io.StdCaptureFD(patchsys=False)
|
||||
print >>sys.stdout, "hello"
|
||||
print >>sys.stderr, "world"
|
||||
os.write(1, "1")
|
||||
os.write(2, "2")
|
||||
sys.stdout.write("hello")
|
||||
sys.stderr.write("world")
|
||||
oswritebytes(1, "1")
|
||||
oswritebytes(2, "2")
|
||||
out, err = cap.reset()
|
||||
assert out == "1"
|
||||
assert err == "2"
|
||||
|
|
@ -276,10 +305,10 @@ def test_capture_no_sys():
|
|||
|
||||
def test_callcapture_nofd():
|
||||
def func(x, y):
|
||||
os.write(1, "hello")
|
||||
os.write(2, "hello")
|
||||
print x
|
||||
print >>sys.stderr, y
|
||||
oswritebytes(1, "hello")
|
||||
oswritebytes(2, "hello")
|
||||
print (x)
|
||||
sys.stderr.write(str(y))
|
||||
return 42
|
||||
|
||||
capfd = py.io.StdCaptureFD(patchsys=False)
|
||||
|
|
|
|||
|
|
@ -44,11 +44,10 @@ def test_terminalwriter_dumb_term_no_markup(monkeypatch):
|
|||
assert not tw.hasmarkup
|
||||
|
||||
def test_unicode_encoding():
|
||||
l = []
|
||||
msg = unicode('b\u00f6y', 'utf8')
|
||||
msg = py.builtin._totext('b\u00f6y', 'utf8')
|
||||
for encoding in 'utf8', 'latin1':
|
||||
tw = py.io.TerminalWriter(l.append)
|
||||
tw._encoding = encoding
|
||||
l = []
|
||||
tw = py.io.TerminalWriter(l.append, encoding=encoding)
|
||||
tw.line(msg)
|
||||
assert l[0] == msg.encode(encoding)
|
||||
|
||||
|
|
@ -64,7 +63,7 @@ class BaseTests:
|
|||
tw = self.getwriter()
|
||||
for encoding in 'utf8', 'latin1':
|
||||
tw._encoding = encoding
|
||||
msg = unicode('b\u00f6y', 'utf8')
|
||||
msg = py.builtin._totext('b\u00f6y', 'utf8')
|
||||
tw.line(msg)
|
||||
l = self.getlines()
|
||||
assert l[0] == msg + "\n"
|
||||
|
|
|
|||
|
|
@ -372,6 +372,11 @@ class LocalPath(FSBase):
|
|||
""" write string content into path. """
|
||||
s = str(content)
|
||||
f = self.open(mode)
|
||||
if not hasattr(s, 'decode'): # byte string
|
||||
encoding = getattr(f, 'encoding', None)
|
||||
if not encoding:
|
||||
encoding = sys.getdefaultencoding()
|
||||
s = s.encode(encoding)
|
||||
try:
|
||||
f.write(s)
|
||||
finally:
|
||||
|
|
|
|||
|
|
@ -111,12 +111,11 @@ class CaptureManager:
|
|||
|
||||
def _maketempfile(self):
|
||||
f = py.std.tempfile.TemporaryFile()
|
||||
newf = py.io.dupfile(f)
|
||||
encoding = getattr(newf, 'encoding', None) or "UTF-8"
|
||||
return EncodedFile(newf, encoding)
|
||||
newf = py.io.dupfile(f, encoding="UTF-8")
|
||||
return newf
|
||||
|
||||
def _makestringio(self):
|
||||
return EncodedFile(py.io.TextIO(), 'UTF-8')
|
||||
return py.io.TextIO()
|
||||
|
||||
def _startcapture(self, method):
|
||||
if method == "fd":
|
||||
|
|
@ -271,25 +270,3 @@ class CaptureFuncarg:
|
|||
self.capture.reset()
|
||||
del self.capture
|
||||
|
||||
class EncodedFile(object):
|
||||
def __init__(self, _stream, encoding):
|
||||
self._stream = _stream
|
||||
self.encoding = encoding
|
||||
|
||||
if py.std.sys.version_info < (3,0):
|
||||
def write(self, obj):
|
||||
if isinstance(obj, unicode):
|
||||
self._stream.write(obj.encode(self.encoding))
|
||||
else:
|
||||
self._stream.write(obj)
|
||||
else:
|
||||
def write(self, obj):
|
||||
self._stream.write(obj.encode(self.encoding))
|
||||
|
||||
def writelines(self, linelist):
|
||||
data = ''.join(linelist)
|
||||
self.write(data)
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self._stream, name)
|
||||
|
||||
|
|
|
|||
|
|
@ -271,7 +271,7 @@ class TmpTestdir:
|
|||
old.chdir()
|
||||
|
||||
def _run(self, *cmdargs):
|
||||
cmdargs = map(str, cmdargs)
|
||||
cmdargs = [str(x) for x in cmdargs]
|
||||
p1 = py.path.local("stdout")
|
||||
p2 = py.path.local("stderr")
|
||||
print_("running", cmdargs, "curdir=", py.path.local())
|
||||
|
|
@ -285,10 +285,10 @@ class TmpTestdir:
|
|||
out, err = p1.readlines(cr=0), p2.readlines(cr=0)
|
||||
if err:
|
||||
for line in err:
|
||||
print >>py.std.sys.stderr, line
|
||||
py.builtin.print_(line, file=sys.stderr)
|
||||
if out:
|
||||
for line in out:
|
||||
print >>py.std.sys.stdout, line
|
||||
py.builtin.print_(line, file=sys.stdout)
|
||||
return RunResult(ret, out, err)
|
||||
|
||||
def runpybin(self, scriptname, *args):
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import py, os, sys
|
||||
from py.__.test.plugin.pytest_capture import CaptureManager, EncodedFile
|
||||
from py.__.test.plugin.pytest_capture import CaptureManager
|
||||
|
||||
class TestCaptureManager:
|
||||
def test_configure_per_fspath(self, testdir):
|
||||
|
|
@ -21,7 +21,7 @@ class TestCaptureManager:
|
|||
try:
|
||||
capman = CaptureManager()
|
||||
capman.resumecapture(method)
|
||||
print "hello"
|
||||
print ("hello")
|
||||
out, err = capman.suspendcapture()
|
||||
if method == "no":
|
||||
assert old == (sys.stdout, sys.stderr, sys.stdin)
|
||||
|
|
@ -41,12 +41,12 @@ class TestCaptureManager:
|
|||
capman.resumecapture("fd")
|
||||
py.test.raises(ValueError, 'capman.resumecapture("fd")')
|
||||
py.test.raises(ValueError, 'capman.resumecapture("sys")')
|
||||
os.write(1, "hello\n")
|
||||
os.write(1, "hello\n".encode('ascii'))
|
||||
out, err = capman.suspendcapture()
|
||||
assert out == "hello\n"
|
||||
capman.resumecapture("sys")
|
||||
os.write(1, "hello\n")
|
||||
print >>sys.stderr, "world"
|
||||
os.write(1, "hello\n".encode('ascii'))
|
||||
py.builtin.print_("world", file=sys.stderr)
|
||||
out, err = capman.suspendcapture()
|
||||
assert not out
|
||||
assert err == "world\n"
|
||||
|
|
@ -55,11 +55,13 @@ class TestCaptureManager:
|
|||
|
||||
@py.test.mark.multi(method=['fd', 'sys'])
|
||||
def test_capturing_unicode(testdir, method):
|
||||
if sys.version_info >= (3,0):
|
||||
py.test.skip("test not applicable on 3k")
|
||||
testdir.makepyfile("""
|
||||
# taken from issue 227 from nosetests
|
||||
def test_unicode():
|
||||
import sys
|
||||
print sys.stdout
|
||||
print (sys.stdout)
|
||||
print u'b\\u00f6y'
|
||||
""")
|
||||
result = testdir.runpytest("--capture=%s" % method)
|
||||
|
|
@ -71,27 +73,16 @@ def test_capturing_unicode(testdir, method):
|
|||
def test_capturing_bytes_in_utf8_encoding(testdir, method):
|
||||
testdir.makepyfile("""
|
||||
def test_unicode():
|
||||
print 'b\\u00f6y'
|
||||
print ('b\\u00f6y')
|
||||
""")
|
||||
result = testdir.runpytest("--capture=%s" % method)
|
||||
result.stdout.fnmatch_lines([
|
||||
"*1 passed*"
|
||||
])
|
||||
|
||||
def test_UnicodeFile(testdir):
|
||||
p = testdir.makepyfile("hello")
|
||||
f = p.open('w')
|
||||
pf = EncodedFile(f, "UTF-8")
|
||||
pf.write(u'b\\00f6y\n')
|
||||
pf.write('b\\00f6y\n')
|
||||
pf.close()
|
||||
assert f.closed
|
||||
lines = p.readlines()
|
||||
assert lines[0] == lines[1]
|
||||
|
||||
def test_collect_capturing(testdir):
|
||||
p = testdir.makepyfile("""
|
||||
print "collect %s failure" % 13
|
||||
print ("collect %s failure" % 13)
|
||||
import xyz42123
|
||||
""")
|
||||
result = testdir.runpytest(p)
|
||||
|
|
@ -104,14 +95,14 @@ class TestPerTestCapturing:
|
|||
def test_capture_and_fixtures(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
def setup_module(mod):
|
||||
print "setup module"
|
||||
print ("setup module")
|
||||
def setup_function(function):
|
||||
print "setup", function.__name__
|
||||
print ("setup " + function.__name__)
|
||||
def test_func1():
|
||||
print "in func1"
|
||||
print ("in func1")
|
||||
assert 0
|
||||
def test_func2():
|
||||
print "in func2"
|
||||
print ("in func2")
|
||||
assert 0
|
||||
""")
|
||||
result = testdir.runpytest(p)
|
||||
|
|
@ -128,14 +119,14 @@ class TestPerTestCapturing:
|
|||
p = testdir.makepyfile("""
|
||||
import sys
|
||||
def setup_module(func):
|
||||
print "module-setup"
|
||||
print ("module-setup")
|
||||
def setup_function(func):
|
||||
print "function-setup"
|
||||
print ("function-setup")
|
||||
def test_func():
|
||||
print "in function"
|
||||
print ("in function")
|
||||
assert 0
|
||||
def teardown_function(func):
|
||||
print "in teardown"
|
||||
print ("in teardown")
|
||||
""")
|
||||
result = testdir.runpytest(p)
|
||||
result.stdout.fnmatch_lines([
|
||||
|
|
@ -151,9 +142,9 @@ class TestPerTestCapturing:
|
|||
def test_no_carry_over(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
def test_func1():
|
||||
print "in func1"
|
||||
print ("in func1")
|
||||
def test_func2():
|
||||
print "in func2"
|
||||
print ("in func2")
|
||||
assert 0
|
||||
""")
|
||||
result = testdir.runpytest(p)
|
||||
|
|
@ -165,12 +156,12 @@ class TestPerTestCapturing:
|
|||
def test_teardown_capturing(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
def setup_function(function):
|
||||
print "setup func1"
|
||||
print ("setup func1")
|
||||
def teardown_function(function):
|
||||
print "teardown func1"
|
||||
print ("teardown func1")
|
||||
assert 0
|
||||
def test_func1():
|
||||
print "in func1"
|
||||
print ("in func1")
|
||||
pass
|
||||
""")
|
||||
result = testdir.runpytest(p)
|
||||
|
|
@ -186,7 +177,7 @@ class TestPerTestCapturing:
|
|||
def test_teardown_final_capturing(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
def teardown_module(mod):
|
||||
print "teardown module"
|
||||
print ("teardown module")
|
||||
assert 0
|
||||
def test_func():
|
||||
pass
|
||||
|
|
@ -203,11 +194,11 @@ class TestPerTestCapturing:
|
|||
p1 = testdir.makepyfile("""
|
||||
import sys
|
||||
def test_capturing():
|
||||
print 42
|
||||
print >>sys.stderr, 23
|
||||
print (42)
|
||||
sys.stderr.write(str(23))
|
||||
def test_capturing_error():
|
||||
print 1
|
||||
print >>sys.stderr, 2
|
||||
print (1)
|
||||
sys.stderr.write(str(2))
|
||||
raise ValueError
|
||||
""")
|
||||
result = testdir.runpytest(p1)
|
||||
|
|
@ -246,7 +237,7 @@ class TestLoggingInteraction:
|
|||
logging.warn("hello1")
|
||||
outerr = cap.suspend()
|
||||
|
||||
print "suspeneded and captured", outerr
|
||||
print ("suspeneded and captured %%s" %% (outerr,))
|
||||
|
||||
logging.warn("hello2")
|
||||
|
||||
|
|
@ -254,7 +245,7 @@ class TestLoggingInteraction:
|
|||
logging.warn("hello3")
|
||||
|
||||
outerr = cap.suspend()
|
||||
print "suspend2 and captured", outerr
|
||||
print ("suspend2 and captured %%s" %% (outerr,))
|
||||
""" % rootdir)
|
||||
result = testdir.runpython(p)
|
||||
assert result.stdout.fnmatch_lines([
|
||||
|
|
@ -279,7 +270,7 @@ class TestLoggingInteraction:
|
|||
assert 0
|
||||
""")
|
||||
for optargs in (('--capture=sys',), ('--capture=fd',)):
|
||||
print optargs
|
||||
print (optargs)
|
||||
result = testdir.runpytest(p, *optargs)
|
||||
s = result.stdout.str()
|
||||
result.stdout.fnmatch_lines([
|
||||
|
|
@ -305,7 +296,7 @@ class TestLoggingInteraction:
|
|||
assert 0
|
||||
""")
|
||||
for optargs in (('--capture=sys',), ('--capture=fd',)):
|
||||
print optargs
|
||||
print (optargs)
|
||||
result = testdir.runpytest(p, *optargs)
|
||||
s = result.stdout.str()
|
||||
result.stdout.fnmatch_lines([
|
||||
|
|
@ -320,7 +311,7 @@ class TestCaptureFuncarg:
|
|||
def test_std_functional(self, testdir):
|
||||
reprec = testdir.inline_runsource("""
|
||||
def test_hello(capsys):
|
||||
print 42
|
||||
print (42)
|
||||
out, err = capsys.readouterr()
|
||||
assert out.startswith("42")
|
||||
""")
|
||||
|
|
@ -330,7 +321,7 @@ class TestCaptureFuncarg:
|
|||
reprec = testdir.inline_runsource("""
|
||||
def test_hello(capfd):
|
||||
import os
|
||||
os.write(1, "42")
|
||||
os.write(1, "42".encode('ascii'))
|
||||
out, err = capfd.readouterr()
|
||||
assert out.startswith("42")
|
||||
capfd.close()
|
||||
|
|
@ -352,7 +343,7 @@ class TestCaptureFuncarg:
|
|||
p = testdir.makepyfile("""
|
||||
def test_hello(capfd):
|
||||
import os
|
||||
os.write(1, "42")
|
||||
os.write(1, str(42).encode('ascii'))
|
||||
raise KeyboardInterrupt()
|
||||
""")
|
||||
result = testdir.runpytest(p)
|
||||
|
|
|
|||
Loading…
Reference in New Issue