This may work around a suspected ipv6-related issue on travis.
self.server_ioloop.close(all_fds=True)
def get_url(self, path):
- return 'http://localhost:%d%s' % (self.port, path)
+ return 'http://127.0.0.1:%d%s' % (self.port, path)
def test_sync_client(self):
response = self.http_client.fetch(self.get_url('/'))
# When given Expect: 100-continue, we get a 100 response after the
# headers, and then the real response after the body.
stream = IOStream(socket.socket(), io_loop=self.io_loop)
- stream.connect(("localhost", self.get_http_port()), callback=self.stop)
+ stream.connect(("127.0.0.1", self.get_http_port()), callback=self.stop)
self.wait()
stream.write(b"\r\n".join([b"POST /hello HTTP/1.1",
b"Content-Length: 1024",
def setUp(self):
super(HTTPServerRawTest, self).setUp()
self.stream = IOStream(socket.socket())
- self.stream.connect(('localhost', self.get_http_port()), self.stop)
+ self.stream.connect(('127.0.0.1', self.get_http_port()), self.stop)
self.wait()
def tearDown(self):
# The next few methods are a crude manual http client
def connect(self):
self.stream = IOStream(socket.socket(), io_loop=self.io_loop)
- self.stream.connect(('localhost', self.get_http_port()), self.stop)
+ self.stream.connect(('127.0.0.1', self.get_http_port()), self.stop)
self.wait()
def read_headers(self):
def connect(self):
stream = IOStream(socket.socket())
- stream.connect(('localhost', self.get_http_port()), self.stop)
+ stream.connect(('127.0.0.1', self.get_http_port()), self.stop)
self.wait()
self.streams.append(stream)
return stream
def test_read_until_close(self):
stream = self._make_client_iostream()
- stream.connect(('localhost', self.get_http_port()), callback=self.stop)
+ stream.connect(('127.0.0.1', self.get_http_port()), callback=self.stop)
self.wait()
stream.write(b"GET / HTTP/1.0\r\n\r\n")
def test_read_zero_bytes(self):
self.stream = self._make_client_iostream()
- self.stream.connect(("localhost", self.get_http_port()),
+ self.stream.connect(("127.0.0.1", self.get_http_port()),
callback=self.stop)
self.wait()
self.stream.write(b"GET / HTTP/1.0\r\n\r\n")
def connected_callback():
connected[0] = True
self.stop()
- stream.connect(("localhost", self.get_http_port()),
+ stream.connect(("127.0.0.1", self.get_http_port()),
callback=connected_callback)
# unlike the previous tests, try to write before the connection
# is complete.
"""Basic test of IOStream's ability to return Futures."""
stream = self._make_client_iostream()
connect_result = yield stream.connect(
- ("localhost", self.get_http_port()))
+ ("127.0.0.1", self.get_http_port()))
self.assertIs(connect_result, stream)
yield stream.write(b"GET / HTTP/1.0\r\n\r\n")
first_line = yield stream.read_until(b"\r\n")
@gen_test
def test_future_close_while_reading(self):
stream = self._make_client_iostream()
- yield stream.connect(("localhost", self.get_http_port()))
+ yield stream.connect(("127.0.0.1", self.get_http_port()))
yield stream.write(b"GET / HTTP/1.0\r\n\r\n")
with self.assertRaises(StreamClosedError):
yield stream.read_bytes(1024 * 1024)
def test_future_read_until_close(self):
# Ensure that the data comes through before the StreamClosedError.
stream = self._make_client_iostream()
- yield stream.connect(("localhost", self.get_http_port()))
+ yield stream.connect(("127.0.0.1", self.get_http_port()))
yield stream.write(b"GET / HTTP/1.0\r\nConnection: close\r\n\r\n")
yield stream.read_until(b"\r\n\r\n")
body = yield stream.read_until_close()
cleanup_func, port = refusing_port()
self.addCleanup(cleanup_func)
with ExpectLog(gen_log, ".*", required=False):
- self.http_client.fetch("http://localhost:%d/" % port, self.stop)
+ self.http_client.fetch("http://127.0.0.1:%d/" % port, self.stop)
response = self.wait()
self.assertEqual(599, response.code)
def testTwistedServerTornadoClientIOLoop(self):
self.start_twisted_server()
response = self.tornado_fetch(
- 'http://localhost:%d' % self.twisted_port, self.run_ioloop)
+ 'http://127.0.0.1:%d' % self.twisted_port, self.run_ioloop)
self.assertEqual(response.body, 'Hello from twisted!')
def testTwistedServerTornadoClientReactor(self):
self.start_twisted_server()
response = self.tornado_fetch(
- 'http://localhost:%d' % self.twisted_port, self.run_reactor)
+ 'http://127.0.0.1:%d' % self.twisted_port, self.run_reactor)
self.assertEqual(response.body, 'Hello from twisted!')
def testTornadoServerTwistedClientIOLoop(self):
self.start_tornado_server()
response = self.twisted_fetch(
- 'http://localhost:%d' % self.tornado_port, self.run_ioloop)
+ 'http://127.0.0.1:%d' % self.tornado_port, self.run_ioloop)
self.assertEqual(response, 'Hello from tornado!')
def testTornadoServerTwistedClientReactor(self):
self.start_tornado_server()
response = self.twisted_fetch(
- 'http://localhost:%d' % self.tornado_port, self.run_reactor)
+ 'http://127.0.0.1:%d' % self.tornado_port, self.run_reactor)
self.assertEqual(response, 'Hello from tornado!')
@skipIfNoSingleDispatch
def testTornadoServerTwistedCoroutineClientIOLoop(self):
self.start_tornado_server()
response = self.twisted_coroutine_fetch(
- 'http://localhost:%d' % self.tornado_port, self.run_ioloop)
+ 'http://127.0.0.1:%d' % self.tornado_port, self.run_ioloop)
self.assertEqual(response, 'Hello from tornado!')
def test_connection_close(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
- s.connect(("localhost", self.get_http_port()))
+ s.connect(("127.0.0.1", self.get_http_port()))
self.stream = IOStream(s, io_loop=self.io_loop)
self.stream.write(b"GET / HTTP/1.0\r\n\r\n")
self.wait()
def connect(self, url, connection_close):
# Use a raw connection so we can control the sending of data.
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
- s.connect(("localhost", self.get_http_port()))
+ s.connect(("127.0.0.1", self.get_http_port()))
stream = IOStream(s, io_loop=self.io_loop)
stream.write(b"GET " + url + b" HTTP/1.1\r\n")
if connection_close:
@gen.coroutine
def ws_connect(self, path, compression_options=None):
ws = yield websocket_connect(
- 'ws://localhost:%d%s' % (self.get_http_port(), path),
+ 'ws://127.0.0.1:%d%s' % (self.get_http_port(), path),
compression_options=compression_options)
raise gen.Return(ws)
def test_websocket_callbacks(self):
websocket_connect(
- 'ws://localhost:%d/echo' % self.get_http_port(),
+ 'ws://127.0.0.1:%d/echo' % self.get_http_port(),
io_loop=self.io_loop, callback=self.stop)
ws = self.wait().result()
ws.write_message('hello')
with self.assertRaises(IOError):
with ExpectLog(gen_log, ".*"):
yield websocket_connect(
- 'ws://localhost:%d/' % port,
+ 'ws://127.0.0.1:%d/' % port,
io_loop=self.io_loop,
connect_timeout=3600)
@gen_test
def test_websocket_close_buffered_data(self):
ws = yield websocket_connect(
- 'ws://localhost:%d/echo' % self.get_http_port())
+ 'ws://127.0.0.1:%d/echo' % self.get_http_port())
ws.write_message('hello')
ws.write_message('world')
# Close the underlying stream.
def test_websocket_headers(self):
# Ensure that arbitrary headers can be passed through websocket_connect.
ws = yield websocket_connect(
- HTTPRequest('ws://localhost:%d/header' % self.get_http_port(),
+ HTTPRequest('ws://127.0.0.1:%d/header' % self.get_http_port(),
headers={'X-Test': 'hello'}))
response = yield ws.read_message()
self.assertEqual(response, 'hello')
def test_check_origin_valid_no_path(self):
port = self.get_http_port()
- url = 'ws://localhost:%d/echo' % port
- headers = {'Origin': 'http://localhost:%d' % port}
+ url = 'ws://127.0.0.1:%d/echo' % port
+ headers = {'Origin': 'http://127.0.0.1:%d' % port}
ws = yield websocket_connect(HTTPRequest(url, headers=headers),
io_loop=self.io_loop)
def test_check_origin_valid_with_path(self):
port = self.get_http_port()
- url = 'ws://localhost:%d/echo' % port
- headers = {'Origin': 'http://localhost:%d/something' % port}
+ url = 'ws://127.0.0.1:%d/echo' % port
+ headers = {'Origin': 'http://127.0.0.1:%d/something' % port}
ws = yield websocket_connect(HTTPRequest(url, headers=headers),
io_loop=self.io_loop)
def test_check_origin_invalid_partial_url(self):
port = self.get_http_port()
- url = 'ws://localhost:%d/echo' % port
- headers = {'Origin': 'localhost:%d' % port}
+ url = 'ws://127.0.0.1:%d/echo' % port
+ headers = {'Origin': '127.0.0.1:%d' % port}
with self.assertRaises(HTTPError) as cm:
yield websocket_connect(HTTPRequest(url, headers=headers),
def test_check_origin_invalid(self):
port = self.get_http_port()
- url = 'ws://localhost:%d/echo' % port
- # Host is localhost, which should not be accessible from some other
+ url = 'ws://127.0.0.1:%d/echo' % port
+ # Host is 127.0.0.1, which should not be accessible from some other
# domain
headers = {'Origin': 'http://somewhereelse.com'}