Merge github.com:grpc/grpc into immolating-conversion

27 files changed, 1372 insertions, 87 deletions

diff --git a/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py b/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py
index 27a5b82e9c..90ad0b9bcb 100644
--- a/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py
+++ b/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py
@@ -191,7 +191,7 @@ class EchoTest(unittest.TestCase):
                      metadata[server_leading_binary_metadata_key])
 
     for datum in test_data:
-      client_call.write(datum, write_tag)
+      client_call.write(datum, write_tag, _low.WriteFlags.WRITE_NO_COMPRESS)
       write_accepted = self.client_events.get()
       self.assertIsNotNone(write_accepted)
       self.assertIs(write_accepted.kind, _low.Event.Kind.WRITE_ACCEPTED)
@@ -206,7 +206,7 @@ class EchoTest(unittest.TestCase):
       self.assertIsNotNone(read_accepted.bytes)
       server_data.append(read_accepted.bytes)
 
-      server_call.write(read_accepted.bytes, write_tag)
+      server_call.write(read_accepted.bytes, write_tag, 0)
       write_accepted = self.server_events.get()
       self.assertIsNotNone(write_accepted)
       self.assertEqual(_low.Event.Kind.WRITE_ACCEPTED, write_accepted.kind)
@@ -370,14 +370,14 @@ class CancellationTest(unittest.TestCase):
     self.assertIsNotNone(metadata_accepted)
 
     for datum in test_data:
-      client_call.write(datum, write_tag)
+      client_call.write(datum, write_tag, 0)
       write_accepted = self.client_events.get()
 
       server_call.read(read_tag)
       read_accepted = self.server_events.get()
       server_data.append(read_accepted.bytes)
 
-      server_call.write(read_accepted.bytes, write_tag)
+      server_call.write(read_accepted.bytes, write_tag, 0)
       write_accepted = self.server_events.get()
       self.assertIsNotNone(write_accepted)
 
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
index 72b1ae5642..cafb6b6eae 100644
--- 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
@@ -27,7 +27,7 @@
 # (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."""
+"""Tests Base interface compliance of the core-over-gRPC-links stack."""
 
 import collections
 import logging
@@ -38,6 +38,7 @@ import unittest
 from grpc._adapter import _intermediary_low
 from grpc._links import invocation
 from grpc._links import service
+from grpc.beta import interfaces as beta_interfaces
 from grpc.framework.core import implementations
 from grpc.framework.interfaces.base import utilities
 from grpc_test import test_common as grpc_test_common
@@ -45,12 +46,6 @@ 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(
@@ -95,10 +90,10 @@ class _Implementation(test_interfaces.Implementation):
     service_grpc_link = service.service_link(
         serialization_behaviors.request_deserializers,
         serialization_behaviors.response_serializers)
-    port = service_grpc_link.add_port(0, None)
+    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',
+        channel, b'localhost', None,
         serialization_behaviors.request_serializers,
         serialization_behaviors.response_deserializers)
 
@@ -114,19 +109,22 @@ class _Implementation(test_interfaces.Implementation):
   def destantiate(self, memo):
     invocation_grpc_link, service_grpc_link = memo
     invocation_grpc_link.stop()
-    service_grpc_link.stop_gracefully()
+    service_grpc_link.begin_stop()
+    service_grpc_link.end_stop()
 
   def invocation_initial_metadata(self):
-    return _INVOCATION_INITIAL_METADATA
+    return grpc_test_common.INVOCATION_INITIAL_METADATA
 
   def service_initial_metadata(self):
