asyncio unittest

asyncio_unittest.py

import unittest
import asyncio
import inspect


def async_test(f):
    def wrapper(*args, **kwargs):
        if inspect.iscoroutinefunction(f):
            future = f(*args, **kwargs)
        else:
            coroutine = asyncio.coroutine(f)
            future = coroutine(*args, **kwargs)
        asyncio.get_event_loop().run_until_complete(future)
    return wrapper


class TestExample(unittest.TestCase):

    def test_upper(self):
        self.assertEqual('foo'.upper(), 'FOO')

    def test_isupper(self):
        self.assertTrue('FOO'.isupper())
        self.assertFalse('Foo'.isupper())

    def test_split(self):
        s = 'hello world'
        self.assertEqual(s.split(), ['hello', 'world'])
        # check that s.split fails when the separator is not a string
        with self.assertRaises(TypeError):
            s.split(2)

    # asyncio test normal function with yield from statements
    @async_test
    def test_tcp1_success(self):
        test = yield from asyncio.open_connection('www.baidu.com', 80)
        print('test_tcp1_success')

    # asyncio test coroutine function with yield from statements
    @async_test
    @asyncio.coroutine
    def test_tcp2_success(self):
        test = yield from asyncio.open_connection('www.baidu.com', 80)
        print('test_tcp2_success')

    # asyncio test with await keywords
    @async_test
    async def test_tcp3_success(self):
        test = await asyncio.open_connection('www.baidu.com', 80)
        print('test_tcp3_success')

    # asyncio test with await keywords
    @async_test
    async def test_tcp3_fail(self):
        test = await asyncio.open_connection('www.baidu.com', 80)
        self.assertTrue(False, 'test_tcp3_fail')

unittest.main()