import unittest
import unittest.mock as mock
import asyncio
import pyx.http as http


def create_dummy_message():
    msg = http.HttpMessage(None)
    msg.headers = [
        http.HttpHeader('Server', 'Pyx'),
        http.HttpHeader('Cookie', 'a'),
        http.HttpHeader('Cookie', 'b'),
    ]
    return msg


def create_dummy_connection():
    loop = asyncio.get_event_loop()

    reader = asyncio.StreamReader(loop=loop)

    @asyncio.coroutine
    def dummy_drain():
        yield from asyncio.sleep(0.001)
    writer = mock.Mock(spec=asyncio.StreamWriter)
    writer.attach_mock(mock.Mock(wraps=dummy_drain), 'drain')

    conn = http.HttpConnection(reader, writer)
    return conn


def create_dummy_request():
    conn = create_dummy_connection()
    req = http.HttpRequest(conn)
    return req


class TestHttpMessage(unittest.TestCase):
    def test_get_header(self):
        msg = create_dummy_message()

        self.assertEqual(msg.get_header("server"), ["Pyx"])
        self.assertEqual(msg.get_header("SERVER"), ["Pyx"])
        self.assertEqual(msg.get_header("pragma"), [])

        self.assertEqual(msg.get_header("cookie"), ["a", "b"])
        self.assertEqual(msg.get_first_header("cookie"), "a")
        self.assertTrue(msg.get_first_header("pragma") is None)

    def test_write_headers(self):
        msg = create_dummy_message()
        self.assertEqual(msg.write_headers(),
                         ['Server: Pyx', 'Cookie: a', 'Cookie: b'])

        msg.headers = []
        self.assertEqual(msg.write_headers(), [])


class TestHttpRequest(unittest.TestCase):
    def test_parse_req_line(self):
        req = create_dummy_request()

        req._parse_req_line(b'POST / HTTP/1.1\r\n')
        self.assertEqual(req.method, 'POST')
        self.assertEqual(req.path, '/')
        self.assertTrue(req.query is None)
        self.assertEqual(req.protocol, 'HTTP')
        self.assertEqual(req.version, (1, 1))

        req._parse_req_line(
            b'GET /some/path?some=query&some_other=query HTTP/1.1\r\n')
        self.assertEqual(req.method, 'GET')
        self.assertEqual(req.path, '/some/path')
        self.assertEqual(req.query, 'some=query&some_other=query')

        with self.assertRaises(http.BadHttpRequestError):
            req._parse_req_line(b'')

        with self.assertRaises(http.BadHttpRequestError):
            req._parse_req_line(b'GET /\r\n')

        with self.assertRaises(http.BadHttpRequestError):
            req._parse_req_line(b'GET / GARBAGE\r\n')

        req._parse_req_line(b'GET / HTTP/1\r\n')
        self.assertEqual(req.version, (1, 0))

    def test_parse_header(self):
        req = create_dummy_request()

        req._parse_header(b'Server: Pyx\r\n')
        self.assertEqual(req.headers, [http.HttpHeader('Server', 'Pyx')])

        req.headers = []
        with self.assertRaises(http.BadHttpHeaderError):
            req._parse_header(b'Server\r\n')

        req.headers = []
        req._parse_header(b'Server:\r\n')
        self.assertEqual(req.headers, [http.HttpHeader('Server', '')])

        req.headers = []
        req._parse_header(b'Server: \r\n')
        self.assertEqual(req.headers, [http.HttpHeader('Server', '')])

        req.headers = []
        req._parse_header(b'Host: some.badasshost.com:8080\r\n')
        self.assertEqual(req.headers, [http.HttpHeader('Host', 'some.badasshost.com:8080')])

        with self.assertRaises(http.BadHttpHeaderError):
            req._parse_header(b': pyx\r\n')

        with self.assertRaises(http.BadHttpHeaderError):
            req._parse_header(b' : pyx')

        with self.assertRaises(http.BadHttpHeaderError):
            req._parse_header(b' \t : pyx')

    def test_parse(self):
        loop = asyncio.get_event_loop()
        conn = create_dummy_connection()

        reader = conn.reader
        reader.feed_data(
            b'GET /?q=p&s=t HTTP/1.1\r\n'
            b'Host: localhost\r\n'
            b'Connection: Keep-Alive\r\n'
            b'Pragma: Test\r\n'
            b' : Test\r\n'
            b'\r\n')

        req = loop.run_until_complete(http.HttpRequest.parse(conn))

        self.assertEqual(req.method, 'GET')
        self.assertEqual(req.path, '/')
        self.assertEqual(req.query, 'q=p&s=t')
        self.assertEqual(req.protocol, 'HTTP')
        self.assertEqual(req.version, (1, 1))
        self.assertEqual(req.headers,
                         [
                             http.HttpHeader('Host', 'localhost'),
                             http.HttpHeader('Connection', 'Keep-Alive'),
                             http.HttpHeader('Pragma', 'Test'),
                         ])

    def test_respond(self):
        req = create_dummy_request()

        req.version = (1, 1)
        resp = req.respond(200)
        self.assertEqual(resp.code, 200)
        self.assertEqual(resp.version, (1, 1))

        req.version = (1, 0)
        resp = req.respond(400)
        self.assertEqual(resp.code, 400)
        self.assertEqual(resp.version, (1, 0))


