sources for test_io.py [rev. 38799]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import py
import sys
WorkerPool = py._thread.WorkerPool
ThreadOut = py._thread.ThreadOut
def test_threadout_install_deinstall():
    old = sys.stdout
    out = ThreadOut(sys, 'stdout')
    out.deinstall()
    assert old == sys.stdout
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):
        l = []
        self.out.setwritefunc(l.append)
        print 42,13,
        x = l.pop(0)
        assert x == '42'
        x = l.pop(0)
        assert x == ' '
        x = l.pop(0)
        assert x == '13'
    def test_threadout_multi_and_default(self):
        num = 3
        defaults = []
        def f(l):
            self.out.setwritefunc(l.append)
            print id(l),
            self.out.delwritefunc()
            print 1
        self.out.setdefaultwriter(defaults.append)
        pool = WorkerPool()
        listlist = []
        for x in range(num):
            l = []
            listlist.append(l)
            pool.dispatch(f, l)
        pool.shutdown()
        for name, value in self.out.__dict__.items():
            print >>sys.stderr, "%s: %s" %(name, value)
        pool.join(2.0)
        for i in range(num):
            item = listlist[i]
            assert item ==[str(id(item))]
        assert not self.out._tid2out
        assert defaults
        expect = ['1' for x in range(num)]
        defaults = [x for x in defaults if x.strip()]
        assert defaults == expect
    def test_threadout_nested(self):
        # we want ThreadOuts to coexist
        last = sys.stdout
        out = ThreadOut(sys, 'stdout')
        assert last == sys.stdout
        out.deinstall()
        assert last == sys.stdout