import hashlib
import warnings
import logging
import io
import ssl
import socket
from itertools import chain

from mock import patch, Mock
import pytest

from urllib3 import add_stderr_logger, disable_warnings
from urllib3.util.request import make_headers, rewind_body, _FAILEDTELL
from urllib3.util.response import assert_header_parsing
from urllib3.util.retry import Retry
from urllib3.util.timeout import Timeout
from urllib3.util.url import (
    get_host,
    parse_url,
    split_first,
    Url,
)
from urllib3.util.ssl_ import (
    resolve_cert_reqs,
    resolve_ssl_version,
    ssl_wrap_socket,
    _const_compare_digest_backport,
)
from urllib3.exceptions import (
    LocationParseError,
    TimeoutStateError,
    InsecureRequestWarning,
    SNIMissingWarning,
    InvalidHeader,
    UnrewindableBodyError,
)
from urllib3.util.connection import (
    allowed_gai_family,
    _has_ipv6
)
from urllib3.util import is_fp_closed, ssl_
import six

from . import clear_warnings

# This number represents a time in seconds, it doesn't mean anything in
# isolation. Setting to a high-ish value to avoid conflicts with the smaller
# numbers used for timeouts
TIMEOUT_EPOCH = 1000


class TestUtil(object):

    url_host_map = [
        # Hosts
        ('http://google.com/mail', ('http', 'google.com', None)),
        ('http://google.com/mail/', ('http', 'google.com', None)),
        ('google.com/mail', ('http', 'google.com', None)),
        ('http://google.com/', ('http', 'google.com', None)),
        ('http://google.com', ('http', 'google.com', None)),
        ('http://www.google.com', ('http', 'www.google.com', None)),
        ('http://mail.google.com', ('http', 'mail.google.com', None)),
        ('http://google.com:8000/mail/', ('http', 'google.com', 8000)),
        ('http://google.com:8000', ('http', 'google.com', 8000)),
        ('https://google.com', ('https', 'google.com', None)),
        ('https://google.com:8000', ('https', 'google.com', 8000)),
        ('http://user:password@127.0.0.1:1234', ('http', '127.0.0.1', 1234)),
        ('http://google.com/foo=http://bar:42/baz', ('http', 'google.com', None)),
        ('http://google.com?foo=http://bar:42/baz', ('http', 'google.com', None)),
        ('http://google.com#foo=http://bar:42/baz', ('http', 'google.com', None)),

        # IPv4
        ('173.194.35.7', ('http', '173.194.35.7', None)),
        ('http://173.194.35.7', ('http', '173.194.35.7', None)),
        ('http://173.194.35.7/test', ('http', '173.194.35.7', None)),
        ('http://173.194.35.7:80', ('http', '173.194.35.7', 80)),
        ('http://173.194.35.7:80/test', ('http', '173.194.35.7', 80)),

        # IPv6
        ('[2a00:1450:4001:c01::67]', ('http', '[2a00:1450:4001:c01::67]', None)),
        ('http://[2a00:1450:4001:c01::67]', ('http', '[2a00:1450:4001:c01::67]', None)),
        ('http://[2a00:1450:4001:c01::67]/test', ('http', '[2a00:1450:4001:c01::67]', None)),
        ('http://[2a00:1450:4001:c01::67]:80', ('http', '[2a00:1450:4001:c01::67]', 80)),
        ('http://[2a00:1450:4001:c01::67]:80/test', ('http', '[2a00:1450:4001:c01::67]', 80)),

        # More IPv6 from http://www.ietf.org/rfc/rfc2732.txt
        ('http://[fedc:ba98:7654:3210:fedc:ba98:7654:3210]:8000/index.html', (
            'http', '[fedc:ba98:7654:3210:fedc:ba98:7654:3210]', 8000)),
        ('http://[1080:0:0:0:8:800:200c:417a]/index.html', (
            'http', '[1080:0:0:0:8:800:200c:417a]', None)),
        ('http://[3ffe:2a00:100:7031::1]', ('http', '[3ffe:2a00:100:7031::1]', None)),
        ('http://[1080::8:800:200c:417a]/foo', ('http', '[1080::8:800:200c:417a]', None)),
        ('http://[::192.9.5.5]/ipng', ('http', '[::192.9.5.5]', None)),
        ('http://[::ffff:129.144.52.38]:42/index.html', ('http', '[::ffff:129.144.52.38]', 42)),
        ('http://[2010:836b:4179::836b:4179]', ('http', '[2010:836b:4179::836b:4179]', None)),

        # Hosts
        ('HTTP://GOOGLE.COM/mail/', ('http', 'google.com', None)),
        ('GOogle.COM/mail', ('http', 'google.com', None)),
        ('HTTP://GoOgLe.CoM:8000/mail/', ('http', 'google.com', 8000)),
        ('HTTP://user:password@EXAMPLE.COM:1234', ('http', 'example.com', 1234)),
        ('173.194.35.7', ('http', '173.194.35.7', None)),
        ('HTTP://173.194.35.7', ('http', '173.194.35.7', None)),
        ('HTTP://[2a00:1450:4001:c01::67]:80/test', ('http', '[2a00:1450:4001:c01::67]', 80)),
        ('HTTP://[FEDC:BA98:7654:3210:FEDC:BA98:7654:3210]:8000/index.html', (
            'http', '[fedc:ba98:7654:3210:fedc:ba98:7654:3210]', 8000)),
        ('HTTPS://[1080:0:0:0:8:800:200c:417A]/index.html', (
            'https', '[1080:0:0:0:8:800:200c:417a]', None)),
        ('abOut://eXamPlE.com?info=1', ('about', 'eXamPlE.com', None)),
        ('http+UNIX://%2fvar%2frun%2fSOCKET/path', (
            'http+unix', '%2fvar%2frun%2fSOCKET', None)),
    ]

    @pytest.mark.parametrize('url, expected_host', url_host_map)
    def test_get_host(self, url, expected_host):
        returned_host = get_host(url)
        assert returned_host == expected_host

    # TODO: Add more tests
    @pytest.mark.parametrize('location', [
        'http://google.com:foo',
        'http://::1/',
        'http://::1:80/',
        'http://google.com:-80',
        six.u('http://google.com:\xb2\xb2'),  # \xb2 = ^2
    ])
    def test_invalid_host(self, location):
        with pytest.raises(LocationParseError):
            get_host(location)

    @pytest.mark.parametrize('url, expected_normalized_url', [
        ('HTTP://GOOGLE.COM/MAIL/', 'http://google.com/MAIL/'),
        ('HTTP://JeremyCline:Hunter2@Example.com:8080/',
         'http://JeremyCline:Hunter2@example.com:8080/'),
        ('HTTPS://Example.Com/?Key=Value', 'https://example.com/?Key=Value'),
        ('Https://Example.Com/#Fragment', 'https://example.com/#Fragment'),
    ])
    def test_parse_url_normalization(self, url, expected_normalized_url):
        """Assert parse_url normalizes the scheme/host, and only the scheme/host"""
        actual_normalized_url = parse_url(url).url
        assert actual_normalized_url == expected_normalized_url

    parse_url_host_map = [
        ('http://google.com/mail', Url('http', host='google.com', path='/mail')),
        ('http://google.com/mail/', Url('http', host='google.com', path='/mail/')),
        ('http://google.com/mail', Url('http', host='google.com', path='mail')),
        ('google.com/mail', Url(host='google.com', path='/mail')),
        ('http://google.com/', Url('http', host='google.com', path='/')),
        ('http://google.com', Url('http', host='google.com')),
        ('http://google.com?foo', Url('http', host='google.com', path='', query='foo')),

        # Path/query/fragment
        ('', Url()),
        ('/', Url(path='/')),
        ('#?/!google.com/?foo#bar', Url(path='', fragment='?/!google.com/?foo#bar')),
        ('/foo', Url(path='/foo')),
        ('/foo?bar=baz', Url(path='/foo', query='bar=baz')),
        ('/foo?bar=baz#banana?apple/orange', Url(path='/foo',
                                                 query='bar=baz',
                                                 fragment='banana?apple/orange')),

        # Port
        ('http://google.com/', Url('http', host='google.com', path='/')),
        ('http://google.com:80/', Url('http', host='google.com', port=80, path='/')),
        ('http://google.com:80', Url('http', host='google.com', port=80)),

        # Auth
        ('http://foo:bar@localhost/', Url('http', auth='foo:bar', host='localhost', path='/')),
        ('http://foo@localhost/', Url('http', auth='foo', host='localhost', path='/')),
        ('http://foo:bar@baz@localhost/', Url('http',
                                              auth='foo:bar@baz',
                                              host='localhost',
                                              path='/')),
        ('http://@', Url('http', host=None, auth=''))
    ]

    non_round_tripping_parse_url_host_map = [
        # Path/query/fragment
        ('?', Url(path='', query='')),
        ('#', Url(path='', fragment='')),

        # Empty Port
        ('http://google.com:', Url('http', host='google.com')),
        ('http://google.com:/', Url('http', host='google.com', path='/')),
    ]

    @pytest.mark.parametrize(
        'url, expected_url',
        chain(parse_url_host_map, non_round_tripping_parse_url_host_map)
    )
    def test_parse_url(self, url, expected_url):
        returned_url = parse_url(url)
        assert returned_url == expected_url

    @pytest.mark.parametrize('url, expected_url', parse_url_host_map)
    def test_unparse_url(self, url, expected_url):
        assert url == expected_url.url

    def test_parse_url_invalid_IPv6(self):
        with pytest.raises(ValueError):
            parse_url('[::1')

    def test_Url_str(self):
        U = Url('http', host='google.com')
        assert str(U) == U.url

    request_uri_map = [
        ('http://google.com/mail', '/mail'),
        ('http://google.com/mail/', '/mail/'),
        ('http://google.com/', '/'),
        ('http://google.com', '/'),
        ('', '/'),
        ('/', '/'),
        ('?', '/?'),
        ('#', '/'),
        ('/foo?bar=baz', '/foo?bar=baz'),
    ]

    @pytest.mark.parametrize('url, expected_request_uri', request_uri_map)
    def test_request_uri(self, url, expected_request_uri):
        returned_url = parse_url(url)
        assert returned_url.request_uri == expected_request_uri

    url_netloc_map = [
        ('http://google.com/mail', 'google.com'),
        ('http://google.com:80/mail', 'google.com:80'),
        ('google.com/foobar', 'google.com'),
        ('google.com:12345', 'google.com:12345'),
    ]

    @pytest.mark.parametrize('url, expected_netloc', url_netloc_map)
    def test_netloc(self, url, expected_netloc):
        assert parse_url(url).netloc == expected_netloc

    @pytest.mark.parametrize('kwargs, expected', [
        ({'accept_encoding': True},
         {'accept-encoding': 'gzip,deflate'}),
        ({'accept_encoding': 'foo,bar'},
         {'accept-encoding': 'foo,bar'}),
        ({'accept_encoding': ['foo', 'bar']},
         {'accept-encoding': 'foo,bar'}),
        ({'accept_encoding': True, 'user_agent': 'banana'},
         {'accept-encoding': 'gzip,deflate', 'user-agent': 'banana'}),
        ({'user_agent': 'banana'},
         {'user-agent': 'banana'}),
        ({'keep_alive': True},
         {'connection': 'keep-alive'}),
        ({'basic_auth': 'foo:bar'},
         {'authorization': 'Basic Zm9vOmJhcg=='}),
        ({'proxy_basic_auth': 'foo:bar'},
         {'proxy-authorization': 'Basic Zm9vOmJhcg=='}),
        ({'disable_cache': True},
         {'cache-control': 'no-cache'}),
    ])
    def test_make_headers(self, kwargs, expected):
        assert make_headers(**kwargs) == expected

    def test_rewind_body(self):
        body = io.BytesIO(b'test data')
        assert body.read() == b'test data'

        # Assert the file object has been consumed
        assert body.read() == b''

        # Rewind it back to just be b'data'
        rewind_body(body, 5)
        assert body.read() == b'data'

    def test_rewind_body_failed_tell(self):
        body = io.BytesIO(b'test data')
        body.read()  # Consume body

        # Simulate failed tell()
        body_pos = _FAILEDTELL
        with pytest.raises(UnrewindableBodyError):
            rewind_body(body, body_pos)

    def test_rewind_body_bad_position(self):
        body = io.BytesIO(b'test data')
        body.read()  # Consume body

        # Pass non-integer position
        with pytest.raises(ValueError):
            rewind_body(body, body_pos=None)
        with pytest.raises(ValueError):
            rewind_body(body, body_pos=object())

    def test_rewind_body_failed_seek(self):
        class BadSeek():

            def seek(self, pos, offset=0):
                raise IOError

        with pytest.raises(UnrewindableBodyError):
            rewind_body(BadSeek(), body_pos=2)

    @pytest.mark.parametrize('input, expected', [
        (('abcd', 'b'),  ('a', 'cd', 'b')),
        (('abcd', 'cb'), ('a', 'cd', 'b')),
        (('abcd', ''),   ('abcd', '', None)),
        (('abcd', 'a'),  ('', 'bcd', 'a')),
        (('abcd', 'ab'), ('', 'bcd', 'a')),
    ])
    def test_split_first(self, input, expected):
        output = split_first(*input)
        assert output == expected

    def test_add_stderr_logger(self):
        handler = add_stderr_logger(level=logging.INFO)  # Don't actually print debug
        logger = logging.getLogger('urllib3')
        assert handler in logger.handlers

        logger.debug('Testing add_stderr_logger')
        logger.removeHandler(handler)

    def test_disable_warnings(self):
        with warnings.catch_warnings(record=True) as w:
            clear_warnings()
            warnings.warn('This is a test.', InsecureRequestWarning)
            assert len(w) == 1
            disable_warnings()
            warnings.warn('This is a test.', InsecureRequestWarning)
            assert len(w) == 1

    def _make_time_pass(self, seconds, timeout, time_mock):
        """ Make some time pass for the timeout object """
        time_mock.return_value = TIMEOUT_EPOCH
        timeout.start_connect()
        time_mock.return_value = TIMEOUT_EPOCH + seconds
        return timeout

    @pytest.mark.parametrize('kwargs, message', [
        ({'total': -1},                 'less than'),
        ({'connect': 2, 'total': -1},   'less than'),
        ({'read': -1},                  'less than'),
        ({'connect': False},            'cannot be a boolean'),
        ({'read': True},                'cannot be a boolean'),
        ({'connect': 0},                'less than or equal'),
        ({'read': 'foo'},               'int, float or None')
    ])
    def test_invalid_timeouts(self, kwargs, message):
        with pytest.raises(ValueError) as e:
            Timeout(**kwargs)
        assert message in str(e.value)

    @patch('urllib3.util.timeout.current_time')
    def test_timeout(self, current_time):
        timeout = Timeout(total=3)

        # make 'no time' elapse
        timeout = self._make_time_pass(seconds=0, timeout=timeout,
                                       time_mock=current_time)
        assert timeout.read_timeout == 3
        assert timeout.connect_timeout == 3

        timeout = Timeout(total=3, connect=2)
        assert timeout.connect_timeout == 2

        timeout = Timeout()
        assert timeout.connect_timeout == Timeout.DEFAULT_TIMEOUT

        # Connect takes 5 seconds, leaving 5 seconds for read
        timeout = Timeout(total=10, read=7)
        timeout = self._make_time_pass(seconds=5, timeout=timeout,
                                       time_mock=current_time)
        assert timeout.read_timeout == 5

        # Connect takes 2 seconds, read timeout still 7 seconds
        timeout = Timeout(total=10, read=7)
        timeout = self._make_time_pass(seconds=2, timeout=timeout,
                                       time_mock=current_time)
        assert timeout.read_timeout == 7

        timeout = Timeout(total=10, read=7)
        assert timeout.read_timeout == 7

        timeout = Timeout(total=None, read=None, connect=None)
        assert timeout.connect_timeout is None
        assert timeout.read_timeout is None
        assert timeout.total is None

        timeout = Timeout(5)
        assert timeout.total == 5

    def test_timeout_str(self):
        timeout = Timeout(connect=1, read=2, total=3)
        assert str(timeout) == "Timeout(connect=1, read=2, total=3)"
        timeout = Timeout(connect=1, read=None, total=3)
        assert str(timeout) == "Timeout(connect=1, read=None, total=3)"

    @patch('urllib3.util.timeout.current_time')
    def test_timeout_elapsed(self, current_time):
        current_time.return_value = TIMEOUT_EPOCH
        timeout = Timeout(total=3)
        with pytest.raises(TimeoutStateError):
            timeout.get_connect_duration()

        timeout.start_connect()
        with pytest.raises(TimeoutStateError):
            timeout.start_connect()

        current_time.return_value = TIMEOUT_EPOCH + 2
        assert timeout.get_connect_duration() == 2
        current_time.return_value = TIMEOUT_EPOCH + 37
        assert timeout.get_connect_duration() == 37

    @pytest.mark.parametrize('candidate, requirements', [
        (None, ssl.CERT_NONE),
        (ssl.CERT_NONE, ssl.CERT_NONE),
        (ssl.CERT_REQUIRED, ssl.CERT_REQUIRED),
        ('REQUIRED', ssl.CERT_REQUIRED),
        ('CERT_REQUIRED', ssl.CERT_REQUIRED),
    ])
    def test_resolve_cert_reqs(self, candidate, requirements):
        assert resolve_cert_reqs(candidate) == requirements

    @pytest.mark.parametrize('candidate, version', [
        (ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1),
        ("PROTOCOL_TLSv1", ssl.PROTOCOL_TLSv1),
        ("TLSv1", ssl.PROTOCOL_TLSv1),
        (ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23),
    ])
    def test_resolve_ssl_version(self, candidate, version):
        assert resolve_ssl_version(candidate) == version

    def test_is_fp_closed_object_supports_closed(self):
        class ClosedFile(object):
            @property
            def closed(self):
                return True

        assert is_fp_closed(ClosedFile())

    def test_is_fp_closed_object_has_none_fp(self):
        class NoneFpFile(object):
            @property
            def fp(self):
                return None

        assert is_fp_closed(NoneFpFile())

    def test_is_fp_closed_object_has_fp(self):
        class FpFile(object):
            @property
            def fp(self):
                return True

        assert not is_fp_closed(FpFile())

    def test_is_fp_closed_object_has_neither_fp_nor_closed(self):
        class NotReallyAFile(object):
            pass

        with pytest.raises(ValueError):
            is_fp_closed(NotReallyAFile())

    def test_ssl_wrap_socket_loads_the_cert_chain(self):
        socket = object()
        mock_context = Mock()
        ssl_wrap_socket(ssl_context=mock_context, sock=socket,
                        certfile='/path/to/certfile')

        mock_context.load_cert_chain.assert_called_once_with(
            '/path/to/certfile', None
        )

    @patch('urllib3.util.ssl_.create_urllib3_context')
    def test_ssl_wrap_socket_creates_new_context(self,
                                                 create_urllib3_context):
        socket = object()
        ssl_wrap_socket(sock=socket, cert_reqs='CERT_REQUIRED')

        create_urllib3_context.assert_called_once_with(
            None, 'CERT_REQUIRED', ciphers=None
        )

    def test_ssl_wrap_socket_loads_verify_locations(self):
        socket = object()
        mock_context = Mock()
        ssl_wrap_socket(ssl_context=mock_context, ca_certs='/path/to/pem',
                        sock=socket)
        mock_context.load_verify_locations.assert_called_once_with(
            '/path/to/pem', None
        )

    def test_ssl_wrap_socket_loads_certificate_directories(self):
        socket = object()
        mock_context = Mock()
        ssl_wrap_socket(ssl_context=mock_context, ca_cert_dir='/path/to/pems',
                        sock=socket)
        mock_context.load_verify_locations.assert_called_once_with(
            None, '/path/to/pems'
        )

    def test_ssl_wrap_socket_with_no_sni_warns(self):
        socket = object()
        mock_context = Mock()
        # Ugly preservation of original value
        HAS_SNI = ssl_.HAS_SNI
        ssl_.HAS_SNI = False
        try:
            with patch('warnings.warn') as warn:
                ssl_wrap_socket(ssl_context=mock_context, sock=socket,
                                server_hostname='www.google.com')
            mock_context.wrap_socket.assert_called_once_with(socket)
            assert warn.call_count >= 1
            warnings = [call[0][1] for call in warn.call_args_list]
            assert SNIMissingWarning in warnings
        finally:
            ssl_.HAS_SNI = HAS_SNI

    def test_const_compare_digest_fallback(self):
        target = hashlib.sha256(b'abcdef').digest()
        assert _const_compare_digest_backport(target, target)

        prefix = target[:-1]
        assert not _const_compare_digest_backport(target, prefix)

        suffix = target + b'0'
        assert not _const_compare_digest_backport(target, suffix)

        incorrect = hashlib.sha256(b'xyz').digest()
        assert not _const_compare_digest_backport(target, incorrect)

    def test_has_ipv6_disabled_on_compile(self):
        with patch('socket.has_ipv6', False):
            assert not _has_ipv6('::1')

    def test_has_ipv6_enabled_but_fails(self):
        with patch('socket.has_ipv6', True):
            with patch('socket.socket') as mock:
                instance = mock.return_value
                instance.bind = Mock(side_effect=Exception('No IPv6 here!'))
                assert not _has_ipv6('::1')

    def test_has_ipv6_enabled_and_working(self):
        with patch('socket.has_ipv6', True):
            with patch('socket.socket') as mock:
                instance = mock.return_value
                instance.bind.return_value = True
                assert _has_ipv6('::1')

    def test_has_ipv6_disabled_on_appengine(self):
        gae_patch = patch(
            'urllib3.contrib._appengine_environ.is_appengine_sandbox',
            return_value=True)
        with gae_patch:
            assert not _has_ipv6('::1')

    def test_ip_family_ipv6_enabled(self):
        with patch('urllib3.util.connection.HAS_IPV6', True):
            assert allowed_gai_family() == socket.AF_UNSPEC

    def test_ip_family_ipv6_disabled(self):
        with patch('urllib3.util.connection.HAS_IPV6', False):
            assert allowed_gai_family() == socket.AF_INET

    @pytest.mark.parametrize('value', [
        "-1",
        "+1",
        "1.0",
        six.u("\xb2"),  # \xb2 = ^2
    ])
    def test_parse_retry_after_invalid(self, value):
        retry = Retry()
        with pytest.raises(InvalidHeader):
            retry.parse_retry_after(value)

    @pytest.mark.parametrize('value, expected', [
        ("0", 0),
        ("1000", 1000),
        ("\t42 ", 42),
    ])
    def test_parse_retry_after(self, value, expected):
        retry = Retry()
        assert retry.parse_retry_after(value) == expected

    @pytest.mark.parametrize('headers', [
        b'foo',
        None,
        object,
    ])
    def test_assert_header_parsing_throws_typeerror_with_non_headers(self, headers):
        with pytest.raises(TypeError):
            assert_header_parsing(headers)