headers = httputil.HTTPHeaders.parse(data[eol:])
# HTTPRequest wants an IP, not a full socket address
- if getattr(self.stream.socket, 'family', socket.AF_INET) in (
- socket.AF_INET, socket.AF_INET6):
- # Jython 2.5.2 doesn't have the socket.family attribute,
- # so just assume IP in that case.
+ if self.stream.socket.family in (socket.AF_INET, socket.AF_INET6):
remote_ip = self.address[0]
else:
# Unix (or other) socket; fake the remote address
f = open(full_path, "r", encoding="utf-8")
except TypeError:
# python 2: files return byte strings, which are decoded below.
- # Once we drop python 2.5, this could use io.open instead
- # on both 2 and 3.
f = open(full_path, "r")
_translations[locale] = {}
for i, row in enumerate(csv.reader(f)):
from tornado.ioloop import IOLoop
+@implementer(IDelayedCall)
class TornadoDelayedCall(object):
"""DelayedCall object for Tornado."""
def __init__(self, reactor, seconds, f, *args, **kw):
def active(self):
return self._active
-# Fake class decorator for python 2.5 compatibility
-TornadoDelayedCall = implementer(IDelayedCall)(TornadoDelayedCall)
+@implementer(IReactorTime, IReactorFDSet)
class TornadoReactor(PosixReactorBase):
"""Twisted reactor built on the Tornado IOLoop.
def mainLoop(self):
self._io_loop.start()
-TornadoReactor = implementer(IReactorTime, IReactorFDSet)(TornadoReactor)
class _TestReactor(TornadoReactor):
return reactor
+@implementer(IReadDescriptor, IWriteDescriptor)
class _FD(object):
def __init__(self, fd, handler):
self.fd = fd
def logPrefix(self):
return ''
-_FD = implementer(IReadDescriptor, IWriteDescriptor)(_FD)
class TwistedIOLoop(tornado.ioloop.IOLoop):
supported. In particular, proxies are not supported, connections
are not reused, and callers cannot select the network interface to be
used.
-
- Python 2.6 or higher is required for HTTPS support. Users of Python 2.5
- should use the curl-based AsyncHTTPClient if HTTPS support is required.
-
"""
def initialize(self, io_loop=None, max_clients=10,
hostname_mapping=None, max_buffer_size=104857600,
if pycurl is not None:
from tornado.curl_httpclient import CurlAsyncHTTPClient
-
+@unittest.skipIf(pycurl is None, "pycurl module not present")
class CurlHTTPClientCommonTestCase(httpclient_test.HTTPClientCommonTestCase):
def get_http_client(self):
client = CurlAsyncHTTPClient(io_loop=self.io_loop)
# make sure AsyncHTTPClient magic doesn't give us the wrong class
self.assertTrue(isinstance(client, CurlAsyncHTTPClient))
return client
-CurlHTTPClientCommonTestCase = unittest.skipIf(pycurl is None,
- "pycurl module not present")(
- CurlHTTPClientCommonTestCase)
+@unittest.skipIf(pycurl is None, "pycurl module not present")
class CurlHTTPClientTestCase(AsyncHTTPTestCase):
def setUp(self):
super(CurlHTTPClientTestCase, self).setUp()
self.wait()
self.assertEqual(1, len(exc_info))
self.assertIs(exc_info[0][0], ZeroDivisionError)
-CurlHTTPClientTestCase = unittest.skipIf(pycurl is None,
- "pycurl module not present")(
- CurlHTTPClientTestCase)
# of SSLv23 allows it.
+@skipIfNoSSL
class SSLv23Test(BaseSSLTest, SSLTestMixin):
def get_ssl_version(self):
return ssl.PROTOCOL_SSLv23
-SSLv23Test = skipIfNoSSL(SSLv23Test)
+@skipIfNoSSL
+@skipIfOldSSL
class SSLv3Test(BaseSSLTest, SSLTestMixin):
def get_ssl_version(self):
return ssl.PROTOCOL_SSLv3
-SSLv3Test = skipIfNoSSL(skipIfOldSSL(SSLv3Test))
+@skipIfNoSSL
+@skipIfOldSSL
class TLSv1Test(BaseSSLTest, SSLTestMixin):
def get_ssl_version(self):
return ssl.PROTOCOL_TLSv1
-TLSv1Test = skipIfNoSSL(skipIfOldSSL(TLSv1Test))
class BadSSLOptionsTest(unittest.TestCase):
self.assertEqual(self.fetch_json('/')['protocol'], 'https')
+@unittest.skipIf(not hasattr(socket, 'AF_UNIX') or sys.platform == 'cygwin',
+ "unix sockets not supported on this platform")
class UnixSocketTest(AsyncTestCase):
"""HTTPServers can listen on Unix sockets too.
self.assertEqual(body, b"Hello world")
stream.close()
server.stop()
-UnixSocketTest = unittest.skipIf(
- not hasattr(socket, 'AF_UNIX') or sys.platform == 'cygwin',
- "unix sockets not supported on this platform")
class KeepAliveTest(AsyncHTTPTestCase):
self.io_loop.add_callback_from_signal(callback, *args, **kwargs)
+@unittest.skipIf(futures is None, "futures module not present")
class TestIOLoopFutures(AsyncTestCase):
def test_add_future_threads(self):
with futures.ThreadPoolExecutor(1) as pool:
self.assertEqual(self.exception.args[0], "callback")
self.assertEqual(self.future.exception().args[0], "worker")
-TestIOLoopFutures = unittest.skipIf(
- futures is None, "futures module not present")(TestIOLoopFutures)
if __name__ == "__main__":
# seconds.
server, client = self.make_iostream_pair()
try:
- try:
- # This test fails on pypy with ssl. I think it's because
- # pypy's gc defeats moves objects, breaking the
- # "frozen write buffer" assumption.
- if (isinstance(server, SSLIOStream) and
- platform.python_implementation() == 'PyPy'):
- raise unittest.SkipTest(
- "pypy gc causes problems with openssl")
- except AttributeError:
- # python 2.5 didn't have platform.python_implementation,
- # but there was no pypy for 2.5
- pass
+ # This test fails on pypy with ssl. I think it's because
+ # pypy's gc defeats moves objects, breaking the
+ # "frozen write buffer" assumption.
+ if (isinstance(server, SSLIOStream) and
+ platform.python_implementation() == 'PyPy'):
+ raise unittest.SkipTest(
+ "pypy gc causes problems with openssl")
NUM_KB = 4096
for i in range(NUM_KB):
client.write(b"A" * 1024)
return IOStream(socket.socket(), io_loop=self.io_loop)
+@skipIfNoSSL
class TestIOStreamWebHTTPS(TestIOStreamWebMixin, AsyncHTTPSTestCase):
def _make_client_iostream(self):
return SSLIOStream(socket.socket(), io_loop=self.io_loop)
-TestIOStreamWebHTTPS = skipIfNoSSL(TestIOStreamWebHTTPS)
class TestIOStream(TestIOStreamMixin, AsyncTestCase):
return IOStream(connection, io_loop=self.io_loop, **kwargs)
+@skipIfNoSSL
class TestIOStreamSSL(TestIOStreamMixin, AsyncTestCase):
def _make_server_iostream(self, connection, **kwargs):
ssl_options = dict(
def _make_client_iostream(self, connection, **kwargs):
return SSLIOStream(connection, io_loop=self.io_loop, **kwargs)
-TestIOStreamSSL = skipIfNoSSL(TestIOStreamSSL)
+@skipIfNonUnix
class TestPipeIOStream(AsyncTestCase):
def test_pipe_iostream(self):
r, w = os.pipe()
self.assertEqual(data, b"ld")
rs.close()
-TestPipeIOStream = skipIfNonUnix(TestPipeIOStream)
@contextlib.contextmanager
def ignore_bytes_warning():
- if not hasattr(warnings, 'catch_warnings'):
- # python 2.5 doesn't have catch_warnings, but it doesn't have
- # BytesWarning either so there's nothing to catch.
- yield
- return
with warnings.catch_warnings():
warnings.simplefilter('ignore', category=BytesWarning)
yield
self.resolver = Resolver(self.io_loop)
+@unittest.skipIf(futures is None, "futures module not present")
class ThreadedResolverTest(AsyncTestCase, _ResolverTestMixin):
def setUp(self):
super(ThreadedResolverTest, self).setUp()
def tearDown(self):
self.executor.shutdown()
super(ThreadedResolverTest, self).tearDown()
-ThreadedResolverTest = unittest.skipIf(
- futures is None, "futures module not present")(ThreadedResolverTest)
# Not using AsyncHTTPTestCase because we need control over the IOLoop.
+@skipIfNonUnix
class ProcessTest(unittest.TestCase):
def get_app(self):
class ProcessHandler(RequestHandler):
except Exception:
logging.error("exception in child process %d", id, exc_info=True)
raise
-ProcessTest = skipIfNonUnix(ProcessTest)
+@skipIfNonUnix
class SubprocessTest(AsyncTestCase):
def test_subprocess(self):
subproc = Subprocess([sys.executable, '-u', '-i'],
ret = self.wait()
self.assertEqual(subproc.returncode, ret)
self.assertEqual(ret, -signal.SIGTERM)
-SubprocessTest = skipIfNonUnix(SubprocessTest)
restore_signal_handlers(self._saved_signals)
+@skipIfNoTwisted
class ReactorWhenRunningTest(ReactorTestCase):
def test_whenRunning(self):
self._whenRunningCalled = False
def anotherWhenRunningCallback(self):
self._anotherWhenRunningCalled = True
-ReactorWhenRunningTest = skipIfNoTwisted(ReactorWhenRunningTest)
+@skipIfNoTwisted
class ReactorCallLaterTest(ReactorTestCase):
def test_callLater(self):
self._laterCalled = False
self._laterCalled = True
self._called = self._reactor.seconds()
self._reactor.stop()
-ReactorCallLaterTest = skipIfNoTwisted(ReactorCallLaterTest)
+@skipIfNoTwisted
class ReactorTwoCallLaterTest(ReactorTestCase):
def test_callLater(self):
self._later1Called = False
self._later2Called = True
self._called2 = self._reactor.seconds()
self._reactor.stop()
-ReactorTwoCallLaterTest = skipIfNoTwisted(ReactorTwoCallLaterTest)
+@skipIfNoTwisted
class ReactorCallFromThreadTest(ReactorTestCase):
def setUp(self):
super(ReactorCallFromThreadTest, self).setUp()
def testCallFromThread(self):
self._reactor.callWhenRunning(self._whenRunningCallback)
self._reactor.run()
-ReactorCallFromThreadTest = skipIfNoTwisted(ReactorCallFromThreadTest)
+@skipIfNoTwisted
class ReactorCallInThread(ReactorTestCase):
def setUp(self):
super(ReactorCallInThread, self).setUp()
def testCallInThread(self):
self._reactor.callWhenRunning(self._whenRunningCallback)
self._reactor.run()
-ReactorCallInThread = skipIfNoTwisted(ReactorCallInThread)
class Reader(object):
Writer = implementer(IWriteDescriptor)(Writer)
+@skipIfNoTwisted
class ReactorReaderWriterTest(ReactorTestCase):
def _set_nonblocking(self, fd):
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
def testNoWriter(self):
self._reactor.callWhenRunning(self._testNoWriter)
self._reactor.run()
-ReactorReaderWriterTest = skipIfNoTwisted(ReactorReaderWriterTest)
# Test various combinations of twisted and tornado http servers,
# http clients, and event loop interfaces.
+@skipIfNoTwisted
class CompatibilityTests(unittest.TestCase):
def setUp(self):
self.saved_signals = save_signal_handlers()
response = self.twisted_fetch(
'http://localhost:%d' % self.tornado_port, self.run_reactor)
self.assertEqual(response, 'Hello from tornado!')
-CompatibilityTests = skipIfNoTwisted(CompatibilityTests)
if have_twisted:
"test_lostFileDescriptor", # incompatible with epoll and kqueue
],
'twisted.internet.test.test_process.ProcessTestsBuilder': [
- # Doesn't work on python 2.5
- 'test_systemCallUninterruptedByChildExit',
],
# Process tests appear to work on OSX 10.7, but not 10.6
#'twisted.internet.test.test_process.PTYProcessTestsBuilder': [
projects / thirdparty / tornado.git / commitdiff
