test_api_callable.py 文件源码

def test_wrap_value_error(self):
        from google.gax.errors import InvalidArgumentError

        invalid_attribute_exc = grpc.RpcError()
        invalid_attribute_exc.code = lambda: grpc.StatusCode.INVALID_ARGUMENT

        def value_error_func(*dummy_args, **dummy_kwargs):
            raise invalid_attribute_exc

        value_error_callable = api_callable.create_api_call(
            value_error_func, _CallSettings())

        with self.assertRaises(ValueError) as exc_info:
            value_error_callable(None)

        self.assertIsInstance(exc_info.exception, InvalidArgumentError)
        self.assertEqual(exc_info.exception.args, (u'RPC failed',))
        self.assertIs(exc_info.exception.cause, invalid_attribute_exc)