The Python interop client.

The server_host_override flag's implementation remaining
outstanding means that this only works over insecure
connections for now.

6 files changed, 299 insertions, 9 deletions

diff --git a/src/python/interop/interop/client.py b/src/python/interop/interop/client.py
new file mode 100644
index 0000000000..f4a449ef9e
--- /dev/null
+++ b/src/python/interop/interop/client.py
@@ -0,0 +1,86 @@
+# 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.
+
+"""The Python implementation of the GRPC interoperability test client."""
+
+import argparse
+
+from grpc.early_adopter import implementations
+
+from interop import methods
+from interop import resources
+
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+
+
+def _args():
+  parser = argparse.ArgumentParser()
+  parser.add_argument(
+      '--server_host', help='the host to which to connect', type=str)
+  parser.add_argument(
+      '--server_host_override',
+      help='the server host to which to claim to connect', type=str)
+  parser.add_argument(
+      '--server_port', help='the port to which to connect', type=int)
+  parser.add_argument(
+      '--test_case', help='the test case to execute', type=str)
+  parser.add_argument(
+      '--use_tls', help='require a secure connection', dest='use_tls',
+      action='store_true')
+  parser.add_argument(
+      '--use_test_ca', help='replace platform root CAs with ca.pem',
+      action='store_true')
+  return parser.parse_args()
+
+
+def _stub(args):
+  if args.use_tls:
+    if args.use_test_ca:
+      root_certificates = resources.test_root_certificates()
+    else:
+      root_certificates = resources.prod_root_certificates()
+    # TODO(nathaniel): server host override.
+
+    stub = implementations.secure_stub(
+        methods.CLIENT_METHODS, args.server_host, args.server_port,
+        root_certificates, None, None)
+  else:
+    stub = implementations.insecure_stub(
+        methods.CLIENT_METHODS, args.server_host, args.server_port)
+  return stub
+
+
+def _test_interoperability():
+  args = _args()
+  stub = _stub(args)
+  methods.test_interoperability(args.test_case, stub)
+
+
+if __name__ == '__main__':
+  _test_interoperability()
diff --git a/src/python/interop/interop/credentials/ca.pem b/src/python/interop/interop/credentials/ca.pem
new file mode 100755
index 0000000000..6c8511a73c
--- /dev/null
+++ b/src/python/interop/interop/credentials/ca.pem
@@ -0,0 +1,15 @@
+-----BEGIN CERTIFICATE-----
+MIICSjCCAbOgAwIBAgIJAJHGGR4dGioHMA0GCSqGSIb3DQEBCwUAMFYxCzAJBgNV
+BAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBX
+aWRnaXRzIFB0eSBMdGQxDzANBgNVBAMTBnRlc3RjYTAeFw0xNDExMTEyMjMxMjla
+Fw0yNDExMDgyMjMxMjlaMFYxCzAJBgNVBAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0
+YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQxDzANBgNVBAMT
+BnRlc3RjYTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAwEDfBV5MYdlHVHJ7
++L4nxrZy7mBfAVXpOc5vMYztssUI7mL2/iYujiIXM+weZYNTEpLdjyJdu7R5gGUu
+g1jSVK/EPHfc74O7AyZU34PNIP4Sh33N+/A5YexrNgJlPY+E3GdVYi4ldWJjgkAd
+Qah2PH5ACLrIIC6tRka9hcaBlIECAwEAAaMgMB4wDAYDVR0TBAUwAwEB/zAOBgNV
+HQ8BAf8EBAMCAgQwDQYJKoZIhvcNAQELBQADgYEAHzC7jdYlzAVmddi/gdAeKPau
+sPBG/C2HCWqHzpCUHcKuvMzDVkY/MP2o6JIW2DBbY64bO/FceExhjcykgaYtCH/m
+oIU63+CFOTtR7otyQAWHqXa7q4SbCDlG7DyRFxqG0txPtGvy12lgldA2+RgcigQG
+Dfcog5wrJytaQ6UA0wE=
+-----END CERTIFICATE-----
diff --git a/src/python/interop/interop/methods.py b/src/python/interop/interop/methods.py
index 6d5990087e..4da28ee775 100644
--- a/src/python/interop/interop/methods.py
+++ b/src/python/interop/interop/methods.py
@@ -29,11 +29,16 @@
 
 """Implementations of interoperability test methods."""
 
