initial commit

.gitignore

@@ -0,0 +1 @@
+*.pyc

README.md

@@ -0,0 +1,21 @@
+#VCR.py
+
+This is a proof-of-concept start at a python version of [Ruby's VCR
+library](https://github.com/myronmarston/vcr).  It
+doesn't actually work.
+
+#What it is supposed to do
+Simplify testing by recording all HTTP interactions and saving them to
+"cassette" files, which are just yaml files.  Then when you run your tests
+again, they all just hit the text files instead of the internet.  This speeds up
+your tests and lets you work offline.
+
+#What it actually does
+Uses up all your memory
+
+
+#Similar libraries in Python
+Neither of these really implement the API I want, but I have cribbed some code
+from them.
+ * https://github.com/bbangert/Dalton
+ * https://github.com/storborg/replaylib

setup.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python
+
+from distutils.core import setup
+
+setup(name='VCR.py',
+      version='1.0',
+      description='VCR.py',
+      author='Kevin McCarthy',
+      author_email='mc@kevinmccarthy.org',
+      packages=['vcr'],
+     )

test.py

@@ -0,0 +1,28 @@
+import os
+import unittest
+import vcr
+import urllib2
+
+TEST_CASSETTE_FILE = 'test/test_req.yaml'
+
+class TestHttpRequest(unittest.TestCase):
+
+    def setUp(self):
+        try:
+            os.remove(TEST_CASSETTE_FILE)
+        except OSError:
+            pass
+
+    def test_response_code(self):
+        with vcr.use_cassette(TEST_CASSETTE_FILE):
+            code = urllib2.urlopen('http://www.iana.org/domains/example/').getcode()
+            self.assertEqual(code,urllib2.urlopen('http://www.iana.org/domains/example/').getcode())
+
+    def test_response_body(self):
+        with vcr.use_cassette('test/synopsis.yaml'):
+            body = urllib2.urlopen('http://www.iana.org/domains/example/').read()
+            self.assertEqual(body,urllib2.urlopen('http://www.iana.org/domains/example/').read())
+
+
+if __name__ == '__main__':
+    unittest.main()

vcr/__init__.py

@@ -0,0 +1 @@
+from .patch import use_cassette

vcr/cassette.py

@@ -0,0 +1,16 @@
+import yaml
+
+
+class Cassette(object):
+    def __init__(self):
+        self.requests = []
+        self.responses = []
+
+    def serialize(self):
+        return yaml.dump([{
+            'request': req,
+            'response': res,
+        }  for req,res in zip(self.requests,self.responses)])
+
+
+

vcr/files.py

@@ -0,0 +1,17 @@
+import os
+import yaml
+
+def load_cassette(cassette_path):
+    try:
+        return yaml.load(open(cassette_path))
+    except IOError:
+        return None
+
+def save_cassette(cassette_path,cassette):
+    dirname,filename = os.path.split(cassette_path)
+    if not os.path.exists(dirname):
+        os.makedirs(dirname)
+    with open(cassette_path,'wc') as cassette_file:
+        cassette_file.write(cassette.serialize())
+
+

vcr/patch.py

@@ -0,0 +1,25 @@
+import httplib
+from contextlib import contextmanager
+from .stubs import VCRHTTPConnection, VCRHTTPSConnection
+
+_HTTPConnection = httplib.HTTPConnection
+_HTTPSConnection = httplib.HTTPSConnection
+
+
+def install(cassette_path):
+    httplib.HTTPConnection = httplib.HTTP._connection_class = VCRHTTPConnection
+    httplib.HTTPSConnection = httplib.HTTPS._connection_class = VCRHTTPSConnection
+    httplib.HTTPConnection._vcr_cassette_path = cassette_path
+    httplib.HTTPSConnection._vcr_cassette_path = cassette_path
+
+def reset():
+    httplib.HTTPConnection = httplib.HTTP._connection_class = _HTTPConnection
+    httplib.HTTPSConnection = httplib.HTTPS._connection_class = \
+            _HTTPSConnection
+
+@contextmanager
+def use_cassette(cassette_path):
+    install(cassette_path)
+    yield
+    reset()
+

vcr/stubs.py

@@ -0,0 +1,55 @@
+from httplib import HTTPConnection
+from .files import save_cassette, load_cassette
+from .cassette import Cassette
+
+class VCRHTTPResponse(object):
+    def __init__(self,recorded_response):
+        self.recorded_response = recorded_response
+        self.msg = recorded_response['status']['message']
+        self.reason = recorded_response['status']['message']
+        self.status = recorded_response['status']['code']
+
+    def read(self,chunked=False):
+        return self.recorded_response['body']['string']
+
+    def getheaders(self):
+        return self.recorded_response['headers']
+
+
+class VCRHTTPConnection(HTTPConnection):
+
+    def __init__(self,*args,**kwargs):
+        self._cassette = Cassette()
+        HTTPConnection.__init__(self,*args,**kwargs)
+
+    def _save_cassette(self):
+        save_cassette(self._vcr_cassette_path,self._cassette)
+
+    def request(self,method,url,body=None,headers={}):
+        old_cassette = load_cassette(self._vcr_cassette_path)
+        if old_cassette:
+            return
+        self._cassette.requests.append(dict(
+            method = method,
+            url = url,
+            body = body,
+            headers = headers
+        ))
+        return HTTPConnection.request(self,method,url,body=body,headers=headers)
+
+    def getresponse(self,buffering=False):
+        old_cassette = load_cassette(self._vcr_cassette_path)
+        if old_cassette:
+            return VCRHTTPResponse(old_cassette[0]['response'])
+        response = HTTPConnection.getresponse(self,buffering=buffering)
+        self._cassette.responses.append({
+            'status':{'code':response.status,'message':response.reason},
+            'headers':dict(response.getheaders()),
+            'body':{'string':response.read()},
+        })
+        self._save_cassette()
+        return response
+
+
+class VCRHTTPSConnection(VCRHTTPConnection):
+    pass