Skip to content

Instantly share code, notes, and snippets.

@ly0
Created December 4, 2015 08:35
Show Gist options
  • Select an option

  • Save ly0/b3be4a5b7708a9a8c3da to your computer and use it in GitHub Desktop.

Select an option

Save ly0/b3be4a5b7708a9a8c3da to your computer and use it in GitHub Desktop.

Revisions

  1. ly0 created this gist Dec 4, 2015.
    59 changes: 59 additions & 0 deletions asyncio_unittest.py
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,59 @@
    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()