floraachy
/
pytest2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix threading io-redirection tests 

--HG--
branch : trunk
This commit is contained in:
holger krekel 2009-06-09 17:28:50 +02:00
parent e61e7b1e26
commit d99f335918
1 changed files with 53 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
py/thread/testing/test_io.py
View File

@ -12,14 +12,11 @@ def test_threadout_install_deinstall():
assert old == sys.stdout assert old == sys.stdout
class TestThreadOut: class TestThreadOut:
def setup_method(self, method):
self.out = ThreadOut(sys, 'stdout')
def teardown_method(self, method):
self.out.deinstall()
def test_threadout_one(self): def test_threadout_one(self):
out = ThreadOut(sys, 'stdout')
try:
l = [] l = []
self.out.setwritefunc(l.append) out.setwritefunc(l.append)
print 42,13, print 42,13,
x = l.pop(0) x = l.pop(0)
assert x == '42' assert x == '42'
@ -27,17 +24,20 @@ class TestThreadOut:
assert x == ' ' assert x == ' '
x = l.pop(0) x = l.pop(0)
assert x == '13' assert x == '13'
finally:
out.deinstall()
def test_threadout_multi_and_default(self): def test_threadout_multi_and_default(self):
out = ThreadOut(sys, 'stdout')
try:
num = 3 num = 3
defaults = [] defaults = []
def f(l): def f(l):
self.out.setwritefunc(l.append) out.setwritefunc(l.append)
print id(l), print id(l),
self.out.delwritefunc() out.delwritefunc()
print 1 print 1
self.out.setdefaultwriter(defaults.append) out.setdefaultwriter(defaults.append)
pool = WorkerPool() pool = WorkerPool()
listlist = [] listlist = []
for x in range(num): for x in range(num):
@ -45,22 +45,28 @@ class TestThreadOut:
listlist.append(l) listlist.append(l)
pool.dispatch(f, l) pool.dispatch(f, l)
pool.shutdown() pool.shutdown()
for name, value in self.out.__dict__.items(): for name, value in out.__dict__.items():
print >>sys.stderr, "%s: %s" %(name, value) print >>sys.stderr, "%s: %s" %(name, value)
pool.join(2.0) pool.join(2.0)
for i in range(num): for i in range(num):
item = listlist[i] item = listlist[i]
assert item ==[str(id(item))] assert item ==[str(id(item))]
assert not self.out._tid2out assert not out._tid2out
assert defaults assert defaults
expect = ['1' for x in range(num)] expect = ['1' for x in range(num)]
defaults = [x for x in defaults if x.strip()] defaults = [x for x in defaults if x.strip()]
assert defaults == expect assert defaults == expect
finally:
out.deinstall()
def test_threadout_nested(self): def test_threadout_nested(self):
out1 = ThreadOut(sys, 'stdout')
try:
# we want ThreadOuts to coexist # we want ThreadOuts to coexist
last = sys.stdout last = sys.stdout
out = ThreadOut(sys, 'stdout') out = ThreadOut(sys, 'stdout')
assert last == sys.stdout assert last == sys.stdout
out.deinstall() out.deinstall()
assert last == sys.stdout assert last == sys.stdout
finally:
out1.deinstall()