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 logging
  17 import unittest
  18
  19 import grpc
  20 import six
  21
  22 from tests.unit import _from_grpc_import_star
  23
  24
  25 class AllTest(unittest.TestCase):
  26
  27     def testAll(self):
  28         expected_grpc_code_elements = (
  29             'FutureTimeoutError',
  30             'FutureCancelledError',
  31             'Future',
  32             'ChannelConnectivity',
  33             'Compression',
  34             'StatusCode',
  35             'Status',
  36             'RpcError',
  37             'RpcContext',
  38             'Call',
  39             'ChannelCredentials',
  40             'CallCredentials',
  41             'AuthMetadataContext',
  42             'AuthMetadataPluginCallback',
  43             'AuthMetadataPlugin',
  44             'ServerCertificateConfiguration',
  45             'ServerCredentials',
  46             'UnaryUnaryMultiCallable',
  47             'UnaryStreamMultiCallable',
  48             'StreamUnaryMultiCallable',
  49             'StreamStreamMultiCallable',
  50             'UnaryUnaryClientInterceptor',
  51             'UnaryStreamClientInterceptor',
  52             'StreamUnaryClientInterceptor',
  53             'StreamStreamClientInterceptor',
  54             'Channel',
  55             'ServicerContext',
  56             'RpcMethodHandler',
  57             'HandlerCallDetails',
  58             'GenericRpcHandler',
  59             'ServiceRpcHandler',
  60             'Server',
  61             'ServerInterceptor',
  62             'LocalConnectionType',
  63             'local_channel_credentials',
  64             'local_server_credentials',
  65             'alts_channel_credentials',
  66             'alts_server_credentials',
  67             'unary_unary_rpc_method_handler',
  68             'unary_stream_rpc_method_handler',
  69             'stream_unary_rpc_method_handler',
  70             'ClientCallDetails',
  71             'stream_stream_rpc_method_handler',
  72             'method_handlers_generic_handler',
  73             'ssl_channel_credentials',
  74             'metadata_call_credentials',
  75             'access_token_call_credentials',
  76             'composite_call_credentials',
  77             'composite_channel_credentials',
  78             'compute_engine_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             'protos',
  88             'services',
  89             'protos_and_services',
  90             'xds_channel_credentials',
  91             'xds_server_credentials',
  92             'insecure_server_credentials',
  93         )
  94
  95         six.assertCountEqual(self, expected_grpc_code_elements,
  96                              _from_grpc_import_star.GRPC_ELEMENTS)
  97
  98
  99 class ChannelConnectivityTest(unittest.TestCase):
 100
 101     def testChannelConnectivity(self):
 102         self.assertSequenceEqual((
 103             grpc.ChannelConnectivity.IDLE,
 104             grpc.ChannelConnectivity.CONNECTING,
 105             grpc.ChannelConnectivity.READY,
 106             grpc.ChannelConnectivity.TRANSIENT_FAILURE,
 107             grpc.ChannelConnectivity.SHUTDOWN,
 108         ), tuple(grpc.ChannelConnectivity))
 109
 110
 111 class ChannelTest(unittest.TestCase):
 112
 113     def test_secure_channel(self):
 114         channel_credentials = grpc.ssl_channel_credentials()
 115         channel = grpc.secure_channel('google.com:443', channel_credentials)
 116         channel.close()
 117
 118
 119 if __name__ == '__main__':
 120     logging.basicConfig()
 121     unittest.main(verbosity=2)