-    return _SERVICE_INITIAL_METADATA
+    return grpc_test_common.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)
+    return utilities.completion(
+        grpc_test_common.SERVICE_TERMINAL_METADATA,
+        beta_interfaces.StatusCode.OK, grpc_test_common.DETAILS)
 
   def metadata_transmitted(self, original_metadata, transmitted_metadata):
     return original_metadata is None or grpc_test_common.metadata_transmitted(
@@ -146,14 +144,6 @@ class _Implementation(test_interfaces.Implementation):
       return True
 
 
-def setUpModule():
-  logging.warn('setUpModule!')
-
-
-def tearDownModule():
-  logging.warn('tearDownModule!')
-
-
 def load_tests(loader, tests, pattern):
   return unittest.TestSuite(
       tests=tuple(
diff --git a/src/python/grpcio_test/grpc_test/_crust_over_core_over_links_face_interface_test.py b/src/python/grpcio_test/grpc_test/_crust_over_core_over_links_face_interface_test.py
new file mode 100644
index 0000000000..a4d4dee38c
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/_crust_over_core_over_links_face_interface_test.py
@@ -0,0 +1,161 @@
+# 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 compliance of the crust-over-core-over-gRPC-links stack."""
+
+import collections
+import unittest
+
+from grpc._adapter import _intermediary_low
+from grpc._links import invocation
+from grpc._links import service
+from grpc.beta import interfaces as beta_interfaces
+from grpc.framework.core import implementations as core_implementations
+from grpc.framework.crust import implementations as crust_implementations
+from grpc.framework.foundation import logging_pool
+from grpc.framework.interfaces.links 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.face import test_cases
+from grpc_test.framework.interfaces.face import test_interfaces
+
+
+class _SerializationBehaviors(
+    collections.namedtuple(
+        '_SerializationBehaviors',
+        ('request_serializers', 'request_deserializers', 'response_serializers',
+         'response_deserializers',))):
+  pass
+
+
+def _serialization_behaviors_from_test_methods(test_methods):
+  request_serializers = {}
+  request_deserializers = {}
+  response_serializers = {}
+  response_deserializers = {}
+  for (group, method), test_method in test_methods.iteritems():
+    request_serializers[group, method] = test_method.serialize_request
+    request_deserializers[group, method] = test_method.deserialize_request
+    response_serializers[group, method] = test_method.serialize_response
+    response_deserializers[group, method] = test_method.deserialize_response
+  return _SerializationBehaviors(
+      request_serializers, request_deserializers, response_serializers,
+      response_deserializers)
+
+
+class _Implementation(test_interfaces.Implementation):
+
+  def instantiate(
+      self, methods, method_implementations, multi_method_implementation):
+    pool = logging_pool.pool(test_constants.POOL_SIZE)
+    servicer = crust_implementations.servicer(
+        method_implementations, multi_method_implementation, pool)
+    serialization_behaviors = _serialization_behaviors_from_test_methods(
+        methods)
+    invocation_end_link = core_implementations.invocation_end_link()
+    service_end_link = core_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', None,
+        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_grpc_link.join_link(service_end_link)
+    service_end_link.join_link(service_grpc_link)
+    service_end_link.start()
+    invocation_end_link.start()
+    invocation_grpc_link.start()
+    service_grpc_link.start()
+
+    generic_stub = crust_implementations.generic_stub(invocation_end_link, pool)
+    # TODO(nathaniel): Add a "groups" attribute to _digest.TestServiceDigest.
+    group = next(iter(methods))[0]
+    # TODO(nathaniel): Add a "cardinalities_by_group" attribute to
+    # _digest.TestServiceDigest.
+    cardinalities = {
+        method: method_object.cardinality()
+        for (group, method), method_object in methods.iteritems()}
+    dynamic_stub = crust_implementations.dynamic_stub(
+        invocation_end_link, group, cardinalities, pool)
+
+    return generic_stub, {group: dynamic_stub}, (
+        invocation_end_link, invocation_grpc_link, service_grpc_link,
+        service_end_link, pool)
+
+  def destantiate(self, memo):
+    (invocation_end_link, invocation_grpc_link, service_grpc_link,
+     service_end_link, pool) = memo
+    invocation_end_link.stop(0).wait()
+    invocation_grpc_link.stop()
+    service_grpc_link.begin_stop()
+    service_end_link.stop(0).wait()
+    service_grpc_link.end_stop()
+    invocation_end_link.join_link(utilities.NULL_LINK)
+    invocation_grpc_link.join_link(utilities.NULL_LINK)
+    service_grpc_link.join_link(utilities.NULL_LINK)
+    service_end_link.join_link(utilities.NULL_LINK)
+    pool.shutdown(wait=True)
+
+  def invocation_metadata(self):
+    return grpc_test_common.INVOCATION_INITIAL_METADATA
+
+  def initial_metadata(self):
+    return grpc_test_common.SERVICE_INITIAL_METADATA
+
+  def terminal_metadata(self):
+    return grpc_test_common.SERVICE_TERMINAL_METADATA
+
+  def code(self):
+    return beta_interfaces.StatusCode.OK
+
+  def details(self):
+    return grpc_test_common.DETAILS
+
+  def metadata_transmitted(self, original_metadata, transmitted_metadata):
+    return original_metadata is None or grpc_test_common.metadata_transmitted(
+        original_metadata, transmitted_metadata)
+
+
+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 373a2b2a1f..8e12e8cc22 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
@@ -45,7 +45,8 @@ class LonelyInvocationLinkTest(unittest.TestCase):
 
   def testUpAndDown(self):
     channel = _intermediary_low.Channel('nonexistent:54321', None)
-    invocation_link = invocation.invocation_link(channel, 'nonexistent', {}, {})
+    invocation_link = invocation.invocation_link(
+        channel, 'nonexistent', None, {}, {})
 
     invocation_link.start()
     invocation_link.stop()
@@ -58,8 +59,7 @@ class LonelyInvocationLinkTest(unittest.TestCase):
 
     channel = _intermediary_low.Channel('nonexistent:54321', None)
     invocation_link = invocation.invocation_link(
-        channel, 'nonexistent', {(test_group, test_method): _NULL_BEHAVIOR},
-        {(test_group, test_method): _NULL_BEHAVIOR})
+        channel, 'nonexistent', None, {}, {})
     invocation_link.join_link(invocation_link_mate)
     invocation_link.start()
 
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 db011bca66..77e83d5561 100644
--- a/src/python/grpcio_test/grpc_test/_links/_transmission_test.py
+++ b/src/python/grpcio_test/grpc_test/_links/_transmission_test.py
@@ -34,6 +34,7 @@ import unittest
 from grpc._adapter import _intermediary_low
 from grpc._links import invocation
 from grpc._links import service
+from grpc.beta import interfaces as beta_interfaces
 from grpc.framework.interfaces.links import links
 from grpc_test import test_common
 from grpc_test._links import _proto_scenarios
@@ -50,11 +51,11 @@ class TransmissionTest(test_cases.TransmissionTest, unittest.TestCase):
     service_link = service.service_link(
         {self.group_and_method(): self.deserialize_request},
         {self.group_and_method(): self.serialize_response})
-    port = service_link.add_port(0, None)
+    port = service_link.add_port('[::]:0', None)
     service_link.start()
     channel = _intermediary_low.Channel('localhost:%d' % port, None)
     invocation_link = invocation.invocation_link(
-        channel, 'localhost',
+        channel, 'localhost', None,
         {self.group_and_method(): self.serialize_request},
         {self.group_and_method(): self.deserialize_response})
     invocation_link.start()
@@ -62,7 +63,8 @@ class TransmissionTest(test_cases.TransmissionTest, unittest.TestCase):
 
   def destroy_transmitting_links(self, invocation_side_link, service_side_link):
     invocation_side_link.stop()
-    service_side_link.stop_gracefully()
+    service_side_link.begin_stop()
+    service_side_link.end_stop()
 
   def create_invocation_initial_metadata(self):
     return (
@@ -92,7 +94,8 @@ class TransmissionTest(test_cases.TransmissionTest, unittest.TestCase):
     return None, None
 
   def create_service_completion(self):
-    return _intermediary_low.Code.OK, 'An exuberant test "details" message!'
+    return (
+        beta_interfaces.StatusCode.OK, b'An exuberant test "details" message!')
 
   def assertMetadataTransmitted(self, original_metadata, transmitted_metadata):
     self.assertTrue(
@@ -109,18 +112,18 @@ class RoundTripTest(unittest.TestCase):
     test_group = 'test package.Test Group'
     test_method = 'test method'
     identity_transformation = {(test_group, test_method): _IDENTITY}
-    test_code = _intermediary_low.Code.OK
+    test_code = beta_interfaces.StatusCode.OK
     test_message = 'a test message'
 
     service_link = service.service_link(
         identity_transformation, identity_transformation)
     service_mate = test_utilities.RecordingLink()
     service_link.join_link(service_mate)
-    port = service_link.add_port(0, None)
+    port = service_link.add_port('[::]:0', None)
     service_link.start()
     channel = _intermediary_low.Channel('localhost:%d' % port, None)
     invocation_link = invocation.invocation_link(
-        channel, 'localhost', identity_transformation, identity_transformation)
+        channel, None, None, identity_transformation, identity_transformation)
     invocation_mate = test_utilities.RecordingLink()
     invocation_link.join_link(invocation_mate)
     invocation_link.start()
@@ -140,7 +143,8 @@ class RoundTripTest(unittest.TestCase):
     invocation_mate.block_until_tickets_satisfy(test_cases.terminated)
 
     invocation_link.stop()
-    service_link.stop_gracefully()
+    service_link.begin_stop()
+    service_link.end_stop()
 
     self.assertIs(
         service_mate.tickets()[-1].termination,
@@ -148,11 +152,13 @@ class RoundTripTest(unittest.TestCase):
     self.assertIs(
         invocation_mate.tickets()[-1].termination,
         links.Ticket.Termination.COMPLETION)
+    self.assertIs(invocation_mate.tickets()[-1].code, test_code)
+    self.assertEqual(invocation_mate.tickets()[-1].message, test_message)
 
   def _perform_scenario_test(self, scenario):
     test_operation_id = object()
     test_group, test_method = scenario.group_and_method()
-    test_code = _intermediary_low.Code.OK
+    test_code = beta_interfaces.StatusCode.OK
     test_message = 'a scenario test message'
 
     service_link = service.service_link(
@@ -160,11 +166,11 @@ class RoundTripTest(unittest.TestCase):
         {(test_group, test_method): scenario.serialize_response})
     service_mate = test_utilities.RecordingLink()
     service_link.join_link(service_mate)
-    port = service_link.add_port(0, None)
+    port = service_link.add_port('[::]:0', None)
     service_link.start()
     channel = _intermediary_low.Channel('localhost:%d' % port, None)
     invocation_link = invocation.invocation_link(
-        channel, 'localhost',
+        channel, 'localhost', None,
         {(test_group, test_method): scenario.serialize_request},
         {(test_group, test_method): scenario.deserialize_response})
     invocation_mate = test_utilities.RecordingLink()
@@ -206,7 +212,8 @@ class RoundTripTest(unittest.TestCase):
     invocation_mate.block_until_tickets_satisfy(test_cases.terminated)
 
     invocation_link.stop()
-    service_link.stop_gracefully()
+    service_link.begin_stop()
+    service_link.end_stop()
 
     observed_requests = tuple(
         ticket.payload for ticket in service_mate.tickets()
diff --git a/src/python/grpcio_test/grpc_test/beta/__init__.py b/src/python/grpcio_test/grpc_test/beta/__init__.py
new file mode 100644
index 0000000000..7086519106
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/beta/__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/beta/_beta_features_test.py b/src/python/grpcio_test/grpc_test/beta/_beta_features_test.py
new file mode 100644
index 0000000000..fad57da9d0
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/beta/_beta_features_test.py
@@ -0,0 +1,232 @@
+# 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 implementations
+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 = implementations.server_options(
+        thread_pool_size=test_constants.POOL_SIZE)
+    self._server = implementations.server(
+        method_implementations, options=server_options)
+    server_credentials = implementations.ssl_server_credentials(
+        [(resources.private_key(), resources.certificate_chain(),),])
+    port = self._server.add_secure_port('[::]:0', server_credentials)
+    self._server.start()
+    self._client_credentials = implementations.ssl_client_credentials(
+        resources.test_root_certificates(), None, None)
+    channel = test_utilities.not_really_secure_channel(
+        'localhost', port, self._client_credentials, _SERVER_HOST_OVERRIDE)
+    stub_options = implementations.stub_options(
+        thread_pool_size=test_constants.POOL_SIZE)
+    self._dynamic_stub = implementations.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)
diff --git a/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py b/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py
new file mode 100644
index 0000000000..b3c05bdb0c
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py
@@ -0,0 +1,191 @@
+# 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 of grpc.beta._connectivity_channel."""
+
+import threading
+import time
+import unittest
+
+from grpc._adapter import _low
+from grpc._adapter import _types
+from grpc.beta import _connectivity_channel
+from grpc.beta import interfaces
+from grpc_test.framework.common import test_constants
+
+
+def _drive_completion_queue(completion_queue):
+  while True:
+    event = completion_queue.next(time.time() + 24 * 60 * 60)
+    if event.type == _types.EventType.QUEUE_SHUTDOWN:
+      break
+
+
+class _Callback(object):
+
+  def __init__(self):
+    self._condition = threading.Condition()
+    self._connectivities = []
+
+  def update(self, connectivity):
+    with self._condition:
+      self._connectivities.append(connectivity)
+      self._condition.notify()
+
+  def connectivities(self):
+    with self._condition:
+      return tuple(self._connectivities)
+
+  def block_until_connectivities_satisfy(self, predicate):
+    with self._condition:
+      while True:
+        connectivities = tuple(self._connectivities)
+        if predicate(connectivities):
+          return connectivities
+        else:
+          self._condition.wait()
+
+
+class ChannelConnectivityTest(unittest.TestCase):
+
+  def test_lonely_channel_connectivity(self):
+    low_channel = _low.Channel('localhost:12345', ())
+    callback = _Callback()
+
+    connectivity_channel = _connectivity_channel.ConnectivityChannel(
+        low_channel)
+    connectivity_channel.subscribe(callback.update, try_to_connect=False)
+    first_connectivities = callback.block_until_connectivities_satisfy(bool)
+    connectivity_channel.subscribe(callback.update, try_to_connect=True)
+    second_connectivities = callback.block_until_connectivities_satisfy(
+        lambda connectivities: 2 <= len(connectivities))
+    # Wait for a connection that will never happen.
+    time.sleep(test_constants.SHORT_TIMEOUT)
+    third_connectivities = callback.connectivities()
+    connectivity_channel.unsubscribe(callback.update)
+    fourth_connectivities = callback.connectivities()
+    connectivity_channel.unsubscribe(callback.update)
+    fifth_connectivities = callback.connectivities()
+
+    self.assertSequenceEqual(
+        (interfaces.ChannelConnectivity.IDLE,), first_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.READY, second_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.READY, third_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.READY, fourth_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.READY, fifth_connectivities)
+
+  def test_immediately_connectable_channel_connectivity(self):
+    server_completion_queue = _low.CompletionQueue()
+    server = _low.Server(server_completion_queue, [])
+    port = server.add_http2_port('[::]:0')
+    server.start()
+    server_completion_queue_thread = threading.Thread(
+        target=_drive_completion_queue, args=(server_completion_queue,))
+    server_completion_queue_thread.start()
+    low_channel = _low.Channel('localhost:%d' % port, ())
+    first_callback = _Callback()
+    second_callback = _Callback()
+
+    connectivity_channel = _connectivity_channel.ConnectivityChannel(
+        low_channel)
+    connectivity_channel.subscribe(first_callback.update, try_to_connect=False)
+    first_connectivities = first_callback.block_until_connectivities_satisfy(
+        bool)
+    # Wait for a connection that will never happen because try_to_connect=True
+    # has not yet been passed.
+    time.sleep(test_constants.SHORT_TIMEOUT)
+    second_connectivities = first_callback.connectivities()
+    connectivity_channel.subscribe(second_callback.update, try_to_connect=True)
+    third_connectivities = first_callback.block_until_connectivities_satisfy(
+        lambda connectivities: 2 <= len(connectivities))
+    fourth_connectivities = second_callback.block_until_connectivities_satisfy(
+        bool)
+    # Wait for a connection that will happen (or may already have happened).
+    first_callback.block_until_connectivities_satisfy(
+        lambda connectivities:
+        interfaces.ChannelConnectivity.READY in connectivities)
+    second_callback.block_until_connectivities_satisfy(
+        lambda connectivities:
+        interfaces.ChannelConnectivity.READY in connectivities)
+    connectivity_channel.unsubscribe(first_callback.update)
+    connectivity_channel.unsubscribe(second_callback.update)
+
+    server.shutdown()
+    server_completion_queue.shutdown()
+    server_completion_queue_thread.join()
+
+    self.assertSequenceEqual(
+        (interfaces.ChannelConnectivity.IDLE,), first_connectivities)
+    self.assertSequenceEqual(
+        (interfaces.ChannelConnectivity.IDLE,), second_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.TRANSIENT_FAILURE, third_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.FATAL_FAILURE, third_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.TRANSIENT_FAILURE,
+        fourth_connectivities)
+    self.assertNotIn(
+        interfaces.ChannelConnectivity.FATAL_FAILURE, fourth_connectivities)
+
+  def test_reachable_then_unreachable_channel_connectivity(self):
+    server_completion_queue = _low.CompletionQueue()
+    server = _low.Server(server_completion_queue, [])
+    port = server.add_http2_port('[::]:0')
+    server.start()
+    server_completion_queue_thread = threading.Thread(
+        target=_drive_completion_queue, args=(server_completion_queue,))
+    server_completion_queue_thread.start()
+    low_channel = _low.Channel('localhost:%d' % port, ())
+    callback = _Callback()
+
+    connectivity_channel = _connectivity_channel.ConnectivityChannel(
+        low_channel)
+    connectivity_channel.subscribe(callback.update, try_to_connect=True)
+    callback.block_until_connectivities_satisfy(
+        lambda connectivities:
+        interfaces.ChannelConnectivity.READY in connectivities)
+    # Now take down the server and confirm that channel readiness is repudiated.
+    server.shutdown()
+    callback.block_until_connectivities_satisfy(
+        lambda connectivities:
+        connectivities[-1] is not interfaces.ChannelConnectivity.READY)
+    connectivity_channel.unsubscribe(callback.update)
+
+    server.shutdown()
+    server_completion_queue.shutdown()
+    server_completion_queue_thread.join()
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/beta/_face_interface_test.py b/src/python/grpcio_test/grpc_test/beta/_face_interface_test.py
new file mode 100644
index 0000000000..aa33e1e6f8
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/beta/_face_interface_test.py
@@ -0,0 +1,138 @@
+# 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 collections
+import unittest
+
+from grpc.beta import implementations
+from grpc.beta import interfaces
+from grpc_test import resources
+from grpc_test import test_common as grpc_test_common
+from grpc_test.beta import test_utilities
+from grpc_test.framework.common import test_constants
+from grpc_test.framework.interfaces.face import test_cases
+from grpc_test.framework.interfaces.face import test_interfaces
+
+_SERVER_HOST_OVERRIDE = 'foo.test.google.fr'
+
+
+class _SerializationBehaviors(
+    collections.namedtuple(
+        '_SerializationBehaviors',
+        ('request_serializers', 'request_deserializers', 'response_serializers',
+         'response_deserializers',))):
+  pass
+
+
+def _serialization_behaviors_from_test_methods(test_methods):
+  request_serializers = {}
+  request_deserializers = {}
+  response_serializers = {}
+  response_deserializers = {}
+  for (group, method), test_method in test_methods.iteritems():
+    request_serializers[group, method] = test_method.serialize_request
+    request_deserializers[group, method] = test_method.deserialize_request
+    response_serializers[group, method] = test_method.serialize_response
+    response_deserializers[group, method] = test_method.deserialize_response
+  return _SerializationBehaviors(
+      request_serializers, request_deserializers, response_serializers,
+      response_deserializers)
+
+
+class _Implementation(test_interfaces.Implementation):
+
+  def instantiate(
+      self, methods, method_implementations, multi_method_implementation):
+    serialization_behaviors = _serialization_behaviors_from_test_methods(
+        methods)
+    # TODO(nathaniel): Add a "groups" attribute to _digest.TestServiceDigest.
+    service = next(iter(methods))[0]
+    # TODO(nathaniel): Add a "cardinalities_by_group" attribute to
+    # _digest.TestServiceDigest.
+    cardinalities = {
+        method: method_object.cardinality()
+        for (group, method), method_object in methods.iteritems()}
+
+    server_options = implementations.server_options(
+        request_deserializers=serialization_behaviors.request_deserializers,
+        response_serializers=serialization_behaviors.response_serializers,
+        thread_pool_size=test_constants.POOL_SIZE)
+    server = implementations.server(
+        method_implementations, options=server_options)
+    server_credentials = implementations.ssl_server_credentials(
+        [(resources.private_key(), resources.certificate_chain(),),])
+    port = server.add_secure_port('[::]:0', server_credentials)
+    server.start()
+    client_credentials = implementations.ssl_client_credentials(
+        resources.test_root_certificates(), None, None)
+    channel = test_utilities.not_really_secure_channel(
+        'localhost', port, client_credentials, _SERVER_HOST_OVERRIDE)
+    stub_options = implementations.stub_options(
+        request_serializers=serialization_behaviors.request_serializers,
+        response_deserializers=serialization_behaviors.response_deserializers,
+        thread_pool_size=test_constants.POOL_SIZE)
+    generic_stub = implementations.generic_stub(channel, options=stub_options)
+    dynamic_stub = implementations.dynamic_stub(
+        channel, service, cardinalities, options=stub_options)
+    return generic_stub, {service: dynamic_stub}, server
+
+  def destantiate(self, memo):
+    memo.stop(test_constants.SHORT_TIMEOUT).wait()
+
+  def invocation_metadata(self):
+    return grpc_test_common.INVOCATION_INITIAL_METADATA
+
+  def initial_metadata(self):
+    return grpc_test_common.SERVICE_INITIAL_METADATA
+
+  def terminal_metadata(self):
+    return grpc_test_common.SERVICE_TERMINAL_METADATA
+
+  def code(self):
+    return interfaces.StatusCode.OK
+
+  def details(self):
+    return grpc_test_common.DETAILS
+
+  def metadata_transmitted(self, original_metadata, transmitted_metadata):
+    return original_metadata is None or grpc_test_common.metadata_transmitted(
+        original_metadata, transmitted_metadata)
+
+
+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/beta/_not_found_test.py b/src/python/grpcio_test/grpc_test/beta/_not_found_test.py
new file mode 100644
index 0000000000..5feb997fef
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/beta/_not_found_test.py
@@ -0,0 +1,75 @@
+# 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 of RPC-method-not-found behavior."""
+
+import unittest
+
+from grpc.beta import implementations
+from grpc.beta import interfaces
+from grpc.framework.interfaces.face import face
+from grpc_test.framework.common import test_constants
+
+
+class NotFoundTest(unittest.TestCase):
+
+  def setUp(self):
+    self._server = implementations.server({})
+    port = self._server.add_insecure_port('[::]:0')
+    channel = implementations.insecure_channel('localhost', port)
+    self._generic_stub = implementations.generic_stub(channel)
+    self._server.start()
+
+  def tearDown(self):
+    self._server.stop(0).wait()
+    self._generic_stub = None
+
+  def test_blocking_unary_unary_not_found(self):
+    with self.assertRaises(face.LocalError) as exception_assertion_context:
+      self._generic_stub.blocking_unary_unary(
+          'groop', 'meffod', b'abc', test_constants.LONG_TIMEOUT,
+          with_call=True)
+    self.assertIs(
+        exception_assertion_context.exception.code,
+        interfaces.StatusCode.UNIMPLEMENTED)
+
+  def test_future_stream_unary_not_found(self):
+    rpc_future = self._generic_stub.future_stream_unary(
+        'grupe', 'mevvod', b'def', test_constants.LONG_TIMEOUT)
+    with self.assertRaises(face.LocalError) as exception_assertion_context:
+      rpc_future.result()
+    self.assertIs(
+        exception_assertion_context.exception.code,
+        interfaces.StatusCode.UNIMPLEMENTED)
+    self.assertIs(
+        rpc_future.exception().code, interfaces.StatusCode.UNIMPLEMENTED)
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/beta/_utilities_test.py b/src/python/grpcio_test/grpc_test/beta/_utilities_test.py
new file mode 100644
index 0000000000..996cea9118
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/beta/_utilities_test.py
@@ -0,0 +1,123 @@
+# 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 of grpc.beta.utilities."""
+
+import threading
+import time
+import unittest
+
+from grpc._adapter import _low
+from grpc._adapter import _types
+from grpc.beta import implementations
+from grpc.beta import utilities
+from grpc.framework.foundation import future
+from grpc_test.framework.common import test_constants
+
+
+def _drive_completion_queue(completion_queue):
+  while True:
+    event = completion_queue.next(time.time() + 24 * 60 * 60)
+    if event.type == _types.EventType.QUEUE_SHUTDOWN:
+      break
+
+
+class _Callback(object):
+
+  def __init__(self):
+    self._condition = threading.Condition()
+    self._value = None
+
+  def accept_value(self, value):
+    with self._condition:
+      self._value = value
+      self._condition.notify_all()
+
+  def block_until_called(self):
+    with self._condition:
+      while self._value is None:
+        self._condition.wait()
+      return self._value
+
+
+class ChannelConnectivityTest(unittest.TestCase):
+
+  def test_lonely_channel_connectivity(self):
+    channel = implementations.insecure_channel('localhost', 12345)
+    callback = _Callback()
+
+    ready_future = utilities.channel_ready_future(channel)
+    ready_future.add_done_callback(callback.accept_value)
+    with self.assertRaises(future.TimeoutError):
+      ready_future.result(test_constants.SHORT_TIMEOUT)
+    self.assertFalse(ready_future.cancelled())
+    self.assertFalse(ready_future.done())
+    self.assertTrue(ready_future.running())
+    ready_future.cancel()
+    value_passed_to_callback = callback.block_until_called()
+    self.assertIs(ready_future, value_passed_to_callback)
+    self.assertTrue(ready_future.cancelled())
+    self.assertTrue(ready_future.done())
+    self.assertFalse(ready_future.running())
+
+  def test_immediately_connectable_channel_connectivity(self):
+    server_completion_queue = _low.CompletionQueue()
+    server = _low.Server(server_completion_queue, [])
+    port = server.add_http2_port('[::]:0')
+    server.start()
+    server_completion_queue_thread = threading.Thread(
+        target=_drive_completion_queue, args=(server_completion_queue,))
+    server_completion_queue_thread.start()
+    channel = implementations.insecure_channel('localhost', port)
+    callback = _Callback()
+
+    try:
+      ready_future = utilities.channel_ready_future(channel)
+      ready_future.add_done_callback(callback.accept_value)
+      self.assertIsNone(
+          ready_future.result(test_constants.SHORT_TIMEOUT))
+      value_passed_to_callback = callback.block_until_called()
+      self.assertIs(ready_future, value_passed_to_callback)
+      self.assertFalse(ready_future.cancelled())
+      self.assertTrue(ready_future.done())
+      self.assertFalse(ready_future.running())
+      # Cancellation after maturity has no effect.
+      ready_future.cancel()
+      self.assertFalse(ready_future.cancelled())
+      self.assertTrue(ready_future.done())
+      self.assertFalse(ready_future.running())
+    finally:
+      ready_future.cancel()
+      server.shutdown()
+      server_completion_queue.shutdown()
+      server_completion_queue_thread.join()
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/beta/test_utilities.py b/src/python/grpcio_test/grpc_test/beta/test_utilities.py
new file mode 100644
index 0000000000..24a8600e12
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/beta/test_utilities.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.
+
+"""Test-appropriate entry points into the gRPC Python Beta API."""
+
+from grpc._adapter import _intermediary_low
+from grpc.beta import implementations
+
+
+def not_really_secure_channel(
+    host, port, client_credentials, server_host_override):
+  """Creates an insecure Channel to a remote host.
+
+  Args:
+    host: The name of the remote host to which to connect.
+    port: The port of the remote host to which to connect.
+    client_credentials: The implementations.ClientCredentials with which to
+      connect.
+    server_host_override: The target name used for SSL host name checking.
+
+  Returns:
+    An implementations.Channel to the remote host through which RPCs may be
+      conducted.
+  """
+  hostport = '%s:%d' % (host, port)
+  intermediary_low_channel = _intermediary_low.Channel(
+      hostport, client_credentials._intermediary_low_credentials,
+      server_host_override=server_host_override)
+  return implementations.Channel(
+      intermediary_low_channel._internal, intermediary_low_channel)
diff --git a/src/python/grpcio_test/grpc_test/credentials/README b/src/python/grpcio_test/grpc_test/credentials/README
new file mode 100644
index 0000000000..cb20dcb49f
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/credentials/README
@@ -0,0 +1 @@
+These are test keys *NOT* to be used in production.
diff --git a/src/python/grpcio_test/grpc_test/credentials/ca.pem b/src/python/grpcio_test/grpc_test/credentials/ca.pem
new file mode 100755
index 0000000000..6c8511a73c
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/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/grpcio_test/grpc_test/credentials/server1.key b/src/python/grpcio_test/grpc_test/credentials/server1.key
new file mode 100755
index 0000000000..143a5b8765
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/credentials/server1.key
@@ -0,0 +1,16 @@
+-----BEGIN PRIVATE KEY-----
+MIICdQIBADANBgkqhkiG9w0BAQEFAASCAl8wggJbAgEAAoGBAOHDFScoLCVJpYDD
+M4HYtIdV6Ake/sMNaaKdODjDMsux/4tDydlumN+fm+AjPEK5GHhGn1BgzkWF+slf
+3BxhrA/8dNsnunstVA7ZBgA/5qQxMfGAq4wHNVX77fBZOgp9VlSMVfyd9N8YwbBY
+AckOeUQadTi2X1S6OgJXgQ0m3MWhAgMBAAECgYAn7qGnM2vbjJNBm0VZCkOkTIWm
+V10okw7EPJrdL2mkre9NasghNXbE1y5zDshx5Nt3KsazKOxTT8d0Jwh/3KbaN+YY
+tTCbKGW0pXDRBhwUHRcuRzScjli8Rih5UOCiZkhefUTcRb6xIhZJuQy71tjaSy0p
+dHZRmYyBYO2YEQ8xoQJBAPrJPhMBkzmEYFtyIEqAxQ/o/A6E+E4w8i+KM7nQCK7q
+K4JXzyXVAjLfyBZWHGM2uro/fjqPggGD6QH1qXCkI4MCQQDmdKeb2TrKRh5BY1LR
+81aJGKcJ2XbcDu6wMZK4oqWbTX2KiYn9GB0woM6nSr/Y6iy1u145YzYxEV/iMwff
+DJULAkB8B2MnyzOg0pNFJqBJuH29bKCcHa8gHJzqXhNO5lAlEbMK95p/P2Wi+4Hd
+aiEIAF1BF326QJcvYKmwSmrORp85AkAlSNxRJ50OWrfMZnBgzVjDx3xG6KsFQVk2
+ol6VhqL6dFgKUORFUWBvnKSyhjJxurlPEahV6oo6+A+mPhFY8eUvAkAZQyTdupP3
+XEFQKctGz+9+gKkemDp7LBBMEMBXrGTLPhpEfcjv/7KPdnFHYmhYeBTBnuVmTVWe
+F98XJ7tIFfJq
+-----END PRIVATE KEY-----
diff --git a/src/python/grpcio_test/grpc_test/credentials/server1.pem b/src/python/grpcio_test/grpc_test/credentials/server1.pem
new file mode 100755
index 0000000000..8e582e571f
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/credentials/server1.pem
@@ -0,0 +1,16 @@
+-----BEGIN CERTIFICATE-----
+MIICmzCCAgSgAwIBAgIBAzANBgkqhkiG9w0BAQUFADBWMQswCQYDVQQGEwJBVTET
+MBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50ZXJuZXQgV2lkZ2l0cyBQ
+dHkgTHRkMQ8wDQYDVQQDDAZ0ZXN0Y2EwHhcNMTQwNzIyMDYwMDU3WhcNMjQwNzE5
+MDYwMDU3WjBkMQswCQYDVQQGEwJVUzERMA8GA1UECBMISWxsaW5vaXMxEDAOBgNV
+BAcTB0NoaWNhZ28xFDASBgNVBAoTC0dvb2dsZSBJbmMuMRowGAYDVQQDFBEqLnRl
+c3QuZ29vZ2xlLmNvbTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEA4cMVJygs
+JUmlgMMzgdi0h1XoCR7+ww1pop04OMMyy7H/i0PJ2W6Y35+b4CM8QrkYeEafUGDO
+RYX6yV/cHGGsD/x02ye6ey1UDtkGAD/mpDEx8YCrjAc1Vfvt8Fk6Cn1WVIxV/J30
+3xjBsFgByQ55RBp1OLZfVLo6AleBDSbcxaECAwEAAaNrMGkwCQYDVR0TBAIwADAL
+BgNVHQ8EBAMCBeAwTwYDVR0RBEgwRoIQKi50ZXN0Lmdvb2dsZS5mcoIYd2F0ZXJ6
+b29pLnRlc3QuZ29vZ2xlLmJlghIqLnRlc3QueW91dHViZS5jb22HBMCoAQMwDQYJ
+KoZIhvcNAQEFBQADgYEAM2Ii0LgTGbJ1j4oqX9bxVcxm+/R5Yf8oi0aZqTJlnLYS
+wXcBykxTx181s7WyfJ49WwrYXo78zTDAnf1ma0fPq3e4mpspvyndLh1a+OarHa1e
+aT0DIIYk7qeEa1YcVljx2KyLd0r1BBAfrwyGaEPVeJQVYWaOJRU2we/KD4ojf9s=
+-----END CERTIFICATE-----
diff --git a/src/python/grpcio_test/grpc_test/framework/_crust_over_core_face_interface_test.py b/src/python/grpcio_test/grpc_test/framework/_crust_over_core_face_interface_test.py
new file mode 100644
index 0000000000..30bb85f6c3
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/framework/_crust_over_core_face_interface_test.py
@@ -0,0 +1,111 @@
+# 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 crust-over-core stack."""
+
+import collections
+import unittest
+
+from grpc.framework.core import implementations as core_implementations
+from grpc.framework.crust import implementations as crust_implementations
+from grpc.framework.foundation import logging_pool
+from grpc.framework.interfaces.links import utilities
+from grpc_test.framework.common import test_constants
+from grpc_test.framework.interfaces.face import test_cases
+from grpc_test.framework.interfaces.face import test_interfaces
+from grpc_test.framework.interfaces.links import test_utilities
+
+
+class _Implementation(test_interfaces.Implementation):
+
+  def instantiate(
+      self, methods, method_implementations, multi_method_implementation):
+    pool = logging_pool.pool(test_constants.POOL_SIZE)
+    servicer = crust_implementations.servicer(
+        method_implementations, multi_method_implementation, pool)
+
+    service_end_link = core_implementations.service_end_link(
+        servicer, test_constants.DEFAULT_TIMEOUT,
+        test_constants.MAXIMUM_TIMEOUT)
+    invocation_end_link = core_implementations.invocation_end_link()
+    invocation_end_link.join_link(service_end_link)
+    service_end_link.join_link(invocation_end_link)
+    service_end_link.start()
+    invocation_end_link.start()
+
+    generic_stub = crust_implementations.generic_stub(invocation_end_link, pool)
+    # TODO(nathaniel): Add a "groups" attribute to _digest.TestServiceDigest.
+    group = next(iter(methods))[0]
+    # TODO(nathaniel): Add a "cardinalities_by_group" attribute to
+    # _digest.TestServiceDigest.
+    cardinalities = {
+        method: method_object.cardinality()
+        for (group, method), method_object in methods.iteritems()}
+    dynamic_stub = crust_implementations.dynamic_stub(
+        invocation_end_link, group, cardinalities, pool)
+
+    return generic_stub, {group: dynamic_stub}, (
+        invocation_end_link, service_end_link, pool)
+
+  def destantiate(self, memo):
+    invocation_end_link, service_end_link, pool = memo
+    invocation_end_link.stop(0).wait()
+    service_end_link.stop(0).wait()
+    invocation_end_link.join_link(utilities.NULL_LINK)
+    service_end_link.join_link(utilities.NULL_LINK)
+    pool.shutdown(wait=True)
+
+  def invocation_metadata(self):
+    return object()
+
+  def initial_metadata(self):
+    return object()
+
+  def terminal_metadata(self):
+    return object()
+
+  def code(self):
+    return object()
+
+  def details(self):
+    return object()
+
+  def metadata_transmitted(self, original_metadata, transmitted_metadata):
+    return original_metadata is transmitted_metadata
+
+
+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/_control.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_control.py
index e4d2a7a0d7..46a01876d8 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/_control.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_control.py
@@ -236,8 +236,8 @@ class Instruction(
     collections.namedtuple(
         'Instruction',
         ('kind', 'advance_args', 'advance_kwargs', 'conclude_success',
-         'conclude_message', 'conclude_invocation_outcome',
-         'conclude_service_outcome',))):
+         'conclude_message', 'conclude_invocation_outcome_kind',
+         'conclude_service_outcome_kind',))):
   """"""
 
   @enum.unique
@@ -532,24 +532,24 @@ class _SequenceController(Controller):
       self._state.service_side_outcome = outcome
       if self._todo is not None or self._remaining_elements:
         self._failed('Premature service-side outcome %s!' % (outcome,))
-      elif outcome is not self._sequence.outcome.service:
+      elif outcome.kind is not self._sequence.outcome_kinds.service:
         self._failed(
-            'Incorrect service-side outcome: %s should have been %s' % (
-                outcome, self._sequence.outcome.service))
+            'Incorrect service-side outcome kind: %s should have been %s' % (
+                outcome.kind, self._sequence.outcome_kinds.service))
       elif self._state.invocation_side_outcome is not None:
-        self._passed(self._state.invocation_side_outcome, outcome)
+        self._passed(self._state.invocation_side_outcome.kind, outcome.kind)
 
   def invocation_on_termination(self, outcome):
     with self._condition:
       self._state.invocation_side_outcome = outcome
       if self._todo is not None or self._remaining_elements:
         self._failed('Premature invocation-side outcome %s!' % (outcome,))
-      elif outcome is not self._sequence.outcome.invocation:
+      elif outcome.kind is not self._sequence.outcome_kinds.invocation:
         self._failed(
-            'Incorrect invocation-side outcome: %s should have been %s' % (
-                outcome, self._sequence.outcome.invocation))
+            'Incorrect invocation-side outcome kind: %s should have been %s' % (
+                outcome.kind, self._sequence.outcome_kinds.invocation))
       elif self._state.service_side_outcome is not None:
-        self._passed(outcome, self._state.service_side_outcome)
+        self._passed(outcome.kind, self._state.service_side_outcome.kind)
 
 
 class _SequenceControllerCreator(ControllerCreator):
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py
index 1d77aaebe6..f547d91681 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py
@@ -103,13 +103,14 @@ class Element(collections.namedtuple('Element', ('kind', 'transmission',))):
     SERVICE_FAILURE = 'service failure'
 
 