+import threading
+
 from grpc.early_adopter import utilities
 
 from interop import empty_pb2
 from interop import messages_pb2
 
+_TIMEOUT = 7
+
+
 def _empty_call(request, unused_context):
   return empty_pb2.Empty()
 
@@ -142,3 +147,134 @@ SERVER_METHODS = {
     FULL_DUPLEX_CALL_METHOD_NAME: _SERVER_FULL_DUPLEX_CALL,
     HALF_DUPLEX_CALL_METHOD_NAME: _SERVER_HALF_DUPLEX_CALL,
 }
+
+
+def _empty_unary(stub):
+  with stub:
+    response = stub.EmptyCall(empty_pb2.Empty(), _TIMEOUT)
+    if not isinstance(response, empty_pb2.Empty):
+      raise TypeError(
+          'response is of type "%s", not empty_pb2.Empty!', type(response))
+
+
+def _large_unary(stub):
+  with stub:
+    request = messages_pb2.SimpleRequest(
+        response_type=messages_pb2.COMPRESSABLE, response_size=314159,
+        payload=messages_pb2.Payload(body=b'\x00' * 271828))
+    response_future = stub.UnaryCall.async(request, _TIMEOUT)
+    response = response_future.result()
+    if response.payload.type is not messages_pb2.COMPRESSABLE:
+      raise ValueError(
+          'response payload type is "%s"!' % type(response.payload.type))
+    if len(response.payload.body) != 314159:
+      raise ValueError(
+          'response body of incorrect size %d!' % len(response.payload.body))
+
+
+def _client_streaming(stub):
+  with stub:
+    payload_body_sizes = (27182, 8, 1828, 45904)
+    payloads = (
+        messages_pb2.Payload(body=b'\x00' * size)
+        for size in payload_body_sizes)
+    requests = (
+        messages_pb2.StreamingInputCallRequest(payload=payload)
+        for payload in payloads)
+    response = stub.StreamingInputCall(requests, _TIMEOUT)
+    if response.aggregated_payload_size != 74922:
+      raise ValueError(
+          'incorrect size %d!' % response.aggregated_payload_size)
+
+
+def _server_streaming(stub):
+  sizes = (31415, 9, 2653, 58979)
+
+  with stub:
+    request = messages_pb2.StreamingOutputCallRequest(
+        response_type=messages_pb2.COMPRESSABLE,
+        response_parameters=(
+            messages_pb2.ResponseParameters(size=sizes[0]),
+            messages_pb2.ResponseParameters(size=sizes[1]),
+            messages_pb2.ResponseParameters(size=sizes[2]),
+            messages_pb2.ResponseParameters(size=sizes[3]),
+        ))
+    response_iterator = stub.StreamingOutputCall(request, _TIMEOUT)
+    for index, response in enumerate(response_iterator):
+      if response.payload.type != messages_pb2.COMPRESSABLE:
+        raise ValueError(
+            'response body of invalid type %s!' % response.payload.type)
+      if len(response.payload.body) != sizes[index]:
+        raise ValueError(
+            'response body of invalid size %d!' % len(response.payload.body))
+
+
+class _Pipe(object):
+
+  def __init__(self):
+    self._condition = threading.Condition()
+    self._values = []
+    self._open = True
+
+  def __iter__(self):
+    return self
+
+  def next(self):
+    with self._condition:
+      while not self._values and self._open:
+        self._condition.wait()
+      if self._values:
+        return self._values.pop(0)
+      else:
+        raise StopIteration()
+
+  def add(self, value):
+    with self._condition:
+      self._values.append(value)
+      self._condition.notify()
+
+  def close(self):
+    with self._condition:
+      self._open = False
+      self._condition.notify()
+
+
+def _ping_pong(stub):
+  request_response_sizes = (31415, 9, 2653, 58979)
+  request_payload_sizes = (27182, 8, 1828, 45904)
+
+  with stub:
+    pipe = _Pipe()
+    response_iterator = stub.FullDuplexCall(pipe, _TIMEOUT)
+    print 'Starting ping-pong with response iterator %s' % response_iterator
+    for response_size, payload_size in zip(
+        request_response_sizes, request_payload_sizes):
+      request = messages_pb2.StreamingOutputCallRequest(
+          response_type=messages_pb2.COMPRESSABLE,
+          response_parameters=(messages_pb2.ResponseParameters(
+              size=response_size),),
+          payload=messages_pb2.Payload(body=b'\x00' * payload_size))
+      pipe.add(request)
+      response = next(response_iterator)
+      if response.payload.type != messages_pb2.COMPRESSABLE:
+        raise ValueError(
+            'response body of invalid type %s!' % response.payload.type)
+      if len(response.payload.body) != response_size:
+        raise ValueError(
+            'response body of invalid size %d!' % len(response.payload.body))
+    pipe.close()
+
+
+def test_interoperability(test_case, stub):
+  if test_case == 'empty_unary':
+    _empty_unary(stub)
+  elif test_case == 'large_unary':
+    _large_unary(stub)
+  elif test_case == 'server_streaming':
+    _server_streaming(stub)
+  elif test_case == 'client_streaming':
+    _client_streaming(stub)
+  elif test_case == 'ping_pong':
+    _ping_pong(stub)
+  else:
+    raise NotImplementedError('Test case "%s" not implemented!')
diff --git a/src/python/interop/interop/resources.py b/src/python/interop/interop/resources.py
new file mode 100644
index 0000000000..2c3045313d
--- /dev/null
+++ b/src/python/interop/interop/resources.py
@@ -0,0 +1,56 @@
+# 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.
+
+"""Constants and functions for data used in interoperability testing."""
+
+import os
+
+import pkg_resources
+
+_ROOT_CERTIFICATES_RESOURCE_PATH = 'credentials/ca.pem'
+_PRIVATE_KEY_RESOURCE_PATH = 'credentials/server1.key'
+_CERTIFICATE_CHAIN_RESOURCE_PATH = 'credentials/server1.pem'
+
+
+def test_root_certificates():
+  return pkg_resources.resource_string(
+      __name__, _ROOT_CERTIFICATES_RESOURCE_PATH)
+
+
+def prod_root_certificates():
+  return open(os.environ['SSL_CERT_FILE'], mode='rb').read()
+
+
+def private_key():
+  return pkg_resources.resource_string(__name__, _PRIVATE_KEY_RESOURCE_PATH)
+
+
+def certificate_chain():
+  return pkg_resources.resource_string(
+      __name__, _CERTIFICATE_CHAIN_RESOURCE_PATH)
diff --git a/src/python/interop/interop/server.py b/src/python/interop/interop/server.py
index 785d482fe5..4e4b127a9a 100644
--- a/src/python/interop/interop/server.py
+++ b/src/python/interop/interop/server.py
@@ -31,18 +31,15 @@
 
 import argparse
 import logging
