1. Ronny Pfannschmidt
  2. pylib-hacks

Commits

holger krekel  committed f5f3e68

fix leakage of file descriptors

  • Participants
  • Parent commits 609cbbc
  • Branches trunk

Comments (0)

Files changed (4)

File py/__init__.py

View file
  • Ignore whitespace
 
 (c) Holger Krekel and others, 2004-2010
 """
-__version__ = "2.0.0.dev1"
+__version__ = '2.0.0.dev3'
 
 from py import _apipkg
 

File py/_io/capture.py

View file
  • Ignore whitespace
             if hasattr(self, '_oldsys'):
                 setattr(sys, patchsysdict[self.targetfd], DontReadFromInput())
         else:
-            fd = self.tmpfile.fileno()
             os.dup2(self.tmpfile.fileno(), self.targetfd)
             if hasattr(self, '_oldsys'):
                 setattr(sys, patchsysdict[self.targetfd], self.tmpfile)
 
     def reset(self):
         """ reset sys.stdout/stderr and return captured output as strings. """
-        outfile, errfile = self.done()
+        if hasattr(self, '_reset'):
+            raise ValueError("was already reset")
+        self._reset = True
+        outfile, errfile = self.done(save=False)
         out, err = "", ""
         if outfile and not outfile.closed:
             out = outfile.read()
         """ resume capturing with original temp files. """
         self.startall()
 
-    def done(self):
+    def done(self, save=True):
         """ return (outfile, errfile) and stop capturing. """
         outfile = errfile = None
         if hasattr(self, 'out') and not self.out.tmpfile.closed:
             errfile = self.err.done()
         if hasattr(self, 'in_'):
             tmpfile = self.in_.done()
-        self._save()
+        if save:
+            self._save()
         return outfile, errfile
 
     def readouterr(self):
         if self.in_:
             sys.stdin  = self.in_  = DontReadFromInput()
 
-    def done(self):
+    def done(self, save=True):
         """ return (outfile, errfile) and stop capturing. """
         outfile = errfile = None
         if self.out and not self.out.closed:

File setup.py

View file
  • Ignore whitespace
         long_description = long_description,
         install_requires=['py>=1.3.9', ], # force newer py version which removes 'py' namespace
         #                                  # so we can occupy it
-        version='2.0.0.dev2',
+        version='2.0.0.dev3',
         url='http://pylib.org',
         license='MIT license',
         platforms=['unix', 'linux', 'osx', 'cygwin', 'win32'],
     )
 
 if __name__ == '__main__':
-    main()
+    main()

File testing/io_/test_capture.py

View file
  • Ignore whitespace
     assert "01234" in repr(s)
     tmpfile.close()
 
+def lsof_check(func):
+    pid = os.getpid()
+    try:
+        out = py.process.cmdexec("lsof -p %d" % pid)
+    except py.process.cmdexec.Error:
+        py.test.skip("could not run 'lsof'")
+    func()
+    out2 = py.process.cmdexec("lsof -p %d" % pid)
+    len1 = len([x for x in out.split("\n") if "REG" in x])
+    len2 = len([x for x in out2.split("\n") if "REG" in x])
+    assert len2 < len1 + 3, out2
+
 class TestFDCapture:
     pytestmark = needsdup
 
             self.test_simple(tmpfile)
 
     def test_simple_many_check_open_files(self, tmpfile):
-        pid = os.getpid()
-        try:
-            out = py.process.cmdexec("lsof -p %d" % pid)
-        except py.process.cmdexec.Error:
-            py.test.skip("could not run 'lsof'")
-        self.test_simple_many(tmpfile)
-        out2 = py.process.cmdexec("lsof -p %d" % pid)
-        len1 = len([x for x in out.split("\n") if "REG" in x])
-        len2 = len([x for x in out2.split("\n") if "REG" in x])
-        assert len2 < len1 + 3, out2
+        lsof_check(lambda: self.test_simple_many(tmpfile))
 
     def test_simple_fail_second_start(self, tmpfile):
         fd = tmpfile.fileno()
         assert out.strip() == "hello world."
         assert not err
 
-    def test_capturing_twice_error(self):
+    def test_reset_twice_error(self):
         cap = self.getcapture()
         print ("hello")
         out, err = cap.reset()
-        print ("world")
-        out2, err = cap.reset()
+        py.test.raises(ValueError, cap.reset)
         assert out == "hello\n"
         assert not err
 
         assert out.startswith("3")
         assert err.startswith("4")
 
+    def test_many(self, capfd):
+        def f():
+            for i in range(10):
+                cap = py.io.StdCaptureFD()
+                cap.reset()
+        lsof_check(f)
+
 class TestStdCaptureFDNotNow(TestStdCaptureFD):
     pytestmark = needsdup
 
     capsys.done()
     capsys.done()
     capsys.reset()
-    capsys.reset()
 
 @needsdup
 def test_capture_no_sys():