diff options
Diffstat (limited to 'tests/aio_test_base.py')
-rw-r--r-- | tests/aio_test_base.py | 63 |
1 files changed, 0 insertions, 63 deletions
diff --git a/tests/aio_test_base.py b/tests/aio_test_base.py deleted file mode 100644 index 7289b6a..0000000 --- a/tests/aio_test_base.py +++ /dev/null @@ -1,63 +0,0 @@ -# Copyright 2015 Vladimir Rutsky <vladimir@rutsky.org> -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Base classes and utility functions for testing asyncio-powered code. -""" - -import unittest -import asyncio -import socket -import functools -import concurrent.futures - - -@asyncio.coroutine -def create_server(protocol_factory, loop=None, sock=None): - """Create server listening on random port""" - - if sock is None: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.bind(("127.0.0.1", 0)) - sock.listen(10) - - if loop is None: - loop = asyncio.get_event_loop() - - return (yield from loop.create_server(protocol_factory, sock=sock)) - - -class AioTestBase(unittest.TestCase): - """Base class for tests that need temporary asyncio event loop""" - def setUp(self): - self.loop = asyncio.new_event_loop() - asyncio.set_event_loop(self.loop) - - self.thread_pool_executor = concurrent.futures.ThreadPoolExecutor(4) - - def tearDown(self): - self.thread_pool_executor.shutdown() - - self.loop.close() - asyncio.set_event_loop(None) - - -def asynctest(test_method): - """Decorator for coroutine tests. - - To be used with `AioTestBase`-based tests""" - @functools.wraps(test_method) - def wrapper(self): - """Synchronously run test method in the event loop""" - self.loop.run_until_complete(test_method(self)) - return wrapper |