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