p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[svn r37741] monster checking for 

* unifying IO capturing methods
* py.io.StdCapture and py.io.StdCaptureFD
  (and both have a classmethod 'call' that is
  a shortcut for capturing output while
  executing a function)
* removing lots of duplicate code
* providing some examples in py/doc/io.txt

at least tests on win32 and linux seem
to pass all for me.

--HG--
branch : trunk
This commit is contained in:
hpk 2007-02-01 16:20:39 +01:00
parent d9572239a8
commit b706ec2f95
18 changed files with 126 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
py/__init__.py
View File

@ -9,14 +9,14 @@ version = "0.8.80-alpha2"
initpkg(__name__,
description = "py.test and the py lib",
revision = int('$LastChangedRevision: 37699 $'.split(':')[1][:-1]),
lastchangedate = '$LastChangedDate: 2007-01-31 23:23:24 +0100 (Wed, 31 Jan 2007) $',
revision = int('$LastChangedRevision: 37741 $'.split(':')[1][:-1]),
lastchangedate = '$LastChangedDate: 2007-02-01 16:20:39 +0100 (Thu, 01 Feb 2007) $',
version = version,
url = "http://codespeak.net/py",
download_url = "http://codespeak.net/download/py/%s.tar.gz" %(version,),
license = "MIT license",
platforms = ['unix', 'linux', 'cygwin'],
author = "holger krekel & others",
author = "holger krekel, Armin Rigo, Guido Wesdorp, Maciej Fijalkowski & others",
author_email = "py-dev@codespeak.net",
long_description = globals()['__doc__'],
@ -94,9 +94,9 @@ initpkg(__name__,
# input-output helping
'io.dupfile' : ('./io/dupfile.py', 'dupfile'),
'io.FDCapture' : ('./io/capture.py', 'FDCapture'),
'io.OutErrCapture' : ('./io/capture.py', 'OutErrCapture'),
'io.callcapture' : ('./io/capture.py', 'callcapture'),
'io.FDCapture' : ('./io/fdcapture.py', 'FDCapture'),
'io.StdCapture' : ('./io/stdcapture.py', 'StdCapture'),
'io.StdCaptureFD' : ('./io/stdcapture.py', 'StdCaptureFD'),
# error module, defining all errno's as Classes
'error' : ('./misc/error.py', 'error'),

2
py/doc/TODO.txt
View File

@ -97,7 +97,7 @@ testing
os.fork to work))
* see why startcapture() used to not use FD-based
"py.io.OutErrCapture" to isolate standard output.
"py.io.StdCaptureFD" to isolate standard output.
use that check if all py and PyPy tests pass
as good as they do without.

4
py/doc/index.txt
View File

@ -11,7 +11,9 @@ py.test and the py lib - documentation
`py.code`_: High-level access/manipulation of Python code and traceback objects.
`py.xml`_ a fast'n'easy way to generate xml/html documents (including CSS-styling)
`py.xml`_ for generating in-memory xml/html object trees
`py.io`_ Helper Classes for Capturing of Input/Output
`py.log`_ an alpha document about the ad-hoc logging facilities

45
py/doc/io.txt Normal file
View File

@ -0,0 +1,45 @@
=======
py.io
=======
.. contents::
.. sectnum::
The 'py' lib provides helper classes for capturing IO during
execution of a program.
IO Capturing examples
===============================================
:api:`py.io.StdCapture`
---------------------------
Basic Example:
>>> import py
>>> capture = py.io.StdCapture()
>>> print "hello"
>>> out,err = capture.reset()
>>> out.strip() == "hello"
True
For calling functions you may use a shortcut:
>>> import py
>>> def f(): print "hello"
>>> res, out, err = py.io.StdCapture.call(f)
>>> out.strip() == "hello"
True
:api:`py.io.StdCaptureFD`
---------------------------
If you also want to capture writes to the stdout/stdin
filedescriptors you may invoke:
>>> import py, sys
>>> capture = py.io.StdCaptureFD()
>>> sys.stdout.write("hello")
>>> sys.stderr.write("world")
>>> out,err = capture.reset()
>>> out.strip() + err.strip() == "helloworld"
True

42
py/io/capture.py → py/io/fdcapture.py
View File

