Some of the code to make mitmproxy code into replayable http test code

auth_and_list_images.json

[
  {
    "request": {
      "url": "http://example.com:8770/v2.0/tokens",
      "method": "POST",
      "content": "{\"auth\": {\"tenantName\": \"termie\", \"passwordCredentials\": {\"username\": \"termie\", \"password\": \"termie\"}}}",
      "headers": {
        "content-length": "103",
        "accept-encoding": "gzip, deflate, compress",
        "accept": "*/*",
        "user-agent": "python-requests/1.1.0 CPython/2.7.2 Darwin/12.2.1",
        "host": "example.com:8770",
        "content-type": "application/json"
      },
      "host": "example.com",
      "query": {},
      "path": "/v2.0/tokens",
      "scheme": "http",
      "port": 8770
    },
    "response": {
      "msg": "OK",
      "headers": {
        "date": "Tue, 05 Feb 2013 19:23:29 GMT",
        "content-type": "application/json",
        "content-length": "1919",
        "vary": "X-Auth-Token"
      },
      "code": 200,
      "content": "{\"access\": {\"token\": {\"expires\": \"2013-02-06T19:23:29Z\", \"id\": \"0b67f1bba1e94cb38d97b8b7fc17ecbc\", \"tenant\": {\"enabled\": true, \"description\": \"\", \"name\": \"termie\", \"id\": \"04244909d66d42f8a03e4e85e05eeecf\"}}, \"serviceCatalog\": [{\"endpoints\": [{\"adminURL\": \"http://proxy:8776/v1/04244909d66d42f8a03e4e85e05eeecf\", \"region\": \"RegionOne\", \"publicURL\": \"http://example.com:8776/v1/04244909d66d42f8a03e4e85e05eeecf\", \"internalURL\": \"http://proxy:8776/v1/04244909d66d42f8a03e4e85e05eeecf\"}], \"endpoints_links\": [], \"type\": \"volume\", \"name\": \"Volume Service\"}, {\"endpoints\": [{\"adminURL\": \"http://proxy:9292/v1\", \"region\": \"RegionOne\", \"publicURL\": \"http://example.com:9292/v1\", \"internalURL\": \"http://proxy:9292/v1\"}], \"endpoints_links\": [], \"type\": \"image\", \"name\": \"Image Service\"}, {\"endpoints\": [{\"adminURL\": \"http://proxy:8774/v2/04244909d66d42f8a03e4e85e05eeecf\", \"region\": \"RegionOne\", \"publicURL\": \"http://example.com:8774/v2/04244909d66d42f8a03e4e85e05eeecf\", \"internalURL\": \"http://proxy:8774/v2/04244909d66d42f8a03e4e85e05eeecf\"}], \"endpoints_links\": [], \"type\": \"compute\", \"name\": \"Compute Service\"}, {\"endpoints\": [{\"adminURL\": \"http://proxy:8773/services/Admin\", \"region\": \"RegionOne\", \"publicURL\": \"http://example.com:8773/services/Cloud\", \"internalURL\": \"http://proxy:8773/services/Cloud\"}], \"endpoints_links\": [], \"type\": \"ec2\", \"name\": \"EC2 Service\"}, {\"endpoints\": [{\"adminURL\": \"http://proxy:35357/v2.0\", \"region\": \"RegionOne\", \"publicURL\": \"http://example.com:8770/v2.0\", \"internalURL\": \"http://proxy:8770/v2.0\"}], \"endpoints_links\": [], \"type\": \"identity\", \"name\": \"Identity Service\"}], \"user\": {\"username\": \"termie\", \"roles_links\": [], \"id\": \"7dfae3229cc640c99075d864725dbfe9\", \"roles\": [{\"name\": \"Admin\"}], \"name\": \"termie\"}, \"metadata\": {\"is_admin\": 0, \"roles\": [\"e73704a31e054b0b9be99ee67a972771\"]}}}"
    }
  },
  {
    "request": {
      "url": "http://example.com:9292/v1/images",
      "method": "GET",
      "content": "",
      "headers": {
        "content-length": "0",
        "accept-encoding": "gzip, deflate, compress",
        "accept": "*/*",
        "x-auth-token": "0b67f1bba1e94cb38d97b8b7fc17ecbc",
        "host": "example.com:9292",
        "user-agent": "python-requests/1.1.0 CPython/2.7.2 Darwin/12.2.1",
        "content-type": "application/json"
      },
      "host": "example.com",
      "query": {},
      "path": "/v1/images",
      "scheme": "http",
      "port": 9292
    },
    "response": {
      "msg": "OK",
      "headers": {
        "date": "Tue, 05 Feb 2013 19:23:32 GMT",
        "content-length": "1027",
        "content-type": "application/json; charset=UTF-8",
        "x-openstack-request-id": "req-4816d397-7d4a-43df-a923-a0b1f7ab58c9"
      },
      "code": 200,
      "content": "{\"images\": [{\"name\": \"CentOS 6.3 x64 machine\", \"container_format\": \"ami\", \"disk_format\": \"ami\", \"checksum\": \"d98b64292013f9ec0474baa7031a4448\", \"id\": \"bf546140-ecd6-49f1-b946-c8443406af75\", \"size\": 2146435072}, {\"name\": \"CentOS 6.3 x64 ramdisk\", \"container_format\": \"ari\", \"disk_format\": \"ari\", \"checksum\": \"61c82250c93be888d3b883cfdc4048a5\", \"id\": \"24d778e5-08b9-480a-97db-cd3a2a3f0288\", \"size\": 12972542}, {\"name\": \"CentOS 6.3 x64 kernel\", \"container_format\": \"aki\", \"disk_format\": \"aki\", \"checksum\": \"98aeda02072f12d8dd5a8ae0ec33cec7\", \"id\": \"d38184ae-cf39-41c4-8609-8525f2fd57c6\", \"size\": 3941776}, {\"name\": \"Ubuntu 12.04 server cloudimg amd64\", \"container_format\": \"ovf\", \"disk_format\": \"qcow2\", \"checksum\": \"bb600526ee29eb8d0f575680decab6eb\", \"id\": \"83187b6f-ccae-4609-8962-5b803aa636f2\", \"size\": 230490112}, {\"name\": \"Ubuntu 11.10 server cloudimg amd64\", \"container_format\": \"ovf\", \"disk_format\": \"qcow2\", \"checksum\": \"e8ff23cf89a0568929e016d8e2786c53\", \"id\": \"8b3a2ba5-6712-4736-9382-1b9cf13c9a10\", \"size\": 228982784}]}"
    }
  }
]

