import asyncio
import unittest
from unittest import mock
from aiohttp.web import Request
from aiohttp.multidict import MultiDict, CIMultiDict
from aiohttp.protocol import HttpVersion
from aiohttp.protocol import RawRequestMessage


class TestWebRequest(unittest.TestCase):

    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)

    def tearDown(self):
        self.loop.close()

    def make_request(self, method, path, headers=CIMultiDict(), *,
                     version=HttpVersion(1, 1), closing=False):
        self.app = mock.Mock()
        message = RawRequestMessage(method, path, version, headers, closing,
                                    False)
        self.payload = mock.Mock()
        self.transport = mock.Mock()
        self.writer = mock.Mock()
        self.reader = mock.Mock()
        req = Request(self.app, message, self.payload,
                      self.transport, self.reader, self.writer)
        return req

    def test_ctor(self):
        req = self.make_request('GET', '/path/to?a=1&b=2')

        self.assertIs(self.app, req.app)
        self.assertEqual('GET', req.method)
        self.assertEqual(HttpVersion(1, 1), req.version)
        self.assertEqual(None, req.host)
        self.assertEqual('/path/to?a=1&b=2', req.path_qs)
        self.assertEqual('/path/to', req.path)
        self.assertEqual('a=1&b=2', req.query_string)

        get = req.GET
        self.assertEqual(MultiDict([('a', '1'), ('b', '2')]), get)
        # second call should return the same object
        self.assertIs(get, req.GET)

        with self.assertWarns(DeprecationWarning):
            self.assertIs(self.payload, req.payload)
        self.assertIs(self.payload, req.content)
        self.assertIs(self.transport, req.transport)
        self.assertTrue(req.keep_alive)

    def test_POST(self):
        req = self.make_request('POST', '/')
        with self.assertRaises(RuntimeError):
            req.POST

        marker = object()
        req._post = marker
        self.assertIs(req.POST, marker)
        self.assertIs(req.POST, marker)

    def test_content_type_not_specified(self):
        req = self.make_request('Get', '/')
        self.assertEqual('application/octet-stream', req.content_type)

    def test_content_type_from_spec(self):
        req = self.make_request(
            'Get', '/',
            CIMultiDict([('CONTENT-TYPE', 'application/json')]))
        self.assertEqual('application/json', req.content_type)

    def test_content_type_from_spec_with_charset(self):
        req = self.make_request(
            'Get', '/',
            CIMultiDict([('CONTENT-TYPE', 'text/html; charset=UTF-8')]))
        self.assertEqual('text/html', req.content_type)
        self.assertEqual('UTF-8', req.charset)

    def test_calc_content_type_on_getting_charset(self):
        req = self.make_request(
            'Get', '/',
            CIMultiDict([('CONTENT-TYPE', 'text/html; charset=UTF-8')]))
        self.assertEqual('UTF-8', req.charset)
        self.assertEqual('text/html', req.content_type)

    def test_urlencoded_querystring(self):
        req = self.make_request(
            'GET',
            '/yandsearch?text=%D1%82%D0%B5%D0%BA%D1%81%D1%82')
        self.assertEqual({'text': 'текст'}, req.GET)

    def test_non_ascii_path(self):
        req = self.make_request('GET', '/путь')
        self.assertEqual('/путь', req.path)

    def test_content_length(self):
        req = self.make_request(
            'Get', '/',
            CIMultiDict([('CONTENT-LENGTH', '123')]))

        self.assertEqual(123, req.content_length)

    def test_non_keepalive_on_http10(self):
        req = self.make_request('GET', '/', version=HttpVersion(1, 0))
        self.assertFalse(req.keep_alive)

    def test_non_keepalive_on_closing(self):
        req = self.make_request('GET', '/', closing=True)
        self.assertFalse(req.keep_alive)

    def test_call_POST_on_GET_request(self):
        req = self.make_request('GET', '/')

        ret = self.loop.run_until_complete(req.post())
        self.assertEqual(CIMultiDict(), ret)

    def test_call_POST_on_weird_content_type(self):
        req = self.make_request(
            'POST', '/',
            headers=CIMultiDict({'CONTENT-TYPE': 'something/weird'}))

        ret = self.loop.run_until_complete(req.post())
        self.assertEqual(CIMultiDict(), ret)

    def test_call_POST_twice(self):
        req = self.make_request('GET', '/')

        ret1 = self.loop.run_until_complete(req.post())
        ret2 = self.loop.run_until_complete(req.post())
        self.assertIs(ret1, ret2)

    def test_no_request_cookies(self):
        req = self.make_request('GET', '/')

        self.assertEqual(req.cookies, {})

        cookies = req.cookies
        self.assertIs(cookies, req.cookies)

    def test_request_cookie(self):
        headers = CIMultiDict(COOKIE='cookie1=value1; cookie2=value2')
        req = self.make_request('GET', '/', headers=headers)

        self.assertEqual(req.cookies, {
            'cookie1': 'value1',
            'cookie2': 'value2',
        })

    def test_request_cookie__set_item(self):
        headers = CIMultiDict(COOKIE='name=value')
        req = self.make_request('GET', '/', headers=headers)

        self.assertEqual(req.cookies, {'name': 'value'})

        with self.assertRaises(TypeError):
            req.cookies['my'] = 'value'

    def test_match_info(self):
        req = self.make_request('GET', '/')
        self.assertIsNone(req.match_info)
        match = {'a': 'b'}
        req._match_info = match
        self.assertIs(match, req.match_info)

    def test_request_is_dict(self):
        req = self.make_request('GET', '/')
        self.assertTrue(isinstance(req, dict))
        req['key'] = 'value'
        self.assertEqual('value', req['key'])

    def test___repr__(self):
        req = self.make_request('GET', '/path/to')
        self.assertEqual("<Request GET /path/to >", repr(req))
