_rpc_test_helpers.py File Reference

Go to the source code of this file.

Classes
class	tests.unit._rpc_test_helpers._GenericHandler
class	tests.unit._rpc_test_helpers._Handler
class	tests.unit._rpc_test_helpers._MethodHandler
class	tests.unit._rpc_test_helpers.BaseRPCTest
class	tests.unit._rpc_test_helpers.Callback

Namespaces
	tests.unit._rpc_test_helpers

Functions
def	tests.unit._rpc_test_helpers.stream_stream_multi_callable (channel)
def	tests.unit._rpc_test_helpers.stream_stream_non_blocking_multi_callable (channel)
def	tests.unit._rpc_test_helpers.stream_unary_multi_callable (channel)
def	tests.unit._rpc_test_helpers.unary_stream_multi_callable (channel)
def	tests.unit._rpc_test_helpers.unary_stream_non_blocking_multi_callable (channel)
def	tests.unit._rpc_test_helpers.unary_unary_multi_callable (channel)

Variables
	tests.unit._rpc_test_helpers._DESERIALIZE_REQUEST = lambda bytestring: bytestring[len(bytestring) // 2:]
	tests.unit._rpc_test_helpers._DESERIALIZE_RESPONSE = lambda bytestring: bytestring[:len(bytestring) // 3]
int	tests.unit._rpc_test_helpers._SERIALIZE_REQUEST = lambda bytestring: bytestring * 2
int	tests.unit._rpc_test_helpers._SERIALIZE_RESPONSE = lambda bytestring: bytestring * 3
string	tests.unit._rpc_test_helpers._STREAM_STREAM = '/test/StreamStream'
string	tests.unit._rpc_test_helpers._STREAM_STREAM_NON_BLOCKING = '/test/StreamStreamNonBlocking'
string	tests.unit._rpc_test_helpers._STREAM_UNARY = '/test/StreamUnary'
string	tests.unit._rpc_test_helpers._UNARY_STREAM = '/test/UnaryStream'
string	tests.unit._rpc_test_helpers._UNARY_STREAM_NON_BLOCKING = '/test/UnaryStreamNonBlocking'
string	tests.unit._rpc_test_helpers._UNARY_UNARY = '/test/UnaryUnary'
	tests.unit._rpc_test_helpers.TIMEOUT_SHORT = datetime.timedelta(seconds=4).total_seconds()