utils/test/result_collection_api/tests/unit/fake_pymongo.py

   1 from bson.objectid import ObjectId
   2 from concurrent.futures import ThreadPoolExecutor
   3
   4 __author__ = 'serena'
   5
   6
   7 class MemCursor(object):
   8     def __init__(self, collection):
   9         self.collection = collection
  10         self.count = len(self.collection)
  11
  12     def _is_next_exist(self):
  13         return self.count != 0
  14
  15     @property
  16     def fetch_next(self):
  17         with ThreadPoolExecutor(max_workers=2) as executor:
  18             result = executor.submit(self._is_next_exist)
  19         return result
  20
  21     def next_object(self):
  22         self.count -= 1
  23         return self.collection.pop()
  24
  25
  26 class MemDb(object):
  27
  28     def __init__(self):
  29         self.contents = []
  30         pass
  31
  32     def _find_one(self, spec_or_id=None, *args):
  33         if spec_or_id is not None and not isinstance(spec_or_id, dict):
  34             spec_or_id = {"_id": spec_or_id}
  35         cursor = self._find(spec_or_id, *args)
  36         for result in cursor:
  37             return result
  38         return None
  39
  40     def find_one(self, spec_or_id=None, *args):
  41         with ThreadPoolExecutor(max_workers=2) as executor:
  42             result = executor.submit(self._find_one, spec_or_id, *args)
  43         return result
  44
  45     def _insert(self, doc_or_docs):
  46
  47         docs = doc_or_docs
  48         return_one = False
  49         if isinstance(docs, dict):
  50             return_one = True
  51             docs = [docs]
  52
  53         ids = []
  54         for doc in docs:
  55             if '_id' not in doc:
  56                 doc['_id'] = ObjectId()
  57             if not self._find_one(doc['_id']):
  58                 ids.append(doc['_id'])
  59                 self.contents.append(doc_or_docs)
  60
  61         if len(ids) == 0:
  62             return None
  63         if return_one:
  64             return ids[0]
  65         else:
  66             return ids
  67
  68     def insert(self, doc_or_docs):
  69         with ThreadPoolExecutor(max_workers=2) as executor:
  70             result = executor.submit(self._insert, doc_or_docs)
  71         return result
  72
  73     @staticmethod
  74     def _in(content, *args):
  75         for arg in args:
  76             for k, v in arg.iteritems():
  77                 if content.get(k, None) != v:
  78                     return False
  79
  80         return True
  81
  82     def _find(self, *args):
  83         res = []
  84         for content in self.contents:
  85             if self._in(content, *args):
  86                 res.append(content)
  87
  88         return res
  89
  90     def find(self, *args):
  91         return MemCursor(self._find(*args))
  92
  93     def _update(self, spec, document):
  94         updated = False
  95         for index in range(len(self.contents)):
  96             content = self.contents[index]
  97             if self._in(content, spec):
  98                 for k, v in document.iteritems():
  99                     updated = True
 100                     content[k] = v
 101             self.contents[index] = content
 102         return updated
 103
 104     def update(self, spec, document):
 105         with ThreadPoolExecutor(max_workers=2) as executor:
 106             result = executor.submit(self._update, spec, document)
 107         return result
 108
 109     def _remove(self, spec_or_id=None):
 110         if spec_or_id is None:
 111             self.contents = []
 112         if not isinstance(spec_or_id, dict):
 113             spec_or_id = {'_id': spec_or_id}
 114         for index in range(len(self.contents)):
 115             content = self.contents[index]
 116             if self._in(content, spec_or_id):
 117                 del self.contents[index]
 118                 return True
 119         return False
 120
 121     def remove(self, spec_or_id=None):
 122         with ThreadPoolExecutor(max_workers=2) as executor:
 123             result = executor.submit(self._remove, spec_or_id)
 124         return result
 125
 126 pod = MemDb()
 127 test_projects = MemDb()
 128 test_cases = MemDb()
 129 test_results = MemDb()