import io
import socket
import unittest


class TestSendRequest(unittest.TestCase):
    def _getTargetClass(self):
        from webob.client import SendRequest

        return SendRequest

    def _makeOne(self, **kw):
        cls = self._getTargetClass()
        return cls(**kw)

    def _makeEnviron(self, extra=None):
        environ = {
            "wsgi.url_scheme": "http",
            "SERVER_NAME": "localhost",
            "HTTP_HOST": "localhost:80",
            "SERVER_PORT": "80",
            "wsgi.input": io.BytesIO(),
            "CONTENT_LENGTH": 0,
            "REQUEST_METHOD": "GET",
        }
        if extra is not None:
            environ.update(extra)
        return environ

    def test___call___unknown_scheme(self):
        environ = self._makeEnviron({"wsgi.url_scheme": "abc"})
        inst = self._makeOne()
        self.assertRaises(ValueError, inst, environ, None)

    def test___call___gardenpath(self):
        environ = self._makeEnviron()
        response = DummyResponse("msg")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "200 OK")
            self.assertEqual(headers, [])
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertEqual(list(iterable), [b"foo"])

    def test___call___no_servername_no_http_host(self):
        environ = self._makeEnviron()
        del environ["SERVER_NAME"]
        del environ["HTTP_HOST"]
        response = DummyResponse("msg")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)
        self.assertRaises(ValueError, inst, environ, None)

    def test___call___no_servername_colon_not_in_host_http(self):
        environ = self._makeEnviron()
        del environ["SERVER_NAME"]
        environ["HTTP_HOST"] = "localhost"
        response = DummyResponse("msg")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "200 OK")
            self.assertEqual(headers, [])
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertEqual(list(iterable), [b"foo"])
        self.assertEqual(environ["SERVER_NAME"], "localhost")
        self.assertEqual(environ["SERVER_PORT"], "80")

    def test___call___no_servername_colon_not_in_host_https(self):
        environ = self._makeEnviron()
        del environ["SERVER_NAME"]
        environ["HTTP_HOST"] = "localhost"
        environ["wsgi.url_scheme"] = "https"
        response = DummyResponse("msg")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPSConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "200 OK")
            self.assertEqual(headers, [])
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertEqual(list(iterable), [b"foo"])
        self.assertEqual(environ["SERVER_NAME"], "localhost")
        self.assertEqual(environ["SERVER_PORT"], "443")

    def test___call___no_content_length(self):
        environ = self._makeEnviron()
        del environ["CONTENT_LENGTH"]
        response = DummyResponse("msg")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "200 OK")
            self.assertEqual(headers, [])
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertEqual(list(iterable), [b"foo"])

    def test___call___with_webob_client_timeout_and_timeout_supported(self):
        environ = self._makeEnviron()
        environ["webob.client.timeout"] = 10
        response = DummyResponse("msg")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "200 OK")
            self.assertEqual(headers, [])
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertEqual(list(iterable), [b"foo"])
        self.assertEqual(conn_factory.kw, {"timeout": 10})

    def test___call___bad_content_length(self):
        environ = self._makeEnviron({"CONTENT_LENGTH": "abc"})
        response = DummyResponse("msg")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "200 OK")
            self.assertEqual(headers, [])
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertEqual(list(iterable), [b"foo"])

    def test___call___with_socket_timeout(self):
        environ = self._makeEnviron()
        response = socket.timeout()
        response.msg = "msg"
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "504 Gateway Timeout")
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertTrue(list(iterable)[0].startswith(b"504"))

    def test___call___with_socket_error_neg2(self):
        environ = self._makeEnviron()
        response = socket.error(-2)
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "502 Bad Gateway")
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertTrue(list(iterable)[0].startswith(b"502"))

    def test___call___with_socket_error_ENODATA(self):
        import errno

        environ = self._makeEnviron()
        if not hasattr(errno, "ENODATA"):
            # no ENODATA on win
            return
        response = socket.error(errno.ENODATA)
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "502 Bad Gateway")
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertTrue(list(iterable)[0].startswith(b"502"))

    def test___call___with_socket_error_unknown(self):
        environ = self._makeEnviron()
        response = socket.error("nope")
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "502 Bad Gateway")
            inst.start_response_called = True

        self.assertRaises(socket.error, inst, environ, start_response)

    def test___call___nolength(self):
        environ = self._makeEnviron()
        response = DummyResponse("msg", None)
        conn_factory = DummyConnectionFactory(response)
        inst = self._makeOne(HTTPConnection=conn_factory)

        def start_response(status, headers):
            self.assertEqual(status, "200 OK")
            self.assertEqual(headers, [])
            inst.start_response_called = True

        iterable = inst(environ, start_response)
        self.assertTrue(inst.start_response_called)
        self.assertEqual(list(iterable), [b"foo"])
        self.assertEqual(response.length, None)


class DummyMessage:
    def __init__(self, msg):
        self.msg = msg
        self.headers = self._headers = {}


class DummyResponse:
    def __init__(self, msg, headerval="10"):
        self.msg = DummyMessage(msg)
        self.status = "200"
        self.reason = "OK"
        self.headerval = headerval

    def getheader(self, name):
        return self.headerval

    def read(self, length=None):
        self.length = length
        return b"foo"


class DummyConnectionFactory:
    def __init__(self, result=None):
        self.result = result
        self.closed = False

    def __call__(self, hostport, **kw):
        self.hostport = hostport
        self.kw = kw
        self.request = DummyRequestFactory(hostport, **kw)
        return self

    def getresponse(self):
        if isinstance(self.result, Exception):
            raise self.result
        return self.result

    def close(self):
        self.closed = True


class DummyRequestFactory:
    def __init__(self, hostport, **kw):
        self.hostport = hostport
        self.kw = kw

    def __call__(self, method, path, body, headers):
        return self