merge with head

9 files changed, 555 insertions, 63 deletions

diff --git a/src/python/grpcio_test/grpc_test/_adapter/_low_test.py b/src/python/grpcio_test/grpc_test/_adapter/_low_test.py
index 44fe760fbc..70149127da 100644
--- a/src/python/grpcio_test/grpc_test/_adapter/_low_test.py
+++ b/src/python/grpcio_test/grpc_test/_adapter/_low_test.py
@@ -52,7 +52,6 @@ def wait_for_events(completion_queues, deadline):
   def set_ith_result(i, completion_queue):
     result = completion_queue.next(deadline)
     with lock:
-      print i, completion_queue, result, time.time() - deadline
       results[i] = result
   for i, completion_queue in enumerate(completion_queues):
     thread = threading.Thread(target=set_ith_result,
@@ -80,10 +79,12 @@ class InsecureServerInsecureClient(unittest.TestCase):
     del self.client_channel
 
     self.client_completion_queue.shutdown()
-    while self.client_completion_queue.next().type != _types.EventType.QUEUE_SHUTDOWN:
+    while (self.client_completion_queue.next().type !=
+           _types.EventType.QUEUE_SHUTDOWN):
       pass
     self.server_completion_queue.shutdown()
-    while self.server_completion_queue.next().type != _types.EventType.QUEUE_SHUTDOWN:
+    while (self.server_completion_queue.next().type !=
+           _types.EventType.QUEUE_SHUTDOWN):
       pass
 
     del self.client_completion_queue
@@ -91,58 +92,68 @@ class InsecureServerInsecureClient(unittest.TestCase):
     del self.server
 
   def testEcho(self):
-    DEADLINE = time.time()+5
-    DEADLINE_TOLERANCE = 0.25
-    CLIENT_METADATA_ASCII_KEY = 'key'
-    CLIENT_METADATA_ASCII_VALUE = 'val'
-    CLIENT_METADATA_BIN_KEY = 'key-bin'
-    CLIENT_METADATA_BIN_VALUE = b'\0'*1000
-    SERVER_INITIAL_METADATA_KEY = 'init_me_me_me'
-    SERVER_INITIAL_METADATA_VALUE = 'whodawha?'
-    SERVER_TRAILING_METADATA_KEY = 'california_is_in_a_drought'
-    SERVER_TRAILING_METADATA_VALUE = 'zomg it is'
-    SERVER_STATUS_CODE = _types.StatusCode.OK
-    SERVER_STATUS_DETAILS = 'our work is never over'
-    REQUEST = 'in death a member of project mayhem has a name'
-    RESPONSE = 'his name is robert paulson'
-    METHOD = 'twinkies'
-    HOST = 'hostess'
+    deadline = time.time() + 5
+    event_time_tolerance = 2
+    deadline_tolerance = 0.25
+    client_metadata_ascii_key = 'key'
+    client_metadata_ascii_value = 'val'
+    client_metadata_bin_key = 'key-bin'
+    client_metadata_bin_value = b'\0'*1000
+    server_initial_metadata_key = 'init_me_me_me'
+    server_initial_metadata_value = 'whodawha?'
+    server_trailing_metadata_key = 'california_is_in_a_drought'
+    server_trailing_metadata_value = 'zomg it is'
+    server_status_code = _types.StatusCode.OK
+    server_status_details = 'our work is never over'
+    request = 'blarghaflargh'
+    response = 'his name is robert paulson'
+    method = 'twinkies'
+    host = 'hostess'
     server_request_tag = object()
-    request_call_result = self.server.request_call(self.server_completion_queue, server_request_tag)
+    request_call_result = self.server.request_call(self.server_completion_queue,
+                                                   server_request_tag)
 
-    self.assertEquals(_types.CallError.OK, request_call_result)
+    self.assertEqual(_types.CallError.OK, request_call_result)
 
     client_call_tag = object()
-    client_call = self.client_channel.create_call(self.client_completion_queue, METHOD, HOST, DEADLINE)
-    client_initial_metadata = [(CLIENT_METADATA_ASCII_KEY, CLIENT_METADATA_ASCII_VALUE), (CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)]
+    client_call = self.client_channel.create_call(
+        self.client_completion_queue, method, host, deadline)
+    client_initial_metadata = [
+        (client_metadata_ascii_key, client_metadata_ascii_value),
+        (client_metadata_bin_key, client_metadata_bin_value)
+    ]
     client_start_batch_result = client_call.start_batch([
         _types.OpArgs.send_initial_metadata(client_initial_metadata),
-        _types.OpArgs.send_message(REQUEST, 0),
+        _types.OpArgs.send_message(request, 0),
         _types.OpArgs.send_close_from_client(),
         _types.OpArgs.recv_initial_metadata(),
         _types.OpArgs.recv_message(),
         _types.OpArgs.recv_status_on_client()
     ], client_call_tag)
-    self.assertEquals(_types.CallError.OK, client_start_batch_result)
+    self.assertEqual(_types.CallError.OK, client_start_batch_result)
 
-    client_no_event, request_event, = wait_for_events([self.client_completion_queue, self.server_completion_queue], time.time() + 2)
-    self.assertEquals(client_no_event, None)
-    self.assertEquals(_types.EventType.OP_COMPLETE, request_event.type)
+    client_no_event, request_event, = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
+    self.assertEqual(client_no_event, None)
+    self.assertEqual(_types.EventType.OP_COMPLETE, request_event.type)
     self.assertIsInstance(request_event.call, _low.Call)
     self.assertIs(server_request_tag, request_event.tag)
-    self.assertEquals(1, len(request_event.results))
+    self.assertEqual(1, len(request_event.results))
     received_initial_metadata = dict(request_event.results[0].initial_metadata)
     # Check that our metadata were transmitted
-    self.assertEquals(
+    self.assertEqual(
         dict(client_initial_metadata),
-        dict((x, received_initial_metadata[x]) for x in zip(*client_initial_metadata)[0]))
+        dict((x, received_initial_metadata[x])
+             for x in zip(*client_initial_metadata)[0]))
     # Check that Python's user agent string is a part of the full user agent
     # string
     self.assertIn('Python-gRPC-{}'.format(_grpcio_metadata.__version__),
                   received_initial_metadata['user-agent'])
-    self.assertEquals(METHOD, request_event.call_details.method)
-    self.assertEquals(HOST, request_event.call_details.host)
-    self.assertLess(abs(DEADLINE - request_event.call_details.deadline), DEADLINE_TOLERANCE)
+    self.assertEqual(method, request_event.call_details.method)
+    self.assertEqual(host, request_event.call_details.host)
+    self.assertLess(abs(deadline - request_event.call_details.deadline),
+                    deadline_tolerance)
 
     # Check that the channel is connected, and that both it and the call have
     # the proper target and peer; do this after the first flurry of messages to
@@ -155,33 +166,43 @@ class InsecureServerInsecureClient(unittest.TestCase):
 
     server_call_tag = object()
     server_call = request_event.call
-    server_initial_metadata = [(SERVER_INITIAL_METADATA_KEY, SERVER_INITIAL_METADATA_VALUE)]
-    server_trailing_metadata = [(SERVER_TRAILING_METADATA_KEY, SERVER_TRAILING_METADATA_VALUE)]
+    server_initial_metadata = [
+        (server_initial_metadata_key, server_initial_metadata_value)
+    ]
+    server_trailing_metadata = [
+        (server_trailing_metadata_key, server_trailing_metadata_value)
+    ]
     server_start_batch_result = server_call.start_batch([
         _types.OpArgs.send_initial_metadata(server_initial_metadata),
         _types.OpArgs.recv_message(),
-        _types.OpArgs.send_message(RESPONSE, 0),
+        _types.OpArgs.send_message(response, 0),
         _types.OpArgs.recv_close_on_server(),
-        _types.OpArgs.send_status_from_server(server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS)
+        _types.OpArgs.send_status_from_server(
+            server_trailing_metadata, server_status_code, server_status_details)
     ], server_call_tag)
-    self.assertEquals(_types.CallError.OK, server_start_batch_result)
+    self.assertEqual(_types.CallError.OK, server_start_batch_result)
 
-    client_event, server_event, = wait_for_events([self.client_completion_queue, self.server_completion_queue], time.time() + 1)
+    client_event, server_event, = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
 
-    self.assertEquals(6, len(client_event.results))
+    self.assertEqual(6, len(client_event.results))
     found_client_op_types = set()
     for client_result in client_event.results:
-      self.assertNotIn(client_result.type, found_client_op_types)  # we expect each op type to be unique
+      # we expect each op type to be unique
+      self.assertNotIn(client_result.type, found_client_op_types)
       found_client_op_types.add(client_result.type)
       if client_result.type == _types.OpType.RECV_INITIAL_METADATA:
-        self.assertEquals(dict(server_initial_metadata), dict(client_result.initial_metadata))
+        self.assertEqual(dict(server_initial_metadata),
+                         dict(client_result.initial_metadata))
       elif client_result.type == _types.OpType.RECV_MESSAGE:
-        self.assertEquals(RESPONSE, client_result.message)
+        self.assertEqual(response, client_result.message)
       elif client_result.type == _types.OpType.RECV_STATUS_ON_CLIENT:
-        self.assertEquals(dict(server_trailing_metadata), dict(client_result.trailing_metadata))
-        self.assertEquals(SERVER_STATUS_DETAILS, client_result.status.details)
-        self.assertEquals(SERVER_STATUS_CODE, client_result.status.code)
-    self.assertEquals(set([
+        self.assertEqual(dict(server_trailing_metadata),
+                         dict(client_result.trailing_metadata))
+        self.assertEqual(server_status_details, client_result.status.details)
+        self.assertEqual(server_status_code, client_result.status.code)
+    self.assertEqual(set([
           _types.OpType.SEND_INITIAL_METADATA,
           _types.OpType.SEND_MESSAGE,
           _types.OpType.SEND_CLOSE_FROM_CLIENT,
@@ -190,16 +211,16 @@ class InsecureServerInsecureClient(unittest.TestCase):
           _types.OpType.RECV_STATUS_ON_CLIENT
       ]), found_client_op_types)
 
-    self.assertEquals(5, len(server_event.results))
+    self.assertEqual(5, len(server_event.results))
     found_server_op_types = set()
     for server_result in server_event.results:
       self.assertNotIn(client_result.type, found_server_op_types)
       found_server_op_types.add(server_result.type)
       if server_result.type == _types.OpType.RECV_MESSAGE:
-        self.assertEquals(REQUEST, server_result.message)
+        self.assertEqual(request, server_result.message)
       elif server_result.type == _types.OpType.RECV_CLOSE_ON_SERVER:
         self.assertFalse(server_result.cancelled)
-    self.assertEquals(set([
+    self.assertEqual(set([
           _types.OpType.SEND_INITIAL_METADATA,
           _types.OpType.RECV_MESSAGE,
           _types.OpType.SEND_MESSAGE,
@@ -211,5 +232,81 @@ class InsecureServerInsecureClient(unittest.TestCase):
     del server_call
 
 
+class HangingServerShutdown(unittest.TestCase):
+
+  def setUp(self):
+    self.server_completion_queue = _low.CompletionQueue()
+    self.server = _low.Server(self.server_completion_queue, [])
+    self.port = self.server.add_http2_port('[::]:0')
+    self.client_completion_queue = _low.CompletionQueue()
+    self.client_channel = _low.Channel('localhost:%d'%self.port, [])
+
+    self.server.start()
+
+  def tearDown(self):
+    self.server.shutdown()
+    del self.client_channel
+
+    self.client_completion_queue.shutdown()
+    self.server_completion_queue.shutdown()
+    while True:
+      client_event, server_event = wait_for_events(
+          [self.client_completion_queue, self.server_completion_queue],
+          float("+inf"))
+      if (client_event.type == _types.EventType.QUEUE_SHUTDOWN and
+          server_event.type == _types.EventType.QUEUE_SHUTDOWN):
+        break
+
+    del self.client_completion_queue
+    del self.server_completion_queue
+    del self.server
+
+  def testHangingServerCall(self):
+    deadline = time.time() + 5
+    deadline_tolerance = 0.25
+    event_time_tolerance = 2
+    cancel_all_calls_time_tolerance = 0.5
+    request = 'blarghaflargh'
+    method = 'twinkies'
+    host = 'hostess'
+    server_request_tag = object()
+    request_call_result = self.server.request_call(self.server_completion_queue,
+                                                   server_request_tag)
+
+    client_call_tag = object()
+    client_call = self.client_channel.create_call(self.client_completion_queue,
+                                                  method, host, deadline)
+    client_start_batch_result = client_call.start_batch([
+        _types.OpArgs.send_initial_metadata([]),
+        _types.OpArgs.send_message(request, 0),
+        _types.OpArgs.send_close_from_client(),
+        _types.OpArgs.recv_initial_metadata(),
+        _types.OpArgs.recv_message(),
+        _types.OpArgs.recv_status_on_client()
+    ], client_call_tag)
+
+    client_no_event, request_event, = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
+
+    # Now try to shutdown the server and expect that we see server shutdown
+    # almost immediately after calling cancel_all_calls.
+    with self.assertRaises(RuntimeError):
+      self.server.cancel_all_calls()
+    shutdown_tag = object()
+    self.server.shutdown(shutdown_tag)
+    pre_cancel_timestamp = time.time()
+    self.server.cancel_all_calls()
+    finish_shutdown_timestamp = None
+    client_call_event, server_shutdown_event = wait_for_events(
+        [self.client_completion_queue, self.server_completion_queue],
+        time.time() + event_time_tolerance)
+    self.assertIs(shutdown_tag, server_shutdown_event.tag)
+    self.assertGreater(pre_cancel_timestamp + cancel_all_calls_time_tolerance,
+                       time.time())
+
+    del client_call
+
+
 if __name__ == '__main__':
   unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py b/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py
new file mode 100644
index 0000000000..72b1ae5642
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py
@@ -0,0 +1,165 @@
+# 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 the RPC Framework Core's implementation of the Base interface."""
+
+import collections
+import logging
+import random
+import time
+import unittest
+
+from grpc._adapter import _intermediary_low
+from grpc._links import invocation
+from grpc._links import service
+from grpc.framework.core import implementations
+from grpc.framework.interfaces.base import utilities
+from grpc_test import test_common as grpc_test_common
+from grpc_test.framework.common import test_constants
+from grpc_test.framework.interfaces.base import test_cases
+from grpc_test.framework.interfaces.base import test_interfaces
+
+_INVOCATION_INITIAL_METADATA = ((b'0', b'abc'), (b'1', b'def'), (b'2', b'ghi'),)
+_SERVICE_INITIAL_METADATA = ((b'3', b'jkl'), (b'4', b'mno'), (b'5', b'pqr'),)
+_SERVICE_TERMINAL_METADATA = ((b'6', b'stu'), (b'7', b'vwx'), (b'8', b'yza'),)
+_CODE = _intermediary_low.Code.OK
+_MESSAGE = b'test message'
+
+
+class _SerializationBehaviors(
+    collections.namedtuple(
+        '_SerializationBehaviors',
+        ('request_serializers', 'request_deserializers', 'response_serializers',
+         'response_deserializers',))):
+  pass
+
+
+class _Links(
+    collections.namedtuple(
+        '_Links',
+        ('invocation_end_link', 'invocation_grpc_link', 'service_grpc_link',
+         'service_end_link'))):
+  pass
+
+
+def _serialization_behaviors_from_serializations(serializations):
+  request_serializers = {}
+  request_deserializers = {}
+  response_serializers = {}
+  response_deserializers = {}
+  for (group, method), serialization in serializations.iteritems():
+    request_serializers[group, method] = serialization.serialize_request
+    request_deserializers[group, method] = serialization.deserialize_request
+    response_serializers[group, method] = serialization.serialize_response
+    response_deserializers[group, method] = serialization.deserialize_response
+  return _SerializationBehaviors(
+      request_serializers, request_deserializers, response_serializers,
+      response_deserializers)
+
+
+class _Implementation(test_interfaces.Implementation):
+
+  def instantiate(self, serializations, servicer):
+    serialization_behaviors = _serialization_behaviors_from_serializations(
+        serializations)
+    invocation_end_link = implementations.invocation_end_link()
+    service_end_link = implementations.service_end_link(
+        servicer, test_constants.DEFAULT_TIMEOUT,
+        test_constants.MAXIMUM_TIMEOUT)
+    service_grpc_link = service.service_link(
+        serialization_behaviors.request_deserializers,
+        serialization_behaviors.response_serializers)
+    port = service_grpc_link.add_port(0, None)
+    channel = _intermediary_low.Channel('localhost:%d' % port, None)
+    invocation_grpc_link = invocation.invocation_link(
+        channel, b'localhost',
+        serialization_behaviors.request_serializers,
+        serialization_behaviors.response_deserializers)
+
+    invocation_end_link.join_link(invocation_grpc_link)
+    invocation_grpc_link.join_link(invocation_end_link)
+    service_end_link.join_link(service_grpc_link)
+    service_grpc_link.join_link(service_end_link)
+    invocation_grpc_link.start()
+    service_grpc_link.start()
+    return invocation_end_link, service_end_link, (
+        invocation_grpc_link, service_grpc_link)
+
+  def destantiate(self, memo):
+    invocation_grpc_link, service_grpc_link = memo
+    invocation_grpc_link.stop()
+    service_grpc_link.stop_gracefully()
+
+  def invocation_initial_metadata(self):
+    return _INVOCATION_INITIAL_METADATA
+
+  def service_initial_metadata(self):
+    return _SERVICE_INITIAL_METADATA
+
+  def invocation_completion(self):
+    return utilities.completion(None, None, None)
+
+  def service_completion(self):
+    return utilities.completion(_SERVICE_TERMINAL_METADATA, _CODE, _MESSAGE)
+
+  def metadata_transmitted(self, original_metadata, transmitted_metadata):
+    return original_metadata is None or grpc_test_common.metadata_transmitted(
+        original_metadata, transmitted_metadata)
+
+  def completion_transmitted(self, original_completion, transmitted_completion):
+    if (original_completion.terminal_metadata is not None and
+        not grpc_test_common.metadata_transmitted(
+            original_completion.terminal_metadata,
+            transmitted_completion.terminal_metadata)):
+        return False
+    elif original_completion.code is not transmitted_completion.code:
+      return False
+    elif original_completion.message != transmitted_completion.message:
+      return False
+    else:
+      return True
+
+
+def setUpModule():
+  logging.warn('setUpModule!')
+
+
+def tearDownModule():
+  logging.warn('tearDownModule!')
+
+
+def load_tests(loader, tests, pattern):
+  return unittest.TestSuite(
+      tests=tuple(
+          loader.loadTestsFromTestCase(test_case_class)
+          for test_case_class in test_cases.test_cases(_Implementation())))
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/_links/_lonely_invocation_link_test.py b/src/python/grpcio_test/grpc_test/_links/_lonely_invocation_link_test.py
index abe240e07a..373a2b2a1f 100644
--- a/src/python/grpcio_test/grpc_test/_links/_lonely_invocation_link_test.py
+++ b/src/python/grpcio_test/grpc_test/_links/_lonely_invocation_link_test.py
@@ -66,7 +66,7 @@ class LonelyInvocationLinkTest(unittest.TestCase):
     ticket = links.Ticket(
         test_operation_id, 0, test_group, test_method,
         links.Ticket.Subscription.FULL, test_constants.SHORT_TIMEOUT, 1, None,
-        None, None, None, None, termination)
+        None, None, None, None, termination, None)
     invocation_link.accept_ticket(ticket)
     invocation_link_mate.block_until_tickets_satisfy(test_cases.terminated)
 
diff --git a/src/python/grpcio_test/grpc_test/_links/_transmission_test.py b/src/python/grpcio_test/grpc_test/_links/_transmission_test.py
index 9cdc9620f0..02ddd512c2 100644
--- a/src/python/grpcio_test/grpc_test/_links/_transmission_test.py
+++ b/src/python/grpcio_test/grpc_test/_links/_transmission_test.py
@@ -128,14 +128,14 @@ class RoundTripTest(unittest.TestCase):
     invocation_ticket = links.Ticket(
         test_operation_id, 0, test_group, test_method,
         links.Ticket.Subscription.FULL, test_constants.LONG_TIMEOUT, None, None,
-        None, None, None, None, links.Ticket.Termination.COMPLETION)
+        None, None, None, None, links.Ticket.Termination.COMPLETION, None)
     invocation_link.accept_ticket(invocation_ticket)
     service_mate.block_until_tickets_satisfy(test_cases.terminated)
 
     service_ticket = links.Ticket(
         service_mate.tickets()[-1].operation_id, 0, None, None, None, None,
         None, None, None, None, test_code, test_message,
-        links.Ticket.Termination.COMPLETION)
+        links.Ticket.Termination.COMPLETION, None)
     service_link.accept_ticket(service_ticket)
     invocation_mate.block_until_tickets_satisfy(test_cases.terminated)
 
@@ -174,33 +174,34 @@ class RoundTripTest(unittest.TestCase):
     invocation_ticket = links.Ticket(
         test_operation_id, 0, test_group, test_method,
         links.Ticket.Subscription.FULL, test_constants.LONG_TIMEOUT, None, None,
-        None, None, None, None, None)
+        None, None, None, None, None, None)
     invocation_link.accept_ticket(invocation_ticket)
     requests = scenario.requests()
     for request_index, request in enumerate(requests):
       request_ticket = links.Ticket(
           test_operation_id, 1 + request_index, None, None, None, None, 1, None,
-          request, None, None, None, None)
+          request, None, None, None, None, None)
       invocation_link.accept_ticket(request_ticket)
       service_mate.block_until_tickets_satisfy(
           test_cases.at_least_n_payloads_received_predicate(1 + request_index))
       response_ticket = links.Ticket(
           service_mate.tickets()[0].operation_id, request_index, None, None,
           None, None, 1, None, scenario.response_for_request(request), None,
-          None, None, None)
+          None, None, None, None)
       service_link.accept_ticket(response_ticket)
       invocation_mate.block_until_tickets_satisfy(
           test_cases.at_least_n_payloads_received_predicate(1 + request_index))
     request_count = len(requests)
     invocation_completion_ticket = links.Ticket(
         test_operation_id, request_count + 1, None, None, None, None, None,
-        None, None, None, None, None, links.Ticket.Termination.COMPLETION)
+        None, None, None, None, None, links.Ticket.Termination.COMPLETION,
+        None)
     invocation_link.accept_ticket(invocation_completion_ticket)
     service_mate.block_until_tickets_satisfy(test_cases.terminated)
     service_completion_ticket = links.Ticket(
         service_mate.tickets()[0].operation_id, request_count, None, None, None,
         None, None, None, None, None, test_code, test_message,
-        links.Ticket.Termination.COMPLETION)
+        links.Ticket.Termination.COMPLETION, None)
     service_link.accept_ticket(service_completion_ticket)
     invocation_mate.block_until_tickets_satisfy(test_cases.terminated)
 
diff --git a/src/python/grpcio_test/grpc_test/framework/core/__init__.py b/src/python/grpcio_test/grpc_test/framework/core/__init__.py
new file mode 100644
index 0000000000..7086519106
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/framework/core/__init__.py
@@ -0,0 +1,30 @@
+# 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.
+
+
diff --git a/src/python/grpcio_test/grpc_test/framework/core/_base_interface_test.py b/src/python/grpcio_test/grpc_test/framework/core/_base_interface_test.py
new file mode 100644
index 0000000000..8d72f131d5
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/framework/core/_base_interface_test.py
@@ -0,0 +1,96 @@
+# 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 the RPC Framework Core's implementation of the Base interface."""
+
+import logging
+import random
+import time
+import unittest
+
+from grpc.framework.core import implementations
+from grpc.framework.interfaces.base import utilities
+from grpc_test.framework.common import test_constants
+from grpc_test.framework.interfaces.base import test_cases
+from grpc_test.framework.interfaces.base import test_interfaces
+
+
+class _Implementation(test_interfaces.Implementation):
+
+  def __init__(self):
+    self._invocation_initial_metadata = object()
+    self._service_initial_metadata = object()
+    self._invocation_terminal_metadata = object()
+    self._service_terminal_metadata = object()
+
+  def instantiate(self, serializations, servicer):
+    invocation = implementations.invocation_end_link()
+    service = implementations.service_end_link(
+        servicer, test_constants.DEFAULT_TIMEOUT,
+        test_constants.MAXIMUM_TIMEOUT)
+    invocation.join_link(service)
+    service.join_link(invocation)
+    return invocation, service, None
+
+  def destantiate(self, memo):
+    pass
+
+  def invocation_initial_metadata(self):
+    return self._invocation_initial_metadata
+
+  def service_initial_metadata(self):
+    return self._service_initial_metadata
+
+  def invocation_completion(self):
+    return utilities.completion(self._invocation_terminal_metadata, None, None)
+
+  def service_completion(self):
+    return utilities.completion(self._service_terminal_metadata, None, None)
+
+  def metadata_transmitted(self, original_metadata, transmitted_metadata):
+    return transmitted_metadata is original_metadata
+
+  def completion_transmitted(self, original_completion, transmitted_completion):
+    return (
+        (original_completion.terminal_metadata is
+         transmitted_completion.terminal_metadata) and
+        original_completion.code is transmitted_completion.code and
+        original_completion.message is transmitted_completion.message
+    )
+
+
+def load_tests(loader, tests, pattern):
+  return unittest.TestSuite(
+      tests=tuple(
+          loader.loadTestsFromTestCase(test_case_class)
+          for test_case_class in test_cases.test_cases(_Implementation())))
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
index dd332fe5dd..5c8b176da4 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
@@ -211,8 +211,10 @@ class _OperationTest(unittest.TestCase):
       elif instruction.kind is _control.Instruction.Kind.CONCLUDE:
         break
 
-    invocation_end.stop_gracefully()
-    service_end.stop_gracefully()
+    invocation_stop_event = invocation_end.stop(0)
+    service_stop_event = service_end.stop(0)
+    invocation_stop_event.wait()
+    service_stop_event.wait()
     invocation_stats = invocation_end.operation_stats()
     service_stats = service_end.operation_stats()
 
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_cases.py
index 1e575d1a9e..ecf49d9cdb 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_cases.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_cases.py
@@ -300,7 +300,7 @@ class TransmissionTest(object):
         invocation_operation_id, 0, _TRANSMISSION_GROUP, _TRANSMISSION_METHOD,
         links.Ticket.Subscription.FULL, timeout, 0, invocation_initial_metadata,
         invocation_payload, invocation_terminal_metadata, invocation_code,
-        invocation_message, links.Ticket.Termination.COMPLETION)
+        invocation_message, links.Ticket.Termination.COMPLETION, None)
     self._invocation_link.accept_ticket(original_invocation_ticket)
 
     self._service_mate.block_until_tickets_satisfy(
@@ -317,7 +317,7 @@ class TransmissionTest(object):
         service_operation_id, 0, None, None, links.Ticket.Subscription.FULL,
         timeout, 0, service_initial_metadata, service_payload,
         service_terminal_metadata, service_code, service_message,
-        links.Ticket.Termination.COMPLETION)
+        links.Ticket.Termination.COMPLETION, None)
     self._service_link.accept_ticket(original_service_ticket)
     self._invocation_mate.block_until_tickets_satisfy(terminated)
     self._assert_is_valid_service_sequence(
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py b/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py
index 6c2e3346aa..39c7f2fc63 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py
@@ -29,9 +29,42 @@
 
 """State and behavior appropriate for use in tests."""
 
+import logging
 import threading
+import time
 
 from grpc.framework.interfaces.links import links
+from grpc.framework.interfaces.links import utilities
+
+# A more-or-less arbitrary limit on the length of raw data values to be logged.
+_UNCOMFORTABLY_LONG = 48
+
+
+def _safe_for_log_ticket(ticket):
+  """Creates a safe-for-printing-to-the-log ticket for a given ticket.
+
+  Args:
+    ticket: Any links.Ticket.
+
+  Returns:
+    A links.Ticket that is as much as can be equal to the given ticket but
+      possibly features values like the string "<payload of length 972321>" in
+      place of the actual values of the given ticket.
+  """
+  if isinstance(ticket.payload, (basestring,)):
+    payload_length = len(ticket.payload)
+  else:
+    payload_length = -1
+  if payload_length < _UNCOMFORTABLY_LONG:
+    return ticket
+  else:
+    return links.Ticket(
+        ticket.operation_id, ticket.sequence_number,
+        ticket.group, ticket.method, ticket.subscription, ticket.timeout,
+        ticket.allowance, ticket.initial_metadata,
+        '<payload of length {}>'.format(payload_length),
+        ticket.terminal_metadata, ticket.code, ticket.message,
+        ticket.termination, None)
 
 
 class RecordingLink(links.Link):
@@ -64,3 +97,71 @@ class RecordingLink(links.Link):
     """Returns a copy of the list of all tickets received by this Link."""
     with self._condition:
       return tuple(self._tickets)
+
+
+class _Pipe(object):
+  """A conduit that logs all tickets passed through it."""
+
+  def __init__(self, name):
+    self._lock = threading.Lock()
+    self._name = name
+    self._left_mate = utilities.NULL_LINK
+    self._right_mate = utilities.NULL_LINK
+
+  def accept_left_to_right_ticket(self, ticket):
+    with self._lock:
+      logging.warning(
+          '%s: moving left to right through %s: %s', time.time(), self._name,
+          _safe_for_log_ticket(ticket))
+      try:
+        self._right_mate.accept_ticket(ticket)
+      except Exception as e:  # pylint: disable=broad-except
+        logging.exception(e)
+
+  def accept_right_to_left_ticket(self, ticket):
+    with self._lock:
+      logging.warning(
+          '%s: moving right to left through %s: %s', time.time(), self._name,
+          _safe_for_log_ticket(ticket))
+      try:
+        self._left_mate.accept_ticket(ticket)
+      except Exception as e:  # pylint: disable=broad-except
+        logging.exception(e)
+
+  def join_left_mate(self, left_mate):
+    with self._lock:
+      self._left_mate = utilities.NULL_LINK if left_mate is None else left_mate
+
+  def join_right_mate(self, right_mate):
+    with self._lock:
+      self._right_mate = (
+          utilities.NULL_LINK if right_mate is None else right_mate)
+
+
+class _Facade(links.Link):
+
+  def __init__(self, accept, join):
+    self._accept = accept
+    self._join = join
+
+  def accept_ticket(self, ticket):
+    self._accept(ticket)
+
+  def join_link(self, link):
+    self._join(link)
+
+
+def logging_links(name):
+  """Creates a conduit that logs all tickets passed through it.
+
+  Args:
+    name: A name to use for the conduit to identify itself in logging output.
+
+  Returns:
+    Two links.Links, the first of which is the "left" side of the conduit
+      and the second of which is the "right" side of the conduit.
+  """
+  pipe = _Pipe(name)
+  left_facade = _Facade(pipe.accept_left_to_right_ticket, pipe.join_left_mate)
+  right_facade = _Facade(pipe.accept_right_to_left_ticket, pipe.join_right_mate)
+  return left_facade, right_facade