-class Outcome(collections.namedtuple('Outcome', ('invocation', 'service',))):
+class OutcomeKinds(
+    collections.namedtuple('Outcome', ('invocation', 'service',))):
   """A description of the expected outcome of an operation test.
 
   Attributes:
-    invocation: The base.Outcome value expected on the invocation side of the
-      operation.
-    service: The base.Outcome value expected on the service side of the
+    invocation: The base.Outcome.Kind value expected on the invocation side of
+      the operation.
+    service: The base.Outcome.Kind value expected on the service side of the
       operation.
   """
 
@@ -117,7 +118,8 @@ class Outcome(collections.namedtuple('Outcome', ('invocation', 'service',))):
 class Sequence(
     collections.namedtuple(
         'Sequence',
-        ('name', 'maximum_duration', 'invocation', 'elements', 'outcome',))):
+        ('name', 'maximum_duration', 'invocation', 'elements',
+         'outcome_kinds',))):
   """Describes at a high level steps to perform in a test.
 
   Attributes:
@@ -128,7 +130,8 @@ class Sequence(
       under test.
     elements: A sequence of Element values describing at coarse granularity
       actions to take during the operation under test.
-    outcome: An Outcome value describing the expected outcome of the test.
+    outcome_kinds: An OutcomeKinds value describing the expected outcome kinds
+      of the test.
   """
 
 _EASY = Sequence(
@@ -139,7 +142,7 @@ _EASY = Sequence(
         Element(
             Element.Kind.SERVICE_TRANSMISSION, Transmission(True, True, True)),
     ),
-    Outcome(base.Outcome.COMPLETED, base.Outcome.COMPLETED))
+    OutcomeKinds(base.Outcome.Kind.COMPLETED, base.Outcome.Kind.COMPLETED))
 
 _PEASY = Sequence(
     'Peasy',
@@ -154,7 +157,7 @@ _PEASY = Sequence(
         Element(
             Element.Kind.SERVICE_TRANSMISSION, Transmission(False, True, True)),
     ),
-    Outcome(base.Outcome.COMPLETED, base.Outcome.COMPLETED))
+    OutcomeKinds(base.Outcome.Kind.COMPLETED, base.Outcome.Kind.COMPLETED))
 
 
 # TODO(issue 2959): Finish this test suite. This tuple of sequences should
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 87332cf612..ddda1018c3 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
@@ -44,7 +44,8 @@ from grpc_test.framework.interfaces.base import test_interfaces
 
 _SYNCHRONICITY_VARIATION = (('Sync', False), ('Async', True))
 
