import asyncio
import os
import socket
import sys
import tempfile
import unittest
import uuid

from uvloop import _testbase as tb


class MyDatagramProto(asyncio.DatagramProtocol):
    done = None

    def __init__(self, loop=None):
        self.state = 'INITIAL'
        self.nbytes = 0
        if loop is not None:
            self.done = asyncio.Future(loop=loop)

    def connection_made(self, transport):
        self.transport = transport
        assert self.state == 'INITIAL', self.state
        self.state = 'INITIALIZED'

    def datagram_received(self, data, addr):
        assert self.state == 'INITIALIZED', self.state
        self.nbytes += len(data)

    def error_received(self, exc):
        assert self.state == 'INITIALIZED', self.state
        raise exc

    def connection_lost(self, exc):
        assert self.state == 'INITIALIZED', self.state
        self.state = 'CLOSED'
        if self.done:
            self.done.set_result(None)


class _TestUDP:

    def _test_create_datagram_endpoint_addrs(self, family, lc_addr):
        class TestMyDatagramProto(MyDatagramProto):
            def __init__(inner_self):
                super().__init__(loop=self.loop)

            def datagram_received(self, data, addr):
                super().datagram_received(data, addr)
                self.transport.sendto(b'resp:' + data, addr)

        coro = self.loop.create_datagram_endpoint(
            TestMyDatagramProto,
            local_addr=lc_addr,
            family=family)

        s_transport, server = self.loop.run_until_complete(coro)

        remote_addr = s_transport.get_extra_info('sockname')
        host, port, *_ = remote_addr

        self.assertIsInstance(server, TestMyDatagramProto)
        self.assertEqual('INITIALIZED', server.state)
        self.assertIs(server.transport, s_transport)

        extra = {}
        if hasattr(socket, 'SO_REUSEPORT'):
            extra['reuse_port'] = True

        coro = self.loop.create_datagram_endpoint(
            lambda: MyDatagramProto(loop=self.loop),
            family=family,
            remote_addr=(host, port),
            **extra)
        transport, client = self.loop.run_until_complete(coro)

        self.assertIsInstance(client, MyDatagramProto)
        self.assertEqual('INITIALIZED', client.state)
        self.assertIs(client.transport, transport)

        transport.sendto(b'xxx')
        tb.run_until(self.loop, lambda: server.nbytes)
        self.assertEqual(3, server.nbytes)
        tb.run_until(self.loop, lambda: client.nbytes)

        # received
        self.assertEqual(8, client.nbytes)

        # https://github.com/MagicStack/uvloop/issues/319
        # uvloop should behave the same as asyncio when given remote_addr
        transport.sendto(b'xxx', remote_addr)
        tb.run_until(
            self.loop, lambda: server.nbytes > 3 or client.done.done())
        self.assertEqual(6, server.nbytes)
        tb.run_until(self.loop, lambda: client.nbytes > 8)

        # received
        self.assertEqual(16, client.nbytes)

        # reject sendto with a different port
        with self.assertRaisesRegex(
            ValueError, "Invalid address.*" + repr(remote_addr)
        ):
            bad_addr = list(remote_addr)
            bad_addr[1] += 1
            bad_addr = tuple(bad_addr)
            transport.sendto(b"xxx", bad_addr)

        # reject sento with unresolved hostname
        if remote_addr[0] != lc_addr[0]:
            with self.assertRaisesRegex(
                ValueError, "Invalid address.*" + repr(remote_addr)
            ):
                bad_addr = list(remote_addr)
                bad_addr[0] = lc_addr[0]
                bad_addr = tuple(bad_addr)
                transport.sendto(b"xxx", bad_addr)

        # extra info is available
        self.assertIsNotNone(transport.get_extra_info('sockname'))

        # close connection
        transport.close()
        self.loop.run_until_complete(client.done)
        self.assertEqual('CLOSED', client.state)
        server.transport.close()
        self.loop.run_until_complete(server.done)

    def test_create_datagram_endpoint_addrs_ipv4(self):
        self._test_create_datagram_endpoint_addrs(
            socket.AF_INET, ('127.0.0.1', 0))

    def test_create_datagram_endpoint_addrs_ipv4_nameaddr(self):
        self._test_create_datagram_endpoint_addrs(
            socket.AF_INET, ('localhost', 0))

    def _test_create_datagram_endpoint_addrs_ipv6(self):
        self._test_create_datagram_endpoint_addrs(
            socket.AF_INET6, ('::1', 0))

    def test_create_datagram_endpoint_ipv6_family(self):
        class TestMyDatagramProto(MyDatagramProto):
            def __init__(inner_self):
                super().__init__(loop=self.loop)

            def datagram_received(self, data, addr):
                super().datagram_received(data, addr)
                self.transport.sendto(b'resp:' + data, addr)

        coro = self.loop.create_datagram_endpoint(
            TestMyDatagramProto, local_addr=None, family=socket.AF_INET6)
        s_transport = None
        try:
            s_transport, server = self.loop.run_until_complete(coro)
        finally:
            if s_transport:
                s_transport.close()
                # let it close
                self.loop.run_until_complete(
                    asyncio.sleep(0.1))

    def test_create_datagram_endpoint_sock(self):
        sock = None
        local_address = ('127.0.0.1', 0)
        infos = self.loop.run_until_complete(
            self.loop.getaddrinfo(
                *local_address, type=socket.SOCK_DGRAM))
        for family, type, proto, cname, address in infos:
            try:
                sock = socket.socket(family=family, type=type, proto=proto)
                sock.setblocking(False)
                sock.bind(address)
            except Exception:
                pass
            else:
                break
        else:
            assert False, 'Can not create socket.'

        with sock:
            f = self.loop.create_datagram_endpoint(
                lambda: MyDatagramProto(loop=self.loop), sock=sock)
            tr, pr = self.loop.run_until_complete(f)
            self.assertIsInstance(pr, MyDatagramProto)
            tr.close()
            self.loop.run_until_complete(pr.done)

    def test_create_datagram_endpoint_sock_unix_domain(self):

        class Proto(asyncio.DatagramProtocol):
            done = None

            def __init__(self, loop):
                self.state = 'INITIAL'
                self.addrs = set()
                self.done = asyncio.Future(loop=loop)
                self.data = b''

            def connection_made(self, transport):
                self.transport = transport
                assert self.state == 'INITIAL', self.state
                self.state = 'INITIALIZED'

            def datagram_received(self, data, addr):
                assert self.state == 'INITIALIZED', self.state
                self.addrs.add(addr)
                self.data += data
                if self.data == b'STOP' and not self.done.done():
                    self.done.set_result(True)

            def error_received(self, exc):
                assert self.state == 'INITIALIZED', self.state
                if not self.done.done():
                    self.done.set_exception(exc or RuntimeError())

            def connection_lost(self, exc):
                assert self.state == 'INITIALIZED', self.state
                self.state = 'CLOSED'
                if self.done and not self.done.done():
                    self.done.set_result(None)

        tmp_file = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
        sock = socket.socket(socket.AF_UNIX, type=socket.SOCK_DGRAM)
        sock.bind(tmp_file)

        with sock:
            pr = Proto(loop=self.loop)
            f = self.loop.create_datagram_endpoint(
                lambda: pr, sock=sock)
            tr, pr_prime = self.loop.run_until_complete(f)
            self.assertIs(pr, pr_prime)

            tmp_file2 = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
            sock2 = socket.socket(socket.AF_UNIX, type=socket.SOCK_DGRAM)
            sock2.bind(tmp_file2)

            with sock2:
                f2 = self.loop.create_datagram_endpoint(
                    asyncio.DatagramProtocol, sock=sock2)
                tr2, pr2 = self.loop.run_until_complete(f2)

                tr2.sendto(b'STOP', tmp_file)

                self.loop.run_until_complete(pr.done)

                tr.close()
                tr2.close()

                # Let transports close
                self.loop.run_until_complete(asyncio.sleep(0.2))

                self.assertIn(tmp_file2, pr.addrs)

    def test_create_datagram_1(self):
        server_addr = ('127.0.0.1', 8888)
        client_addr = ('127.0.0.1', 0)

        async def run():
            server_transport, client_protocol = \
                await self.loop.create_datagram_endpoint(
                    asyncio.DatagramProtocol,
                    local_addr=server_addr)

            client_transport, client_conn = \
                await self.loop.create_datagram_endpoint(
                    asyncio.DatagramProtocol,
                    remote_addr=server_addr,
                    local_addr=client_addr)

            client_transport.close()
            server_transport.close()

            # let transports close
            await asyncio.sleep(0.1)

        self.loop.run_until_complete(run())

    def test_socketpair(self):
        peername = asyncio.Future(loop=self.loop)

        class Proto(MyDatagramProto):
            def datagram_received(self, data, addr):
                super().datagram_received(data, addr)
                peername.set_result(addr)

        s1, s2 = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM, 0)

        with s1, s2:
            f = self.loop.create_datagram_endpoint(
                lambda: Proto(loop=self.loop), sock=s1)
            tr, pr = self.loop.run_until_complete(f)
            self.assertIsInstance(pr, Proto)

            s2.send(b'hello, socketpair')
            addr = self.loop.run_until_complete(
                asyncio.wait_for(peername, 1))
            if sys.platform.startswith('linux'):
                self.assertEqual(addr, None)
            else:
                self.assertEqual(addr, '')
            self.assertEqual(pr.nbytes, 17)

            if not self.is_asyncio_loop():
                # asyncio doesn't support sendto(xx) on UDP sockets
                # https://git.io/Jfqbw
                data = b'from uvloop'
                tr.sendto(data)
                result = self.loop.run_until_complete(asyncio.wait_for(
                    self.loop.run_in_executor(None, s2.recv, 1024),
                    1))
                self.assertEqual(data, result)

            tr.close()
            self.loop.run_until_complete(pr.done)

    def _skip_create_datagram_endpoint_reuse_addr(self):
        if self.implementation == 'asyncio':
            if sys.version_info[:2] >= (3, 11):
                raise unittest.SkipTest()
            if (3, 8, 0) <= sys.version_info < (3, 8, 1):
                raise unittest.SkipTest()
            if (3, 7, 0) <= sys.version_info < (3, 7, 6):
                raise unittest.SkipTest()

    def test_create_datagram_endpoint_reuse_address_error(self):
        # bpo-37228: Ensure that explicit passing of `reuse_address=True`
        # raises an error, as it is not safe to use SO_REUSEADDR when using UDP

        self._skip_create_datagram_endpoint_reuse_addr()

        coro = self.loop.create_datagram_endpoint(
            lambda: MyDatagramProto(loop=self.loop),
            local_addr=('127.0.0.1', 0),
            reuse_address=True)

        with self.assertRaises(ValueError):
            self.loop.run_until_complete(coro)

    def test_create_datagram_endpoint_reuse_address_warning(self):
        # bpo-37228: Deprecate *reuse_address* parameter

        self._skip_create_datagram_endpoint_reuse_addr()

        coro = self.loop.create_datagram_endpoint(
            lambda: MyDatagramProto(loop=self.loop),
            local_addr=('127.0.0.1', 0),
            reuse_address=False)

        with self.assertWarns(DeprecationWarning):
            tr, pr = self.loop.run_until_complete(coro)

        tr.close()
        self.loop.run_until_complete(pr.done)


