import datetime
import json
import re
from unittest import mock

import pytest
from multidict import CIMultiDict

from aiohttp import hdrs, signals
from aiohttp.protocol import (HttpVersion, HttpVersion10, HttpVersion11,
                              RawRequestMessage)
from aiohttp.web import (ContentCoding, Request, Response, StreamResponse,
                         json_response)


def make_request(method, path, headers=CIMultiDict(),
                 version=HttpVersion11, **kwargs):
    message = RawRequestMessage(method, path, version, headers,
                                [(k.encode('utf-8'), v.encode('utf-8'))
                                 for k, v in headers.items()],
                                False, False)
    return request_from_message(message, **kwargs)


def request_from_message(message, **kwargs):
    app = kwargs.get('app') or mock.Mock()
    app._debug = False
    app.on_response_prepare = signals.Signal(app)
    payload = mock.Mock()
    transport = mock.Mock()
    reader = mock.Mock()
    writer = kwargs.get('writer') or mock.Mock()
    req = Request(app, message, payload,
                  transport, reader, writer)
    return req


def test_stream_response_ctor():
    resp = StreamResponse()
    assert 200 == resp.status
    assert resp.keep_alive is None


def test_content_length():
    resp = StreamResponse()
    assert resp.content_length is None


def test_content_length_setter():
    resp = StreamResponse()

    resp.content_length = 234
    assert 234 == resp.content_length


def test_drop_content_length_header_on_setting_len_to_None():
    resp = StreamResponse()

    resp.content_length = 1
    assert "1" == resp.headers['Content-Length']
    resp.content_length = None
    assert 'Content-Length' not in resp.headers


def test_set_content_length_to_None_on_non_set():
    resp = StreamResponse()

    resp.content_length = None
    assert 'Content-Length' not in resp.headers
    resp.content_length = None
    assert 'Content-Length' not in resp.headers


def test_setting_content_type():
    resp = StreamResponse()

    resp.content_type = 'text/html'
    assert 'text/html' == resp.headers['content-type']


def test_setting_charset():
    resp = StreamResponse()

    resp.content_type = 'text/html'
    resp.charset = 'koi8-r'
    assert 'text/html; charset=koi8-r' == resp.headers['content-type']


def test_default_charset():
    resp = StreamResponse()

    assert resp.charset is None


def test_reset_charset():
    resp = StreamResponse()

    resp.content_type = 'text/html'
    resp.charset = None
    assert resp.charset is None


def test_reset_charset_after_setting():
    resp = StreamResponse()

    resp.content_type = 'text/html'
    resp.charset = 'koi8-r'
    resp.charset = None
    assert resp.charset is None


def test_charset_without_content_type():
    resp = StreamResponse()

    with pytest.raises(RuntimeError):
        resp.charset = 'koi8-r'


def test_last_modified_initial():
    resp = StreamResponse()
    assert resp.last_modified is None


def test_last_modified_string():
    resp = StreamResponse()

    dt = datetime.datetime(1990, 1, 2, 3, 4, 5, 0, datetime.timezone.utc)
    resp.last_modified = 'Mon, 2 Jan 1990 03:04:05 GMT'
    assert resp.last_modified == dt


def test_last_modified_timestamp():
    resp = StreamResponse()

    dt = datetime.datetime(1970, 1, 1, 0, 0, 0, 0, datetime.timezone.utc)

    resp.last_modified = 0
    assert resp.last_modified == dt

    resp.last_modified = 0.0
    assert resp.last_modified == dt


def test_last_modified_datetime():
    resp = StreamResponse()

    dt = datetime.datetime(2001, 2, 3, 4, 5, 6, 0, datetime.timezone.utc)
    resp.last_modified = dt
    assert resp.last_modified == dt


def test_last_modified_reset():
    resp = StreamResponse()

    resp.last_modified = 0
    resp.last_modified = None
    assert resp.last_modified is None


@pytest.mark.run_loop
def test_start():
    req = make_request('GET', '/')
    resp = StreamResponse()
    assert resp.keep_alive is None

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)

        assert msg.send_headers.called
        msg2 = yield from resp.prepare(req)
        assert msg is msg2

        assert resp.keep_alive

    req2 = make_request('GET', '/')
    with pytest.raises(RuntimeError):
        yield from resp.prepare(req2)


@pytest.mark.run_loop
def test_chunked_encoding():
    req = make_request('GET', '/')
    resp = StreamResponse()
    assert not resp.chunked

    resp.enable_chunked_encoding()
    assert resp.chunked

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)
        assert msg.chunked


@pytest.mark.run_loop
def test_chunk_size():
    req = make_request('GET', '/')
    resp = StreamResponse()
    assert not resp.chunked

    resp.enable_chunked_encoding(chunk_size=8192)
    assert resp.chunked

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)
        assert msg.chunked
        msg.add_chunking_filter.assert_called_with(8192)
        assert msg.filter is not None


@pytest.mark.run_loop
def test_chunked_encoding_forbidden_for_http_10():
    req = make_request('GET', '/', version=HttpVersion10)
    resp = StreamResponse()
    resp.enable_chunked_encoding()

    with pytest.raises(RuntimeError) as ctx:
        yield from resp.prepare(req)
    assert re.match("Using chunked encoding is forbidden for HTTP/1.0",
                    str(ctx.value))


@pytest.mark.run_loop
def test_compression_no_accept():
    req = make_request('GET', '/')
    resp = StreamResponse()
    assert not resp.chunked

    assert not resp.compression
    resp.enable_compression()
    assert resp.compression

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)
        assert not msg.add_compression_filter.called


@pytest.mark.run_loop
def test_force_compression_no_accept_backwards_compat():
    req = make_request('GET', '/')
    resp = StreamResponse()
    assert not resp.chunked

    assert not resp.compression
    resp.enable_compression(force=True)
    assert resp.compression

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)
    assert msg.add_compression_filter.called
    assert msg.filter is not None


@pytest.mark.run_loop
def test_force_compression_false_backwards_compat():
    req = make_request('GET', '/')
    resp = StreamResponse()

    assert not resp.compression
    resp.enable_compression(for
ce=False)
    assert resp.compression

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)
    assert
 not msg.add_compression_filter.called


@pytest.mark.run_loop
def test_compression_default_coding():
    req = make_request(
        'GET', '/',
        headers=CIMultiDict({hdrs.ACCEPT_ENCODING: 'gzip, deflate'}))
    resp = StreamResponse()
    assert not resp.chunked

    assert not resp.compression
    resp.enable_compression()
    assert resp.compression

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)

    msg.add_compression_filter.assert_called_with('deflate')
    assert 'deflate' == resp.headers.get(hdrs.CONTENT_ENCODING)
    assert msg.filter is not None


@pytest.mark.run_loop
def test_force_compression_deflate():
    req = make_request(
        'GET', '/',
        headers=CIMultiDict({hdrs.ACCEPT_ENCODING: 'gzip, deflate'}))
    resp = StreamResponse()

    resp.enable_compression(ContentCoding.deflate)
    assert resp.compression

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yield from resp.prepare(req)
    msg.add_compression_filter.assert_called_with('deflate')
    assert 'deflate' == resp.headers.get(hdrs.CONTENT_ENCODING)


@pytest.mark.run_loop
def test_force_compression_no_accept_deflate():
    req = make_request('GET', '/')
    resp = StreamResponse()

    resp.enable_compression(ContentCoding.deflate)
    assert resp.compression

    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
        msg = yiel