"""
Tests for kqueue wrapper.
"""
import errno
import os
import select
import socket
from test import support
import time
import unittest

from test.support import warnings_helper

if not hasattr(select, "kqueue"):
    raise unittest.SkipTest("test works only on BSD")

class TestKQueue(unittest.TestCase):
    def test_create_queue(self):
        kq = select.kqueue()
        self.assertTrue(kq.fileno() > 0, kq.fileno())
        self.assertTrue(not kq.closed)
        kq.close()
        self.assertTrue(kq.closed)
        self.assertRaises(ValueError, kq.fileno)

    def test_control_overflowing_timeout(self):
        # gh-154836: out-of-range timeouts must raise OverflowError,
        # not a (misleading) TypeError, like select(), poll() and
        # epoll() do.
        kq = select.kqueue()
        self.addCleanup(kq.close)
        for timeout in (1e300, float('inf'), 2**200):
            with self.subTest(timeout=timeout):
                with self.assertRaises(OverflowError):
                    kq.control(None, 0, timeout)
        # Non-numbers still raise TypeError.
        with self.assertRaises(TypeError):
            kq.control(None, 0, "0.1")

    def test_create_event(self):
        from operator import lt, le, gt, ge

        fd = os.open(os.devnull, os.O_WRONLY)
        self.addCleanup(os.close, fd)

        ev = select.kevent(fd)
        other = select.kevent(1000)
        self.assertEqual(ev.ident, fd)
        self.assertEqual(ev.filter, select.KQ_FILTER_READ)
        self.assertEqual(ev.flags, select.KQ_EV_ADD)
        self.assertEqual(ev.fflags, 0)
        self.assertEqual(ev.data, 0)
        self.assertEqual(ev.udata, 0)
        self.assertEqual(ev, ev)
        self.assertNotEqual(ev, other)
        self.assertTrue(ev < other)
        self.assertTrue(other >= ev)
        for op in lt, le, gt, ge:
            self.assertRaises(TypeError, op, ev, None)
            self.assertRaises(TypeError, op, ev, 1)
            self.assertRaises(TypeError, op, ev, "ev")

        ev = select.kevent(fd, select.KQ_FILTER_WRITE)
        self.assertEqual(ev.ident, fd)
        self.assertEqual(ev.filter, select.KQ_FILTER_WRITE)
        self.assertEqual(ev.flags, select.KQ_EV_ADD)
        self.assertEqual(ev.fflags, 0)
        self.assertEqual(ev.data, 0)
        self.assertEqual(ev.udata, 0)
        self.assertEqual(ev, ev)
        self.assertNotEqual(ev, other)

        ev = select.kevent(fd, select.KQ_FILTER_WRITE, select.KQ_EV_ONESHOT)
        self.assertEqual(ev.ident, fd)
        self.assertEqual(ev.filter, select.KQ_FILTER_WRITE)
        self.assertEqual(ev.flags, select.KQ_EV_ONESHOT)
        self.assertEqual(ev.fflags, 0)
        self.assertEqual(ev.data, 0)
        self.assertEqual(ev.udata, 0)
        self.assertEqual(ev, ev)
        self.assertNotEqual(ev, other)

        ev = select.kevent(1, 2, 3, 4, 5, 6)
        self.assertEqual(ev.ident, 1)
        self.assertEqual(ev.filter, 2)
        self.assertEqual(ev.flags, 3)
        self.assertEqual(ev.fflags, 4)
        self.assertEqual(ev.data, 5)
        self.assertEqual(ev.udata, 6)
        self.assertEqual(ev, ev)
        self.assertNotEqual(ev, other)

        bignum = 0x7fff
        ev = select.kevent(bignum, 1, 2, 3, bignum - 1, bignum)
        self.assertEqual(ev.ident, bignum)
        self.assertEqual(ev.filter, 1)
        self.assertEqual(ev.flags, 2)
        self.assertEqual(ev.fflags, 3)
        self.assertEqual(ev.data, bignum - 1)
        self.assertEqual(ev.udata, bignum)
        self.assertEqual(ev, ev)
        self.assertNotEqual(ev, other)

        # Issue 11973
        bignum = 0xffff
        ev = select.kevent(0, 1, bignum)
        self.assertEqual(ev.ident, 0)
        self.assertEqual(ev.filter, 1)
        self.assertEqual(ev.flags, bignum)
        self.assertEqual(ev.fflags, 0)
        self.assertEqual(ev.data, 0)
        self.assertEqual(ev.udata, 0)
        self.assertEqual(ev, ev)
        self.assertNotEqual(ev, other)

        # Issue 11973
        bignum = 0xffffffff
        ev = select.kevent(0, 1, 2, bignum)
        self.assertEqual(ev.ident, 0)
        self.assertEqual(ev.filter, 1)
        self.assertEqual(ev.flags, 2)
        self.assertEqual(ev.fflags, bignum)
        self.assertEqual(ev.data, 0)
        self.assertEqual(ev.udata, 0)
        self.assertEqual(ev, ev)
        self.assertNotEqual(ev, other)


    def test_event_attributes(self):
        fd = os.open(os.devnull, os.O_WRONLY)
        self.addCleanup(os.close, fd)

        ev = select.kevent(fd)
        # All attributes are numeric members: they can be set and cannot be
        # deleted.
        for name, value in (('ident', 1), ('filter', select.KQ_FILTER_WRITE),
                            ('flags', select.KQ_EV_DELETE), ('fflags', 2),
                            ('data', 3), ('udata', 4)):
            with self.subTest(name=name):
                setattr(ev, name, value)
                self.assertEqual(getattr(ev, name), value)
                with self.assertRaises(TypeError):
                    setattr(ev, name, 'not a number')
                with self.assertRaises(OverflowError):
                    setattr(ev, name, 2**1000)
                with self.assertRaises(OverflowError):
                    setattr(ev, name, -2**1000)
                with self.assertRaisesRegex(
                        TypeError, "can't delete numeric/char attribute"):
                    delattr(ev, name)
                # a failed assignment does not change the value
                self.assertEqual(getattr(ev, name), value)

    def test_queue_event(self):
        serverSocket = socket.create_server(('127.0.0.1', 0))
        client = socket.socket()
        client.setblocking(False)
        try:
            client.connect(('127.0.0.1', serverSocket.getsockname()[1]))
        except OSError as e:
            self.assertEqual(e.args[0], errno.EINPROGRESS)
        else:
            #raise AssertionError("Connect should have raised EINPROGRESS")
            pass # FreeBSD doesn't raise an exception here
        server, addr = serverSocket.accept()

        kq = select.kqueue()
        kq2 = select.kqueue.fromfd(kq.fileno())

        ev = select.kevent(server.fileno(),
                           select.KQ_FILTER_WRITE,
                           select.KQ_EV_ADD | select.KQ_EV_ENABLE)
        kq.control([ev], 0)
        ev = select.kevent(server.fileno(),
                           select.KQ_FILTER_READ,
                           select.KQ_EV_ADD | select.KQ_EV_ENABLE)
        kq.control([ev], 0)
        ev = select.kevent(client.fileno(),
                           select.KQ_FILTER_WRITE,
                           select.KQ_EV_ADD | select.KQ_EV_ENABLE)
        kq2.control([ev], 0)
        ev = select.kevent(client.fileno(),
                           select.KQ_FILTER_READ,
                           select.KQ_EV_ADD | select.KQ_EV_ENABLE)
        kq2.control([ev], 0)

        events = kq.control(None, 4, 1)
        events = set((e.ident, e.filter) for e in events)
        self.assertEqual(events, set([
            (client.fileno(), select.KQ_FILTER_WRITE),
            (server.fileno(), select.KQ_FILTER_WRITE)]))

        client.send(b"Hello!")
        server.send(b"world!!!")

        # We may need to call it several times
        for i in range(10):
            events = kq.control(None, 4, 1)
            if len(events) == 4:
                break
            time.sleep(1.0)
        else:
            self.fail('timeout waiting for event notifications')

        events = set((e.ident, e.filter) for e in events)
        self.assertEqual(events, set([
            (client.fileno(), select.KQ_FILTER_WRITE),
            (client.fileno(), select.KQ_FILTER_READ),
            (server.fileno(), select.KQ_FILTER_WRITE),
            (server.fileno(), select.KQ_FILTER_READ)]))

        # Remove completely client, and server read part
        ev = select.kevent(client.fileno(),
                           select.KQ_FILTER_WRITE,
                           select.KQ_EV_DELETE)
        kq.control([ev], 0)
        ev = select.kevent(client.fileno(),
                           select.KQ_FILTER_READ,
                           select.KQ_EV_DELETE)
        kq.control([ev], 0)
        ev = select.kevent(server.fileno(),
                           select.KQ_FILTER_READ,
                           select.KQ_EV_DELETE)
        kq.control([ev], 0, 0)

        events = kq.control([], 4, 0.99)
        events = set((e.ident, e.filter) for e in events)
        self.assertEqual(events, set([
            (server.fileno(), select.KQ_FILTER_WRITE)]))

        client.close()
        server.close()
        serverSocket.close()

    def testPair(self):
        kq = select.kqueue()
        a, b = socket.socketpair()

        a.send(b'foo')
        event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
        event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
        r = kq.control([event1, event2], 1, 1)
        self.assertTrue(r)
        self.assertFalse(r[0].flags & select.KQ_EV_ERROR)
        self.assertEqual(b.recv(r[0].data), b'foo')

        a.close()
        b.close()
        kq.close()

    def test_issue30058(self):
        # changelist must be an iterable
        kq = select.kqueue()
        a, b = socket.socketpair()
        ev = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)

        kq.control([ev], 0)
        # not a list
        kq.control((ev,), 0)
        # __len__ is not consistent with __iter__
        class BadList:
            def __len__(self):
                return 0
            def __iter__(self):
                for i in range(100):
                    yield ev
        kq.control(BadList(), 0)
        # doesn't have __len__
        kq.control(iter([ev]), 0)

        a.close()
        b.close()
        kq.close()

    def test_close(self):
        open_file = open(__file__, "rb")
        self.addCleanup(open_file.close)
        fd = open_file.fileno()
        kqueue = select.kqueue()

        # test fileno() method and closed attribute
        self.assertIsInstance(kqueue.fileno(), int)
        self.assertFalse(kqueue.closed)

        # test close()
        kqueue.close()
        self.assertTrue(kqueue.closed)
        self.assertRaises(ValueError, kqueue.fileno)

        # close() can be called more than once
        kqueue.close()

        # operations must fail with ValueError("I/O operation on closed ...")
        self.assertRaises(ValueError, kqueue.control, None, 4)

    def test_close_error(self):
        # gh-146205: close() should raise OSError if underlying fd is invalid
        kqueue = select.kqueue()
        fd = kqueue.fileno()
        os.close(fd)
        with self.assertRaises(OSError) as cm:
            kqueue.close()
        self.assertEqual(cm.exception.errno, errno.EBADF)

    def test_fd_non_inheritable(self):
        kqueue = select.kqueue()
        self.addCleanup(kqueue.close)
        self.assertEqual(os.get_inheritable(kqueue.fileno()), False)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    @support.requires_fork()
    def test_fork(self):
        # gh-110395: kqueue objects must be closed after fork
        kqueue = select.kqueue()
        if (pid := os.fork()) == 0:
            try:
                self.assertTrue(kqueue.closed)
                with self.assertRaisesRegex(ValueError, "closed kqueue"):
                    kqueue.fileno()
            except:
                os._exit(1)
            finally:
                os._exit(0)
        else:
            support.wait_process(pid, exitcode=0)
            self.assertFalse(kqueue.closed)  # child done, we're still open.


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