import datetime
import functools
import socket
+import subprocess
import sys
import threading
import time
import types
+from tornado.escape import native_str
from tornado import gen
from tornado.ioloop import IOLoop, TimeoutError, PollIOLoop, PeriodicCallback
from tornado.log import app_log
except ImportError:
futures = None
+try:
+ import asyncio
+except ImportError:
+ asyncio = None
+
+try:
+ import twisted
+except ImportError:
+ twisted = None
+
class FakeTimeSelect(_Select):
def __init__(self):
self.assertEqual(calls, expected)
+class TestIOLoopConfiguration(unittest.TestCase):
+ def run_python(self, *statements):
+ statements = [
+ 'from tornado.ioloop import IOLoop, PollIOLoop',
+ 'classname = lambda x: x.__class__.__name__',
+ ] + list(statements)
+ args = [sys.executable, '-c', '; '.join(statements)]
+ return native_str(subprocess.check_output(args)).strip()
+
+ def test_default(self):
+ # The default is a subclass of PollIOLoop
+ is_poll = self.run_python(
+ 'print(isinstance(IOLoop.current(), PollIOLoop))')
+ self.assertEqual(is_poll, 'True')
+
+ def test_explicit_select(self):
+ # SelectIOLoop can always be configured explicitly.
+ default_class = self.run_python(
+ 'IOLoop.configure("tornado.platform.select.SelectIOLoop")',
+ 'print(classname(IOLoop.current()))')
+ self.assertEqual(default_class, 'SelectIOLoop')
+
+ @unittest.skipIf(asyncio is None, "asyncio module not present")
+ def test_asyncio(self):
+ cls = self.run_python(
+ 'IOLoop.configure("tornado.platform.asyncio.AsyncIOLoop")',
+ 'print(classname(IOLoop.current()))')
+ self.assertEqual(cls, 'AsyncIOLoop')
+
+ @unittest.skipIf(asyncio is None, "asyncio module not present")
+ def test_asyncio_main(self):
+ cls = self.run_python(
+ 'from tornado.platform.asyncio import AsyncIOMainLoop',
+ 'AsyncIOMainLoop().install()',
+ 'print(classname(IOLoop.current()))')
+ self.assertEqual(cls, 'AsyncIOMainLoop')
+
+ @unittest.skipIf(twisted is None, "twisted module not present")
+ def test_twisted(self):
+ cls = self.run_python(
+ 'from tornado.platform.twisted import TwistedIOLoop',
+ 'TwistedIOLoop().install()',
+ 'print(classname(IOLoop.current()))')
+ self.assertEqual(cls, 'TwistedIOLoop')
+
+
if __name__ == "__main__":
unittest.main()