test_poll.py 4.25 KB
Newer Older
1
# Test case for the os.poll() function
2

3
import sys, os, select, random
4
from test_support import verify, verbose, TestSkipped, TESTFN
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

try:
    select.poll
except AttributeError:
    raise TestSkipped, "select.poll not defined -- skipping test_poll"


def find_ready_matching(ready, flag):
    match = []
    for fd, mode in ready:
        if mode & flag:
            match.append(fd)
    return match

def test_poll1():
    """Basic functional test of poll object

    Create a bunch of pipe and test that poll works with them.
    """
    print 'Running poll test 1'
    p = select.poll()

    NUM_PIPES = 12
    MSG = " This is a test."
    MSG_LEN = len(MSG)
    readers = []
    writers = []
    r2w = {}
    w2r = {}

    for i in range(NUM_PIPES):
        rd, wr = os.pipe()
        p.register(rd, select.POLLIN)
        p.register(wr, select.POLLOUT)
        readers.append(rd)
        writers.append(wr)
        r2w[rd] = wr
        w2r[wr] = rd

    while writers:
        ready = p.poll()
        ready_writers = find_ready_matching(ready, select.POLLOUT)
        if not ready_writers:
            raise RuntimeError, "no pipes ready for writing"
        wr = random.choice(ready_writers)
        os.write(wr, MSG)

        ready = p.poll()
        ready_readers = find_ready_matching(ready, select.POLLIN)
        if not ready_readers:
            raise RuntimeError, "no pipes ready for reading"
        rd = random.choice(ready_readers)
        buf = os.read(rd, MSG_LEN)
58
        verify(len(buf) == MSG_LEN)
59
        print buf
60 61 62
        os.close(r2w[rd]) ; os.close( rd )
        p.unregister( r2w[rd] )
        p.unregister( rd )
63 64 65 66 67 68 69 70 71 72 73 74 75 76 77
        writers.remove(r2w[rd])

    poll_unit_tests()
    print 'Poll test 1 complete'

def poll_unit_tests():
    # returns NVAL for invalid file descriptor
    FD = 42
    try:
        os.close(FD)
    except OSError:
        pass
    p = select.poll()
    p.register(FD)
    r = p.poll()
78
    verify(r[0] == (FD, select.POLLNVAL))
79 80 81 82 83 84

    f = open(TESTFN, 'w')
    fd = f.fileno()
    p = select.poll()
    p.register(f)
    r = p.poll()
85
    verify(r[0][0] == fd)
86 87
    f.close()
    r = p.poll()
88
    verify(r[0] == (fd, select.POLLNVAL))
89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149
    os.unlink(TESTFN)

    # type error for invalid arguments
    p = select.poll()
    try:
        p.register(p)
    except TypeError:
        pass
    else:
        print "Bogus register call did not raise TypeError"
    try:
        p.unregister(p)
    except TypeError:
        pass
    else:
        print "Bogus unregister call did not raise TypeError"

    # can't unregister non-existent object
    p = select.poll()
    try:
        p.unregister(3)
    except KeyError:
        pass
    else:
        print "Bogus unregister call did not raise KeyError"

    # Test error cases
    pollster = select.poll()
    class Nope:
        pass

    class Almost:
        def fileno(self):
            return 'fileno'

    try:
        pollster.register( Nope(), 0 )
    except TypeError: pass
    else: print 'expected TypeError exception, not raised'

    try:
        pollster.register( Almost(), 0 )
    except TypeError: pass
    else: print 'expected TypeError exception, not raised'


# Another test case for poll().  This is copied from the test case for
# select(), modified to use poll() instead.

def test_poll2():
    print 'Running poll test 2'
    cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'
    p = os.popen(cmd, 'r')
    pollster = select.poll()
    pollster.register( p, select.POLLIN )
    for tout in (0, 1000, 2000, 4000, 8000, 16000) + (-1,)*10:
        if verbose:
            print 'timeout =', tout
        fdlist = pollster.poll(tout)
        if (fdlist == []):
            continue
150 151
        fd, flags = fdlist[0]
        if flags & select.POLLHUP:
152 153 154 155 156
            line = p.readline()
            if line != "":
                print 'error: pipe seems to be closed, but still returns data'
            continue

157
        elif flags & select.POLLIN:
158 159 160 161 162 163 164 165 166 167 168 169 170 171 172
            line = p.readline()
            if verbose:
                print `line`
            if not line:
                if verbose:
                    print 'EOF'
                break
            continue
        else:
            print 'Unexpected return value from select.poll:', fdlist
    p.close()
    print 'Poll test 2 complete'

test_poll1()
test_poll2()