import sys
import socket

from waitress.compat import (
    PY2,
    WIN,
    )

if sys.version_info[:2] == (2, 6): # pragma: no cover
    import unittest2 as unittest
else: # pragma: no cover
    import unittest

class Test_asbool(unittest.TestCase):

    def _callFUT(self, s):
        from waitress.adjustments import asbool
        return asbool(s)

    def test_s_is_None(self):
        result = self._callFUT(None)
        self.assertEqual(result, False)

    def test_s_is_True(self):
        result = self._callFUT(True)
        self.assertEqual(result, True)

    def test_s_is_False(self):
        result = self._callFUT(False)
        self.assertEqual(result, False)

    def test_s_is_true(self):
        result = self._callFUT('True')
        self.assertEqual(result, True)

    def test_s_is_false(self):
        result = self._callFUT('False')
        self.assertEqual(result, False)

    def test_s_is_yes(self):
        result = self._callFUT('yes')
        self.assertEqual(result, True)

    def test_s_is_on(self):
        result = self._callFUT('on')
        self.assertEqual(result, True)

    def test_s_is_1(self):
        result = self._callFUT(1)
        self.assertEqual(result, True)

class Test_as_socket_list(unittest.TestCase):

    def test_only_sockets_in_list(self):
        from waitress.adjustments import as_socket_list
        sockets = [
            socket.socket(socket.AF_INET, socket.SOCK_STREAM),
            socket.socket(socket.AF_INET6, socket.SOCK_STREAM)]
        if hasattr(socket, 'AF_UNIX'):
            sockets.append(socket.socket(socket.AF_UNIX, socket.SOCK_STREAM))
        new_sockets = as_socket_list(sockets)
        self.assertEqual(sockets, new_sockets)
        for sock in sockets:
            sock.close()

    def test_not_only_sockets_in_list(self):
        from waitress.adjustments import as_socket_list
        sockets = [
            socket.socket(socket.AF_INET, socket.SOCK_STREAM),
            socket.socket(socket.AF_INET6, socket.SOCK_STREAM),
            {'something': 'else'}]
        new_sockets = as_socket_list(sockets)
        self.assertEqual(new_sockets, [sockets[0], sockets[1]])
        for sock in [sock for sock in sockets if isinstance(sock, socket.socket)]:
            sock.close()