@ -57,45 +57,3 @@ class FDCapture:
finally:
tempfp.close()
class OutErrCapture:
""" capture Stdout and Stderr both on filedescriptor
and sys.stdout/stderr level.
"""
def __init__(self, out=True, err=True, patchsys=True):
if out:
self.out = FDCapture(1)
if patchsys:
self.out.setasfile('stdout')
if err:
self.err = FDCapture(2)
if patchsys:
self.err.setasfile('stderr')
def reset(self):
""" reset sys.stdout and sys.stderr
returns a tuple of file objects (out, err) for the captured
data
"""
out = err = ""
if hasattr(self, 'out'):
outfile = self.out.done()
out = outfile.read()
if hasattr(self, 'err'):
errfile = self.err.done()
err = errfile.read()
return out, err
def callcapture(func, *args, **kwargs):
""" call the given function with args/kwargs
and return a (res, out, err) tuple where
out and err represent the output/error output
during function execution.
"""
so = OutErrCapture()
try:
res = func(*args, **kwargs)
finally:
out, err = so.reset()
return res, out, err

62
py/misc/simplecapture.py → py/io/stdcapture.py
View File

@ -1,13 +1,55 @@
"""
capture stdout/stderr
"""
import os
import sys
import py
try: from cStringIO import StringIO
except ImportError: from StringIO import StringIO
class SimpleOutErrCapture:
class Capture(object):
def call(cls, func, *args, **kwargs):
""" return a (res, out, err) tuple where
out and err represent the output/error output
during function execution.
call the given function with args/kwargs
and capture output/error during its execution.
"""
so = cls()
try:
res = func(*args, **kwargs)
finally:
out, err = so.reset()
return res, out, err
call = classmethod(call)
class StdCaptureFD(Capture):
""" capture Stdout and Stderr both on filedescriptor
and sys.stdout/stderr level.
"""
def __init__(self, out=True, err=True, patchsys=True):
if out:
self.out = py.io.FDCapture(1)
if patchsys:
self.out.setasfile('stdout')
if err:
self.err = py.io.FDCapture(2)
if patchsys:
self.err.setasfile('stderr')
def reset(self):
""" reset sys.stdout and sys.stderr
returns a tuple of file objects (out, err) for the captured
data
"""
out = err = ""
if hasattr(self, 'out'):
outfile = self.out.done()
out = outfile.read()
if hasattr(self, 'err'):
errfile = self.err.done()
err = errfile.read()
return out, err
class StdCapture(Capture):
""" capture sys.stdout/sys.stderr (but not system level fd 1 and 2).
this captures only "In-Memory" and is currently intended to be
@ -48,11 +90,3 @@ class DontReadFromInput:
readline = read
readlines = read
__iter__ = read
def callcapture(func, *args, **kwargs):
so = SimpleOutErrCapture()
try:
res = func(*args, **kwargs)
finally:
out, err = so.reset()
return res, out, err

10
py/io/test/test_capture.py
View File

@ -59,7 +59,7 @@ class TestFDCapture:
class TestCapturing:
def getcapture(self):
return py.io.OutErrCapture()
return py.io.StdCaptureFD()
def test_capturing_simple(self):
cap = self.getcapture()
@ -120,13 +120,13 @@ def test_callcapture():
print >>py.std.sys.stderr, y
return 42
res, out, err = py.io.callcapture(func, 3, y=4)
res, out, err = py.io.StdCaptureFD.call(func, 3, y=4)
assert res == 42
assert out.startswith("3")
assert err.startswith("4")
def test_just_out_capture():
cap = py.io.OutErrCapture(out=True, err=False)
cap = py.io.StdCaptureFD(out=True, err=False)
print >>sys.stdout, "hello"
print >>sys.stderr, "world"
out, err = cap.reset()
@ -134,7 +134,7 @@ def test_just_out_capture():
assert not err
def test_just_err_capture():
cap = py.io.OutErrCapture(out=False, err=True)
cap = py.io.StdCaptureFD(out=False, err=True)
print >>sys.stdout, "hello"
print >>sys.stderr, "world"
out, err = cap.reset()
@ -142,7 +142,7 @@ def test_just_err_capture():
assert not out
def test_capture_no_sys():
cap = py.io.OutErrCapture(patchsys=False)
cap = py.io.StdCaptureFD(patchsys=False)
print >>sys.stdout, "hello"
print >>sys.stderr, "world"
os.write(1, "1")

31
py/misc/testing/test_simplecapture.py → py/io/test/test_simplecapture.py
View File

@ -1,29 +1,10 @@
import os, sys
import py
from py.__.misc.simplecapture import SimpleOutErrCapture, callcapture
from py.__.misc.capture import Capture, FDCapture
class TestFDCapture:
def test_basic(self):
tmpfile = py.std.os.tmpfile()
fd = tmpfile.fileno()
cap = FDCapture(fd)
os.write(fd, "hello")
f = cap.done()
s = f.read()
assert s == "hello"
def test_stderr(self):
cap = FDCapture(2, 'stderr')
print >>sys.stderr, "hello"
f = cap.done()
s = f.read()
assert s == "hello\n"
class TestCapturingOnSys:
def getcapture(self):
return SimpleOutErrCapture()
return py.io.StdCapture()
def test_capturing_simple(self):
cap = self.getcapture()
@ -73,20 +54,14 @@ class TestCapturingOnSys:
finally:
cap.reset()
def test_callcapture():
def test_callcapture_nofd():
def func(x, y):
print x
print >>py.std.sys.stderr, y
return 42
res, out, err = callcapture(func, 3, y=4)
res, out, err = py.io.StdCapture.call(func, 3, y=4)
assert res == 42
assert out.startswith("3")
assert err.startswith("4")
class TestCapturingOnFDs(TestCapturingOnSys):
def test_reading_stdin_while_captured_doesnt_hang(self):
py.test.skip("Hangs in py.test --session=R")
def getcapture(self):
return Capture()

3
py/log/testing/test_log.py
View File

@ -1,7 +1,8 @@
import py
from py.__.misc.simplecapture import callcapture
import sys
callcapture = py.io.StdCapture.call
def setup_module(mod):
mod.tempdir = py.test.ensuretemp("py.log-test")
mod.logstate = py.log._getstate()

49
py/misc/capture.py
View File

@ -1,49 +0,0 @@
import os, sys
class FDCapture:
def __init__(self, targetfd, sysattr=None):
self.targetfd = targetfd
self.tmpfile = self.maketmpfile()
self._savefd = os.dup(targetfd)
os.dup2(self.tmpfile.fileno(), targetfd)
if sysattr is not None:
self._reset = (lambda oldval=getattr(sys, sysattr):
setattr(sys, sysattr, oldval))
setattr(sys, sysattr, self.tmpfile)
def done(self):
os.dup2(self._savefd, self.targetfd)
if hasattr(self, '_reset'):
self._reset()
del self._reset
os.close(self._savefd)
f = self.tmpfile
f.seek(0)
del self._savefd
del self.tmpfile
return f
def maketmpfile(self):
f = os.tmpfile()
fd = f.fileno()
newfd = os.dup(fd)
newf = os.fdopen(newfd, 'w+b', 0)
f.close()
return newf
class Capture:
def __init__(self):
self._out = FDCapture(1, 'stdout')
self._oldsysout = sys.stdout
sys.stdout = self._out.tmpfile
self._err = FDCapture(2, 'stderr')
self._olderrout = sys.stderr
sys.stderr = self._err.tmpfile
def reset(self):
outfile = self._out.done()
errfile = self._err.done()
return outfile.read(), errfile.read()

6
py/test/collect.py
View File

@ -379,11 +379,7 @@ class Module(FSCollector, PyCollectorMixin):
def startcapture(self):
if not self.config.option.nocapture:
assert not hasattr(self, '_capture')
self._capture = py.io.OutErrCapture()
# XXX integrate this into py.io / refactor
# execnet/py.test capturing mechanisms
#from py.__.misc.simplecapture import SimpleOutErrCapture
#self._capture = SimpleOutErrCapture()
self._capture = py.io.StdCaptureFD()
def finishcapture(self):
if hasattr(self, '_capture'):

5
py/test/item.py
View File

@ -32,10 +32,7 @@ class SetupState(object):
class Item(py.test.collect.Collector):
def startcapture(self):
if not self.config.option.nocapture:
# XXX refactor integrate capturing
self._capture = py.io.OutErrCapture()
#from py.__.misc.simplecapture import SimpleOutErrCapture
#self._capture = SimpleOutErrCapture()
self._capture = py.io.StdCaptureFD()
def finishcapture(self):
if hasattr(self, '_capture'):

7
py/test/rsession/local.py
View File

@ -2,6 +2,7 @@
""" local-only operations
"""
import py
from py.__.test.rsession.executor import BoxExecutor, RunExecutor,\
ApigenExecutor
from py.__.test.rsession import report
@ -9,10 +10,8 @@ from py.__.test.rsession.outcome import ReprOutcome
# XXX copied from session.py
def startcapture(session):
if not session.config.option.nocapture and not session.config.option.usepdb:
# XXX refactor integrate capturing
from py.__.misc.simplecapture import SimpleOutErrCapture
session._capture = SimpleOutErrCapture()
if not session.config.option.nocapture:
session._capture = py.io.StdCapture()
def finishcapture(session):
if hasattr(session, '_capture'):

2
py/test/rsession/rsession.py
View File

@ -259,7 +259,7 @@ class LSession(AbstractSession):
raise NotImplementedError("%s does not contain 'build' "
"function" %(apigen,))
print >>sys.stderr, 'building documentation'
capture = py.io.OutErrCapture()
capture = py.io.StdCaptureFD()
try:
pkgdir = self.getpkgdir(self.config.args[0])
apigen.build(pkgdir,

10
py/test/rsession/testing/test_reporter.py
View File

@ -74,7 +74,7 @@ class AbstractTestReporter(object):
for outcome in outcomes:
r.report(report.ReceivedItemOutcome(ch, item, outcome))
cap = py.io.OutErrCapture()
cap = py.io.StdCaptureFD()
boxfun(config, item, outcomes)
out, err = cap.reset()
assert not err
@ -97,7 +97,7 @@ class AbstractTestReporter(object):
for outcome in outcomes:
r.report(report.ReceivedItemOutcome(ch, funcitem, outcome))
cap = py.io.OutErrCapture()
cap = py.io.StdCaptureFD()
boxfun(self.pkgdir, config, moditem, funcitem, outcomes)
out, err = cap.reset()
assert not err
@ -125,7 +125,7 @@ class AbstractTestReporter(object):
r = self.reporter(config, hosts)
list(rootcol.tryiter(reporterror=lambda x : AbstractSession.reporterror(r.report, x)))
cap = py.io.OutErrCapture()
cap = py.io.StdCaptureFD()
boxfun()
out, err = cap.reset()
assert not err
@ -147,7 +147,7 @@ class AbstractTestReporter(object):
list(rootcol.tryiter(reporterror=lambda x : AbstractSession.reporterror(r.report, x)))
r.report(report.TestFinished())
cap = py.io.OutErrCapture()
cap = py.io.StdCaptureFD()
boxfun()
out, err = cap.reset()
assert not err
@ -156,7 +156,7 @@ class AbstractTestReporter(object):
def _test_still_to_go(self):
tmpdir = py.test.ensuretemp("stilltogo")
tmpdir.ensure("__init__.py")
cap = py.io.OutErrCapture()
cap = py.io.StdCaptureFD()
config = py.test.config._reparse([str(tmpdir)])
hosts = [HostInfo(i) for i in ["host1", "host2", "host3"]]
r = self.reporter(config, hosts)

2
py/test/testing/test_session.py
View File

@ -189,7 +189,7 @@ class TestTerminalSession:
import py
class Function(py.test.Function):
def startcapture(self):
self._mycapture = py.io.OutErrCapture()
self._mycapture = py.io.StdCaptureFD()
def finishcapture(self):
self._testmycapture = self._mycapture.reset()

10
py/test/todo-cleanup.txt
View File

@ -1,10 +0,0 @@
fix --looponfailing: currently in case of failures
all tests are re-run. the problem was introduced
while cleaning up session.main() calls ...
the setup of remote and local config objects
and collectors needs to be reviewed anyway
(and the terminalsession and rsession handling
of such config object should be unified with
respect to this configuration/failure communication)

2
py/thread/testing/test_pool.py
View File

@ -77,7 +77,7 @@ def test_join_timeout():
pool.join(timeout=0.1)
def test_pool_clean_shutdown():
capture = py.io.OutErrCapture()
capture = py.io.StdCaptureFD()
pool = WorkerPool()
def f():
pass