本文整理汇总了Python中tchannel.TChannel类的典型用法代码示例。如果您正苦于以下问题：Python TChannel类的具体用法？Python TChannel怎么用？Python TChannel使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了TChannel类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_binary
def test_binary():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testBinary(request):
return request.body.thing
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
resp = yield tchannel.thrift(
service.testBinary(
# this is ThriftTest.Xtruct(string_thing='hi')
'\x0c\x00\x00\x0b\x00\x01\x00\x00\x00\x0bhi\x00\x00'
)
)
assert resp.headers == {}
assert (
resp.body ==
'\x0c\x00\x00\x0b\x00\x01\x00\x00\x00\x0bhi\x00\x00'
)
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:34，代码来源:test_thrift.py
示例2: test_headers_and_body_should_be_optional
def test_headers_and_body_should_be_optional():
# Given this test server:
server = TChannel(name='server')
@server.register(scheme=schemes.RAW)
def endpoint(request):
# assert request.headers is None # TODO uncomment
# assert request.body is None # TODO uncomment
pass
server.listen()
# Make a call:
tchannel = TChannel(name='client')
resp = yield tchannel.call(
scheme=schemes.RAW,
service='server',
arg1='endpoint',
hostport=server.hostport,
)
# verify response
assert isinstance(resp, Response)
assert resp.headers == '' # TODO should be None to match server
assert resp.body == '' # TODO should be None to match server
开发者ID:Willyham，项目名称:tchannel-python，代码行数:29，代码来源:test_tchannel.py
示例3: keyvalue_server
def keyvalue_server(io_loop, keyvalue, keyvalue_data):
server = TChannel(name='keyvalue')
server.listen()
@server.thrift.register(keyvalue.KeyValue)
def getItem(request):
assert request.service == 'keyvalue'
key = request.body.key
if key in keyvalue_data:
assert request.headers == {'expect': 'success'}
return keyvalue_data[key]
else:
assert request.headers == {'expect': 'failure'}
raise keyvalue.ItemDoesNotExist(key)
@server.json.register('putItem')
def json_put_item(request):
assert request.service == 'keyvalue'
assert request.timeout == 0.5
key = request.body['key']
value = request.body['value']
keyvalue_data[key] = value
return {'success': True}
return server
开发者ID:dnathe4th，项目名称:tchannel-python，代码行数:25，代码来源:test_forwarding.py
示例4: test_call_response_should_contain_transport_headers
def test_call_response_should_contain_transport_headers(
server, service, ThriftTest
):
# Given this test server:
@server.thrift.register(ThriftTest)
def testString(request):
return request.body.thing
# Make a call:
tchannel = TChannel(name='client')
resp = yield tchannel.thrift(service.testString('hi'))
# verify response
assert isinstance(resp, Response)
assert resp.headers == {}
assert resp.body == 'hi'
# verify response transport headers
assert isinstance(resp.transport, TransportHeaders)
assert resp.transport.scheme == schemes.THRIFT
assert resp.transport.failure_domain is None
开发者ID:encrylife，项目名称:tchannel-python，代码行数:25，代码来源:test_thrift.py
示例5: test_client_connection_change_callback
def test_client_connection_change_callback():
server = TChannel('server')
server.listen()
@server.raw.register
def hello(request):
return 'hi'
client = TChannel('client')
count = [0]
def test_cb(peer):
count[0] += 1
client._dep_tchannel.peers.get(
server.hostport)._on_conn_change_cb = test_cb
yield client.raw(
hostport=server.hostport,
body='work',
endpoint='hello',
service='server'
)
# 1: connection built, 1: sending request, 1: finish sending request
assert count[0] == 3
开发者ID:uber，项目名称:tchannel-python，代码行数:25，代码来源:test_connection.py
示例6: test_nest
def test_nest(server, service, ThriftTest, client_ttypes):
# Given this test server:
@server.thrift.register(ThriftTest)
def testNest(request):
return request.body.thing
# Make a call:
tchannel = TChannel(name='client')
xstruct = client_ttypes.Xtruct(
string_thing='hi',
byte_thing=1,
i32_thing=-1,
i64_thing=-34359738368,
)
xstruct2 = client_ttypes.Xtruct2(
byte_thing=1,
struct_thing=xstruct,
i32_thing=1,
)
resp = yield tchannel.thrift(
service.testNest(thing=xstruct2)
)
assert resp.headers == {}
assert resp.body == xstruct2
开发者ID:encrylife，项目名称:tchannel-python，代码行数:30，代码来源:test_thrift.py
示例7: test_map_map
def test_map_map(server, service, ThriftTest):
# Given this test server:
map_map = {
-4: {
-4: -4,
-3: -3,
-2: -2,
-1: -1,
},
4: {
1: 1,
2: 2,
3: 3,
4: 4,
},
}
@server.thrift.register(ThriftTest)
def testMapMap(request):
return map_map
# Make a call:
tchannel = TChannel(name='client')
resp = yield tchannel.thrift(
service.testMapMap(1)
)
assert resp.headers == {}
assert resp.body == map_map
开发者ID:encrylife，项目名称:tchannel-python，代码行数:33，代码来源:test_thrift.py
示例8: test_type_def
def test_type_def():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testTypedef(request):
return request.body.thing
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
x = 0xffffffffffffff # 7 bytes of 0xff
resp = yield tchannel.thrift(
service.testTypedef(thing=x)
)
assert resp.headers == {}
assert resp.body == x
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:29，代码来源:test_thrift.py
示例9: test_void_with_headers
def test_void_with_headers():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testVoid(request):
assert request.headers == {'req': 'header'}
return Response(headers={'resp': 'header'})
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
resp = yield tchannel.thrift(
service.testVoid(),
headers={'req': 'header'},
)
assert resp.headers == {
'resp': 'header'
}
assert resp.body is None
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:32，代码来源:test_thrift.py
示例10: test_void
def test_void():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testVoid(request):
pass
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
resp = yield tchannel.thrift(service.testVoid())
assert resp.headers == {}
assert resp.body is None
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:26，代码来源:test_thrift.py
示例11: test_enum
def test_enum():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testEnum(request):
return request.body.thing
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
x = ThriftTest.Numberz.FIVE
resp = yield tchannel.thrift(
service.testEnum(thing=x)
)
assert resp.headers == {}
assert resp.body == x
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:29，代码来源:test_thrift.py
示例12: test_string_map
def test_string_map():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testStringMap(request):
return request.body.thing
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
x = {
'hello': 'there',
'my': 'name',
'is': 'shirly',
}
resp = yield tchannel.thrift(
service.testStringMap(thing=x)
)
assert resp.headers == {}
assert resp.body == x
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:33，代码来源:test_thrift.py
示例13: test_map
def test_map():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testMap(request):
return request.body.thing
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
x = {
0: 1,
1: 2,
2: 3,
3: 4,
-1: -2,
}
resp = yield tchannel.thrift(
service.testMap(thing=x)
)
assert resp.headers == {}
assert resp.body == x
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:35，代码来源:test_thrift.py
示例14: test_struct
def test_struct():
# Given this test server:
server = TChannel(name='server')
@server.thrift.register(ThriftTest)
def testStruct(request):
assert request.body.thing.string_thing == 'req string'
return ThriftTest.Xtruct(
string_thing="resp string"
)
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='service',
thrift_module=ThriftTest,
hostport=server.hostport
)
resp = yield tchannel.thrift(
service.testStruct(ThriftTest.Xtruct("req string"))
)
# verify response
assert isinstance(resp, Response)
assert resp.headers == {}
assert resp.body == ThriftTest.Xtruct("resp string")
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:35，代码来源:test_thrift.py
示例15: test_i32
def test_i32(server, service, ThriftTest):
# Given this test server:
@server.thrift.register(ThriftTest)
def testI32(request):
return request.body.thing
# Make a call:
tchannel = TChannel(name='client')
# case #1
resp = yield tchannel.thrift(
service.testI32(-1)
)
assert resp.headers == {}
assert resp.body == -1
# case #2
resp = yield tchannel.thrift(
service.testI32(1)
)
assert resp.headers == {}
assert resp.body == 1
开发者ID:encrylife，项目名称:tchannel-python，代码行数:25，代码来源:test_thrift.py
示例16: test_oneway
def test_oneway():
# Given this test server:
server = TChannel(name='server')
# TODO - server should raise same exception as client
with pytest.raises(AssertionError):
@server.thrift.register(ThriftTest)
def testOneway(request):
pass
server.listen()
# Make a call:
tchannel = TChannel(name='client')
service = thrift_request_builder(
service='server',
thrift_module=ThriftTest,
hostport=server.hostport,
)
with pytest.raises(OneWayNotSupportedError):
yield tchannel.thrift(service.testOneway(1))
开发者ID:MadanThangavelu，项目名称:tchannel-python，代码行数:26，代码来源:test_thrift.py
示例17: test_struct_with_headers
def test_struct_with_headers(
server, service, ThriftTest, server_ttypes, client_ttypes
):
# Given this test server:
@server.thrift.register(ThriftTest)
def testStruct(request):
assert isinstance(request, Request)
assert request.headers == {'req': 'header'}
assert request.body.thing.string_thing == 'req string'
return Response(
server_ttypes.Xtruct(
string_thing="resp string"
),
headers={'resp': 'header'},
)
# Make a call:
tchannel = TChannel(name='client')
resp = yield tchannel.thrift(
service.testStruct(client_ttypes.Xtruct("req string")),
headers={'req': 'header'},
)
# verify response
assert isinstance(resp, Response)
assert resp.headers == {'resp': 'header'}
assert resp.body == client_ttypes.Xtruct("resp string")
开发者ID:encrylife，项目名称:tchannel-python，代码行数:33，代码来源:test_thrift.py
示例18: test_endpoint_can_return_just_body
def test_endpoint_can_return_just_body():
# Given this test server:
server = TChannel(name='server')
@server.json.register
def endpoint(request):
return {'resp': 'body'}
server.listen()
# Make a call:
tchannel = TChannel(name='client')
resp = yield tchannel.json(
service='server',
endpoint='endpoint',
hostport=server.hostport,
)
# verify response
assert isinstance(resp, Response)
assert resp.body == {'resp': 'body'}
开发者ID:dnathe4th，项目名称:tchannel-python，代码行数:25，代码来源:test_json.py
示例19: test_map
def test_map(server, service, ThriftTest):
# Given this test server:
@server.thrift.register(ThriftTest)
def testMap(request):
return request.body.thing
# Make a call:
tchannel = TChannel(name='client')
x = {
0: 1,
1: 2,
2: 3,
3: 4,
-1: -2,
}
resp = yield tchannel.thrift(
service.testMap(thing=x)
)
assert resp.headers == {}
assert resp.body == x
开发者ID:encrylife，项目名称:tchannel-python，代码行数:26，代码来源:test_thrift.py
示例20: test_timeout_during_handshake_is_retried
def test_timeout_during_handshake_is_retried(timeout_server):
tchannel = TChannel(name='client', known_peers=[timeout_server])
# We want the client to look for other peers if an INIT times out rather
# than raising a timeout error so we expect a NoAvailablePeerError here.
with patch.object(connection, 'DEFAULT_INIT_TIMEOUT_SECS', 0.1):
with pytest.raises(errors.NoAvailablePeerError):
yield tchannel.raw(service='server', endpoint='endpoint')
开发者ID:uber，项目名称:tchannel-python，代码行数:7，代码来源:test_tchannel.py
注：本文中的tchannel.TChannel类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论