src/python/grpcio_tests/tests/unit/_api_test.py

   1 # Copyright 2016 gRPC authors.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 """Test of gRPC Python's application-layer API."""
  15
  16 import unittest
  17 import logging
  18
  19 import six
  20
  21 import grpc
  22
  23 from tests.unit import _from_grpc_import_star
  24
  25
  26 class AllTest(unittest.TestCase):
  27
  28     def testAll(self):
  29         expected_grpc_code_elements = (
  30             'FutureTimeoutError',
  31             'FutureCancelledError',
  32             'Future',
  33             'ChannelConnectivity',
  34             'Compression',
  35             'StatusCode',
  36             'Status',
  37             'RpcError',
  38             'RpcContext',
  39             'Call',
  40             'ChannelCredentials',
  41             'CallCredentials',
  42             'AuthMetadataContext',
  43             'AuthMetadataPluginCallback',
  44             'AuthMetadataPlugin',
  45             'ServerCertificateConfiguration',
  46             'ServerCredentials',
  47             'UnaryUnaryMultiCallable',
  48             'UnaryStreamMultiCallable',
  49             'StreamUnaryMultiCallable',
  50             'StreamStreamMultiCallable',
  51             'UnaryUnaryClientInterceptor',
  52             'UnaryStreamClientInterceptor',
  53             'StreamUnaryClientInterceptor',
  54             'StreamStreamClientInterceptor',
  55             'Channel',
  56             'ServicerContext',
  57             'RpcMethodHandler',
  58             'HandlerCallDetails',
  59             'GenericRpcHandler',
  60             'ServiceRpcHandler',
  61             'Server',
  62             'ServerInterceptor',
  63             'LocalConnectionType',
  64             'local_channel_credentials',
  65             'local_server_credentials',
  66             'alts_channel_credentials',
  67             'alts_server_credentials',
  68             'unary_unary_rpc_method_handler',
  69             'unary_stream_rpc_method_handler',
  70             'stream_unary_rpc_method_handler',
  71             'ClientCallDetails',
  72             'stream_stream_rpc_method_handler',
  73             'method_handlers_generic_handler',
  74             'ssl_channel_credentials',
  75             'metadata_call_credentials',
  76             'access_token_call_credentials',
  77             'composite_call_credentials',
  78             'composite_channel_credentials',
  79             'ssl_server_credentials',
  80             'ssl_server_certificate_configuration',
  81             'dynamic_ssl_server_credentials',
  82             'channel_ready_future',
  83             'insecure_channel',
  84             'secure_channel',
  85             'intercept_channel',
  86             'server',
  87         )
  88
  89         six.assertCountEqual(self, expected_grpc_code_elements,
  90                              _from_grpc_import_star.GRPC_ELEMENTS)
  91
  92
  93 class ChannelConnectivityTest(unittest.TestCase):
  94
  95     def testChannelConnectivity(self):
  96         self.assertSequenceEqual((
  97             grpc.ChannelConnectivity.IDLE,
  98             grpc.ChannelConnectivity.CONNECTING,
  99             grpc.ChannelConnectivity.READY,
 100             grpc.ChannelConnectivity.TRANSIENT_FAILURE,
 101             grpc.ChannelConnectivity.SHUTDOWN,
 102         ), tuple(grpc.ChannelConnectivity))
 103
 104
 105 class ChannelTest(unittest.TestCase):
 106
 107     def test_secure_channel(self):
 108         channel_credentials = grpc.ssl_channel_credentials()
 109         channel = grpc.secure_channel('google.com:443', channel_credentials)
 110         channel.close()
 111
 112
 113 if __name__ == '__main__':
 114     logging.basicConfig()
 115     unittest.main(verbosity=2)