test_tornado_transport.py 文件源码

def test_post(self, mock_httpclient_fetch):
        cache = stub(get=lambda url: None, add=lambda url, content: None)

        response = HTTPResponse(HTTPRequest('http://tests.python-zeep.org/test.xml'), 200)
        response.buffer = True
        response._body = 'x'
        http_fetch_future = Future()
        http_fetch_future.set_result(response)
        mock_httpclient_fetch.return_value = http_fetch_future

        transport = TornadoAsyncTransport(cache=cache)

        envelope = etree.Element('Envelope')

        result = yield transport.post_xml(
            'http://tests.python-zeep.org/test.xml',
            envelope=envelope,
            headers={})

        assert result.content == 'x'
        assert result.status_code == 200