Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Classic Tornado unittest for AsyncMongo
from tornado import ioloop, testing
import asyncmongo
import asyncmongo.errors
class MyTestCase(testing.AsyncTestCase):
def get_new_ioloop(self):
# See discussion at
# https://groups.google.com/d/msg/python-tornado/2xLfv0hhi54/CFIfX4WqLaAJ
return ioloop.IOLoop.instance()
def test_stuff(self):
db = asyncmongo.Client(
pool_id='test_query',
host='127.0.0.1',
port=27017,
dbname='test',
mincached=3
)
def cb(result, error):
self.stop((result, error))
db.collection.remove(safe=True, callback=cb)
self.wait()
db.collection.insert({"_id" : 1}, safe=True, callback=cb)
self.wait()
# Verify the document was inserted
db.collection.find(callback=cb)
result, error = self.wait()
self.assertEqual([{'_id': 1}], result)
# MongoDB has a unique index on _id
db.collection.insert({"_id" : 1}, safe=True, callback=cb)
result, error = self.wait()
self.assertTrue(isinstance(error, asyncmongo.errors.IntegrityError))
# Apparently needed in this module for testing.main()
def all():
return MyTestCase('test_stuff')
if __name__ == '__main__':
# Use Tornado's own testing.main(), not unittest
testing.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.