utils/test/result_collection_api/opnfv_testapi/tests/unit/test_fake_pymongo.py

   1 import unittest
   2 from tornado.web import Application
   3 from tornado import gen
   4 from tornado.testing import AsyncHTTPTestCase, gen_test
   5
   6 import fake_pymongo
   7
   8
   9 class MyTest(AsyncHTTPTestCase):
  10     def setUp(self):
  11         super(MyTest, self).setUp()
  12         self.db = fake_pymongo
  13         self.addCleanup(self._clear)
  14         self.io_loop.run_sync(self.fixture_setup)
  15
  16     def get_app(self):
  17         return Application()
  18
  19     @gen.coroutine
  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'})
  25
  26     @gen_test
  27     def test_find_one(self):
  28         user = yield self.db.pods.find_one({'name': 'test1'})
  29         self.assertEqual(user, self.test1)
  30         self.db.pods.remove()
  31
  32     @gen_test
  33     def test_find(self):
  34         cursor = self.db.pods.find()
  35         names = []
  36         while (yield cursor.fetch_next):
  37             ob = cursor.next_object()
  38             names.append(ob.get('name'))
  39         self.assertItemsEqual(names, ['test1', 'test2'])
  40
  41     @gen_test
  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')
  46
  47     @gen_test
  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)
  52
  53     @gen_test
  54     def test_insert_check_keys(self):
  55         yield self.db.pods.insert({'_id': '1', 'name': 'test1'},
  56                                   check_keys=False)
  57         cursor = self.db.pods.find({'_id': '1'})
  58         names = []
  59         while (yield cursor.fetch_next):
  60             ob = cursor.next_object()
  61             names.append(ob.get('name'))
  62         self.assertItemsEqual(names, ['test1', 'test1'])
  63
  64     def _clear(self):
  65         self.db.pods.clear()
  66
  67 if __name__ == '__main__':
  68     unittest.main()