class TestAdjustments(unittest.TestCase):

    def _hasIPv6(self): # pragma: nocover
        if not socket.has_ipv6:
            return False

        try:
            socket.getaddrinfo(
                '::1',
                0,
                socket.AF_UNSPEC,
                socket.SOCK_STREAM,
                socket.IPPROTO_TCP,
                socket.AI_PASSIVE | socket.AI_ADDRCONFIG
            )

            return True
        except socket.gaierror as e:
            # Check to see what the error is
            if e.errno == socket.EAI_ADDRFAMILY:
                return False
            else:
                raise e

    def _makeOne(self, **kw):
        from waitress.adjustments import Adjustments
        return Adjustments(**kw)

    def test_goodvars(self):
        inst = self._makeOne(
            host='localhost',
            port='8080',
            threads='5',
            trusted_proxy='192.168.1.1',
            url_scheme='https',
            backlog='20',
            recv_bytes='200',
            send_bytes='300',
            outbuf_overflow='400',
            inbuf_overflow='500',
            connection_limit='1000',
            cleanup_interval='1100',
            channel_timeout='1200',
            log_socket_errors='true',
            max_request_header_size='1300',
            max_request_body_size='1400',
            expose_tracebacks='true',
            ident='abc',
            asyncore_loop_timeout='5',
            asyncore_use_poll=True,
            unix_socket_perms='777',
            url_prefix='///foo/',
            ipv4=True,
            ipv6=False,
        )

        self.assertEqual(inst.host, 'localhost')
        self.assertEqual(inst.port, 8080)
        self.assertEqual(inst.threads, 5)
        self.assertEqual(inst.trusted_proxy, '192.168.1.1')
        self.assertEqual(inst.url_scheme, 'https')
        self.assertEqual(inst.backlog, 20)
        self.assertEqual(inst.recv_bytes, 200)
        self.assertEqual(inst.send_bytes, 300)
        self.assertEqual(inst.outbuf_overflow, 400)
        self.assertEqual(inst.inbuf_overflow, 500)
        self.assertEqual(inst.connection_limit, 1000)
        self.assertEqual(inst.cleanup_interval, 1100)
        self.assertEqual(inst.channel_timeout, 1200)
        self.assertEqual(inst.log_socket_errors, True)
        self.assertEqual(inst.max_request_header_size, 1300)
        self.assertEqual(inst.max_request_body_size, 1400)
        self.assertEqual(inst.expose_tracebacks, True)
        self.assertEqual(inst.asyncore_loop_timeout, 5)
        self.assertEqual(inst.asyncore_use_poll, True)
        self.assertEqual(inst.ident, 'abc')
        self.assertEqual(inst.unix_socket_perms, 0o777)
        self.assertEqual(inst.url_prefix, '/foo')
        self.assertEqual(inst.ipv4, True)
        self.assertEqual(inst.ipv6, False)

        bind_pairs = [
            sockaddr[:2]
            for (family, _, _, sockaddr) in inst.listen
            if family == socket.AF_INET
        ]

        # On Travis, somehow we start listening to two sockets when resolving
        # localhost...
        self.assertEqual(('127.0.0.1', 8080), bind_pairs[0])

    def test_goodvar_listen(self):
        inst = self._makeOne(listen='127.0.0.1')

        bind_pairs = [(host, port) for (_, _, _, (host, port)) in inst.listen]

        self.assertEqual(bind_pairs, [('127.0.0.1', 8080)])

    def test_default_listen(self):
        inst = self._makeOne()

        bind_pairs = [(host, port) for (_, _, _, (host, port)) in inst.listen]

        self.assertEqual(bind_pairs, [('0.0.0.0', 8080)])

    def test_multiple_listen(self):
        inst = self._makeOne(listen='127.0.0.1:9090 127.0.0.1:8080')

        bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]

        self.assertEqual(bind_pairs,
                         [('127.0.0.1', 9090),
                          ('127.0.0.1', 8080)])

    def test_wildcard_listen(self):
        inst = self._makeOne(listen='*:8080')

        bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]

        self.assertTrue(len(bind_pairs) >= 1)

    def test_ipv6_no_port(self): # pragma: nocover
        if not self._hasIPv6():
            return

        inst = self._makeOne(listen='[::1]')

        bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]

        self.assertEqual(bind_pairs, [('::1', 8080)])

    def test_bad_port(self):
        self.assertRaises(ValueError, self._makeOne, listen='127.0.0.1:test')

    def test_service_port(self):
        if WIN and PY2: # pragma: no cover
            # On Windows and Python 2 this is broken, so we raise a ValueError
            self.assertRaises(
                ValueError,
                self._makeOne,
                listen='127.0.0.1:http',
            )
            return

        inst = self._makeOne(listen='127.0.0.1:http 0.0.0.0:https')

        bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]

        self.assertEqual(bind_pairs, [('127.0.0.1', 80), ('0.0.0.0', 443)])

    def test_dont_mix_host_port_listen(self):
        self.assertRaises(
            ValueError,
            self._makeOne,
            host='localhost',
            port='8080',
            listen='127.0.0.1:8080',
        )

    def test_good_sockets(self):
        sockets = [
            socket.socket(socket.AF_INET6, socket.SOCK_STREAM),
            socket.socket(socket.AF_INET, socket.SOCK_STREAM)]
        inst = self._makeOne(sockets=sockets)
        self.assertEqual(inst.sockets, sockets)
        sockets[0].close()
        sockets[1].close()

    def test_dont_mix_sockets_and_listen(self):
        sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM)]
        self.assertRaises(
            ValueError,
            self._makeOne,
            listen='127.0.0.1:8080',
            sockets=sockets)
        sockets[0].close()

    def test_dont_mix_sockets_and_host_port(self):
        sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM)]
        self.assertRaises(
            ValueError,
            self._makeOne,
            host='localhost',
            port='8080',
            sockets=sockets)
        sockets[0].close()

    def test_dont_mix_sockets_and_unix_socket(self):
        sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM)]
        self.assertRaises(
            ValueError,
            self._makeOne,
            unix_socket='./tmp/test',
            sockets=sockets)
        sockets[0].close()

    def test_dont_mix_unix_socket_and_host_port(self):
        self.assertRaises(
            ValueError,
            self._makeOne,
            unix_socket='./tmp/test',
            host='localhost',
            port='8080')

    def test_dont_mix_unix_socket_and_listen(self):
        self.assertRaises(
            ValueError,
            self._makeOne,
            unix_socket='./tmp/test',
            listen='127.0.0.1:8080')

    def test_dont_use_unsupported_socket_types(self):
        sockets = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]
        self.assertRaises(
            ValueError,
            self._makeOne,
            sockets=sockets)
        sockets[0].close()

    def test_badvar(self):
        self.assertRaises(ValueError, self._makeOne, nope=True)

    def test_ipv4_disabled(self):
        self.assertRaises(ValueError, self._makeOne, ipv4=False, listen="127.0.0.1:8080")

    def test_ipv6_disabled(self):
        self.assertRaises(ValueError, self._makeOne, ipv6=False, listen="[::]:8080")

    def test_server_header_removable(self):
        inst = self._makeOne(ident=None)
        self.assertEqual(inst.ident, None)

        inst = self._makeOne(ident='')
        self.assertEqual(inst.ident, None)

        inst = self._makeOne(ident='specific_header')
        self.assertEqual(inst.ident, 'specific_header')


