|
|
|
@ -544,7 +544,9 @@ class SimpleStubsPluginTest(unittest.TestCase): |
|
|
|
|
|
|
|
|
|
def testUnaryCall(self): |
|
|
|
|
request = request_pb2.SimpleRequest(response_size=13) |
|
|
|
|
response = service_pb2_grpc.TestService.UnaryCall(request, self._target) |
|
|
|
|
response = service_pb2_grpc.TestService.UnaryCall(request, |
|
|
|
|
self._target, |
|
|
|
|
wait_for_ready=True) |
|
|
|
|
expected_response = self.servicer_methods.UnaryCall( |
|
|
|
|
request, 'not a real context!') |
|
|
|
|
self.assertEqual(expected_response, response) |
|
|
|
@ -554,21 +556,23 @@ class SimpleStubsPluginTest(unittest.TestCase): |
|
|
|
|
expected_responses = self.servicer_methods.StreamingOutputCall( |
|
|
|
|
request, 'not a real RpcContext!') |
|
|
|
|
responses = service_pb2_grpc.TestService.StreamingOutputCall( |
|
|
|
|
request, self._target) |
|
|
|
|
request, self._target, wait_for_ready=True) |
|
|
|
|
for expected_response, response in moves.zip_longest( |
|
|
|
|
expected_responses, responses): |
|
|
|
|
self.assertEqual(expected_response, response) |
|
|
|
|
|
|
|
|
|
def testStreamingInputCall(self): |
|
|
|
|
response = service_pb2_grpc.TestService.StreamingInputCall( |
|
|
|
|
_streaming_input_request_iterator(), self._target) |
|
|
|
|
_streaming_input_request_iterator(), |
|
|
|
|
self._target, |
|
|
|
|
wait_for_ready=True) |
|
|
|
|
expected_response = self.servicer_methods.StreamingInputCall( |
|
|
|
|
_streaming_input_request_iterator(), 'not a real RpcContext!') |
|
|
|
|
self.assertEqual(expected_response, response) |
|
|
|
|
|
|
|
|
|
def testFullDuplexCall(self): |
|
|
|
|
responses = service_pb2_grpc.TestService.FullDuplexCall( |
|
|
|
|
_full_duplex_request_iterator(), self._target) |
|
|
|
|
_full_duplex_request_iterator(), self._target, wait_for_ready=True) |
|
|
|
|
expected_responses = self.servicer_methods.FullDuplexCall( |
|
|
|
|
_full_duplex_request_iterator(), 'not a real RpcContext!') |
|
|
|
|
for expected_response, response in moves.zip_longest( |
|
|
|
@ -587,7 +591,7 @@ class SimpleStubsPluginTest(unittest.TestCase): |
|
|
|
|
yield request |
|
|
|
|
|
|
|
|
|
responses = service_pb2_grpc.TestService.HalfDuplexCall( |
|
|
|
|
half_duplex_request_iterator(), self._target) |
|
|
|
|
half_duplex_request_iterator(), self._target, wait_for_ready=True) |
|
|
|
|
expected_responses = self.servicer_methods.HalfDuplexCall( |
|
|
|
|
half_duplex_request_iterator(), 'not a real RpcContext!') |
|
|
|
|
for expected_response, response in moves.zip_longest( |
|
|
|
|