refactor lsof checking and fix an lsof leak in pypy

This commit is contained in:
holger krekel 2014-01-29 11:18:15 +01:00
parent 505a34bb85
commit 4eabfed651
1 changed files with 18 additions and 13 deletions

View File

@ -5,6 +5,7 @@ import os
import sys import sys
import py import py
import pytest import pytest
import contextlib
from _pytest import capture from _pytest import capture
from _pytest.capture import CaptureManager from _pytest.capture import CaptureManager
@ -121,9 +122,10 @@ class TestCaptureManager:
capouter.reset() capouter.reset()
@pytest.mark.xfail("hasattr(sys, 'pypy_version_info')")
@pytest.mark.parametrize("method", ['fd', 'sys']) @pytest.mark.parametrize("method", ['fd', 'sys'])
def test_capturing_unicode(testdir, method): def test_capturing_unicode(testdir, method):
if hasattr(sys, "pypy_version_info") and sys.pypy_version_info < (2,2):
pytest.xfail("does not work on pypy < 2.2")
if sys.version_info >= (3, 0): if sys.version_info >= (3, 0):
obj = "'b\u00f6y'" obj = "'b\u00f6y'"
else: else:
@ -605,12 +607,12 @@ def test_dontreadfrominput():
f.close() # just for completeness f.close() # just for completeness
def pytest_funcarg__tmpfile(request): @pytest.yield_fixture
testdir = request.getfuncargvalue("testdir") def tmpfile(testdir):
f = testdir.makepyfile("").open('wb+') f = testdir.makepyfile("").open('wb+')
request.addfinalizer(f.close) yield f
return f if not f.closed:
f.close()
@needsosdup @needsosdup
def test_dupfile(tmpfile): def test_dupfile(tmpfile):
@ -645,13 +647,14 @@ def test_dupfile_no_mode():
capture.dupfile(tmpfile, raising=True) capture.dupfile(tmpfile, raising=True)
def lsof_check(func): @contextlib.contextmanager
def lsof_check():
pid = os.getpid() pid = os.getpid()
try: try:
out = py.process.cmdexec("lsof -p %d" % pid) out = py.process.cmdexec("lsof -p %d" % pid)
except py.process.cmdexec.Error: except py.process.cmdexec.Error:
pytest.skip("could not run 'lsof'") pytest.skip("could not run 'lsof'")
func() yield
out2 = py.process.cmdexec("lsof -p %d" % pid) out2 = py.process.cmdexec("lsof -p %d" % pid)
len1 = len([x for x in out.split("\n") if "REG" in x]) 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]) len2 = len([x for x in out2.split("\n") if "REG" in x])
@ -668,6 +671,7 @@ class TestFDCapture:
os.write(fd, data) os.write(fd, data)
f = cap.done() f = cap.done()
s = f.read() s = f.read()
f.close()
assert not s assert not s
cap = capture.FDCapture(fd) cap = capture.FDCapture(fd)
cap.start() cap.start()
@ -675,13 +679,16 @@ class TestFDCapture:
f = cap.done() f = cap.done()
s = f.read() s = f.read()
assert s == "hello" assert s == "hello"
f.close()
def test_simple_many(self, tmpfile): def test_simple_many(self, tmpfile):
for i in range(10): for i in range(10):
self.test_simple(tmpfile) self.test_simple(tmpfile)
def test_simple_many_check_open_files(self, tmpfile): def test_simple_many_check_open_files(self, testdir):
lsof_check(lambda: self.test_simple_many(tmpfile)) with lsof_check():
with testdir.makepyfile("").open('wb+') as tmpfile:
self.test_simple_many(tmpfile)
def test_simple_fail_second_start(self, tmpfile): def test_simple_fail_second_start(self, tmpfile):
fd = tmpfile.fileno() fd = tmpfile.fileno()
@ -888,12 +895,10 @@ class TestStdCaptureFD(TestStdCapture):
assert err == "abc" assert err == "abc"
def test_many(self, capfd): def test_many(self, capfd):
def f(): with lsof_check():
for i in range(10): for i in range(10):
cap = capture.StdCaptureFD() cap = capture.StdCaptureFD()
cap.reset() cap.reset()
lsof_check(f)
@needsosdup @needsosdup