Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implemented async_unittest for writing tests for asynchronous code
Added logging call if main loop is exited unexpectedly
- Loading branch information
1 parent
f468d47
commit 2ff4fdc
Showing
4 changed files
with
49 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,42 @@ | ||
import asyncio | ||
from unittest import case, loader, mock, result, runner, signals, suite, util | ||
from unittest import * # import everything that IS defined in __all__ | ||
|
||
|
||
def AsyncMock(mock_func, *args, **kwargs): | ||
''' | ||
Returns an asynchronous function coroutine. The provided "mock_func" | ||
argument is the function to be called when the coroutine is run. | ||
All args and kwargs are passed to the mock when it is called. | ||
''' | ||
async def mock_coroutine(*args, **kwargs): | ||
return mock_func(*args, **kwargs) | ||
|
||
mock_coroutine.mock = mock_func | ||
return mock_coroutine | ||
|
||
|
||
def AsyncMagicMock(*args, **kwargs): | ||
''' | ||
Returns an asynchronous function coroutine. The returned coroutine | ||
contains a "mock" attribute, which is the MagicMock object to be called | ||
when the coroutine is run. All args and kwargs are passed to the mock. | ||
''' | ||
return AsyncMock(mock.MagicMock(*args, **kwargs), *args, **kwargs) | ||
|
||
|
||
class AsyncTestCase(TestCase): | ||
_loop = None | ||
|
||
@property | ||
def loop(self): | ||
if self._loop is None: | ||
self._loop = asyncio.get_event_loop() | ||
return self._loop | ||
|
||
@loop.setter | ||
def loop(self, new_val): | ||
self._loop = new_val | ||
|
||
def run_async(self, coroutine): | ||
return self.loop.run_until_complete(coroutine) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters