]> git.ipfire.org Git - thirdparty/tornado.git/commitdiff
ioloop: Add tests for IOLoop configuration
authorBen Darnell <ben@bendarnell.com>
Sat, 3 Jun 2017 21:08:51 +0000 (17:08 -0400)
committerBen Darnell <ben@bendarnell.com>
Sat, 3 Jun 2017 21:14:17 +0000 (17:14 -0400)
tornado/test/ioloop_test.py

index 1601813f44c5285fc66af16aeb05b440cbc6d935..318f3fad0932377d8d1c63e72a8ac8a1414538ce 100644 (file)
@@ -6,11 +6,13 @@ import contextlib
 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
@@ -24,6 +26,16 @@ try:
 except ImportError:
     futures = None
 
+try:
+    import asyncio
+except ImportError:
+    asyncio = None
+
+try:
+    import twisted
+except ImportError:
+    twisted = None
+
 
 class FakeTimeSelect(_Select):
     def __init__(self):
@@ -677,5 +689,51 @@ class TestPeriodicCallback(unittest.TestCase):
         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()