Nathaniel Manista
10 years ago
parent
13db8e517d
commit
41abb052b8
8 changed files with 389 additions and 24 deletions
@ -0,0 +1,231 @@ |
||||
# Copyright 2015, Google Inc. |
||||
# All rights reserved. |
||||
# |
||||
# Redistribution and use in source and binary forms, with or without |
||||
# modification, are permitted provided that the following conditions are |
||||
# met: |
||||
# |
||||
# * Redistributions of source code must retain the above copyright |
||||
# notice, this list of conditions and the following disclaimer. |
||||
# * Redistributions in binary form must reproduce the above |
||||
# copyright notice, this list of conditions and the following disclaimer |
||||
# in the documentation and/or other materials provided with the |
||||
# distribution. |
||||
# * Neither the name of Google Inc. nor the names of its |
||||
# contributors may be used to endorse or promote products derived from |
||||
# this software without specific prior written permission. |
||||
# |
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
|
||||
"""Tests Face interface compliance of the gRPC Python Beta API.""" |
||||
|
||||
import threading |
||||
import unittest |
||||
|
||||
from grpc.beta import beta |
||||
from grpc.beta import interfaces |
||||
from grpc.framework.common import cardinality |
||||
from grpc.framework.interfaces.face import utilities |
||||
from grpc_test import resources |
||||
from grpc_test.beta import test_utilities |
||||
from grpc_test.framework.common import test_constants |
||||
|
||||
_SERVER_HOST_OVERRIDE = 'foo.test.google.fr' |
||||
|
||||
_GROUP = 'group' |
||||
_UNARY_UNARY = 'unary-unary' |
||||
_UNARY_STREAM = 'unary-stream' |
||||
_STREAM_UNARY = 'stream-unary' |
||||
_STREAM_STREAM = 'stream-stream' |
||||
|
||||
_REQUEST = b'abc' |
||||
_RESPONSE = b'123' |
||||
|
||||
|
||||
class _Servicer(object): |
||||
|
||||
def __init__(self): |
||||
self._condition = threading.Condition() |
||||
self._peer = None |
||||
self._serviced = False |
||||
|
||||
def unary_unary(self, request, context): |
||||
with self._condition: |
||||
self._request = request |
||||
self._peer = context.protocol_context().peer() |
||||
context.protocol_context().disable_next_response_compression() |
||||
self._serviced = True |
||||
self._condition.notify_all() |
||||
return _RESPONSE |
||||
|
||||
def unary_stream(self, request, context): |
||||
with self._condition: |
||||
self._request = request |
||||
self._peer = context.protocol_context().peer() |
||||
context.protocol_context().disable_next_response_compression() |
||||
self._serviced = True |
||||
self._condition.notify_all() |
||||
return |
||||
yield |
||||
|
||||
def stream_unary(self, request_iterator, context): |
||||
for request in request_iterator: |
||||
self._request = request |
||||
with self._condition: |
||||
self._peer = context.protocol_context().peer() |
||||
context.protocol_context().disable_next_response_compression() |
||||
self._serviced = True |
||||
self._condition.notify_all() |
||||
return _RESPONSE |
||||
|
||||
def stream_stream(self, request_iterator, context): |
||||
for request in request_iterator: |
||||
with self._condition: |
||||
self._peer = context.protocol_context().peer() |
||||
context.protocol_context().disable_next_response_compression() |
||||
yield _RESPONSE |
||||
with self._condition: |
||||
self._serviced = True |
||||
self._condition.notify_all() |
||||
|
||||
def peer(self): |
||||
with self._condition: |
||||
return self._peer |
||||
|
||||
def block_until_serviced(self): |
||||
with self._condition: |
||||
while not self._serviced: |
||||
self._condition.wait() |
||||
|
||||
|
||||
class _BlockingIterator(object): |
||||
|
||||
def __init__(self, upstream): |
||||
self._condition = threading.Condition() |
||||
self._upstream = upstream |
||||
self._allowed = [] |
||||
|
||||
def __iter__(self): |
||||
return self |
||||
|
||||
def next(self): |
||||
with self._condition: |
||||
while True: |
||||
if self._allowed is None: |
||||
raise StopIteration() |
||||
elif self._allowed: |
||||
return self._allowed.pop(0) |
||||
else: |
||||
self._condition.wait() |
||||
|
||||
def allow(self): |
||||
with self._condition: |
||||
try: |
||||
self._allowed.append(next(self._upstream)) |
||||
except StopIteration: |
||||
self._allowed = None |
||||
self._condition.notify_all() |
||||
|
||||
|
||||
class BetaFeaturesTest(unittest.TestCase): |
||||
|
||||
def setUp(self): |
||||
self._servicer = _Servicer() |
||||
method_implementations = { |
||||
(_GROUP, _UNARY_UNARY): |
||||
utilities.unary_unary_inline(self._servicer.unary_unary), |
||||
(_GROUP, _UNARY_STREAM): |
||||
utilities.unary_stream_inline(self._servicer.unary_stream), |
||||
(_GROUP, _STREAM_UNARY): |
||||
utilities.stream_unary_inline(self._servicer.stream_unary), |
||||
(_GROUP, _STREAM_STREAM): |
||||
utilities.stream_stream_inline(self._servicer.stream_stream), |
||||
} |
||||
|
||||
cardinalities = { |
||||
_UNARY_UNARY: cardinality.Cardinality.UNARY_UNARY, |
||||
_UNARY_STREAM: cardinality.Cardinality.UNARY_STREAM, |
||||
_STREAM_UNARY: cardinality.Cardinality.STREAM_UNARY, |
||||
_STREAM_STREAM: cardinality.Cardinality.STREAM_STREAM, |
||||
} |
||||
|
||||
server_options = beta.server_options( |
||||
thread_pool_size=test_constants.POOL_SIZE) |
||||
self._server = beta.server(method_implementations, options=server_options) |
||||
server_credentials = beta.ssl_server_credentials( |
||||
[(resources.private_key(), resources.certificate_chain(),),]) |
||||
port = self._server.add_secure_port('[::]:0', server_credentials) |
||||
self._server.start() |
||||
self._client_credentials = beta.ssl_client_credentials( |
||||
resources.test_root_certificates(), None, None) |
||||
channel = test_utilities.create_not_really_secure_channel( |
||||
'localhost', port, self._client_credentials, _SERVER_HOST_OVERRIDE) |
||||
stub_options = beta.stub_options( |
||||
thread_pool_size=test_constants.POOL_SIZE) |
||||
self._dynamic_stub = beta.dynamic_stub( |
||||
channel, _GROUP, cardinalities, options=stub_options) |
||||
|
||||
def tearDown(self): |
||||
self._dynamic_stub = None |
||||
self._server.stop(test_constants.SHORT_TIMEOUT).wait() |
||||
|
||||
def test_unary_unary(self): |
||||
call_options = interfaces.grpc_call_options( |
||||
disable_compression=True, credentials=self._client_credentials) |
||||
response = getattr(self._dynamic_stub, _UNARY_UNARY)( |
||||
_REQUEST, test_constants.LONG_TIMEOUT, protocol_options=call_options) |
||||
self.assertEqual(_RESPONSE, response) |
||||
self.assertIsNotNone(self._servicer.peer()) |
||||
|
||||
def test_unary_stream(self): |
||||
call_options = interfaces.grpc_call_options( |
||||
disable_compression=True, credentials=self._client_credentials) |
||||
response_iterator = getattr(self._dynamic_stub, _UNARY_STREAM)( |
||||
_REQUEST, test_constants.LONG_TIMEOUT, protocol_options=call_options) |
||||
self._servicer.block_until_serviced() |
||||
self.assertIsNotNone(self._servicer.peer()) |
||||
|
||||
def test_stream_unary(self): |
||||
call_options = interfaces.grpc_call_options( |
||||
credentials=self._client_credentials) |
||||
request_iterator = _BlockingIterator(iter((_REQUEST,))) |
||||
response_future = getattr(self._dynamic_stub, _STREAM_UNARY).future( |
||||
request_iterator, test_constants.LONG_TIMEOUT, |
||||
protocol_options=call_options) |
||||
response_future.protocol_context().disable_next_request_compression() |
||||
request_iterator.allow() |
||||
response_future.protocol_context().disable_next_request_compression() |
||||
request_iterator.allow() |
||||
self._servicer.block_until_serviced() |
||||
self.assertIsNotNone(self._servicer.peer()) |
||||
self.assertEqual(_RESPONSE, response_future.result()) |
||||
|
||||
def test_stream_stream(self): |
||||
call_options = interfaces.grpc_call_options( |
||||
credentials=self._client_credentials) |
||||
request_iterator = _BlockingIterator(iter((_REQUEST,))) |
||||
response_iterator = getattr(self._dynamic_stub, _STREAM_STREAM)( |
||||
request_iterator, test_constants.SHORT_TIMEOUT, |
||||
protocol_options=call_options) |
||||
response_iterator.protocol_context().disable_next_request_compression() |
||||
request_iterator.allow() |
||||
response = next(response_iterator) |
||||
response_iterator.protocol_context().disable_next_request_compression() |
||||
request_iterator.allow() |
||||
self._servicer.block_until_serviced() |
||||
self.assertIsNotNone(self._servicer.peer()) |
||||
self.assertEqual(_RESPONSE, response) |
||||
|
||||
|
||||
if __name__ == '__main__': |
||||
unittest.main(verbosity=2) |
||||
Loading…
Reference in new issue