[svn r37729] Added some code to py.io.FDCapture and py.io.OutErrCapture to allow writing

to the original (patched) file descriptor. Also made that the capturing object
is passed to apigen.py's build() function (from rsession.py), which uses the
new methods to print progress information.

--HG--
branch : trunk

py/io/capture.py

@@ -1,9 +1,12 @@
 
-import os, sys
+import os
+import sys
+import thread
 import py
 
 class FDCapture: 
     """ Capture IO to/from a given os-level filedescriptor. """
+    
     def __init__(self, targetfd, tmpfile=None): 
         self.targetfd = targetfd
         if tmpfile is None: 
@@ -14,16 +17,22 @@ class FDCapture:
         self._patched = []
 
     def setasfile(self, name, module=sys): 
+        """ patch <module>.<name> to self.tmpfile
+        """
         key = (module, name)
         self._patched.append((key, getattr(module, name)))
         setattr(module, name, self.tmpfile) 
 
     def unsetfiles(self): 
+        """ unpatch all patched items
+        """
         while self._patched: 
             (module, name), value = self._patched.pop()
             setattr(module, name, value) 
 
     def done(self): 
+        """ unpatch and clean up, returns the self.tmpfile (file object)
+        """
         os.dup2(self._savefd, self.targetfd) 
         self.unsetfiles() 
         os.close(self._savefd) 
@@ -31,11 +40,23 @@ class FDCapture:
         return self.tmpfile 
 
     def maketmpfile(self): 
+        """ create a temporary file
+        """
         f = os.tmpfile()
         newf = py.io.dupfile(f) 
         f.close()
         return newf 
 
+    def writeorg(self, str):
+        """ write a string to the original file descriptor
+        """
+        tempfp = os.tmpfile()
+        try:
+            os.dup2(self._savefd, tempfp.fileno())
+            tempfp.write(str)
+        finally:
+            tempfp.close()
+
 class OutErrCapture: 
     """ capture Stdout and Stderr both on filedescriptor 
         and sys.stdout/stderr level. 
@@ -51,6 +72,11 @@ class OutErrCapture:
                 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() 
@@ -60,6 +86,20 @@ class OutErrCapture:
             err = errfile.read()
         return out, err 
 
+    def writeorgout(self, str):
+        """ write something to the original stdout
+        """
+        if not hasattr(self, 'out'):
+            raise IOError('stdout not patched')
+        self.out.writeorg(str)
+
+    def writeorgerr(self, str):
+        """ write something to the original stderr
+        """
+        if not hasattr(self, 'err'):
+            raise IOError('stderr not patched')
+        self.err.writeorg(str)
+
 def callcapture(func, *args, **kwargs): 
     """ call the given function with args/kwargs
         and return a (res, out, err) tuple where

py/io/test/test_capture.py

@@ -30,6 +30,33 @@ class TestFDCapture:
         f = cap.done()
         assert x == "3"
 
+    def test_writeorg(self):
+        tmppath = py.test.ensuretemp('test_writeorg').ensure('stderr',
+                                                             file=True)
+        tmpfp = tmppath.open('w+b')
+        try:
+            cap = py.io.FDCapture(tmpfp.fileno())
+            print >>tmpfp, 'foo'
+            cap.writeorg('bar\n')
+        finally:
+            tmpfp.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()
+
 class TestCapturing: 
     def getcapture(self): 
         return py.io.OutErrCapture()