| # Copyright 2016 gRPC authors. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| """Test of gRPC Python's application-layer API.""" |
| |
| import unittest |
| |
| import six |
| |
| import grpc |
| |
| from tests.unit import _from_grpc_import_star |
| |
| |
| class AllTest(unittest.TestCase): |
| |
| def testAll(self): |
| expected_grpc_code_elements = ( |
| 'FutureTimeoutError', 'FutureCancelledError', 'Future', |
| 'ChannelConnectivity', 'StatusCode', 'RpcError', 'RpcContext', |
| 'Call', 'ChannelCredentials', 'CallCredentials', |
| 'AuthMetadataContext', 'AuthMetadataPluginCallback', |
| 'AuthMetadataPlugin', 'ServerCertificateConfiguration', |
| 'ServerCredentials', 'UnaryUnaryMultiCallable', |
| 'UnaryStreamMultiCallable', 'StreamUnaryMultiCallable', |
| 'StreamStreamMultiCallable', 'UnaryUnaryClientInterceptor', |
| 'UnaryStreamClientInterceptor', 'StreamUnaryClientInterceptor', |
| 'StreamStreamClientInterceptor', 'Channel', 'ServicerContext', |
| 'RpcMethodHandler', 'HandlerCallDetails', 'GenericRpcHandler', |
| 'ServiceRpcHandler', 'Server', 'ServerInterceptor', |
| 'unary_unary_rpc_method_handler', 'unary_stream_rpc_method_handler', |
| 'stream_unary_rpc_method_handler', 'ClientCallDetails', |
| 'stream_stream_rpc_method_handler', |
| 'method_handlers_generic_handler', 'ssl_channel_credentials', |
| 'metadata_call_credentials', 'access_token_call_credentials', |
| 'composite_call_credentials', 'composite_channel_credentials', |
| 'ssl_server_credentials', 'ssl_server_certificate_configuration', |
| 'dynamic_ssl_server_credentials', 'channel_ready_future', |
| 'insecure_channel', 'secure_channel', 'intercept_channel', |
| 'server',) |
| |
| six.assertCountEqual(self, expected_grpc_code_elements, |
| _from_grpc_import_star.GRPC_ELEMENTS) |
| |
| |
| class ChannelConnectivityTest(unittest.TestCase): |
| |
| def testChannelConnectivity(self): |
| self.assertSequenceEqual( |
| (grpc.ChannelConnectivity.IDLE, grpc.ChannelConnectivity.CONNECTING, |
| grpc.ChannelConnectivity.READY, |
| grpc.ChannelConnectivity.TRANSIENT_FAILURE, |
| grpc.ChannelConnectivity.SHUTDOWN,), |
| tuple(grpc.ChannelConnectivity)) |
| |
| |
| class ChannelTest(unittest.TestCase): |
| |
| def test_secure_channel(self): |
| channel_credentials = grpc.ssl_channel_credentials() |
| channel = grpc.secure_channel('google.com:443', channel_credentials) |
| |
| |
| if __name__ == '__main__': |
| unittest.main(verbosity=2) |