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

   1 ##############################################################################
   2 # Copyright (c) 2017 ZTE Corp
   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 functools
  10 import httplib
  11
  12 from concurrent.futures import ThreadPoolExecutor
  13 import mock
  14
  15
  16 O_get_secure_cookie = (
  17     'opnfv_testapi.handlers.base_handlers.GenericApiHandler.get_secure_cookie')
  18
  19
  20 def thread_execute(method, *args, **kwargs):
  21         with ThreadPoolExecutor(max_workers=2) as executor:
  22             result = executor.submit(method, *args, **kwargs)
  23         return result
  24
  25
  26 def mock_invalid_lfid():
  27     def _mock_invalid_lfid(xstep):
  28         def wrap(self, *args, **kwargs):
  29             with mock.patch(O_get_secure_cookie) as m_cookie:
  30                 m_cookie.return_value = 'InvalidUser'
  31                 return xstep(self, *args, **kwargs)
  32         return wrap
  33     return _mock_invalid_lfid
  34
  35
  36 def mock_valid_lfid():
  37     def _mock_valid_lfid(xstep):
  38         def wrap(self, *args, **kwargs):
  39             with mock.patch(O_get_secure_cookie) as m_cookie:
  40                 m_cookie.return_value = 'ValidUser'
  41                 return xstep(self, *args, **kwargs)
  42         return wrap
  43     return _mock_valid_lfid
  44
  45
  46 def upload(excepted_status, excepted_response):
  47     def _upload(create_request):
  48         @functools.wraps(create_request)
  49         def wrap(self):
  50             request = create_request(self)
  51             status, body = self.upload(request)
  52             if excepted_status == httplib.OK:
  53                 getattr(self, excepted_response)(body)
  54             else:
  55                 self.assertIn(excepted_response, body)
  56         return wrap
  57     return _upload
  58
  59
  60 def create(excepted_status, excepted_response):
  61     def _create(create_request):
  62         @functools.wraps(create_request)
  63         def wrap(self):
  64             request = create_request(self)
  65             status, body = self.create(request)
  66             if excepted_status == httplib.OK:
  67                 getattr(self, excepted_response)(body)
  68             else:
  69                 self.assertIn(excepted_response, body)
  70         return wrap
  71     return _create
  72
  73
  74 def get(excepted_status, excepted_response):
  75     def _get(get_request):
  76         @functools.wraps(get_request)
  77         def wrap(self):
  78             request = get_request(self)
  79             status, body = self.get(request)
  80             if excepted_status == httplib.OK:
  81                 getattr(self, excepted_response)(body)
  82             else:
  83                 self.assertIn(excepted_response, body)
  84         return wrap
  85     return _get
  86
  87
  88 def update(excepted_status, excepted_response):
  89     def _update(update_request):
  90         @functools.wraps(update_request)
  91         def wrap(self):
  92             request, resource = update_request(self)
  93             status, body = self.update(request, resource)
  94             if excepted_status == httplib.OK:
  95                 getattr(self, excepted_response)(request, body)
  96             else:
  97                 self.assertIn(excepted_response, body)
  98         return wrap
  99     return _update
 100
 101
 102 def delete(excepted_status, excepted_response):
 103     def _delete(delete_request):
 104         @functools.wraps(delete_request)
 105         def wrap(self):
 106             request = delete_request(self)
 107             if isinstance(request, tuple):
 108                 status, body = self.delete(request[0], *(request[1]))
 109             else:
 110                 status, body = self.delete(request)
 111             if excepted_status == httplib.OK:
 112                 getattr(self, excepted_response)(body)
 113             else:
 114                 self.assertIn(excepted_response, body)
 115         return wrap
 116     return _delete
 117
 118
 119 def query(excepted_status, excepted_response, number=0):
 120     def _query(get_request):
 121         @functools.wraps(get_request)
 122         def wrap(self):
 123             request = get_request(self)
 124             status, body = self.query(request)
 125             if excepted_status == httplib.OK:
 126                 getattr(self, excepted_response)(body, number)
 127             else:
 128                 self.assertIn(excepted_response, body)
 129         return wrap
 130     return _query