Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_awaited_wait(self):
mock = asynctest.mock.CoroutineFunctionMock()
t = asyncio.ensure_future(mock.awaited.wait())
yield from mock()
yield from mock()
yield from t
mock.reset_mock()
t = asyncio.ensure_future(mock.awaited.wait(skip=1))
yield from mock()
self.assertFalse(t.done())
yield from mock()
yield from t
def test_patch_as_context_manager_uses_CoroutineFunctionMock_on_coroutine_function(self):
with asynctest.mock.patch('test.test_mock.Test.a_coroutine') as mock:
import test.test_mock
self.assertIs(test.test_mock.Test.a_coroutine, mock)
self.assertIsInstance(mock, asynctest.mock.CoroutineFunctionMock)
if not isinstance(original, unittest.mock.FunctionTypes):
new = unittest.mock._SpecState(original, spec_set, mock, entry,
instance)
mock._mock_children[entry] = new
else:
parent = mock
if isinstance(spec, unittest.mock.FunctionTypes):
parent = mock.mock
skipfirst = unittest.mock._must_skip(spec, entry, is_type)
kwargs['_eat_self'] = skipfirst
if asyncio.iscoroutinefunction(original):
child_klass = CoroutineFunctionMock
else:
child_klass = MagicMock
new = child_klass(parent=parent, name=entry, _new_name=entry,
_new_parent=parent, **kwargs)
mock._mock_children[entry] = new
unittest.mock._check_signature(original, new, skipfirst=skipfirst)
if isinstance(new, unittest.mock.FunctionTypes):
setattr(mock, entry, new)
return mock
def _get_child_mock(self, *args, **kwargs):
_new_name = kwargs.get("_new_name")
if _new_name in self.__dict__['_spec_coroutines']:
return CoroutineFunctionMock(*args, **kwargs)
_type = type(self)
if issubclass(_type, MagicMock) and _new_name in async_magic_coroutines:
klass = CoroutineFunctionMock
elif issubclass(_type, CoroutineFunctionMock):
klass = MagicMock
elif not issubclass(_type, unittest.mock.CallableMixin):
if issubclass(_type, unittest.mock.NonCallableMagicMock):
klass = MagicMock
elif issubclass(_type, NonCallableMock):
klass = Mock
else:
klass = _type.__mro__[1]
return klass(*args, **kwargs)
def test_await_args(self):
with self.subTest('in order'):
mock = asynctest.mock.CoroutineFunctionMock()
t1 = mock('foo')
t2 = mock('bar')
yield from t1
yield from t2
self.assertEqual(mock.await_args, asynctest.call('bar'))
with self.subTest('out of order'):
mock = asynctest.mock.CoroutineFunctionMock()
t1 = mock('foo')
t2 = mock('bar')
yield from t2
yield from t1
self.assertEqual(mock.await_args, asynctest.call('foo'))
def test_assert_awaited(self):
mock = asynctest.mock.CoroutineFunctionMock()
with self.assertRaises(AssertionError):
mock.assert_awaited()
yield from mock()
mock.assert_awaited()
def test_assert_awaited_with(self):
mock = asynctest.mock.CoroutineFunctionMock()
with self.assertRaises(AssertionError):
mock.assert_awaited_with('foo')
yield from mock('foo')
mock.assert_awaited_with('foo')
yield from mock('bar')
with self.assertRaises(AssertionError):
mock.assert_awaited_with('foo')
yield from mock('foo')
with self.assertRaises(AssertionError):
mock.assert_has_awaits(calls)
yield from mock('bar')
with self.assertRaises(AssertionError):
mock.assert_has_awaits(calls)
yield from mock('baz')
mock.assert_has_awaits(calls)
yield from mock('qux')
mock.assert_has_awaits(calls)
with self.subTest('any_order=True'):
mock = asynctest.mock.CoroutineFunctionMock()
with self.assertRaises(AssertionError):
mock.assert_has_awaits(calls, any_order=True)
yield from mock('baz')
with self.assertRaises(AssertionError):
mock.assert_has_awaits(calls, any_order=True)
yield from mock('foo')
with self.assertRaises(AssertionError):
mock.assert_has_awaits(calls, any_order=True)
yield from mock('bar')
mock.assert_has_awaits(calls, any_order=True)
yield from mock('qux')
def test_asyncio_iscoroutinefunction(self):
mock = asynctest.mock.CoroutineFunctionMock()
self.assertTrue(asyncio.iscoroutinefunction(mock))
_type = type(self)
if issubclass(_type, MagicMock) and _new_name in async_magic_coroutines:
klass = CoroutineFunctionMock
elif issubclass(_type, CoroutineFunctionMock):
klass = MagicMock
elif not issubclass(_type, unittest.mock.CallableMixin):
if issubclass(_type, unittest.mock.NonCallableMagicMock):
klass = MagicMock
elif issubclass(_type, NonCallableMock):
klass = Mock
else:
klass = _type.__mro__[1]
return klass(*args, **kwargs)