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))