1# Copyright 2019 The 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 15import argparse 16import asyncio 17import logging 18import os 19import sys 20 21import grpc 22from grpc.experimental import aio 23 24from tests.interop import client as interop_client_lib 25from tests_aio.interop import methods 26 27_LOGGER = logging.getLogger(__name__) 28_LOGGER.setLevel(logging.DEBUG) 29 30 31def _create_channel(args): 32 target = f"{args.server_host}:{args.server_port}" 33 34 if ( 35 args.use_tls 36 or args.use_alts 37 or args.custom_credentials_type is not None 38 ): 39 ( 40 channel_credentials, 41 options, 42 ) = interop_client_lib.get_secure_channel_parameters(args) 43 return aio.secure_channel(target, channel_credentials, options) 44 else: 45 return aio.insecure_channel(target) 46 47 48def _test_case_from_arg(test_case_arg): 49 for test_case in methods.TestCase: 50 if test_case_arg == test_case.value: 51 return test_case 52 else: 53 raise ValueError('No test case "%s"!' % test_case_arg) 54 55 56async def test_interoperability(): 57 args = interop_client_lib.parse_interop_client_args(sys.argv) 58 channel = _create_channel(args) 59 stub = interop_client_lib.create_stub(channel, args) 60 test_case = _test_case_from_arg(args.test_case) 61 await methods.test_interoperability(test_case, stub, args) 62 63 64if __name__ == "__main__": 65 logging.basicConfig(level=logging.DEBUG) 66 asyncio.get_event_loop().set_debug(True) 67 asyncio.get_event_loop().run_until_complete(test_interoperability()) 68