glance_tests.py

class GlanceTests(test.ApiTestCase):

  def test_auth_and_list_images(self):
    flow = test.load_flow('auth_and_list_images')
    glance_auto = glance.GlanceAuto()

    with test.Flow('auth_and_list_images') as flow:
      auth_ref = self.authenticate()
      #print auth_ref.catalog
      images = glance_auto.list_images(auth_ref=auth_ref)
      #print images

test.py

import cStringIO as StringIO
import json
import unittest

import mock
import requests
from requests import structures


def load_flow(name):
  flow = json.load(open('tests/flows/%s.json' % name))
  return flow


class TestCase(unittest.TestCase):
  def assertDictShape(self, actual, expected):
    for k, v in expected.iteritems():
      if k not in actual:
        raise AssertionError('%s not present in dict %s' % (k, actual))
      if type(v) == type({}):
        self.assertDictShape(actual[k], expected[k])


def fake_response(res):
  response = requests.Response()
  response.raw = StringIO.StringIO(res['content'])
  response.headers = structures.CaseInsensitiveDict(res['headers'])
  response.status_code = res['code']
  #response.url = '%s://%s:%s%s' % (res['scheme'], res['host'], res['port'], res['path'])
  return response


class Flow(TestCase):
  def __init__(self, flow):
    self.flow = load_flow(flow)
    self.index = 0
    self._patches = []

  def __enter__(self):
    self._patches.append(mock.patch('requests.get', self.get))
    self._patches.append(mock.patch('requests.post', self.post))
    self._patches.append(mock.patch('requests.put', self.put))
    self._patches.append(mock.patch('requests.head', self.head))
    self._patches.append(mock.patch('requests.delete', self.delete))
    for patcher in self._patches:
      patcher.start()
    return self

  def __exit__(self, exc_type, exc_value, traceback):
    for patcher in self._patches:
      patcher.stop()
    self.assert_(self.index == len(self.flow), 'did not reach end of flow')

  def _do_request(self, method, url, data=None, params=None, headers=None):
    req = self.flow[self.index]['request']
    res = self.flow[self.index]['response']
    self.assertEquals(method, req['method'])

    if method in ('POST', 'PUT'):
      actual = json.loads(data)
      expected = json.loads(req['content'])
      self.assertDictShape(actual, expected)
    else:
      expected = req['query']
      actual = params
      for k, v in expected:
        self.assertIn(k, actual)

    self.index += 1
    return fake_response(res)

  def get(self, url, data=None, params=None, headers=None):
    return self._do_request('GET', url, params=params, headers=headers)

  def post(self, url, data=None, params=None, headers=None):
    return self._do_request('POST', url, data=data, headers=headers)

  def put(self, url, data=None, params=None, headers=None):
    return self._do_request('PUT', url, data=data, headers=headers)

  def head(self, url, data=None, params=None, headers=None):
    return self._do_request('HEAD', url, params=params, headers=headers)

  def delete(self, url, data=None, params=None, headers=None):
    return self._do_request('delete', url, params=params, headers=headers)

to_requests.py

import json

def request_to_dict(req):
  o = {}
  d = dict((k.lower(), v) for k, v in req.headers.lst)
  o['headers'] = d
  o['content'] = req.content
  o['path'] = req.path
  o['method'] = req.method
  o['port'] = req.port
  o['host'] = req.host
  o['scheme'] = req.scheme
  o['url'] = req.get_url()

  q = dict((k, v) for k, v in req.get_query().lst)
  o['query'] = q

  return o


def response_to_dict(res):
  o = {}
  d = dict((k.lower(), v) for k, v in res.headers.lst)
  o['headers'] = d
  o['code'] = res.code
  o['msg'] = res.msg
  o['content'] = res.content
  return o


# Global so we can dump everything at the end
final_output = []

def response(context, flow):
  try:
    req = request_to_dict(flow.response.request)
    res = response_to_dict(flow.response)
    final_output.append({'request': req,
                         'response': res})
  except Exception as e:
    print e

def done(context):
  print json.dumps(final_output, indent=2)

to_requests.sh

#!/bin/sh
mitmdump -ns to_requests.py -r $1 -q > $1.json