2 from tornado.web import Application
3 from tornado import gen
4 from tornado.testing import AsyncHTTPTestCase, gen_test
9 class MyTest(AsyncHTTPTestCase):
11 super(MyTest, self).setUp()
12 self.db = fake_pymongo
13 self.addCleanup(self._clear)
14 self.io_loop.run_sync(self.fixture_setup)
20 def fixture_setup(self):
21 self.test1 = {'_id': '1', 'name': 'test1'}
22 self.test2 = {'name': 'test2'}
23 yield self.db.pods.insert({'_id': '1', 'name': 'test1'})
24 yield self.db.pods.insert({'name': 'test2'})
27 def test_find_one(self):
28 user = yield self.db.pods.find_one({'name': 'test1'})
29 self.assertEqual(user, self.test1)
34 cursor = self.db.pods.find()
36 while (yield cursor.fetch_next):
37 ob = cursor.next_object()
38 names.append(ob.get('name'))
39 self.assertItemsEqual(names, ['test1', 'test2'])
42 def test_update(self):
43 yield self.db.pods.update({'_id': '1'}, {'name': 'new_test1'})
44 user = yield self.db.pods.find_one({'_id': '1'})
45 self.assertEqual(user.get('name', None), 'new_test1')
48 def test_remove(self):
49 yield self.db.pods.remove({'_id': '1'})
50 user = yield self.db.pods.find_one({'_id': '1'})
51 self.assertIsNone(user)
54 def test_insert_check_keys(self):
55 yield self.db.pods.insert({'_id': '1', 'name': 'test1'},
57 cursor = self.db.pods.find({'_id': '1'})
59 while (yield cursor.fetch_next):
60 ob = cursor.next_object()
61 names.append(ob.get('name'))
62 self.assertItemsEqual(names, ['test1', 'test1'])
67 if __name__ == '__main__':