From: Ben Darnell Date: Sat, 3 Jun 2017 21:08:51 +0000 (-0400) Subject: ioloop: Add tests for IOLoop configuration X-Git-Tag: v5.0.0~76^2~1 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ebf18ec84242935f84046c8ed5140f9164f703ef;p=thirdparty%2Ftornado.git ioloop: Add tests for IOLoop configuration --- diff --git a/tornado/test/ioloop_test.py b/tornado/test/ioloop_test.py index 1601813f4..318f3fad0 100644 --- a/tornado/test/ioloop_test.py +++ b/tornado/test/ioloop_test.py @@ -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()