-_EMPTY_OUTCOME_DICT = {outcome: 0 for outcome in base.Outcome}
+_EMPTY_OUTCOME_KIND_DICT = {
+    outcome_kind: 0 for outcome_kind in base.Outcome.Kind}
 
 
 class _Serialization(test_interfaces.Serialization):
@@ -118,8 +119,19 @@ class _Operator(base.Operator):
           'Deliberately raised exception from Operator.advance (in a test)!')
 
 
+class _ProtocolReceiver(base.ProtocolReceiver):
+
+  def __init__(self):
+    self._condition = threading.Condition()
+    self._contexts = []
+
+  def context(self, protocol_context):
+    with self._condition:
+      self._contexts.append(protocol_context)
+
+
 class _Servicer(base.Servicer):
-  """An base.Servicer with instrumented for testing."""
+  """A base.Servicer with instrumented for testing."""
 
   def __init__(self, group, method, controllers, pool):
     self._condition = threading.Condition()
@@ -143,7 +155,7 @@ class _Servicer(base.Servicer):
             controller.service_on_termination)
         if outcome is not None:
           controller.service_on_termination(outcome)
-        return utilities.full_subscription(operator)
+        return utilities.full_subscription(operator, _ProtocolReceiver())
 
 
 class _OperationTest(unittest.TestCase):
