utils/test/testapi/opnfv_testapi/tests/unit/fake_pymongo.py

   1 ##############################################################################
   2 # Copyright (c) 2016 ZTE Corporation
   3 # feng.xiaowei@zte.com.cn
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9 import re
  10
  11 from operator import itemgetter
  12
  13 from bson.objectid import ObjectId
  14 from concurrent.futures import ThreadPoolExecutor
  15
  16
  17 def thread_execute(method, *args, **kwargs):
  18         with ThreadPoolExecutor(max_workers=2) as executor:
  19             result = executor.submit(method, *args, **kwargs)
  20         return result
  21
  22
  23 class MemCursor(object):
  24     def __init__(self, collection):
  25         self.collection = collection
  26         self.length = len(self.collection)
  27         self.sorted = []
  28
  29     def _is_next_exist(self):
  30         return self.length != 0
  31
  32     @property
  33     def fetch_next(self):
  34         return thread_execute(self._is_next_exist)
  35
  36     def next_object(self):
  37         self.length -= 1
  38         return self.collection.pop()
  39
  40     def sort(self, key_or_list):
  41         for k, v in key_or_list.iteritems():
  42             if v == -1:
  43                 reverse = True
  44             else:
  45                 reverse = False
  46
  47             self.collection = sorted(self.collection,
  48                                      key=itemgetter(k), reverse=reverse)
  49         return self
  50
  51     def limit(self, limit):
  52         if limit != 0 and limit < len(self.collection):
  53             self.collection = self.collection[0: limit]
  54             self.length = limit
  55         return self
  56
  57     def skip(self, skip):
  58         if skip < self.length and (skip > 0):
  59             self.collection = self.collection[self.length - skip: -1]
  60             self.length -= skip
  61         elif skip >= self.length:
  62             self.collection = []
  63             self.length = 0
  64         return self
  65
  66     def _count(self):
  67         return self.length
  68
  69     def count(self):
  70         return thread_execute(self._count)
  71
  72
  73 class MemDb(object):
  74
  75     def __init__(self, name):
  76         self.name = name
  77         self.contents = []
  78         pass
  79
  80     def _find_one(self, spec_or_id=None, *args):
  81         if spec_or_id is not None and not isinstance(spec_or_id, dict):
  82             spec_or_id = {"_id": spec_or_id}
  83         if '_id' in spec_or_id:
  84             spec_or_id['_id'] = str(spec_or_id['_id'])
  85         cursor = self._find(spec_or_id, *args)
  86         for result in cursor:
  87             return result
  88         return None
  89
  90     def find_one(self, spec_or_id=None, *args):
  91         return thread_execute(self._find_one, spec_or_id, *args)
  92
  93     def _insert(self, doc_or_docs, check_keys=True):
  94
  95         docs = doc_or_docs
  96         return_one = False
  97         if isinstance(docs, dict):
  98             return_one = True
  99             docs = [docs]
 100
 101         if check_keys:
 102             for doc in docs:
 103                 self._check_keys(doc)
 104
 105         ids = []
 106         for doc in docs:
 107             if '_id' not in doc:
 108                 doc['_id'] = str(ObjectId())
 109             if not self._find_one(doc['_id']):
 110                 ids.append(doc['_id'])
 111                 self.contents.append(doc_or_docs)
 112
 113         if len(ids) == 0:
 114             return None
 115         if return_one:
 116             return ids[0]
 117         else:
 118             return ids
 119
 120     def insert(self, doc_or_docs, check_keys=True):
 121         return thread_execute(self._insert, doc_or_docs, check_keys)
 122
 123     @staticmethod
 124     def _compare_date(spec, value):
 125         gte = True
 126         lt = False
 127         for k, v in spec.iteritems():
 128             if k == '$gte' and value < v:
 129                 gte = False
 130             elif k == '$lt' and value < v:
 131                 lt = True
 132         return gte and lt
 133
 134     def _in(self, content, *args):
 135         if self.name == 'scenarios':
 136             return self._in_scenarios(content, *args)
 137         else:
 138             return self._in_others(content, *args)
 139
 140     def _in_scenarios_installer(self, installer, content):
 141         hit = False
 142         for s_installer in content['installers']:
 143             if installer == s_installer['installer']:
 144                 hit = True
 145
 146         return hit
 147
 148     def _in_scenarios_version(self, version, content):
 149         hit = False
 150         for s_installer in content['installers']:
 151             for s_version in s_installer['versions']:
 152                 if version == s_version['version']:
 153                     hit = True
 154         return hit
 155
 156     def _in_scenarios_project(self, project, content):
 157         hit = False
 158         for s_installer in content['installers']:
 159             for s_version in s_installer['versions']:
 160                 for s_project in s_version['projects']:
 161                     if project == s_project['project']:
 162                         hit = True
 163
 164         return hit
 165
 166     def _in_scenarios(self, content, *args):
 167         for arg in args:
 168             for k, v in arg.iteritems():
 169                 if k == 'installers':
 170                     for inner in v.values():
 171                         for i_k, i_v in inner.iteritems():
 172                             if i_k == 'installer':
 173                                 return self._in_scenarios_installer(i_v,
 174                                                                     content)
 175                             elif i_k == 'versions.version':
 176                                 return self._in_scenarios_version(i_v,
 177                                                                   content)
 178                             elif i_k == 'versions.projects.project':
 179                                 return self._in_scenarios_project(i_v,
 180                                                                   content)
 181                 elif content.get(k, None) != v:
 182                     return False
 183
 184         return True
 185
 186     def _in_others(self, content, *args):
 187         for arg in args:
 188             for k, v in arg.iteritems():
 189                 if k == 'start_date':
 190                     if not MemDb._compare_date(v, content.get(k)):
 191                         return False
 192                 elif k == 'trust_indicator.current':
 193                     if content.get('trust_indicator').get('current') != v:
 194                         return False
 195                 elif not isinstance(v, dict):
 196                     if isinstance(v, re._pattern_type):
 197                         if v.match(content.get(k, None)) is None:
 198                             return False
 199                     else:
 200                         if content.get(k, None) != v:
 201                             return False
 202         return True
 203
 204     def _find(self, *args):
 205         res = []
 206         for content in self.contents:
 207             if self._in(content, *args):
 208                 res.append(content)
 209         return res
 210
 211     def find(self, *args):
 212         return MemCursor(self._find(*args))
 213
 214     def _aggregate(self, *args, **kwargs):
 215         res = self.contents
 216         print args
 217         for arg in args[0]:
 218             for k, v in arg.iteritems():
 219                 if k == '$match':
 220                     res = self._find(v)
 221         cursor = MemCursor(res)
 222         for arg in args[0]:
 223             for k, v in arg.iteritems():
 224                 if k == '$sort':
 225                     cursor = cursor.sort(v)
 226                 elif k == '$skip':
 227                     cursor = cursor.skip(v)
 228                 elif k == '$limit':
 229                     cursor = cursor.limit(v)
 230         return cursor
 231
 232     def aggregate(self, *args, **kwargs):
 233         return self._aggregate(*args, **kwargs)
 234
 235     def _update(self, spec, document, check_keys=True):
 236         updated = False
 237
 238         if check_keys:
 239             self._check_keys(document)
 240
 241         for index in range(len(self.contents)):
 242             content = self.contents[index]
 243             if self._in(content, spec):
 244                 for k, v in document.iteritems():
 245                     updated = True
 246                     content[k] = v
 247             self.contents[index] = content
 248         return updated
 249
 250     def update(self, spec, document, check_keys=True):
 251         return thread_execute(self._update, spec, document, check_keys)
 252
 253     def _remove(self, spec_or_id=None):
 254         if spec_or_id is None:
 255             self.contents = []
 256         if not isinstance(spec_or_id, dict):
 257             spec_or_id = {'_id': spec_or_id}
 258         for index in range(len(self.contents)):
 259             content = self.contents[index]
 260             if self._in(content, spec_or_id):
 261                 del self.contents[index]
 262                 return True
 263         return False
 264
 265     def remove(self, spec_or_id=None):
 266         return thread_execute(self._remove, spec_or_id)
 267
 268     def clear(self):
 269         self._remove()
 270
 271     def _check_keys(self, doc):
 272         for key in doc.keys():
 273             if '.' in key:
 274                 raise NameError('key {} must not contain .'.format(key))
 275             if key.startswith('$'):
 276                 raise NameError('key {} must not start with $'.format(key))
 277             if isinstance(doc.get(key), dict):
 278                 self._check_keys(doc.get(key))
 279
 280
 281 def __getattr__(name):
 282     return globals()[name]
 283
 284
 285 pods = MemDb('pods')
 286 projects = MemDb('projects')
 287 testcases = MemDb('testcases')
 288 results = MemDb('results')
 289 scenarios = MemDb('scenarios')
 290 tokens = MemDb('tokens')
 291 users = MemDb('users')