python类_RequestProxy()的实例源码

def raw_fetch(self, headers, body):
        with closing(Resolver(io_loop=self.io_loop)) as resolver:
            with closing(SimpleAsyncHTTPClient(self.io_loop,
                                               resolver=resolver)) as client:
                conn = RawRequestHTTPConnection(
                    self.io_loop, client,
                    httpclient._RequestProxy(
                        httpclient.HTTPRequest(self.get_url("/")),
                        dict(httpclient.HTTPRequest._DEFAULTS)),
                    None, self.stop,
                    1024 * 1024, resolver)
                conn.set_request(
                    b"\r\n".join(headers +
                                 [utf8("Content-Length: %d\r\n" % len(body))]) +
                    b"\r\n" + body)
                response = self.wait()
                response.rethrow()
                return response

def websocket_connect(url, io_loop=None, callback=None, connect_timeout=None):
    """Client-side websocket support.

    Takes a url and returns a Future whose result is a
    `WebSocketClientConnection`.

    .. versionchanged:: 3.2
       Also accepts ``HTTPRequest`` objects in place of urls.
    """
    if io_loop is None:
        io_loop = IOLoop.current()
    if isinstance(url, httpclient.HTTPRequest):
        assert connect_timeout is None
        request = url
        # Copy and convert the headers dict/object (see comments in
        # AsyncHTTPClient.fetch)
        request.headers = httputil.HTTPHeaders(request.headers)
    else:
        request = httpclient.HTTPRequest(url, connect_timeout=connect_timeout)
    request = httpclient._RequestProxy(
        request, httpclient.HTTPRequest._DEFAULTS)
    conn = WebSocketClientConnection(io_loop, request)
    if callback is not None:
        io_loop.add_future(conn.connect_future, callback)
    return conn.connect_future

def raw_fetch(self, headers, body):
        client = SimpleAsyncHTTPClient(self.io_loop)
        conn = RawRequestHTTPConnection(
            self.io_loop, client,
            httpclient._RequestProxy(
                httpclient.HTTPRequest(self.get_url("/")),
                dict(httpclient.HTTPRequest._DEFAULTS)),
            None, self.stop,
            1024 * 1024, Resolver(io_loop=self.io_loop))
        conn.set_request(
            b"\r\n".join(headers +
                         [utf8("Content-Length: %d\r\n" % len(body))]) +
            b"\r\n" + body)
        response = self.wait()
        client.close()
        response.rethrow()
        return response

def test_reuse_request_from_response(self):
        # The response.request attribute should be an HTTPRequest, not
        # a _RequestProxy.
        # This test uses self.http_client.fetch because self.fetch calls
        # self.get_url on the input unconditionally.
        url = self.get_url('/hello')
        response = yield self.http_client.fetch(url)
        self.assertEqual(response.request.url, url)
        self.assertTrue(isinstance(response.request, HTTPRequest))
        response2 = yield self.http_client.fetch(response.request)
        self.assertEqual(response2.body, b'Hello world!')

def test_request_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          user_agent='foo'),
                              dict())
        self.assertEqual(proxy.user_agent, 'foo')

def test_default_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict(network_interface='foo'))
        self.assertEqual(proxy.network_interface, 'foo')

def test_both_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          proxy_host='foo'),
                              dict(proxy_host='bar'))
        self.assertEqual(proxy.proxy_host, 'foo')

def test_bad_attribute(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict())
        with self.assertRaises(AttributeError):
            proxy.foo

def test_defaults_none(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'), None)
        self.assertIs(proxy.auth_username, None)

def test_reuse_request_from_response(self):
        # The response.request attribute should be an HTTPRequest, not
        # a _RequestProxy.
        # This test uses self.http_client.fetch because self.fetch calls
        # self.get_url on the input unconditionally.
        url = self.get_url('/hello')
        response = yield self.http_client.fetch(url)
        self.assertEqual(response.request.url, url)
        self.assertTrue(isinstance(response.request, HTTPRequest))
        response2 = yield self.http_client.fetch(response.request)
        self.assertEqual(response2.body, b'Hello world!')

def test_default_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict(network_interface='foo'))
        self.assertEqual(proxy.network_interface, 'foo')

def test_both_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          proxy_host='foo'),
                              dict(proxy_host='bar'))
        self.assertEqual(proxy.proxy_host, 'foo')

def test_neither_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict())
        self.assertIs(proxy.auth_username, None)

def test_bad_attribute(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict())
        with self.assertRaises(AttributeError):
            proxy.foo

def test_defaults_none(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'), None)
        self.assertIs(proxy.auth_username, None)

def test_reuse_request_from_response(self):
        # The response.request attribute should be an HTTPRequest, not
        # a _RequestProxy.
        # This test uses self.http_client.fetch because self.fetch calls
        # self.get_url on the input unconditionally.
        url = self.get_url('/hello')
        response = yield self.http_client.fetch(url)
        self.assertEqual(response.request.url, url)
        self.assertTrue(isinstance(response.request, HTTPRequest))
        response2 = yield self.http_client.fetch(response.request)
        self.assertEqual(response2.body, b'Hello world!')

def test_request_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          user_agent='foo'),
                              dict())
        self.assertEqual(proxy.user_agent, 'foo')

def test_default_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict(network_interface='foo'))
        self.assertEqual(proxy.network_interface, 'foo')

def test_both_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          proxy_host='foo'),
                              dict(proxy_host='bar'))
        self.assertEqual(proxy.proxy_host, 'foo')

def test_bad_attribute(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict())
        with self.assertRaises(AttributeError):
            proxy.foo

def test_defaults_none(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'), None)
        self.assertIs(proxy.auth_username, None)

def test_reuse_request_from_response(self):
        # The response.request attribute should be an HTTPRequest, not
        # a _RequestProxy.
        # This test uses self.http_client.fetch because self.fetch calls
        # self.get_url on the input unconditionally.
        url = self.get_url('/hello')
        response = yield self.http_client.fetch(url)
        self.assertEqual(response.request.url, url)
        self.assertTrue(isinstance(response.request, HTTPRequest))
        response2 = yield self.http_client.fetch(response.request)
        self.assertEqual(response2.body, b'Hello world!')

def test_request_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          user_agent='foo'),
                              dict())
        self.assertEqual(proxy.user_agent, 'foo')

def test_default_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict(network_interface='foo'))
        self.assertEqual(proxy.network_interface, 'foo')

def test_both_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          proxy_host='foo'),
                              dict(proxy_host='bar'))
        self.assertEqual(proxy.proxy_host, 'foo')

def test_bad_attribute(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict())
        with self.assertRaises(AttributeError):
            proxy.foo

def test_defaults_none(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'), None)
        self.assertIs(proxy.auth_username, None)

def test_reuse_request_from_response(self):
        # The response.request attribute should be an HTTPRequest, not
        # a _RequestProxy.
        # This test uses self.http_client.fetch because self.fetch calls
        # self.get_url on the input unconditionally.
        url = self.get_url('/hello')
        response = yield self.http_client.fetch(url)
        self.assertEqual(response.request.url, url)
        self.assertTrue(isinstance(response.request, HTTPRequest))
        response2 = yield self.http_client.fetch(response.request)
        self.assertEqual(response2.body, b'Hello world!')

def test_request_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/',
                                          user_agent='foo'),
                              dict())
        self.assertEqual(proxy.user_agent, 'foo')

def test_default_set(self):
        proxy = _RequestProxy(HTTPRequest('http://example.com/'),
                              dict(network_interface='foo'))
        self.assertEqual(proxy.network_interface, 'foo')