class TestHttpResponse(unittest.TestCase):
    def test_write(self):
        resp = http.HttpResponse(200, None)
        resp.headers = [
            http.HttpHeader('Server', 'Pyx'),
            http.HttpHeader('Connection', 'keep-alive')
        ]
        self.assertEqual(resp.write(),
                         ['HTTP/1.1 200 OK',
                          'Server: Pyx',
                          'Connection: keep-alive',
                          '\r\n'])
        self.assertEqual(str(resp),
                         'HTTP/1.1 200 OK\r\n'
                         'Server: Pyx\r\n'
                         'Connection: keep-alive\r\n'
                         '\r\n')

    def test_send(self):
        loop = asyncio.get_event_loop()
        req = create_dummy_request()
        resp = req.respond(200)
        self.assertEqual(resp.code, 200)
        self.assertFalse(req.responded)

        resp.headers = [
            http.HttpHeader('Server', 'Pyx'),
            http.HttpHeader('Content-Length', '100'),
            http.HttpHeader('Content-Type', 'text/plain'),
        ]
        loop.run_until_complete(resp.send())
        resp.connection.writer.write.assert_called_with(str(resp).encode())
        self.assertTrue(req.responded)

    def test_send_body(self):
        loop = asyncio.get_event_loop()
        req = create_dummy_request()
        resp = req.respond(200)

        loop.run_until_complete(resp.send())
        self.assertTrue(req.responded)

        loop.run_until_complete(resp.send_body(b'Yes, this is the body.'))
        resp.connection.writer.write.assert_called_with(b'Yes, this is the body.')

        loop.run_until_complete(resp.send_body('This is another string body.'))
        resp.connection.writer.write.assert_called_with(b'This is another string body.')


class DummyResource(http.UrlResource):
    def get_child(self, key):
        if key == 'hello':
            return self
        elif key == "static":
            return http.StaticRootResource('.')
        else:
            raise http.HttpError(404, '{} not found'.format(key))


class TestUrlResource(unittest.TestCase):
    def test_traverse(self):
        res = DummyResource()
        self.assertEqual(res.traverse(''), res)
        self.assertEqual(res.traverse('/'), res)
        self.assertEqual(res.traverse('/hello'), res)

        with self.assertRaises(http.HttpError):
            res.traverse('/does/not/exist')

        sres = res.traverse('/static')
        self.assertEqual(sres.root, '.')
        self.assertEqual(sres._build_real_path(), '.')

        sres = res.traverse('/static/')
        self.assertEqual(sres._build_real_path(), '.')

        sres = res.traverse('/static/some/path')
        self.assertEqual(sres._build_real_path(), './some/path')

    def test_not_implemented(self):
        res = http.UrlResource()

        with self.assertRaises(NotImplementedError):
            res.traverse('/hello')

        req = create_dummy_request()
        with self.assertRaises(NotImplementedError):
            res.handle_request(req)

class TestStaticRootResource(unittest.TestCase):
    def test_build_real_path(self):
        res = http.StaticRootResource('local_root')
        res = res.traverse('/some/long/path/where/ever/it/leads/')
        self.assertEqual(res._build_real_path(),
                         'local_root/some/long/path/where/ever/it/leads')

        res = http.StaticRootResource('local_root')
        res = res.traverse('/some/../dangerous/path')
        self.assertEqual(res._build_real_path(),
                         'local_root/dangerous/path')

        res = http.StaticRootResource('local_root')
        res = res.traverse('/some/../../dangerous/path')
        self.assertEqual(res._build_real_path(),
                         'local_root/dangerous/path')

        res = http.StaticRootResource('local_root')
        res = res.traverse('/some/%2e%2e%2f%2e%2e/dangerous/path')
        self.assertEqual(res._build_real_path(),
                         'local_root/dangerous/path')

