# -*- coding: utf-8 -*-
"""
test_http11.py
~~~~~~~~~~~~~~

Unit tests for hyper's HTTP/1.1 implementation.
"""
import os
import zlib

from collections import namedtuple
from io import BytesIO, StringIO

import mock
import pytest

import hyper
from hyper.http11.connection import HTTP11Connection
from hyper.http11.response import HTTP11Response
from hyper.common.headers import HTTPHeaderMap
from hyper.common.exceptions import ChunkedDecodeError, ConnectionResetError
from hyper.common.util import HTTPVersion
from hyper.compat import bytes, zlib_compressobj


class TestHTTP11Connection(object):
    def test_pycohttpparser_installs_correctly(self):
        # This test is a debugging tool: if pycohttpparser is being tested by
        # Travis, we need to confirm it imports correctly. Hyper will normally
        # hide the import failure, so let's discover it here.
        # Alternatively, if we are *not* testing with nghttp2, this test should
        # confirm that it's not available.
        if os.environ.get('HYPER_FAST_PARSE') == 'true':
            import pycohttpparser
        else:
            with pytest.raises(ImportError):
                import pycohttpparser  # noqa

        assert True

    def test_initialization_no_port(self):
        c = HTTP11Connection('httpbin.org')

        assert c.host == 'httpbin.org'
        assert c.port == 80
        assert not c.secure
        assert not c.proxy_host

    def test_initialization_inline_port(self):
        c = HTTP11Connection('httpbin.org:443')

        assert c.host == 'httpbin.org'
        assert c.port == 443
        assert c.secure
        assert not c.proxy_host

    def test_initialization_separate_port(self):
        c = HTTP11Connection('localhost', 8080)

        assert c.host == 'localhost'
        assert c.port == 8080
        assert not c.secure
        assert not c.proxy_host

    def test_can_override_security(self):
        c = HTTP11Connection('localhost', 443, secure=False)

        assert c.host == 'localhost'
        assert c.port == 443
        assert not c.secure
        assert not c.proxy_host

    def test_initialization_proxy(self):
        c = HTTP11Connection('httpbin.org', proxy_host='localhost')

        assert c.host == 'httpbin.org'
        assert c.port == 80
        assert not c.secure
        assert c.proxy_host == 'localhost'
        assert c.proxy_port == 8080

    def test_initialization_proxy_with_inline_port(self):
        c = HTTP11Connection('httpbin.org', proxy_host='localhost:8443')

        assert c.host == 'httpbin.org'
        assert c.port == 80
        assert not c.secure
        assert c.proxy_host == 'localhost'
        assert c.proxy_port == 8443

    def test_initialization_proxy_with_separate_port(self):
        c = HTTP11Connection(
            'httpbin.org', proxy_host='localhost', proxy_port=8443
        )

        assert c.host == 'httpbin.org'
        assert c.port == 80
        assert not c.secure
        assert c.proxy_host == 'localhost'
        assert c.proxy_port == 8443

    def test_initialization_with_ipv6_addresses_proxy_inline_port(self):
        c = HTTP11Connection(
            '[abcd:dcba::1234]', proxy_host='[ffff:aaaa::1]:8443'
        )

        assert c.host == 'abcd:dcba::1234'
        assert c.port == 80
        assert not c.secure
        assert c.proxy_host == 'ffff:aaaa::1'
        assert c.proxy_port == 8443

    def test_basic_request(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        c.request('GET', '/get', headers={'User-Agent': 'hyper'})

        expected = (
            b"GET /get HTTP/1.1\r\n"
            b"User-Agent: hyper\r\n"
            b"connection: Upgrade, HTTP2-Settings\r\n"
            b"upgrade: h2c\r\n"
            b"HTTP2-Settings: AAQAAP__\r\n"
            b"host: httpbin.org\r\n"
            b"\r\n"
        )
        received = b''.join(sock.queue)

        assert received == expected

    def test_iterable_header(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        c.request('GET', '/get', headers=(
            ('User-Agent', 'hyper'),
            ('Custom-field', 'test'),
            ('Custom-field2', 'test'),
            ('Custom-field', 'test2'),
        ))

        expected = (
            b"GET /get HTTP/1.1\r\n"
            b"User-Agent: hyper\r\n"
            b"Custom-field: test\r\n"
            b"Custom-field2: test\r\n"
            b"Custom-field: test2\r\n"
            b"connection: Upgrade, HTTP2-Settings\r\n"
            b"upgrade: h2c\r\n"
            b"HTTP2-Settings: AAQAAP__\r\n"
            b"host: httpbin.org\r\n"
            b"\r\n"
        )
        received = b''.join(sock.queue)

        assert received == expected

    def test_invalid_header(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = DummySocket()

        with pytest.raises(ValueError):
            c.request('GET', '/get', headers=42)

    def test_proxy_request(self):
        c = HTTP11Connection('httpbin.org', proxy_host='localhost')
        c._sock = sock = DummySocket()

        c.request('GET', '/get', headers={'User-Agent': 'hyper'})

        expected = (
            b"GET /get HTTP/1.1\r\n"
            b"User-Agent: hyper\r\n"
            b"connection: Upgrade, HTTP2-Settings\r\n"
            b"upgrade: h2c\r\n"
            b"HTTP2-Settings: AAQAAP__\r\n"
            b"host: httpbin.org\r\n"
            b"\r\n"
        )
        received = b''.join(sock.queue)

        assert received == expected

    def test_request_with_bytestring_body(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        c.request(
            'POST',
            '/post',
            headers=HTTPHeaderMap([('User-Agent', 'hyper')]),
            body=b'hi'
        )

        expected = (
            b"POST /post HTTP/1.1\r\n"
            b"User-Agent: hyper\r\n"
            b"connection: Upgrade, HTTP2-Settings\r\n"
            b"upgrade: h2c\r\n"
            b"HTTP2-Settings: AAQAAP__\r\n"
            b"content-length: 2\r\n"
            b"host: httpbin.org\r\n"
            b"\r\n"
            b"hi"
        )
        received = b''.join(sock.queue)

        assert received == expected

    def test_request_with_file_body(self):
        # Testing this is tricksy: in practice, we do this by passing a fake
        # file and monkeypatching out 'os.fstat'. This makes it look like a
        # real file.
        FstatRval = namedtuple('FstatRval', ['st_size'])

        def fake_fstat(*args):
            return FstatRval(16)

        old_fstat = hyper.http11.connection.os.fstat

        try:
            hyper.http11.connection.os.fstat = fake_fstat
            c = HTTP11Connection('httpbin.org')
            c._sock = sock = DummySocket()

            f = DummyFile(b'some binary data')
            c.request('POST', '/post',  body=f)

            expected = (
                b"POST /post HTTP/1.1\r\n"
                b"connection: Upgrade, HTTP2-Settings\r\n"
                b"upgrade: h2c\r\n"
                b"HTTP2-Settings: AAQAAP__\r\n"
                b"content-length: 16\r\n"
                b"host: httpbin.org\r\n"
                b"\r\n"
                b"some binary data"
            )
            received = b''.join(sock.queue)

            assert received == expected

        finally:
            # Put back the monkeypatch.
            hyper.http11.connection.os.fstat = old_fstat

    def test_request_with_generator_body(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        def body():
            yield b'hi'
            yield b'there'
            yield b'sir'

        c.request('POST', '/post', body=body())

        expected = (
            b"POST /post HTTP/1.1\r\n"
            b"connection: Upgrade, HTTP2-Settings\r\n"
            b"upgrade: h2c\r\n"
            b"HTTP2-Settings: AAQAAP__\r\n"
            b"transfer-encoding: chunked\r\n"
            b"host: httpbin.org\r\n"
            b"\r\n"
            b"2\r\nhi\r\n"
            b"5\r\nthere\r\n"
            b"3\r\nsir\r\n"
            b"0\r\n\r\n"
        )
        received = b''.join(sock.queue)

        assert received == expected

    def test_content_length_overrides_generator(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        def body():
            yield b'hi'
            yield b'there'
            yield b'sir'

        c.request(
            'POST', '/post', body=body(), headers={b'content-length': b'10'}
        )

        expected = (
            b"POST /post HTTP/1.1\r\n"
            b"content-length: 10\r\n"
            b"connection: Upgrade, HTTP2-Settings\r\n"
            b"upgrade: h2c\r\n"
            b"HTTP2-Settings: AAQAAP__\r\n"
            b"host: httpbin.org\r\n"
            b"\r\n"
            b"hitheresir"
        )

        received = b''.join(sock.queue)
        assert received == expected

    def test_chunked_overrides_body(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        f = DummyFile(b'oneline\nanotherline')

        c.request(
            'POST',
            '/post',
            headers={'transfer-encoding': 'chunked'},
            body=f
        )

        expected = (
            b"POST /post HTTP/1.1\r\n"
            b"transfer-encoding: chunked\r\n"
            b"connection: Upgrade, HTTP2-Settings\r\n"
            b"upgrade: h2c\r\n"
            b"HTTP2-Settings: AAQAAP__\r\n"
            b"host: httpbin.org\r\n"
            b"\r\n"
            b"8\r\noneline\n\r\n"
            b"b\r\nanotherline\r\n"
            b"0\r\n\r\n"
        )
        received = b''.join(sock.queue)

        assert received == expected

    def test_response_with_empty_reason(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        sock._buffer = BytesIO(
            b"HTTP/1.1 201 \r\n"
            b"Connection: close\r\n"
            b"Server: Socket\r\n"
            b"Content-Length: 0\r\n"
            b"\r\n"
        )

        r = c.get_response()

        assert r.status == 201
        assert r.reason == b''

    def test_get_response(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        sock._buffer = BytesIO(
            b"HTTP/1.1 201 No Content\r\n"
            b"Connection: close\r\n"
            b"Server: Socket\r\n"
            b"Content-Length: 0\r\n"
            b"\r\n"
        )

        r = c.get_response()

        assert r.status == 201
        assert r.reason == b'No Content'
        assert list(r.headers.iter_raw()) == [
            (b'Connection', b'close'),
            (b'Server', b'Socket'),
            (b'Content-Length', b'0')
        ]
        assert r.read() == b''

    def test_response_short_reads(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = sock = DummySocket()

        sock._buffer = BytesIO(
            b"HTTP/1.1 200 OK\r\n"
            b"Content-Length: 15\r\n"
            b"\r\n"
            b"hellotherechamp"
        )

        r = c.get_response()

        assert r.status == 200
        assert r.reason == b'OK'
        assert r.read(5) == b'hello'
        assert r.read(5) == b'there'
        assert r.read(5) == b'champ'
        assert r.read(5) == b''

    def test_request_with_unicodestring_body(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = DummySocket()

        with pytest.raises(ValueError):
            c.request(
                'POST',
                '/post',
                headers=HTTPHeaderMap([('User-Agent', 'hyper')]),
                body=u'hi'
            )

    def test_request_with_file_body_in_text_mode(self):
        # Testing this is tricksy: in practice, we do this by passing a fake
        # file and monkeypatching out 'os.fstat'. This makes it look like a
        # real file.
        FstatRval = namedtuple('FstatRval', ['st_size'])

        def fake_fstat(*args):
            return FstatRval(16)

        old_fstat = hyper.http11.connection.os.fstat

        try:
            hyper.http11.connection.os.fstat = fake_fstat
            c = HTTP11Connection('httpbin.org')
            c._sock = DummySocket()

            f = DummyFile(b'')
            f.buffer = StringIO(u'some binary data')

            with pytest.raises(ValueError):
                c.request('POST', '/post',  body=f)
        finally:
            # Put back the monkeypatch.
            hyper.http11.connection.os.fstat = old_fstat

    def test_request_with_unicode_generator_body(self):
        c = HTTP11Connection('httpbin.org')
        c._sock = DummySocket()

        def body():
            yield u'hi'