@@ -168,7 +180,8 @@ class _OperationTest(unittest.TestCase):
       test_operator = _Operator(
           self._controller, self._controller.on_invocation_advance,
           self._pool, None)
-      subscription = utilities.full_subscription(test_operator)
+      subscription = utilities.full_subscription(
+          test_operator, _ProtocolReceiver())
     else:
       # TODO(nathaniel): support and test other subscription kinds.
       self.fail('Non-full subscriptions not yet supported!')
@@ -223,11 +236,12 @@ class _OperationTest(unittest.TestCase):
     self.assertTrue(
         instruction.conclude_success, msg=instruction.conclude_message)
 
-    expected_invocation_stats = dict(_EMPTY_OUTCOME_DICT)
-    expected_invocation_stats[instruction.conclude_invocation_outcome] += 1
+    expected_invocation_stats = dict(_EMPTY_OUTCOME_KIND_DICT)
+    expected_invocation_stats[
+        instruction.conclude_invocation_outcome_kind] += 1
     self.assertDictEqual(expected_invocation_stats, invocation_stats)
-    expected_service_stats = dict(_EMPTY_OUTCOME_DICT)
-    expected_service_stats[instruction.conclude_service_outcome] += 1
+    expected_service_stats = dict(_EMPTY_OUTCOME_KIND_DICT)
+    expected_service_stats[instruction.conclude_service_outcome_kind] += 1
     self.assertDictEqual(expected_service_stats, service_stats)
 
 
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_3069_test_constant.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_3069_test_constant.py
new file mode 100644
index 0000000000..363d9ce8f1
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_3069_test_constant.py
@@ -0,0 +1,37 @@
+# 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.
+
+"""A test constant working around issue 3069."""
+
+# test_constants is referenced from specification in this module.
+from grpc_test.framework.common import test_constants  # pylint: disable=unused-import
+
+# TODO(issue 3069): Replace uses of this constant with
+# test_constants.SHORT_TIMEOUT.
+REALLY_SHORT_TIMEOUT = 0.1
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py
index 857ad5cf3e..2d2a081955 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py
@@ -37,6 +37,7 @@ from grpc.framework.interfaces.face import face
 from grpc_test.framework.common import test_constants
 from grpc_test.framework.common import test_control
 from grpc_test.framework.common import test_coverage
