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"""Tests of channel arguments on client/server side.""" 15 16from concurrent import futures 17import logging 18import unittest 19 20import grpc 21 22 23class TestPointerWrapper(object): 24 def __int__(self): 25 return 123456 26 27 28TEST_CHANNEL_ARGS = ( 29 ("arg1", b"bytes_val"), 30 ("arg2", "str_val"), 31 ("arg3", 1), 32 (b"arg4", "str_val"), 33 ("arg6", TestPointerWrapper()), 34) 35 36INVALID_TEST_CHANNEL_ARGS = [ 37 {"foo": "bar"}, 38 (("key",),), 39 "str", 40] 41 42 43class ChannelArgsTest(unittest.TestCase): 44 def test_client(self): 45 grpc.insecure_channel("localhost:8080", options=TEST_CHANNEL_ARGS) 46 47 def test_server(self): 48 grpc.server( 49 futures.ThreadPoolExecutor(max_workers=1), options=TEST_CHANNEL_ARGS 50 ) 51 52 def test_invalid_client_args(self): 53 for invalid_arg in INVALID_TEST_CHANNEL_ARGS: 54 self.assertRaises( 55 ValueError, 56 grpc.insecure_channel, 57 "localhost:8080", 58 options=invalid_arg, 59 ) 60 61 62if __name__ == "__main__": 63 logging.basicConfig() 64 unittest.main(verbosity=2) 65