run(main())
self.assertTrue(iscoroutinefunction(spec))
+ self.assertTrue(inspect.iscoroutinefunction(spec))
self.assertTrue(asyncio.iscoroutine(awaitable))
+ self.assertTrue(inspect.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)])
with self.assertRaises(AssertionError):
spec.assert_any_await(e=1)
+ def test_autospec_checks_signature(self):
+ spec = create_autospec(async_func_args)
+ # signature is not checked when called
+ awaitable = spec()
+ self.assertListEqual(spec.mock_calls, [])
+
+ async def main():
+ await awaitable
+
+ # but it is checked when awaited
+ with self.assertRaises(TypeError):
+ run(main())
+
+ # _checksig_ raises before running or awaiting the mock
+ self.assertListEqual(spec.mock_calls, [])
+ self.assertEqual(spec.await_count, 0)
+ self.assertIsNone(spec.await_args)
+ self.assertEqual(spec.await_args_list, [])
+ spec.assert_not_awaited()
def test_patch_with_autospec(self):
self.assertIsInstance(mock_method.mock, AsyncMock)
self.assertTrue(iscoroutinefunction(mock_method))
+ self.assertTrue(inspect.iscoroutinefunction(mock_method))
self.assertTrue(asyncio.iscoroutine(awaitable))
+ self.assertTrue(inspect.iscoroutine(awaitable))
self.assertTrue(inspect.isawaitable(awaitable))
# Verify the default values during mock setup
_setup_func(funcopy, mock, sig)
return funcopy
+def _set_async_signature(mock, original, instance=False, is_async_mock=False):
+ # creates an async function with signature (*args, **kwargs) that delegates to a
+ # mock. It still does signature checking by calling a lambda with the same
+ # signature as the original.
+
+ skipfirst = isinstance(original, type)
+ result = _get_signature_object(original, instance, skipfirst)
+ if result is None:
+ return mock
+ func, sig = result
+ def checksig(*args, **kwargs):
+ sig.bind(*args, **kwargs)
+ _copy_func_details(func, checksig)
+
+ name = original.__name__
+ if not name.isidentifier():
+ name = 'funcopy'
+ context = {'_checksig_': checksig, 'mock': mock}
+ src = """async def %s(*args, **kwargs):
+ _checksig_(*args, **kwargs)
+ return await mock(*args, **kwargs)""" % name
+ exec (src, context)
+ funcopy = context[name]
+ _setup_func(funcopy, mock, sig)
+ _setup_async_mock(funcopy)
+ return funcopy
+
def _setup_func(funcopy, mock, sig):
funcopy.mock = mock
if isinstance(spec, FunctionTypes):
# should only happen at the top level because we don't
# recurse for functions
- mock = _set_signature(mock, spec)
if is_async_func:
- _setup_async_mock(mock)
+ mock = _set_async_signature(mock, spec)
+ else:
+ mock = _set_signature(mock, spec)
else:
_check_signature(spec, mock, is_type, instance)