Method test_control_and_wait

Lib/test/test_epoll.py:163–214

(self)

            self.fail("epoll on closed fd didn't raise EBADF")

    def test_control_and_wait(self):
        # create the epoll object
        client, server = self._connected_pair()
        ep = select.epoll(16)
        ep.register(server.fileno(),
                    select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)
        ep.register(client.fileno(),
                    select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)

        # EPOLLOUT
        now = time.monotonic()
        events = ep.poll(1, 4)
        then = time.monotonic()
        self.assertFalse(then - now > 0.1, then - now)

        expected = [(client.fileno(), select.EPOLLOUT),
                    (server.fileno(), select.EPOLLOUT)]
        self.assertEqual(sorted(events), sorted(expected))

        # no event
        events = ep.poll(timeout=0.1, maxevents=4)
        self.assertFalse(events)

        # send: EPOLLIN and EPOLLOUT
        client.sendall(b"Hello!")
        server.sendall(b"world!!!")

        # we might receive events one at a time, necessitating multiple calls to
        # poll
        events = []
        for _ in support.busy_retry(support.SHORT_TIMEOUT):
            now = time.monotonic()
            events += ep.poll(1.0, 4)
            then = time.monotonic()
            self.assertFalse(then - now > 0.01)
            if len(events) >= 2:
                break

        expected = [(client.fileno(), select.EPOLLIN | select.EPOLLOUT),
                    (server.fileno(), select.EPOLLIN | select.EPOLLOUT)]
        self.assertEqual(sorted(events), sorted(expected))

        # unregister, modify
        ep.unregister(client.fileno())
        ep.modify(server.fileno(), select.EPOLLOUT)
        now = time.monotonic()
        events = ep.poll(1, 4)
        then = time.monotonic()
        self.assertFalse(then - now > 0.01)

        expected = [(server.fileno(), select.EPOLLOUT)]
        self.assertEqual(events, expected)

    def test_errors(self):
        self.assertRaises(ValueError, select.epoll, -2)