class Test_UV_UDP(_TestUDP, tb.UVTestCase):

    def test_create_datagram_endpoint_wrong_sock(self):
        sock = socket.socket(socket.AF_INET)
        with sock:
            coro = self.loop.create_datagram_endpoint(lambda: None, sock=sock)
            with self.assertRaisesRegex(ValueError,
                                        'A UDP Socket was expected'):
                self.loop.run_until_complete(coro)

    def test_udp_sendto_dns(self):
        coro = self.loop.create_datagram_endpoint(
            asyncio.DatagramProtocol,
            local_addr=('127.0.0.1', 0),
            family=socket.AF_INET)

        s_transport, server = self.loop.run_until_complete(coro)

        with self.assertRaisesRegex(ValueError, 'DNS lookup'):
            s_transport.sendto(b'aaaa', ('example.com', 80))

        with self.assertRaisesRegex(ValueError, 'socket family mismatch'):
            s_transport.sendto(b'aaaa', ('::1', 80))

        s_transport.close()
        self.loop.run_until_complete(asyncio.sleep(0.01))

    def test_send_after_close(self):
        coro = self.loop.create_datagram_endpoint(
            asyncio.DatagramProtocol,
            local_addr=('127.0.0.1', 0),
            family=socket.AF_INET)

        s_transport, _ = self.loop.run_until_complete(coro)

        s_transport.close()
        s_transport.sendto(b'aaaa', ('127.0.0.1', 80))
        self.loop.run_until_complete(asyncio.sleep(0.01))
        s_transport.sendto(b'aaaa', ('127.0.0.1', 80))

    @unittest.skipUnless(tb.has_IPv6, 'no IPv6')
    def test_create_datagram_endpoint_addrs_ipv6(self):
        self._test_create_datagram_endpoint_addrs_ipv6()


class Test_AIO_UDP(_TestUDP, tb.AIOTestCase):
    @unittest.skipUnless(tb.has_IPv6, 'no IPv6')
    @unittest.skipIf(
        sys.version_info[:3] == (3, 7, 0),
        'bpo-27500: bug fixed in Python 3.7.1 and above.',
    )
    def test_create_datagram_endpoint_addrs_ipv6(self):
        self._test_create_datagram_endpoint_addrs_ipv6()