test case for the os poll function basic functional test of poll object create a bunch of pipe and test that poll works with them returns nval for invalid file descriptor type error for invalid arguments can t unregister nonexistent object test error cases another test case for poll this is copied from the test case for select modified to use poll instead test int overflow issues 15989 17919 issues 15989 17919 trigger ufds array reallocation and make the call to poll from the thread return create two file descriptors this will be used to unlock the blocking call to poll poll inside the thread write to the pipe so pollster poll unblocks and the thread ends test case for the os poll function basic functional test of poll object create a bunch of pipe and test that poll works with them returns nval for invalid file descriptor type error for invalid arguments can t unregister non existent object test error cases another test case for poll this is copied from the test case for select modified to use poll instead test int overflow issues 15989 17919 issues 15989 17919 trigger ufds array reallocation and make the call to poll from the thread return create two file descriptors this will be used to unlock the blocking call to poll poll inside the thread write to the pipe so pollster poll unblocks and the thread ends
import os
import subprocess
import random
import select
import threading
import time
import unittest
from test.support import (
    cpython_only, requires_subprocess, requires_working_socket, requires_resource
)
from test.support import threading_helper
from test.support.os_helper import TESTFN


try:
    select.poll
except AttributeError:
    raise unittest.SkipTest("select.poll not defined")

requires_working_socket(module=True)

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):
        
        

        p = select.poll()

        NUM_PIPES = 12
        MSG = b" 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 test_poll_unit_tests(self):
        
        FD, w = os.pipe()
        os.close(FD)
        os.close(w)
        p = select.poll()
        p.register(FD)
        r = p.poll()
        self.assertEqual(r[0], (FD, select.POLLNVAL))

        with open(TESTFN, 'w') as f:
            fd = f.fileno()
            p = select.poll()
            p.register(f)
            r = p.poll()
            self.assertEqual(r[0][0], fd)
        r = p.poll()
        self.assertEqual(r[0], (fd, select.POLLNVAL))
        os.unlink(TESTFN)

        
        p = select.poll()
        self.assertRaises(TypeError, p.register, p)
        self.assertRaises(TypeError, p.unregister, p)

        
        p = select.poll()
        self.assertRaises(KeyError, p.unregister, 3)

        
        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)

    
    

    @requires_subprocess()
    @requires_resource('walltime')
    def test_poll2(self):
        cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'
        proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
                                bufsize=0)
        self.enterContext(proc)
        p = proc.stdout
        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 != b"":
                    self.fail('error: pipe seems to be closed, but still returns data')
                continue

            elif flags & select.POLLIN:
                line = p.readline()
                if not line:
                    break
                self.assertEqual(line, b'testing...\n')
                continue
            else:
                self.fail('Unexpected return value from select.poll: %s' % fdlist)

    def test_poll3(self):
        
        pollster = select.poll()
        pollster.register(1)

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

        x = 2 + 3
        if x != 5:
            self.fail('Overflow must have occurred')

        
        self.assertRaises(ValueError, pollster.register, 0, -1)
        self.assertRaises(OverflowError, pollster.register, 0, 1 << 64)
        self.assertRaises(ValueError, pollster.modify, 1, -1)
        self.assertRaises(OverflowError, pollster.modify, 1, 1 << 64)

    @cpython_only
    def test_poll_c_limits(self):
        from _testcapi import USHRT_MAX, INT_MAX, UINT_MAX
        pollster = select.poll()
        pollster.register(1)

        
        self.assertRaises(OverflowError, pollster.register, 0, USHRT_MAX + 1)
        self.assertRaises(OverflowError, pollster.modify, 1, USHRT_MAX + 1)
        self.assertRaises(OverflowError, pollster.poll, INT_MAX + 1)
        self.assertRaises(OverflowError, pollster.poll, UINT_MAX + 1)

    @threading_helper.reap_threads
    def test_threaded_poll(self):
        r, w = os.pipe()
        self.addCleanup(os.close, r)
        self.addCleanup(os.close, w)
        rfds = []
        for i in range(10):
            fd = os.dup(r)
            self.addCleanup(os.close, fd)
            rfds.append(fd)
        pollster = select.poll()
        for fd in rfds:
            pollster.register(fd, select.POLLIN)

        t = threading.Thread(target=pollster.poll)
        t.start()
        try:
            time.sleep(0.5)
            
            for fd in rfds:
                pollster.unregister(fd)
            pollster.register(w, select.POLLOUT)
            self.assertRaises(RuntimeError, pollster.poll)
        finally:
            
            os.write(w, b'spam')
            t.join()

    @unittest.skipUnless(threading, 'Threading required for this test.')
    @threading_helper.reap_threads
    def test_poll_blocks_with_negative_ms(self):
        for timeout_ms in [None, -1000, -1, -1.0, -0.1, -1e-100]:
            
            
            r, w = os.pipe()
            pollster = select.poll()
            pollster.register(r, select.POLLIN)

            poll_thread = threading.Thread(target=pollster.poll, args=(timeout_ms,))
            poll_thread.start()
            poll_thread.join(timeout=0.1)
            self.assertTrue(poll_thread.is_alive())

            
            os.write(w, b'spam')
            poll_thread.join()
            self.assertFalse(poll_thread.is_alive())
            os.close(r)
            os.close(w)


if __name__ == '__main__':
    unittest.main()