test_inbound.py 文件源码

def test_inbound_headers(http_headers, rpc_headers, http_resp_headers):
    class Handler(object):
        def handle(self, request):
            assert rpc_headers == request.headers
            return Response(headers=rpc_headers, body=request.body)

    inbound = HTTPInbound()
    inbound.start(Handler())

    client = AsyncHTTPClient()
    req = HTTPRequest(
        url='http://localhost:%s' % inbound.port,
        method='POST',
        headers=http_headers,
        body='',
    )

    res = yield client.fetch(req)
    assert 200 == res.code
    for k, v in http_resp_headers.items():
        assert v == res.headers[k]