HAR Player - replay your browser's HAR files

harplayer.py

#!/usr/bin/env python


"""

    harplayer

    Use it with python harplayer.py <file.har> <*ports>

    REQUIRES:

        * tornado 3.2

    NOTE:

        * In the future this should be an executable ALA unicorn or ipython
        * It would be even better written in node ?
        * Need of a config.py or something of the sort to map
            accounts.locarise.com: localhost:8001
            sads.locarise.com: localhost:8000
            etc etc
        * 'harplayer parse <file>' outputs the relevant sub-har to stdout
        * 'harplayer server <file>' multiserves the harfile according to settings
"""

import os
import sys
import json

from urlparse import urlparse
import tornado.ioloop
import tornado.web

mock_store = dict()


def load_har_entries(path):
    with open(path, 'r') as f:
        raw_har = f.read()
    har = json.loads(raw_har)
    return har['log']['entries']


def get_complete_path(url):
    path = url.path
    if url.query != '':
        path += '?%s' % url.query
    return path


class MainHandler(tornado.web.RequestHandler):

    def set_default_headers(self):
        self.set_header('Access-Control-Allow-Origin', '*')
        self.set_header('Access-Control-Allow-Credentials', 'true')
        self.set_header('Access-Control-Allow-Methods', 'GET,PUT,POST,DELETE,OPTIONS')
        self.set_header('Access-Control-Allow-Headers',
                        'Content-Type, Depth, User-Agent, X-File-Size, X-Requested-With, X-Requested-By, If-Modified-Since, X-File-Name, Cache-Control')

    def check_xsrf_cookie(self):
        pass

    def send_answer(self):
        port = self.request.host.split(':')[1]
        e = mock_store[port][self.request.uri][self.request.method]
        self.set_status(e['status'], e['statusText'])
        for h in e['headers']:
            if h['name'] == 'Content-Encoding':
                continue  # always plain text
            self.set_header(h['name'], h['value'])
        self.write(e['content']['text'])

    def options(self, *args, **kwargs):
        port = self.request.host.split(':')[1]
        try:
            e = mock_store[port][self.request.path][self.request.method]
            self.set_status(e['status'], e['statusText'])
            for h in e['headers']:
                self.set_header(h['name'], h['value'])
        except KeyError:
            self.set_status(200, 'OK')
            headers = [
                {'name': 'Date', 'value': 'Wed, 05 Mar 2014 16:46:34 GMT'},
                {'name': 'Server', 'value': 'WSGIServer/0.1 Python/2.7.4'},
                {'name': 'Access-Control-Max-Age', 'value': '86400'},
                {'name': 'Access-Control-Allow-Methods', 'value': 'GET, POST, PUT, PATCH, DELETE, OPTIONS'},
                {'name': 'Content-Type', 'value': 'text/html; charset=utf-8'},
                {'name': 'Access-Control-Allow-Origin', 'value': '*'},
                {'name': 'Access-Control-Allow-Credentials', 'value': 'true'},
                {'name': 'Access-Control-Allow-Headers', 'value': 'x-requested-with, content-type, accept, origin, authorization'},
            ]
            for h in headers:
                self.set_header(h['name'], h['value'])

    def get(self, *args, **kwargs):
        self.send_answer()

    def post(self, *args, **kwargs):
        self.send_answer()


application = tornado.web.Application([
    (r'/(.*)', MainHandler),
], debug='True', autoreload='True')

if __name__ == '__main__':

    ports = map(int, sys.argv[2:])

    entries = load_har_entries(sys.argv[1])
    for e in entries:
        url = urlparse(e['request']['url'])
        path = get_complete_path(url)
        port = str(url.port)
        try:
            if not int(port) in ports:
                continue
        except ValueError:
            continue
        if not port in mock_store:
            mock_store[port] = dict()
        mock = mock_store[port]
        if not path in mock:
            mock[path] = dict()
        method = e['request']['method']
        mock[path][method] = e['response']
        # print 'Inserted %s:%s:%s' % (port, method, path)

    for port in ports:
        application.listen(port)
    tornado.ioloop.IOLoop.instance().start()