+from grpc_test.framework.interfaces.face import _3069_test_constant
 from grpc_test.framework.interfaces.face import _digest
 from grpc_test.framework.interfaces.face import _stock_service
 from grpc_test.framework.interfaces.face import test_interfaces  # pylint: disable=unused-import
@@ -72,6 +73,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
     Overriding implementations must call this implementation.
     """
+    self._invoker = None
     self.implementation.destantiate(self._memo)
 
   def testSuccessfulUnaryRequestUnaryResponse(self):
@@ -80,8 +82,8 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
       for test_messages in test_messages_sequence:
         request = test_messages.request()
 
-        response = self._invoker.blocking(group, method)(
-            request, test_constants.LONG_TIMEOUT)
+        response, call = self._invoker.blocking(group, method)(
+            request, test_constants.LONG_TIMEOUT, with_call=True)
 
         test_messages.verify(request, response, self)
 
@@ -103,8 +105,8 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
       for test_messages in test_messages_sequence:
         requests = test_messages.requests()
 
-        response = self._invoker.blocking(group, method)(
-            iter(requests), test_constants.LONG_TIMEOUT)
+        response, call = self._invoker.blocking(group, method)(
+            iter(requests), test_constants.LONG_TIMEOUT, with_call=True)
 
         test_messages.verify(requests, response, self)
 
@@ -170,7 +172,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         with self._control.pause(), self.assertRaises(
             face.ExpirationError):
           self._invoker.blocking(group, method)(
-              request, test_constants.SHORT_TIMEOUT)
+              request, _3069_test_constant.REALLY_SHORT_TIMEOUT)
 
   def testExpiredUnaryRequestStreamResponse(self):
     for (group, method), test_messages_sequence in (
@@ -181,7 +183,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         with self._control.pause(), self.assertRaises(
             face.ExpirationError):
           response_iterator = self._invoker.blocking(group, method)(
-              request, test_constants.SHORT_TIMEOUT)
+              request, _3069_test_constant.REALLY_SHORT_TIMEOUT)
           list(response_iterator)
 
   def testExpiredStreamRequestUnaryResponse(self):
@@ -193,7 +195,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         with self._control.pause(), self.assertRaises(
             face.ExpirationError):
           self._invoker.blocking(group, method)(
-              iter(requests), test_constants.SHORT_TIMEOUT)
+              iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT)
 
   def testExpiredStreamRequestStreamResponse(self):
     for (group, method), test_messages_sequence in (
@@ -204,7 +206,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         with self._control.pause(), self.assertRaises(
             face.ExpirationError):
           response_iterator = self._invoker.blocking(group, method)(
-              iter(requests), test_constants.SHORT_TIMEOUT)
+              iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT)
           list(response_iterator)
 
   def testFailedUnaryRequestUnaryResponse(self):
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py
index ea5cdeaea3..7cb273bf78 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py
@@ -37,6 +37,7 @@ from grpc.framework.interfaces.face import face
 from grpc_test.framework.common import test_constants
 from grpc_test.framework.common import test_control
 from grpc_test.framework.common import test_coverage
+from grpc_test.framework.interfaces.face import _3069_test_constant
 from grpc_test.framework.interfaces.face import _digest
 from grpc_test.framework.interfaces.face import _receiver
 from grpc_test.framework.interfaces.face import _stock_service
@@ -73,6 +74,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
     Overriding implementations must call this implementation.
     """
