diff options
Diffstat (limited to 'mock/tests/testasync.py')
| -rw-r--r-- | mock/tests/testasync.py | 1074 |
1 files changed, 0 insertions, 1074 deletions
diff --git a/mock/tests/testasync.py b/mock/tests/testasync.py deleted file mode 100644 index 9fd2b65..0000000 --- a/mock/tests/testasync.py +++ /dev/null @@ -1,1074 +0,0 @@ - -import asyncio -import gc -import inspect -import re -import unittest -from contextlib import contextmanager - -from mock import (ANY, call, AsyncMock, patch, MagicMock, Mock, - create_autospec, sentinel) -from mock.backports import IsolatedAsyncioTestCase, iscoroutinefunction -from mock.mock import _CallList - - -try: - from asyncio import run -except ImportError: - def run(main): - loop = asyncio.new_event_loop() - try: - return_value = loop.run_until_complete(main) - finally: - loop.close() - return return_value - - -def tearDownModule(): - asyncio.set_event_loop_policy(None) - - -class AsyncClass: - def __init__(self): pass - async def async_method(self): pass - def normal_method(self): pass - - @classmethod - async def async_class_method(cls): pass - - @staticmethod - async def async_static_method(): pass - - -class AwaitableClass: - def __await__(self): yield - -async def async_func(): pass - -async def async_func_args(a, b, *, c): pass - -def normal_func(): pass - -class NormalClass(object): - def a(self): pass - - -async_foo_name = f'{__name__}.AsyncClass' -normal_foo_name = f'{__name__}.NormalClass' - - -@contextmanager -def assertNeverAwaited(test): - with test.assertWarnsRegex(RuntimeWarning, "was never awaited$"): - yield - # In non-CPython implementations of Python, this is needed because timely - # deallocation is not guaranteed by the garbage collector. - gc.collect() - - -class AsyncPatchDecoratorTest(unittest.TestCase): - def test_is_coroutine_function_patch(self): - @patch.object(AsyncClass, 'async_method') - def test_async(mock_method): - self.assertTrue(iscoroutinefunction(mock_method)) - test_async() - - def test_is_async_patch(self): - @patch.object(AsyncClass, 'async_method') - def test_async(mock_method): - m = mock_method() - self.assertTrue(inspect.isawaitable(m)) - run(m) - - @patch(f'{async_foo_name}.async_method') - def test_no_parent_attribute(mock_method): - m = mock_method() - self.assertTrue(inspect.isawaitable(m)) - run(m) - - test_async() - test_no_parent_attribute() - - def test_is_AsyncMock_patch(self): - @patch.object(AsyncClass, 'async_method') - def test_async(mock_method): - self.assertIsInstance(mock_method, AsyncMock) - - test_async() - - def test_is_AsyncMock_patch_staticmethod(self): - @patch.object(AsyncClass, 'async_static_method') - def test_async(mock_method): - self.assertIsInstance(mock_method, AsyncMock) - - test_async() - - def test_is_AsyncMock_patch_classmethod(self): - @patch.object(AsyncClass, 'async_class_method') - def test_async(mock_method): - self.assertIsInstance(mock_method, AsyncMock) - - test_async() - - def test_async_def_patch(self): - @patch(f"{__name__}.async_func", return_value=1) - @patch(f"{__name__}.async_func_args", return_value=2) - async def test_async(func_args_mock, func_mock): - self.assertEqual(func_args_mock._mock_name, "async_func_args") - self.assertEqual(func_mock._mock_name, "async_func") - - self.assertIsInstance(async_func, AsyncMock) - self.assertIsInstance(async_func_args, AsyncMock) - - self.assertEqual(await async_func(), 1) - self.assertEqual(await async_func_args(1, 2, c=3), 2) - - run(test_async()) - self.assertTrue(inspect.iscoroutinefunction(async_func)) - - -class AsyncPatchCMTest(unittest.TestCase): - def test_is_async_function_cm(self): - def test_async(): - with patch.object(AsyncClass, 'async_method') as mock_method: - self.assertTrue(iscoroutinefunction(mock_method)) - - test_async() - - def test_is_async_cm(self): - def test_async(): - with patch.object(AsyncClass, 'async_method') as mock_method: - m = mock_method() - self.assertTrue(inspect.isawaitable(m)) - run(m) - - test_async() - - def test_is_AsyncMock_cm(self): - def test_async(): - with patch.object(AsyncClass, 'async_method') as mock_method: - self.assertIsInstance(mock_method, AsyncMock) - - test_async() - - def test_async_def_cm(self): - async def test_async(): - with patch(f"{__name__}.async_func", AsyncMock()): - self.assertIsInstance(async_func, AsyncMock) - self.assertTrue(inspect.iscoroutinefunction(async_func)) - - run(test_async()) - - -class AsyncMockTest(unittest.TestCase): - def test_iscoroutinefunction_default(self): - mock = AsyncMock() - self.assertTrue(iscoroutinefunction(mock)) - - def test_iscoroutinefunction_function(self): - async def foo(): pass - mock = AsyncMock(foo) - self.assertTrue(iscoroutinefunction(mock)) - self.assertTrue(inspect.iscoroutinefunction(mock)) - - def test_isawaitable(self): - mock = AsyncMock() - m = mock() - self.assertTrue(inspect.isawaitable(m)) - run(m) - self.assertIn('assert_awaited', dir(mock)) - - def test_iscoroutinefunction_normal_function(self): - def foo(): pass - mock = AsyncMock(foo) - self.assertTrue(iscoroutinefunction(mock)) - self.assertTrue(inspect.iscoroutinefunction(mock)) - - def test_future_isfuture(self): - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - fut = asyncio.Future() - loop.stop() - loop.close() - mock = AsyncMock(fut) - self.assertIsInstance(mock, asyncio.Future) - - -class AsyncAutospecTest(unittest.TestCase): - def test_is_AsyncMock_patch(self): - @patch(async_foo_name, autospec=True) - def test_async(mock_method): - self.assertIsInstance(mock_method.async_method, AsyncMock) - self.assertIsInstance(mock_method, MagicMock) - - @patch(async_foo_name, autospec=True) - def test_normal_method(mock_method): - self.assertIsInstance(mock_method.normal_method, MagicMock) - - test_async() - test_normal_method() - - def test_create_autospec_instance(self): - with self.assertRaises(RuntimeError): - create_autospec(async_func, instance=True) - - def test_create_autospec_awaitable_class(self): - awaitable_mock = create_autospec(spec=AwaitableClass()) - self.assertIsInstance(create_autospec(awaitable_mock), AsyncMock) - - def test_create_autospec(self): - spec = create_autospec(async_func_args) - awaitable = spec(1, 2, c=3) - async def main(): - await awaitable - - self.assertEqual(spec.await_count, 0) - self.assertIsNone(spec.await_args) - self.assertEqual(spec.await_args_list, []) - spec.assert_not_awaited() - - run(main()) - - self.assertTrue(iscoroutinefunction(spec)) - self.assertTrue(asyncio.iscoroutine(awaitable)) - self.assertEqual(spec.await_count, 1) - self.assertEqual(spec.await_args, call(1, 2, c=3)) - self.assertEqual(spec.await_args_list, [call(1, 2, c=3)]) - spec.assert_awaited_once() - spec.assert_awaited_once_with(1, 2, c=3) - spec.assert_awaited_with(1, 2, c=3) - spec.assert_awaited() - - with self.assertRaises(AssertionError): - spec.assert_any_await(e=1) - - - def test_patch_with_autospec(self): - - async def test_async(): - with patch(f"{__name__}.async_func_args", autospec=True) as mock_method: - awaitable = mock_method(1, 2, c=3) - self.assertIsInstance(mock_method.mock, AsyncMock) - - self.assertTrue(iscoroutinefunction(mock_method)) - self.assertTrue(asyncio.iscoroutine(awaitable)) - self.assertTrue(inspect.isawaitable(awaitable)) - - # Verify the default values during mock setup - self.assertEqual(mock_method.await_count, 0) - self.assertEqual(mock_method.await_args_list, []) - self.assertIsNone(mock_method.await_args) - mock_method.assert_not_awaited() - - await awaitable - - self.assertEqual(mock_method.await_count, 1) - self.assertEqual(mock_method.await_args, call(1, 2, c=3)) - self.assertEqual(mock_method.await_args_list, [call(1, 2, c=3)]) - mock_method.assert_awaited_once() - mock_method.assert_awaited_once_with(1, 2, c=3) - mock_method.assert_awaited_with(1, 2, c=3) - mock_method.assert_awaited() - - mock_method.reset_mock() - self.assertEqual(mock_method.await_count, 0) - self.assertIsNone(mock_method.await_args) - self.assertEqual(mock_method.await_args_list, []) - - run(test_async()) - - -class AsyncSpecTest(unittest.TestCase): - def test_spec_normal_methods_on_class(self): - def inner_test(mock_type): - mock = mock_type(AsyncClass) - self.assertIsInstance(mock.async_method, AsyncMock) - self.assertIsInstance(mock.normal_method, MagicMock) - - for mock_type in [AsyncMock, MagicMock]: - with self.subTest(f"test method types with {mock_type}"): - inner_test(mock_type) - - def test_spec_normal_methods_on_class_with_mock(self): - mock = Mock(AsyncClass) - self.assertIsInstance(mock.async_method, AsyncMock) - self.assertIsInstance(mock.normal_method, Mock) - - def test_spec_mock_type_kw(self): - def inner_test(mock_type): - async_mock = mock_type(spec=async_func) - self.assertIsInstance(async_mock, mock_type) - with assertNeverAwaited(self): - self.assertTrue(inspect.isawaitable(async_mock())) - - sync_mock = mock_type(spec=normal_func) - self.assertIsInstance(sync_mock, mock_type) - - for mock_type in [AsyncMock, MagicMock, Mock]: - with self.subTest(f"test spec kwarg with {mock_type}"): - inner_test(mock_type) - - def test_spec_mock_type_positional(self): - def inner_test(mock_type): - async_mock = mock_type(async_func) - self.assertIsInstance(async_mock, mock_type) - with assertNeverAwaited(self): - self.assertTrue(inspect.isawaitable(async_mock())) - - sync_mock = mock_type(normal_func) - self.assertIsInstance(sync_mock, mock_type) - - for mock_type in [AsyncMock, MagicMock, Mock]: - with self.subTest(f"test spec positional with {mock_type}"): - inner_test(mock_type) - - def test_spec_as_normal_kw_AsyncMock(self): - mock = AsyncMock(spec=normal_func) - self.assertIsInstance(mock, AsyncMock) - m = mock() - self.assertTrue(inspect.isawaitable(m)) - run(m) - - def test_spec_as_normal_positional_AsyncMock(self): - mock = AsyncMock(normal_func) - self.assertIsInstance(mock, AsyncMock) - m = mock() - self.assertTrue(inspect.isawaitable(m)) - run(m) - - def test_spec_async_mock(self): - @patch.object(AsyncClass, 'async_method', spec=True) - def test_async(mock_method): - self.assertIsInstance(mock_method, AsyncMock) - - test_async() - - def test_spec_parent_not_async_attribute_is(self): - @patch(async_foo_name, spec=True) - def test_async(mock_method): - self.assertIsInstance(mock_method, MagicMock) - self.assertIsInstance(mock_method.async_method, AsyncMock) - - test_async() - - def test_target_async_spec_not(self): - @patch.object(AsyncClass, 'async_method', spec=NormalClass.a) - def test_async_attribute(mock_method): - self.assertIsInstance(mock_method, MagicMock) - self.assertFalse(inspect.iscoroutine(mock_method)) - self.assertFalse(inspect.isawaitable(mock_method)) - - test_async_attribute() - - def test_target_not_async_spec_is(self): - @patch.object(NormalClass, 'a', spec=async_func) - def test_attribute_not_async_spec_is(mock_async_func): - self.assertIsInstance(mock_async_func, AsyncMock) - test_attribute_not_async_spec_is() - - def test_spec_async_attributes(self): - @patch(normal_foo_name, spec=AsyncClass) - def test_async_attributes_coroutines(MockNormalClass): - self.assertIsInstance(MockNormalClass.async_method, AsyncMock) - self.assertIsInstance(MockNormalClass, MagicMock) - - test_async_attributes_coroutines() - - -class AsyncSpecSetTest(unittest.TestCase): - def test_is_AsyncMock_patch(self): - @patch.object(AsyncClass, 'async_method', spec_set=True) - def test_async(async_method): - self.assertIsInstance(async_method, AsyncMock) - test_async() - - def test_is_async_AsyncMock(self): - mock = AsyncMock(spec_set=AsyncClass.async_method) - self.assertTrue(iscoroutinefunction(mock)) - self.assertIsInstance(mock, AsyncMock) - - def test_is_child_AsyncMock(self): - mock = MagicMock(spec_set=AsyncClass) - self.assertTrue(iscoroutinefunction(mock.async_method)) - self.assertFalse(iscoroutinefunction(mock.normal_method)) - self.assertIsInstance(mock.async_method, AsyncMock) - self.assertIsInstance(mock.normal_method, MagicMock) - self.assertIsInstance(mock, MagicMock) - - def test_magicmock_lambda_spec(self): - mock_obj = MagicMock() - mock_obj.mock_func = MagicMock(spec=lambda x: x) - - with patch.object(mock_obj, "mock_func") as cm: - self.assertIsInstance(cm, MagicMock) - - -class AsyncArguments(IsolatedAsyncioTestCase): - async def test_add_return_value(self): - async def addition(self, var): pass - - mock = AsyncMock(addition, return_value=10) - output = await mock(5) - - self.assertEqual(output, 10) - - async def test_add_side_effect_exception(self): - async def addition(var): pass - mock = AsyncMock(addition, side_effect=Exception('err')) - with self.assertRaises(Exception): - await mock(5) - - async def test_add_side_effect_coroutine(self): - async def addition(var): - return var + 1 - mock = AsyncMock(side_effect=addition) - result = await mock(5) - self.assertEqual(result, 6) - - async def test_add_side_effect_normal_function(self): - def addition(var): - return var + 1 - mock = AsyncMock(side_effect=addition) - result = await mock(5) - self.assertEqual(result, 6) - - async def test_add_side_effect_iterable(self): - vals = [1, 2, 3] - mock = AsyncMock(side_effect=vals) - for item in vals: - self.assertEqual(await mock(), item) - - with self.assertRaises(StopAsyncIteration) as e: - await mock() - - async def test_add_side_effect_exception_iterable(self): - class SampleException(Exception): - pass - - vals = [1, SampleException("foo")] - mock = AsyncMock(side_effect=vals) - self.assertEqual(await mock(), 1) - - with self.assertRaises(SampleException) as e: - await mock() - - async def test_return_value_AsyncMock(self): - value = AsyncMock(return_value=10) - mock = AsyncMock(return_value=value) - result = await mock() - self.assertIs(result, value) - - async def test_return_value_awaitable(self): - fut = asyncio.Future() - fut.set_result(None) - mock = AsyncMock(return_value=fut) - result = await mock() - self.assertIsInstance(result, asyncio.Future) - - async def test_side_effect_awaitable_values(self): - fut = asyncio.Future() - fut.set_result(None) - - mock = AsyncMock(side_effect=[fut]) - result = await mock() - self.assertIsInstance(result, asyncio.Future) - - with self.assertRaises(StopAsyncIteration): - await mock() - - async def test_side_effect_is_AsyncMock(self): - effect = AsyncMock(return_value=10) - mock = AsyncMock(side_effect=effect) - - result = await mock() - self.assertEqual(result, 10) - - async def test_wraps_coroutine(self): - value = asyncio.Future() - - ran = False - async def inner(): - nonlocal ran - ran = True - return value - - mock = AsyncMock(wraps=inner) - result = await mock() - self.assertEqual(result, value) - mock.assert_awaited() - self.assertTrue(ran) - - async def test_wraps_normal_function(self): - value = 1 - - ran = False - def inner(): - nonlocal ran - ran = True - return value - - mock = AsyncMock(wraps=inner) - result = await mock() - self.assertEqual(result, value) - mock.assert_awaited() - self.assertTrue(ran) - - async def test_await_args_list_order(self): - async_mock = AsyncMock() - mock2 = async_mock(2) - mock1 = async_mock(1) - await mock1 - await mock2 - async_mock.assert_has_awaits([call(1), call(2)]) - self.assertEqual(async_mock.await_args_list, [call(1), call(2)]) - self.assertEqual(async_mock.call_args_list, [call(2), call(1)]) - - -class AsyncMagicMethods(unittest.TestCase): - def test_async_magic_methods_return_async_mocks(self): - m_mock = MagicMock() - self.assertIsInstance(m_mock.__aenter__, AsyncMock) - self.assertIsInstance(m_mock.__aexit__, AsyncMock) - self.assertIsInstance(m_mock.__anext__, AsyncMock) - # __aiter__ is actually a synchronous object - # so should return a MagicMock - self.assertIsInstance(m_mock.__aiter__, MagicMock) - - def test_sync_magic_methods_return_magic_mocks(self): - a_mock = AsyncMock() - self.assertIsInstance(a_mock.__enter__, MagicMock) - self.assertIsInstance(a_mock.__exit__, MagicMock) - self.assertIsInstance(a_mock.__next__, MagicMock) - self.assertIsInstance(a_mock.__len__, MagicMock) - - def test_magicmock_has_async_magic_methods(self): - m_mock = MagicMock() - self.assertTrue(hasattr(m_mock, "__aenter__")) - self.assertTrue(hasattr(m_mock, "__aexit__")) - self.assertTrue(hasattr(m_mock, "__anext__")) - - def test_asyncmock_has_sync_magic_methods(self): - a_mock = AsyncMock() - self.assertTrue(hasattr(a_mock, "__enter__")) - self.assertTrue(hasattr(a_mock, "__exit__")) - self.assertTrue(hasattr(a_mock, "__next__")) - self.assertTrue(hasattr(a_mock, "__len__")) - - def test_magic_methods_are_async_functions(self): - m_mock = MagicMock() - self.assertIsInstance(m_mock.__aenter__, AsyncMock) - self.assertIsInstance(m_mock.__aexit__, AsyncMock) - # AsyncMocks are also coroutine functions - self.assertTrue(iscoroutinefunction(m_mock.__aenter__)) - self.assertTrue(iscoroutinefunction(m_mock.__aexit__)) - -class AsyncContextManagerTest(unittest.TestCase): - - class WithAsyncContextManager: - async def __aenter__(self, *args, **kwargs): pass - - async def __aexit__(self, *args, **kwargs): pass - - class WithSyncContextManager: - def __enter__(self, *args, **kwargs): pass - - def __exit__(self, *args, **kwargs): pass - - class ProductionCode: - # Example real-world(ish) code - def __init__(self): - self.session = None - - async def main(self): - async with self.session.post('https://python.org') as response: - val = await response.json() - return val - - def test_set_return_value_of_aenter(self): - def inner_test(mock_type): - pc = self.ProductionCode() - pc.session = MagicMock(name='sessionmock') - cm = mock_type(name='magic_cm') - response = AsyncMock(name='response') - response.json = AsyncMock(return_value={'json': 123}) - cm.__aenter__.return_value = response - pc.session.post.return_value = cm - result = run(pc.main()) - self.assertEqual(result, {'json': 123}) - - for mock_type in [AsyncMock, MagicMock]: - with self.subTest(f"test set return value of aenter with {mock_type}"): - inner_test(mock_type) - - def test_mock_supports_async_context_manager(self): - - def inner_test(mock_type): - called = False - cm = self.WithAsyncContextManager() - cm_mock = mock_type(cm) - - async def use_context_manager(): - nonlocal called - async with cm_mock as result: - called = True - return result - - cm_result = run(use_context_manager()) - self.assertTrue(called) - self.assertTrue(cm_mock.__aenter__.called) - self.assertTrue(cm_mock.__aexit__.called) - cm_mock.__aenter__.assert_awaited() - cm_mock.__aexit__.assert_awaited() - # We mock __aenter__ so it does not return self - self.assertIsNot(cm_mock, cm_result) - - for mock_type in [AsyncMock, MagicMock]: - with self.subTest(f"test context manager magics with {mock_type}"): - inner_test(mock_type) - - def test_mock_customize_async_context_manager(self): - instance = self.WithAsyncContextManager() - mock_instance = MagicMock(instance) - - expected_result = object() - mock_instance.__aenter__.return_value = expected_result - - async def use_context_manager(): - async with mock_instance as result: - return result - - self.assertIs(run(use_context_manager()), expected_result) - - def test_mock_customize_async_context_manager_with_coroutine(self): - enter_called = False - exit_called = False - - async def enter_coroutine(*args): - nonlocal enter_called - enter_called = True - - async def exit_coroutine(*args): - nonlocal exit_called - exit_called = True - - instance = self.WithAsyncContextManager() - mock_instance = MagicMock(instance) - - mock_instance.__aenter__ = enter_coroutine - mock_instance.__aexit__ = exit_coroutine - - async def use_context_manager(): - async with mock_instance: - pass - - run(use_context_manager()) - self.assertTrue(enter_called) - self.assertTrue(exit_called) - - def test_context_manager_raise_exception_by_default(self): - async def raise_in(context_manager): - async with context_manager: - raise TypeError() - - instance = self.WithAsyncContextManager() - mock_instance = MagicMock(instance) - with self.assertRaises(TypeError): - run(raise_in(mock_instance)) - - -class AsyncIteratorTest(unittest.TestCase): - class WithAsyncIterator(object): - def __init__(self): - self.items = ["foo", "NormalFoo", "baz"] - - def __aiter__(self): pass - - async def __anext__(self): pass - - def test_aiter_set_return_value(self): - mock_iter = AsyncMock(name="tester") - mock_iter.__aiter__.return_value = [1, 2, 3] - async def main(): - return [i async for i in mock_iter] - result = run(main()) - self.assertEqual(result, [1, 2, 3]) - - def test_mock_aiter_and_anext_asyncmock(self): - def inner_test(mock_type): - instance = self.WithAsyncIterator() - mock_instance = mock_type(instance) - # Check that the mock and the real thing bahave the same - # __aiter__ is not actually async, so not a coroutinefunction - self.assertFalse(iscoroutinefunction(instance.__aiter__)) - self.assertFalse(iscoroutinefunction(mock_instance.__aiter__)) - # __anext__ is async - self.assertTrue(iscoroutinefunction(instance.__anext__)) - self.assertTrue(iscoroutinefunction(mock_instance.__anext__)) - - for mock_type in [AsyncMock, MagicMock]: - with self.subTest(f"test aiter and anext corourtine with {mock_type}"): - inner_test(mock_type) - - - def test_mock_async_for(self): - async def iterate(iterator): - accumulator = [] - async for item in iterator: - accumulator.append(item) - - return accumulator - - expected = ["FOO", "BAR", "BAZ"] - def test_default(mock_type): - mock_instance = mock_type(self.WithAsyncIterator()) - self.assertEqual(run(iterate(mock_instance)), []) - - - def test_set_return_value(mock_type): - mock_instance = mock_type(self.WithAsyncIterator()) - mock_instance.__aiter__.return_value = expected[:] - self.assertEqual(run(iterate(mock_instance)), expected) - - def test_set_return_value_iter(mock_type): - mock_instance = mock_type(self.WithAsyncIterator()) - mock_instance.__aiter__.return_value = iter(expected[:]) - self.assertEqual(run(iterate(mock_instance)), expected) - - for mock_type in [AsyncMock, MagicMock]: - with self.subTest(f"default value with {mock_type}"): - test_default(mock_type) - - with self.subTest(f"set return_value with {mock_type}"): - test_set_return_value(mock_type) - - with self.subTest(f"set return_value iterator with {mock_type}"): - test_set_return_value_iter(mock_type) - - -class AsyncMockAssert(unittest.TestCase): - def setUp(self): - self.mock = AsyncMock() - - async def _runnable_test(self, *args, **kwargs): - await self.mock(*args, **kwargs) - - async def _await_coroutine(self, coroutine): - return await coroutine - - def test_assert_called_but_not_awaited(self): - mock = AsyncMock(AsyncClass) - with assertNeverAwaited(self): - mock.async_method() - self.assertTrue(iscoroutinefunction(mock.async_method)) - mock.async_method.assert_called() - mock.async_method.assert_called_once() - mock.async_method.assert_called_once_with() - with self.assertRaises(AssertionError): - mock.assert_awaited() - with self.assertRaises(AssertionError): - mock.async_method.assert_awaited() - - def test_assert_called_then_awaited(self): - mock = AsyncMock(AsyncClass) - mock_coroutine = mock.async_method() - mock.async_method.assert_called() - mock.async_method.assert_called_once() - mock.async_method.assert_called_once_with() - with self.assertRaises(AssertionError): - mock.async_method.assert_awaited() - - run(self._await_coroutine(mock_coroutine)) - # Assert we haven't re-called the function - mock.async_method.assert_called_once() - mock.async_method.assert_awaited() - mock.async_method.assert_awaited_once() - mock.async_method.assert_awaited_once_with() - - def test_assert_called_and_awaited_at_same_time(self): - with self.assertRaises(AssertionError): - self.mock.assert_awaited() - - with self.assertRaises(AssertionError): - self.mock.assert_called() - - run(self._runnable_test()) - self.mock.assert_called_once() - self.mock.assert_awaited_once() - - def test_assert_called_twice_and_awaited_once(self): - mock = AsyncMock(AsyncClass) - coroutine = mock.async_method() - # The first call will be awaited so no warning there - # But this call will never get awaited, so it will warn here - with assertNeverAwaited(self): - mock.async_method() - with self.assertRaises(AssertionError): - mock.async_method.assert_awaited() - mock.async_method.assert_called() - run(self._await_coroutine(coroutine)) - mock.async_method.assert_awaited() - mock.async_method.assert_awaited_once() - - def test_assert_called_once_and_awaited_twice(self): - mock = AsyncMock(AsyncClass) - coroutine = mock.async_method() - mock.async_method.assert_called_once() - run(self._await_coroutine(coroutine)) - with self.assertRaises(RuntimeError): - # Cannot reuse already awaited coroutine - run(self._await_coroutine(coroutine)) - mock.async_method.assert_awaited() - - def test_assert_awaited_but_not_called(self): - with self.assertRaises(AssertionError): - self.mock.assert_awaited() - with self.assertRaises(AssertionError): - self.mock.assert_called() - with self.assertRaises(TypeError): - # You cannot await an AsyncMock, it must be a coroutine - run(self._await_coroutine(self.mock)) - - with self.assertRaises(AssertionError): - self.mock.assert_awaited() - with self.assertRaises(AssertionError): - self.mock.assert_called() - - def test_assert_has_calls_not_awaits(self): - kalls = [call('foo')] - with assertNeverAwaited(self): - self.mock('foo') - self.mock.assert_has_calls(kalls) - with self.assertRaises(AssertionError): - self.mock.assert_has_awaits(kalls) - - def test_assert_has_mock_calls_on_async_mock_no_spec(self): - with assertNeverAwaited(self): - self.mock() - kalls_empty = [('', (), {})] - self.assertEqual(self.mock.mock_calls, kalls_empty) - - with assertNeverAwaited(self): - self.mock('foo') - with assertNeverAwaited(self): - self.mock('baz') - mock_kalls = ([call(), call('foo'), call('baz')]) - self.assertEqual(self.mock.mock_calls, mock_kalls) - - def test_assert_has_mock_calls_on_async_mock_with_spec(self): - a_class_mock = AsyncMock(AsyncClass) - with assertNeverAwaited(self): - a_class_mock.async_method() - kalls_empty = [('', (), {})] - self.assertEqual(a_class_mock.async_method.mock_calls, kalls_empty) - self.assertEqual(a_class_mock.mock_calls, [call.async_method()]) - - with assertNeverAwaited(self): - a_class_mock.async_method(1, 2, 3, a=4, b=5) - method_kalls = [call(), call(1, 2, 3, a=4, b=5)] - mock_kalls = [call.async_method(), call.async_method(1, 2, 3, a=4, b=5)] - self.assertEqual(a_class_mock.async_method.mock_calls, method_kalls) - self.assertEqual(a_class_mock.mock_calls, mock_kalls) - - def test_async_method_calls_recorded(self): - with assertNeverAwaited(self): - self.mock.something(3, fish=None) - with assertNeverAwaited(self): - self.mock.something_else.something(6, cake=sentinel.Cake) - - self.assertEqual(self.mock.method_calls, [ - ("something", (3,), {'fish': None}), - ("something_else.something", (6,), {'cake': sentinel.Cake}) - ], - "method calls not recorded correctly") - self.assertEqual(self.mock.something_else.method_calls, - [("something", (6,), {'cake': sentinel.Cake})], - "method calls not recorded correctly") - - def test_async_arg_lists(self): - def assert_attrs(mock): - names = ('call_args_list', 'method_calls', 'mock_calls') - for name in names: - attr = getattr(mock, name) - self.assertIsInstance(attr, _CallList) - self.assertIsInstance(attr, list) - self.assertEqual(attr, []) - - assert_attrs(self.mock) - with assertNeverAwaited(self): - self.mock() - with assertNeverAwaited(self): - self.mock(1, 2) - with assertNeverAwaited(self): - self.mock(a=3) - - self.mock.reset_mock() - assert_attrs(self.mock) - - a_mock = AsyncMock(AsyncClass) - with assertNeverAwaited(self): - a_mock.async_method() - with assertNeverAwaited(self): - a_mock.async_method(1, a=3) - - a_mock.reset_mock() - assert_attrs(a_mock) - - def test_assert_awaited(self): - with self.assertRaises(AssertionError): - self.mock.assert_awaited() - - run(self._runnable_test()) - self.mock.assert_awaited() - - def test_assert_awaited_once(self): - with self.assertRaises(AssertionError): - self.mock.assert_awaited_once() - - run(self._runnable_test()) - self.mock.assert_awaited_once() - - run(self._runnable_test()) - with self.assertRaises(AssertionError): - self.mock.assert_awaited_once() - - def test_assert_awaited_with(self): - msg = 'Not awaited' - with self.assertRaisesRegex(AssertionError, msg): - self.mock.assert_awaited_with('foo') - - run(self._runnable_test()) - msg = 'expected await not found' - with self.assertRaisesRegex(AssertionError, msg): - self.mock.assert_awaited_with('foo') - - run(self._runnable_test('foo')) - self.mock.assert_awaited_with('foo') - - run(self._runnable_test('SomethingElse')) - with self.assertRaises(AssertionError): - self.mock.assert_awaited_with('foo') - - def test_assert_awaited_once_with(self): - with self.assertRaises(AssertionError): - self.mock.assert_awaited_once_with('foo') - - run(self._runnable_test('foo')) - self.mock.assert_awaited_once_with('foo') - - run(self._runnable_test('foo')) - with self.assertRaises(AssertionError): - self.mock.assert_awaited_once_with('foo') - - def test_assert_any_wait(self): - with self.assertRaises(AssertionError): - self.mock.assert_any_await('foo') - - run(self._runnable_test('baz')) - with self.assertRaises(AssertionError): - self.mock.assert_any_await('foo') - - run(self._runnable_test('foo')) - self.mock.assert_any_await('foo') - - run(self._runnable_test('SomethingElse')) - self.mock.assert_any_await('foo') - - def test_assert_has_awaits_no_order(self): - calls = [call('foo'), call('baz')] - - with self.assertRaises(AssertionError) as cm: - self.mock.assert_has_awaits(calls) - self.assertEqual(len(cm.exception.args), 1) - - run(self._runnable_test('foo')) - with self.assertRaises(AssertionError): - self.mock.assert_has_awaits(calls) - - run(self._runnable_test('foo')) - with self.assertRaises(AssertionError): - self.mock.assert_has_awaits(calls) - - run(self._runnable_test('baz')) - self.mock.assert_has_awaits(calls) - - run(self._runnable_test('SomethingElse')) - self.mock.assert_has_awaits(calls) - - def test_awaits_asserts_with_any(self): - class Foo: - def __eq__(self, other): pass - - run(self._runnable_test(Foo(), 1)) - - self.mock.assert_has_awaits([call(ANY, 1)]) - self.mock.assert_awaited_with(ANY, 1) - self.mock.assert_any_await(ANY, 1) - - def test_awaits_asserts_with_spec_and_any(self): - class Foo: - def __eq__(self, other): pass - - mock_with_spec = AsyncMock(spec=Foo) - - async def _custom_mock_runnable_test(*args): - await mock_with_spec(*args) - - run(_custom_mock_runnable_test(Foo(), 1)) - mock_with_spec.assert_has_awaits([call(ANY, 1)]) - mock_with_spec.assert_awaited_with(ANY, 1) - mock_with_spec.assert_any_await(ANY, 1) - - def test_assert_has_awaits_ordered(self): - calls = [call('foo'), call('baz')] - with self.assertRaises(AssertionError): - self.mock.assert_has_awaits(calls, any_order=True) - - run(self._runnable_test('baz')) - with self.assertRaises(AssertionError): - self.mock.assert_has_awaits(calls, any_order=True) - - run(self._runnable_test('bamf')) - with self.assertRaises(AssertionError): - self.mock.assert_has_awaits(calls, any_order=True) - - run(self._runnable_test('foo')) - self.mock.assert_has_awaits(calls, any_order=True) - - run(self._runnable_test('qux')) - self.mock.assert_has_awaits(calls, any_order=True) - - def test_assert_not_awaited(self): - self.mock.assert_not_awaited() - - run(self._runnable_test()) - with self.assertRaises(AssertionError): - self.mock.assert_not_awaited() - - def test_assert_has_awaits_not_matching_spec_error(self): - async def f(x=None): pass - - self.mock = AsyncMock(spec=f) - run(self._runnable_test(1)) - - with self.assertRaisesRegex( - AssertionError, - '^{}$'.format( - re.escape('Awaits not found.\n' - 'Expected: [call()]\n' - 'Actual: [call(1)]'))) as cm: - self.mock.assert_has_awaits([call()]) - self.assertIsNone(cm.exception.__cause__) - - with self.assertRaisesRegex( - AssertionError, - '^{}$'.format( - re.escape( - 'Error processing expected awaits.\n' - "Errors: [None, TypeError('too many positional " - "arguments')]\n" - 'Expected: [call(), call(1, 2)]\n' - 'Actual: [call(1)]').replace( - "arguments\\'", "arguments\\',?") - )) as cm: - self.mock.assert_has_awaits([call(), call(1, 2)]) - self.assertIsInstance(cm.exception.__cause__, TypeError) |