-import pkg_resources
 import time
 
 from grpc.early_adopter import implementations
 
 from interop import methods
+from interop import resources
 
 _ONE_DAY_IN_SECONDS = 60 * 60 * 24
 
-_PRIVATE_KEY_RESOURCE_PATH = 'credentials/server1.key'
-_CERTIFICATE_CHAIN_RESOURCE_PATH = 'credentials/server1.pem'
-
 
 def serve():
   parser = argparse.ArgumentParser()
@@ -54,10 +51,8 @@ def serve():
   args = parser.parse_args()
 
   if args.use_tls:
-    private_key = pkg_resources.resource_string(
-        __name__, _PRIVATE_KEY_RESOURCE_PATH)
-    certificate_chain = pkg_resources.resource_string(
-        __name__, _CERTIFICATE_CHAIN_RESOURCE_PATH)
+    private_key = resources.private_key()
+    certificate_chain = resources.certificate_chain()
     server = implementations.secure_server(
         methods.SERVER_METHODS, args.port, private_key, certificate_chain)
   else:
diff --git a/src/python/interop/setup.py b/src/python/interop/setup.py
index 4b7709f234..6db5435090 100644
--- a/src/python/interop/setup.py
+++ b/src/python/interop/setup.py
@@ -40,7 +40,9 @@ _PACKAGE_DIRECTORIES = {
 }
 
 _PACKAGE_DATA = {
-    'interop': ['credentials/server1.key', 'credentials/server1.pem',]
+    'interop': [
+        'credentials/ca.pem', 'credentials/server1.key',
+        'credentials/server1.pem',]
 }
 
 _INSTALL_REQUIRES = ['grpc-2015>=0.0.1']