class TestCLI(unittest.TestCase):

    def parse(self, argv):
        from waitress.adjustments import Adjustments
        return Adjustments.parse_args(argv)

    def test_noargs(self):
        opts, args = self.parse([])
        self.assertDictEqual(opts, {'call': False, 'help': False})
        self.assertSequenceEqual(args, [])

    def test_help(self):
        opts, args = self.parse(['--help'])
        self.assertDictEqual(opts, {'call': False, 'help': True})
        self.assertSequenceEqual(args, [])

    def test_call(self):
        opts, args = self.parse(['--call'])
        self.assertDictEqual(opts, {'call': True, 'help': False})
        self.assertSequenceEqual(args, [])

    def test_both(self):
        opts, args = self.parse(['--call', '--help'])
        self.assertDictEqual(opts, {'call': True, 'help': True})
        self.assertSequenceEqual(args, [])

    def test_positive_boolean(self):
        opts, args = self.parse(['--expose-tracebacks'])
        self.assertDictContainsSubset({'expose_tracebacks': 'true'}, opts)
        self.assertSequenceEqual(args, [])

    def test_negative_boolean(self):
        opts, args = self.parse(['--no-expose-tracebacks'])
        self.assertDictContainsSubset({'expose_tracebacks': 'false'}, opts)
        self.assertSequenceEqual(args, [])

    def test_cast_params(self):
        opts, args = self.parse([
            '--host=localhost',
            '--port=80',
            '--unix-socket-perms=777'
        ])
        self.assertDictContainsSubset({
            'host': 'localhost',
            'port': '80',
            'unix_socket_perms': '777',
        }, opts)
        self.assertSequenceEqual(args, [])

    def test_listen_params(self):
        opts, args = self.parse([
            '--listen=test:80',
            ])

        self.assertDictContainsSubset({
            'listen': ' test:80'
            }, opts)
        self.assertSequenceEqual(args, [])

    def test_multiple_listen_params(self):
        opts, args = self.parse([
            '--listen=test:80',
            '--listen=test:8080',
            ])

        self.assertDictContainsSubset({
            'listen': ' test:80 test:8080'
            }, opts)
        self.assertSequenceEqual(args, [])

    def test_bad_param(self):
        import getopt
        self.assertRaises(getopt.GetoptError, self.parse, ['--no-host'])


if hasattr(socket, 'AF_UNIX'):
    class TestUnixSocket(unittest.TestCase):
        def _makeOne(self, **kw):
            from waitress.adjustments import Adjustments
            return Adjustments(**kw)

        def test_dont_mix_internet_and_unix_sockets(self):
            sockets = [
                socket.socket(socket.AF_INET, socket.SOCK_STREAM),
                socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)]
            self.assertRaises(
                ValueError,
                self._makeOne,
                sockets=sockets)
            sockets[0].close()
            sockets[1].close()