+    self._invoker = None
     self.implementation.destantiate(self._memo)
 
   def testSuccessfulUnaryRequestUnaryResponse(self):
@@ -264,7 +266,8 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.pause():
           self._invoker.event(group, method)(
-              request, receiver, receiver.abort, test_constants.SHORT_TIMEOUT)
+              request, receiver, receiver.abort,
+              _3069_test_constant.REALLY_SHORT_TIMEOUT)
           receiver.block_until_terminated()
 
         self.assertIs(face.Abortion.Kind.EXPIRED, receiver.abortion().kind)
@@ -278,7 +281,8 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.pause():
           self._invoker.event(group, method)(
-              request, receiver, receiver.abort, test_constants.SHORT_TIMEOUT)
+              request, receiver, receiver.abort,
+              _3069_test_constant.REALLY_SHORT_TIMEOUT)
           receiver.block_until_terminated()
 
         self.assertIs(face.Abortion.Kind.EXPIRED, receiver.abortion().kind)
@@ -290,7 +294,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         receiver = _receiver.Receiver()
 
         self._invoker.event(group, method)(
-            receiver, receiver.abort, test_constants.SHORT_TIMEOUT)
+            receiver, receiver.abort, _3069_test_constant.REALLY_SHORT_TIMEOUT)
         receiver.block_until_terminated()
 
         self.assertIs(face.Abortion.Kind.EXPIRED, receiver.abortion().kind)
@@ -303,7 +307,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         receiver = _receiver.Receiver()
 
         call_consumer = self._invoker.event(group, method)(
-            receiver, receiver.abort, test_constants.SHORT_TIMEOUT)
+            receiver, receiver.abort, _3069_test_constant.REALLY_SHORT_TIMEOUT)
         for request in requests:
           call_consumer.consume(request)
         receiver.block_until_terminated()
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py
index a649362cef..272a37f15f 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py
@@ -40,6 +40,7 @@ from grpc.framework.interfaces.face import face
 from grpc_test.framework.common import test_constants
 from grpc_test.framework.common import test_control
 from grpc_test.framework.common import test_coverage
+from grpc_test.framework.interfaces.face import _3069_test_constant
 from grpc_test.framework.interfaces.face import _digest
 from grpc_test.framework.interfaces.face import _stock_service
 from grpc_test.framework.interfaces.face import test_interfaces  # pylint: disable=unused-import
@@ -102,6 +103,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
     Overriding implementations must call this implementation.
     """
+    self._invoker = None
     self.implementation.destantiate(self._memo)
     self._digest_pool.shutdown(wait=True)
 
@@ -265,7 +267,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.pause():
           response_future = self._invoker.future(
-              group, method)(request, test_constants.SHORT_TIMEOUT)
+              group, method)(request, _3069_test_constant.REALLY_SHORT_TIMEOUT)
           self.assertIsInstance(
               response_future.exception(), face.ExpirationError)
           with self.assertRaises(face.ExpirationError):
@@ -279,7 +281,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.pause():
           response_iterator = self._invoker.future(group, method)(
-              request, test_constants.SHORT_TIMEOUT)
+              request, _3069_test_constant.REALLY_SHORT_TIMEOUT)
           with self.assertRaises(face.ExpirationError):
             list(response_iterator)
 
@@ -291,7 +293,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.pause():
           response_future = self._invoker.future(group, method)(
-              iter(requests), test_constants.SHORT_TIMEOUT)
+              iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT)
           self.assertIsInstance(
               response_future.exception(), face.ExpirationError)
           with self.assertRaises(face.ExpirationError):
@@ -305,7 +307,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.pause():
           response_iterator = self._invoker.future(group, method)(
-              iter(requests), test_constants.SHORT_TIMEOUT)
+              iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT)
           with self.assertRaises(face.ExpirationError):
             list(response_iterator)
 
@@ -317,7 +319,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.fail():
           response_future = self._invoker.future(group, method)(
-              request, test_constants.SHORT_TIMEOUT)
+              request, _3069_test_constant.REALLY_SHORT_TIMEOUT)
 
           # Because the servicer fails outside of the thread from which the
           # servicer-side runtime called into it its failure is
@@ -340,7 +342,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         # expiration of the RPC.
         with self._control.fail(), self.assertRaises(face.ExpirationError):
           response_iterator = self._invoker.future(group, method)(
-              request, test_constants.SHORT_TIMEOUT)
+              request, _3069_test_constant.REALLY_SHORT_TIMEOUT)
           list(response_iterator)
 
   def testFailedStreamRequestUnaryResponse(self):
@@ -351,7 +353,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         with self._control.fail():
           response_future = self._invoker.future(group, method)(
-              iter(requests), test_constants.SHORT_TIMEOUT)
+              iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT)
 
           # Because the servicer fails outside of the thread from which the
           # servicer-side runtime called into it its failure is
@@ -374,5 +376,5 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         # expiration of the RPC.
         with self._control.fail(), self.assertRaises(face.ExpirationError):
           response_iterator = self._invoker.future(group, method)(
-              iter(requests), test_constants.SHORT_TIMEOUT)
+              iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT)
           list(response_iterator)
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py
index 1dd2ec3633..808e2c4e36 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py
@@ -1,4 +1,4 @@
-B# Copyright 2015, Google Inc.
+# Copyright 2015, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
diff --git a/src/python/grpcio_test/grpc_test/resources.py b/src/python/grpcio_test/grpc_test/resources.py
new file mode 100644
index 0000000000..2c3045313d
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/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/grpcio_test/grpc_test/test_common.py b/src/python/grpcio_test/grpc_test/test_common.py
index f8e1f1e43f..44284be88b 100644
--- a/src/python/grpcio_test/grpc_test/test_common.py
+++ b/src/python/grpcio_test/grpc_test/test_common.py
@@ -31,6 +31,11 @@
 
 import collections
 
+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'),)
+DETAILS = b'test details'
+
 
 def metadata_transmitted(original_metadata, transmitted_metadata):
   """Judges whether or not metadata was acceptably transmitted.