# Test case for the os.poll() function

import os, select, random, unittest

from test.test_support import TestSkipped, TESTFN, run_unittest

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

class PollTests(unittest.TestCase):

    def test_poll1(self):

        # Basic functional test of poll object

        # Create a bunch of pipe and test that poll works with them.

        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)

            p.modify(rd, select.POLLIN)

            p.register(wr, select.POLLOUT)

            readers.append(rd)

            writers.append(wr)

            r2w[rd] = wr

            w2r[wr] = rd

        bufs = []

        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)

            self.assertEqual(len(buf), MSG_LEN)

            bufs.append(buf)

            os.close(r2w[rd]) ; os.close( rd )

            p.unregister( r2w[rd] )

            p.unregister( rd )

            writers.remove(r2w[rd])

        self.assertEqual(bufs, [MSG] * NUM_PIPES)

    def poll_unit_tests(self):

        # returns NVAL for invalid file descriptor

        FD = 42

        try:

            os.close(FD)

        except OSError:

            pass

        p = select.poll()

        p.register(FD)

        r = p.poll()

        self.assertEqual(r[0], (FD, select.POLLNVAL))

        f = open(TESTFN, 'w')

        fd = f.fileno()

        p = select.poll()

        p.register(f)

        r = p.poll()

        self.assertEqual(r[0][0], fd)

        f.close()

        r = p.poll()

        self.assertEqual(r[0], (fd, select.POLLNVAL))

        os.unlink(TESTFN)

        # type error for invalid arguments

        p = select.poll()

        self.assertRaises(TypeError, p.register, p)

        self.assertRaises(TypeError, p.unregister, p)

        # can't unregister non-existent object

        p = select.poll()

        self.assertRaises(KeyError, p.unregister, 3)

        # Test error cases

        pollster = select.poll()

        class Nope:

            pass

        class Almost:

            def fileno(self):

                return 'fileno'

        self.assertRaises(TypeError, pollster.register, Nope(), 0)

        self.assertRaises(TypeError, pollster.register, Almost(), 0)

    # Another test case for poll().  This is copied from the test case for

    # select(), modified to use poll() instead.

    def test_poll2(self):

        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:

            fdlist = pollster.poll(tout)

            if (fdlist == []):

                continue

            fd, flags = fdlist[0]

            if flags & select.POLLHUP:

                line = p.readline()

                if line != "":

                    self.fail('error: pipe seems to be closed, but still returns data')

                continue

            elif flags & select.POLLIN:

                line = p.readline()

                if not line:

                    break

                continue

            else:

                self.fail('Unexpected return value from select.poll: %s' % fdlist)

        p.close()

    def test_poll3(self):

        # test int overflow

        pollster = select.poll()

        pollster.register(1)

        self.assertRaises(OverflowError, pollster.poll, 1L << 64)

        x = 2 + 3

        if x != 5:

            self.fail('Overflow must have occurred')

def test_main():

    run_unittest(PollTests)

if __name__ == '__main__':

    test_main()