[(8, 'pyx.http.HttpMessage', 'http.HttpMessage', (['None'], {}), True, 'import pyx.http as http\n'), (18, 'asyncio.get_event_loop', 'asyncio.get_event_loop', ([], {}), False, 'import asyncio\n'), (20, 'asyncio.StreamReader', 'asyncio.StreamReader', ([], {'loop': 'loop'}), False, 'import asyncio\n'), (25, 'unittest.mock.Mock', 'mock.Mock', ([], {'spec': 'asyncio.StreamWriter'}), True, 'import unittest.mock as mock\n'), (28, 'pyx.http.HttpConnection', 'http.HttpConnection', (['reader', 'writer'], {}), True, 'import pyx.http as http\n'), (34, 'pyx.http.HttpRequest', 'http.HttpRequest', (['conn'], {}), True, 'import pyx.http as http\n'), (10, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Server"""', '"""Pyx"""'], {}), True, 'import pyx.http as http\n'), (11, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Cookie"""', '"""a"""'], {}), True, 'import pyx.http as http\n'), (12, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Cookie"""', '"""b"""'], {}), True, 'import pyx.http as http\n'), (26, 'unittest.mock.Mock', 'mock.Mock', ([], {'wraps': 'dummy_drain'}), True, 'import unittest.mock as mock\n'), (120, 'asyncio.get_event_loop', 'asyncio.get_event_loop', ([], {}), False, 'import asyncio\n'), (162, 'pyx.http.HttpResponse', 'http.HttpResponse', (['(200)', 'None'], {}), True, 'import pyx.http as http\n'), (179, 'asyncio.get_event_loop', 'asyncio.get_event_loop', ([], {}), False, 'import asyncio\n'), (195, 'asyncio.get_event_loop', 'asyncio.get_event_loop', ([], {}), False, 'import asyncio\n'), (240, 'pyx.http.UrlResource', 'http.UrlResource', ([], {}), True, 'import pyx.http as http\n'), (251, 'pyx.http.StaticRootResource', 'http.StaticRootResource', (['"""local_root"""'], {}), True, 'import pyx.http as http\n'), (256, 'pyx.http.StaticRootResource', 'http.StaticRootResource', (['"""local_root"""'], {}), True, 'import pyx.http as http\n'), (261, 'pyx.http.StaticRootResource', 'http.StaticRootResource', (['"""local_root"""'], {}), True, 'import pyx.http as http\n'), (266, 'pyx.http.StaticRootResource', 'http.StaticRootResource', (['"""local_root"""'], {}), True, 'import pyx.http as http\n'), (24, 'asyncio.sleep', 'asyncio.sleep', (['(0.001)'], {}), False, 'import asyncio\n'), (132, 'pyx.http.HttpRequest.parse', 'http.HttpRequest.parse', (['conn'], {}), True, 'import pyx.http as http\n'), (164, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Server"""', '"""Pyx"""'], {}), True, 'import pyx.http as http\n'), (165, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Connection"""', '"""keep-alive"""'], {}), True, 'import pyx.http as http\n'), (186, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Server"""', '"""Pyx"""'], {}), True, 'import pyx.http as http\n'), (187, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Content-Length"""', '"""100"""'], {}), True, 'import pyx.http as http\n'), (188, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Content-Type"""', '"""text/plain"""'], {}), True, 'import pyx.http as http\n'), (92, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Server"""', '"""Pyx"""'], {}), True, 'import pyx.http as http\n'), (100, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Server"""', '""""""'], {}), True, 'import pyx.http as http\n'), (104, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Server"""', '""""""'], {}), True, 'import pyx.http as http\n'), (108, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Host"""', '"""some.badasshost.com:8080"""'], {}), True, 'import pyx.http as http\n'), (141, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Host"""', '"""localhost"""'], {}), True, 'import pyx.http as http\n'), (142, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Connection"""', '"""Keep-Alive"""'], {}), True, 'import pyx.http as http\n'), (143, 'pyx.http.HttpHeader', 'http.HttpHeader', (['"""Pragma"""', '"""Test"""'], {}), True, 'import pyx.http as http\n'), (214, 'pyx.http.StaticRootResource', 'http.StaticRootResource', (['"""."""'], {}), True, 'import pyx.http as http\n')]