Permalink
Browse files

added Cushion library, which simplifies CouchDB and Tornado interacti…

…on as a layer on top of trombi

NOTE - Cushion currently lacks full test coverage.  Use with caution, patches
gleefully accepted.
  • Loading branch information...
1 parent 00a07c5 commit c378b31f724a77ee3c379947a9c7b8e80e125466 Jeremy Kelley committed Feb 24, 2011
Showing with 562 additions and 0 deletions.
  1. +24 −0 README.md
  2. +174 −0 tests/test_cushion.py
  3. +64 −0 tests/test_cushion_mixin.py
  4. +300 −0 tornado_addons/cushion.py
View
@@ -77,3 +77,27 @@ does cleanup your RequestHandlers quite nicely and really streamlines workflow.
You begin to see the real power of this when you start to have handlers that make
multiple async calls.
+
+### CushionDBMixin
+
+Cushion is a bit of an extraction layer around
+[trombi](http://github.com/inoi/trombi), an asynchronous CouchDB driver for
+Tornado.
+
+It provides a mixin for RequestHandler that simplifies database access,
+especially when used in conjunction with async_yield.
+
+ from tornado.web import RequestHandler
+ from tornado_addons.cushion import CushionDBMixin
+ from tornado_addons.async_yield import async_yield, AsyncYieldMixin
+
+ class SomeHandler(RequestHandler, CushionDBMixin, AsyncYieldMixin):
+
+ def prepare(self):
+ yield self.db_setup('someDB', uri_to_couchdb, self.yield_cb)
+
+ @async_yield
+ def get(self):
+ x = yield self.db_one('some_key')
+ # ... do stuff wth your data in x now
+
View
@@ -0,0 +1,174 @@
+
+"""
+Tests the base cushion class. Gets skipped if trombi doesn't import.
+"""
+
+try:
+ import trombi
+ no_trombi = False
+except:
+ no_trombi = True
+
+from unittest import skipIf
+from random import randint
+from tornado.testing import AsyncTestCase
+from ..tornado_addons.cushion import Cushion, CushionException, CushionDBNotReady
+
+baseurl = 'http://localhost:5984'
+
+@skipIf(no_trombi, "not testing Cushion, trombi failed to import")
+class CushionTests(AsyncTestCase):
+
+ def setUp(self):
+ AsyncTestCase.setUp(self)
+
+ # now create our test cushion object
+ self.cushion = Cushion(baseurl, io_loop=self.io_loop)
+ assert isinstance(self.cushion._server, trombi.Server)
+
+ # create a test db
+ self.dbname = 'test_db' + str(randint(100, 100000))
+ # connect to our database
+ # this tests our open(..) method on cushion. I'd rather have it
+ # in a separate test, but we need this database and if open fails here,
+ # everything else should tank, so setUp is as good a place as any.
+ self.cushion.create(self.dbname, self.stop)
+ self.wait()
+ self.cushion.open(self.dbname, self.stop)
+ self.wait()
+ # we're after the db has been added
+ assert isinstance(self.cushion._pool[self.dbname], trombi.Database)
+
+ def tearDown(self):
+ # just blow away our test database using standard trombi fare
+ self.cushion._server.delete(self.dbname, self.stop)
+ self.wait()
+
+ def test_db_open_with_callback(self):
+ # note, this creates and deletes a bogus db
+ bogus_db = 'test_db_trash_' + str(randint(10,99))
+ self.cushion.create(bogus_db, self.stop)
+ self.wait()
+ self.cushion.open(bogus_db, self.stop)
+ self.wait()
+ self.assertTrue(bogus_db in self.cushion)
+ self.cushion._server.delete(self.dbname, self.stop)
+ self.wait()
+
+ def test_db_not_exists(self):
+ # note, this creates and deletes a bogus db
+ bogus_db = 'test_db_not_exists_' + str(randint(10,99))
+ self.cushion.exists(bogus_db, self.stop)
+ is_there = self.wait()
+ self.assertTrue( not is_there )
+
+ def test_db_exists(self):
+ # note, this creates and deletes a bogus db
+ self.cushion.exists(self.dbname, self.stop)
+ is_there = self.wait()
+ self.assertTrue( is_there )
+
+ def test_db_exists_make_one(self):
+ # note, this creates and deletes a bogus db
+ bogus_db = 'test_db_exists_' + str(randint(10,99))
+ self.cushion.create(bogus_db, self.stop)
+ self.wait()
+ self.cushion.exists(bogus_db, self.stop)
+ is_there = self.wait()
+ self.assertTrue( is_there )
+
+ def test_db_get(self):
+ # check for a bogus database first
+ self.assertRaises(
+ CushionDBNotReady,
+ self.cushion.get, 'bogus-not-there' )
+
+ # now check for a good one
+ self.assertTrue(
+ isinstance(
+ self.cushion.get(self.dbname),
+ trombi.Database
+ )
+ )
+
+ def test_db_ready(self):
+ self.assertTrue( self.cushion.ready(self.dbname) )
+
+ def test_db_shorthand_in(self):
+ self.assertTrue( self.dbname in self.cushion )
+
+ def _save_some_data(self, data):
+ self.cushion.save( self.dbname, data, self.stop)
+ return self.wait()
+
+ def test_save(self):
+ data = {'shoesize': 11}
+ doc = self._save_some_data(data)
+ self.assertTrue( '_id' in doc.raw() )
+ self.saving_data = doc.raw()
+
+ def test_delete(self):
+
+ # try to delete bogus data
+ try:
+ self.cushion.delete(self.dbname, {'bogus':'yep'}, self.stop)
+ self.wait()
+ except Exception, e:
+ self.assertTrue(isinstance(e, CushionException))
+
+ # delete real data
+ data = {'shoesize': 11}
+ doc = self._save_some_data(data)
+ self.cushion.delete(self.dbname, doc.raw(), self.stop)
+ retval = self.wait()
+ self.assertFalse(retval.error)
+
+ def test_one(self):
+ doc = self._save_some_data({'shoes':11, 'hat':'fitted'}).raw()
+ self.cushion.one(self.dbname, doc['_id'], self.stop )
+ retval = self.wait()
+ self.assertEqual(retval['shoes'], doc['shoes'])
+
+ def test_one_fail(self):
+ self.cushion.one(self.dbname, 'just_not_there', self.stop )
+ self.assertTrue( not self.wait() )
+
+ def test_one_return_type(self):
+ doc = self._save_some_data({'shoes':11, 'hat':'fitted'}).raw()
+ self.cushion.one(self.dbname, doc['_id'], self.stop )
+ retval = self.wait()
+ # should be a dict
+ self.assertTrue( type({}) == type(retval) )
+
+ def test_view(self):
+ # This test does quite a bit. First, create 4 test records.
+ # Then, create a view that will emit those records and insert that into
+ # the db. Finally, call our cushion.view object and compare results.
+
+ self._save_some_data({'foo': 1, 'bar': 'a'})
+ self._save_some_data({'foo': 2, 'bar': 'a'})
+ self._save_some_data({'foo': 3, 'bar': 'b'})
+ self._save_some_data({'foo': 4, 'bar': 'b'})
+
+ fake_map = """ function (doc) { emit(doc['bar'], doc); } """
+
+ # we're going to use python-couchdb's dynamic view loader stuff here
+ from couchdb.design import ViewDefinition
+ from couchdb.client import Server
+ global baseurl
+ cdb = Server(baseurl)
+ couchdb = cdb[self.dbname]
+
+ view_defn = ViewDefinition(
+ 'test', 'view',
+ map_fun = fake_map,
+ language = 'javascript' )
+ view_defn.sync(couchdb)
+
+ self.cushion.view(self.dbname, 'test/view', self.stop, key='b')
+ records = self.wait()
+
+ self.assertTrue(len(records) == 2)
+
+ # OPTIMIZE: do more to ensure we're getting back what we want
+
@@ -0,0 +1,64 @@
+
+"""
+test the CushionDBMixin on RequestHandlers
+
+NOTE - CushionDBMIxin lacks FULL TEST COVERAGE! FIXME
+"""
+
+try:
+ import trombi
+ no_trombi = False
+except:
+ no_trombi = True
+
+from unittest import skipIf
+from random import randint
+from ..tornado_addons.cushion import Cushion, CushionException, CushionDBNotReady
+
+baseurl = 'http://localhost:5984'
+
+from ..tornado_addons.async_yield import AsyncYieldMixin
+from ..tornado_addons.cushion import CushionDBMixin
+
+import tornado
+from random import randint
+
+from tornado.testing import AsyncTestCase
+
+
+class CushionHandler(CushionDBMixin, AsyncYieldMixin):
+ """
+ very basic handler for writing async yield tests on a RequestHandler
+ """
+
+
+
+@skipIf(no_trombi, "not testing Cushion, trombi failed to import")
+class CushionMixinTests(AsyncTestCase):
+
+ def setUp(self):
+ AsyncTestCase.setUp(self)
+ dbname = 'test_db' + str(randint(100, 100000))
+ print "WORKING ON", dbname
+ self.handler = CushionHandler()
+ # typically, this would be called in the Handler.prepare()
+ self.handler.db_setup(
+ dbname, baseurl,
+ io_loop=self.io_loop, callback=self.stop, create=True )
+ self.wait()
+
+ # create one test record
+ self.handler.cushion.save(self.handler.db_default, {'fake':'data'}, callback=self.stop)
+ rec = self.wait()
+ self.record = rec.raw()
+
+ def tearDown(self):
+ self.handler.cushion._server.delete(self.handler.db_default, self.stop)
+ self.wait()
+ del self.handler
+
+ def test_db_one(self):
+ self.handler.db_one(self.record['_id'], callback=self.stop)
+ rec = self.wait()
+ self.assertTrue(self.record['fake'] == rec['fake'])
+
Oops, something went wrong.

0 comments on commit c378b31

Please sign in to comment.