Tornado¶
HoverPy can be used to make virtualise asynchronous requests made from Tornado’s AsyncHTTPClient.
Capturing traffic¶
from tornado import gen, web, ioloop, httpclient
class MainHandler(web.RequestHandler):
@gen.coroutine
def get(self):
http_client = httpclient.AsyncHTTPClient()
time = yield http_client.fetch("http://time.ioloop.io?format=json")
self.write(time.body)
from hoverpy import capture, lib
@lib.tornado(proxyPort=8500, proxyHost="localhost")
@capture(dbpath="tornado.db")
def start():
app = web.Application([("/", MainHandler)])
app.listen(8080)
ioloop.IOLoop.current().start()
start()
Making a request to our server now captures the requests.
curl http://localhost:8080
In fact you may notice your directory now contains a tornado.db
.
Simulating traffic¶
We can how switch our server to simulate mode:
from tornado import gen, web, ioloop, httpclient
class MainHandler(web.RequestHandler):
@gen.coroutine
def get(self):
http_client = httpclient.AsyncHTTPClient()
time = yield http_client.fetch("http://time.ioloop.io?format=json")
self.write(time.body)
from hoverpy import simulate, lib
@lib.tornado
@simulate(dbpath="tornado.db")
def start():
app = web.Application([("/", MainHandler)])
app.listen(8080)
ioloop.IOLoop.current().start()
start()
Which means we are no longer hitting the real downstream dependency.
Modifying traffic¶
HoverPy can also be used to modify your requests, to introduce failures, or build tolerant readers
from tornado import gen, web, ioloop, httpclient
class MainHandler(web.RequestHandler):
@gen.coroutine
def get(self):
http_client = httpclient.AsyncHTTPClient()
time = yield http_client.fetch("http://time.ioloop.io?format=json")
self.write(time.body)
from hoverpy import modify, lib
@lib.tornado
@modify(middleware="python examples/tornado/middleware.py")
def start():
app = web.Application([("/", MainHandler)])
app.listen(8080)
ioloop.IOLoop.current().start()
start()
This is our middleware:
# middleware.py
import sys
import logging
import json
logging.basicConfig(filename='middleware.log', level=logging.DEBUG)
data = sys.stdin.readlines()
payload = data[0]
doc = json.loads(payload)
logging.debug(json.dumps(doc, indent=4, separators=(',', ': ')))
if "request" in doc:
doc["request"]["headers"]["Accept-Encoding"] = ["identity"]
if "response" in doc and doc["response"]["status"] == 200:
if doc["request"]["destination"] == "time.ioloop.io":
body = json.loads(doc["response"]["body"])
body["epoch"] = 101010
doc["response"]["body"] = json.dumps(body)
doc["response"]["headers"]["Content-Length"] = [str(len(json.dumps(body)))]
print(json.dumps(doc))