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             'unary_unary_rpc_method_handler',
  67             'unary_stream_rpc_method_handler',
  68             'stream_unary_rpc_method_handler',
  69             'ClientCallDetails',
  70             'stream_stream_rpc_method_handler',
  71             'method_handlers_generic_handler',
  72             'ssl_channel_credentials',
  73             'metadata_call_credentials',
  74             'access_token_call_credentials',
  75             'composite_call_credentials',
  76             'composite_channel_credentials',
  77             'ssl_server_credentials',
  78             'ssl_server_certificate_configuration',
  79             'dynamic_ssl_server_credentials',
  80             'channel_ready_future',
  81             'insecure_channel',
  82             'secure_channel',
  83             'intercept_channel',
  84             'server',
  85         )
  86
  87         six.assertCountEqual(self, expected_grpc_code_elements,
  88                              _from_grpc_import_star.GRPC_ELEMENTS)
  89
  90
  91 class ChannelConnectivityTest(unittest.TestCase):
  92
  93     def testChannelConnectivity(self):
  94         self.assertSequenceEqual((
  95             grpc.ChannelConnectivity.IDLE,
  96             grpc.ChannelConnectivity.CONNECTING,
  97             grpc.ChannelConnectivity.READY,
  98             grpc.ChannelConnectivity.TRANSIENT_FAILURE,
  99             grpc.ChannelConnectivity.SHUTDOWN,
 100         ), tuple(grpc.ChannelConnectivity))
 101
 102
 103 class ChannelTest(unittest.TestCase):
 104
 105     def test_secure_channel(self):
 106         channel_credentials = grpc.ssl_channel_credentials()
 107         channel = grpc.secure_channel('google.com:443', channel_credentials)
 108         channel.close()
 109
 110
 111 if __name__ == '__main__':
 112     logging.basicConfig()
 113     unittest.main(verbosity=2)