Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

That line changed count seemed a bit big

Browse Source
pull/23163/head
Richard Belleville 5 years ago
parent 5811df0fd0
commit 93477ae55f
1 changed files with 195 additions and 195 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 390
      src/python/grpcio_tests/tests_py3_only/unit/_simple_stubs_test.py

390
src/python/grpcio_tests/tests_py3_only/unit/_simple_stubs_test.py
Unescape View File

@ -180,201 +180,201 @@ class SimpleStubsTest(unittest.TestCase):
else:
self.fail(message() + " after " + str(timeout))
# def test_unary_unary_insecure(self):
# with _server(None) as port:
# target = f'localhost:{port}'
# response = grpc.experimental.unary_unary(
# _REQUEST,
# target,
# _UNARY_UNARY,
# channel_credentials=grpc.experimental.
# insecure_channel_credentials())
# self.assertEqual(_REQUEST, response)
# def test_unary_unary_secure(self):
# with _server(grpc.local_server_credentials()) as port:
# target = f'localhost:{port}'
# response = grpc.experimental.unary_unary(
# _REQUEST,
# target,
# _UNARY_UNARY,
# channel_credentials=grpc.local_channel_credentials())
# self.assertEqual(_REQUEST, response)
# def test_channels_cached(self):
# with _server(grpc.local_server_credentials()) as port:
# target = f'localhost:{port}'
# test_name = inspect.stack()[0][3]
# args = (_REQUEST, target, _UNARY_UNARY)
# kwargs = {"channel_credentials": grpc.local_channel_credentials()}
# def _invoke(seed: str):
# run_kwargs = dict(kwargs)
# run_kwargs["options"] = ((test_name + seed, ""),)
# grpc.experimental.unary_unary(*args, **run_kwargs)
# self.assert_cached(_invoke)
# def test_channels_evicted(self):
# with _server(grpc.local_server_credentials()) as port:
# target = f'localhost:{port}'
# response = grpc.experimental.unary_unary(
# _REQUEST,
# target,
# _UNARY_UNARY,
# channel_credentials=grpc.local_channel_credentials())
# self.assert_eventually(
# lambda: grpc._simple_stubs.ChannelCache.get(
# )._test_only_channel_count() == 0,
# message=lambda:
# f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} remain"
# )
# def test_total_channels_enforced(self):
# with _server(grpc.local_server_credentials()) as port:
# target = f'localhost:{port}'
# for i in range(_STRESS_EPOCHS):
# # Ensure we get a new channel each time.
# options = (("foo", str(i)),)
# # Send messages at full blast.
# grpc.experimental.unary_unary(
# _REQUEST,
# target,
# _UNARY_UNARY,
# options=options,
# channel_credentials=grpc.local_channel_credentials())
# self.assert_eventually(
# lambda: grpc._simple_stubs.ChannelCache.get(
# )._test_only_channel_count() <= _MAXIMUM_CHANNELS + 1,
# message=lambda:
# f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} channels remain"
# )
# def test_unary_stream(self):
# with _server(grpc.local_server_credentials()) as port:
# target = f'localhost:{port}'
# for response in grpc.experimental.unary_stream(
# _REQUEST,
# target,
# _UNARY_STREAM,
# channel_credentials=grpc.local_channel_credentials()):
# self.assertEqual(_REQUEST, response)
# def test_stream_unary(self):
# def request_iter():
# for _ in range(_CLIENT_REQUEST_COUNT):
# yield _REQUEST
# with _server(grpc.local_server_credentials()) as port:
# target = f'localhost:{port}'
# response = grpc.experimental.stream_unary(
# request_iter(),
# target,
# _STREAM_UNARY,
# channel_credentials=grpc.local_channel_credentials())
# self.assertEqual(_REQUEST, response)
# def test_stream_stream(self):
# def request_iter():
# for _ in range(_CLIENT_REQUEST_COUNT):
# yield _REQUEST
# with _server(grpc.local_server_credentials()) as port:
# target = f'localhost:{port}'
# for response in grpc.experimental.stream_stream(
# request_iter(),
# target,
# _STREAM_STREAM,
# channel_credentials=grpc.local_channel_credentials()):
# self.assertEqual(_REQUEST, response)
# def test_default_ssl(self):
# _private_key = resources.private_key()
# _certificate_chain = resources.certificate_chain()
# _server_certs = ((_private_key, _certificate_chain),)
# _server_host_override = 'foo.test.google.fr'
# _test_root_certificates = resources.test_root_certificates()
# _property_options = ((
# 'grpc.ssl_target_name_override',
# _server_host_override,
# ),)
# cert_dir = os.path.join(os.path.dirname(resources.__file__),
# "credentials")
# cert_file = os.path.join(cert_dir, "ca.pem")
# with _env("GRPC_DEFAULT_SSL_ROOTS_FILE_PATH", cert_file):
# server_creds = grpc.ssl_server_credentials(_server_certs)
# with _server(server_creds) as port:
# target = f'localhost:{port}'
# response = grpc.experimental.unary_unary(
# _REQUEST, target, _UNARY_UNARY, options=_property_options)
# def test_insecure_sugar(self):
# with _server(None) as port:
# target = f'localhost:{port}'
# response = grpc.experimental.unary_unary(_REQUEST,
# target,
# _UNARY_UNARY,
# insecure=True)
# self.assertEqual(_REQUEST, response)
# def test_insecure_sugar_mutually_exclusive(self):
# with _server(None) as port:
# target = f'localhost:{port}'
# with self.assertRaises(ValueError):
# response = grpc.experimental.unary_unary(
# _REQUEST,
# target,
# _UNARY_UNARY,
# insecure=True,
# channel_credentials=grpc.local_channel_credentials())
# def test_default_wait_for_ready(self):
# addr, port, sock = get_socket()
# sock.close()
# target = f'{addr}:{port}'
# channel = grpc._simple_stubs.ChannelCache.get().get_channel(
# target, (), None, True, None)
# rpc_finished_event = threading.Event()
# rpc_failed_event = threading.Event()
# server = None
# def _on_connectivity_changed(connectivity):
# nonlocal server
# if connectivity is grpc.ChannelConnectivity.TRANSIENT_FAILURE:
# self.assertFalse(rpc_finished_event.is_set())
# self.assertFalse(rpc_failed_event.is_set())
# server = test_common.test_server()
# server.add_insecure_port(target)
# server.add_generic_rpc_handlers((_GenericHandler(),))
# server.start()
# channel.unsubscribe(_on_connectivity_changed)
# elif connectivity in (grpc.ChannelConnectivity.IDLE,
# grpc.ChannelConnectivity.CONNECTING):
# pass
# else:
# self.fail("Encountered unknown state.")
# channel.subscribe(_on_connectivity_changed)
# def _send_rpc():
# try:
# response = grpc.experimental.unary_unary(_REQUEST,
# target,
# _UNARY_UNARY,
# insecure=True)
# rpc_finished_event.set()
# except Exception as e:
# rpc_failed_event.set()
# t = threading.Thread(target=_send_rpc)
# t.start()
# t.join()
# self.assertFalse(rpc_failed_event.is_set())
# self.assertTrue(rpc_finished_event.is_set())
# if server is not None:
# server.stop(None)
def test_unary_unary_insecure(self):
with _server(None) as port:
target = f'localhost:{port}'
response = grpc.experimental.unary_unary(
_REQUEST,
target,
_UNARY_UNARY,
channel_credentials=grpc.experimental.
insecure_channel_credentials())
self.assertEqual(_REQUEST, response)
def test_unary_unary_secure(self):
with _server(grpc.local_server_credentials()) as port:
target = f'localhost:{port}'
response = grpc.experimental.unary_unary(
_REQUEST,
target,
_UNARY_UNARY,
channel_credentials=grpc.local_channel_credentials())
self.assertEqual(_REQUEST, response)
def test_channels_cached(self):
with _server(grpc.local_server_credentials()) as port:
target = f'localhost:{port}'
test_name = inspect.stack()[0][3]
args = (_REQUEST, target, _UNARY_UNARY)
kwargs = {"channel_credentials": grpc.local_channel_credentials()}
def _invoke(seed: str):
run_kwargs = dict(kwargs)
run_kwargs["options"] = ((test_name + seed, ""),)
grpc.experimental.unary_unary(*args, **run_kwargs)
self.assert_cached(_invoke)
def test_channels_evicted(self):
with _server(grpc.local_server_credentials()) as port:
target = f'localhost:{port}'
response = grpc.experimental.unary_unary(
_REQUEST,
target,
_UNARY_UNARY,
channel_credentials=grpc.local_channel_credentials())
self.assert_eventually(
lambda: grpc._simple_stubs.ChannelCache.get(
)._test_only_channel_count() == 0,
message=lambda:
f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} remain"
)
def test_total_channels_enforced(self):
with _server(grpc.local_server_credentials()) as port:
target = f'localhost:{port}'
for i in range(_STRESS_EPOCHS):
# Ensure we get a new channel each time.
options = (("foo", str(i)),)
# Send messages at full blast.
grpc.experimental.unary_unary(
_REQUEST,
target,
_UNARY_UNARY,
options=options,
channel_credentials=grpc.local_channel_credentials())
self.assert_eventually(
lambda: grpc._simple_stubs.ChannelCache.get(
)._test_only_channel_count() <= _MAXIMUM_CHANNELS + 1,
message=lambda:
f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} channels remain"
)
def test_unary_stream(self):
with _server(grpc.local_server_credentials()) as port:
target = f'localhost:{port}'
for response in grpc.experimental.unary_stream(
_REQUEST,
target,
_UNARY_STREAM,
channel_credentials=grpc.local_channel_credentials()):
self.assertEqual(_REQUEST, response)
def test_stream_unary(self):
def request_iter():
for _ in range(_CLIENT_REQUEST_COUNT):
yield _REQUEST
with _server(grpc.local_server_credentials()) as port:
target = f'localhost:{port}'
response = grpc.experimental.stream_unary(
request_iter(),
target,
_STREAM_UNARY,
channel_credentials=grpc.local_channel_credentials())
self.assertEqual(_REQUEST, response)
def test_stream_stream(self):
def request_iter():
for _ in range(_CLIENT_REQUEST_COUNT):
yield _REQUEST
with _server(grpc.local_server_credentials()) as port:
target = f'localhost:{port}'
for response in grpc.experimental.stream_stream(
request_iter(),
target,
_STREAM_STREAM,
channel_credentials=grpc.local_channel_credentials()):
self.assertEqual(_REQUEST, response)
def test_default_ssl(self):
_private_key = resources.private_key()
_certificate_chain = resources.certificate_chain()
_server_certs = ((_private_key, _certificate_chain),)
_server_host_override = 'foo.test.google.fr'
_test_root_certificates = resources.test_root_certificates()
_property_options = ((
'grpc.ssl_target_name_override',
_server_host_override,
),)
cert_dir = os.path.join(os.path.dirname(resources.__file__),
"credentials")
cert_file = os.path.join(cert_dir, "ca.pem")
with _env("GRPC_DEFAULT_SSL_ROOTS_FILE_PATH", cert_file):
server_creds = grpc.ssl_server_credentials(_server_certs)
with _server(server_creds) as port:
target = f'localhost:{port}'
response = grpc.experimental.unary_unary(
_REQUEST, target, _UNARY_UNARY, options=_property_options)
def test_insecure_sugar(self):
with _server(None) as port:
target = f'localhost:{port}'
response = grpc.experimental.unary_unary(_REQUEST,
target,
_UNARY_UNARY,
insecure=True)
self.assertEqual(_REQUEST, response)
def test_insecure_sugar_mutually_exclusive(self):
with _server(None) as port:
target = f'localhost:{port}'
with self.assertRaises(ValueError):
response = grpc.experimental.unary_unary(
_REQUEST,
target,
_UNARY_UNARY,
insecure=True,
channel_credentials=grpc.local_channel_credentials())
def test_default_wait_for_ready(self):
addr, port, sock = get_socket()
sock.close()
target = f'{addr}:{port}'
channel = grpc._simple_stubs.ChannelCache.get().get_channel(
target, (), None, True, None)
rpc_finished_event = threading.Event()
rpc_failed_event = threading.Event()
server = None
def _on_connectivity_changed(connectivity):
nonlocal server
if connectivity is grpc.ChannelConnectivity.TRANSIENT_FAILURE:
self.assertFalse(rpc_finished_event.is_set())
self.assertFalse(rpc_failed_event.is_set())
server = test_common.test_server()
server.add_insecure_port(target)
server.add_generic_rpc_handlers((_GenericHandler(),))
server.start()
channel.unsubscribe(_on_connectivity_changed)
elif connectivity in (grpc.ChannelConnectivity.IDLE,
grpc.ChannelConnectivity.CONNECTING):
pass
else:
self.fail("Encountered unknown state.")
channel.subscribe(_on_connectivity_changed)
def _send_rpc():
try:
response = grpc.experimental.unary_unary(_REQUEST,
target,
_UNARY_UNARY,
insecure=True)
rpc_finished_event.set()
except Exception as e:
rpc_failed_event.set()
t = threading.Thread(target=_send_rpc)
t.start()
t.join()
self.assertFalse(rpc_failed_event.is_set())
self.assertTrue(rpc_finished_event.is_set())
if server is not None:
server.stop(None)
def assert_times_out(self, invocation_args):
with _server(None) as port:

Write Preview
Loading…
Cancel
Save