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
  13 def upload(excepted_status, excepted_response):
  14     def _upload(create_request):
  15         @functools.wraps(create_request)
  16         def wrap(self):
  17             request = create_request(self)
  18             status, body = self.upload(request)
  19             if excepted_status == httplib.OK:
  20                 getattr(self, excepted_response)(body)
  21             else:
  22                 self.assertIn(excepted_response, body)
  23         return wrap
  24     return _upload
  25
  26
  27 def create(excepted_status, excepted_response):
  28     def _create(create_request):
  29         @functools.wraps(create_request)
  30         def wrap(self):
  31             request = create_request(self)
  32             status, body = self.create(request)
  33             if excepted_status == httplib.OK:
  34                 getattr(self, excepted_response)(body)
  35             else:
  36                 self.assertIn(excepted_response, body)
  37         return wrap
  38     return _create
  39
  40
  41 def get(excepted_status, excepted_response):
  42     def _get(get_request):
  43         @functools.wraps(get_request)
  44         def wrap(self):
  45             request = get_request(self)
  46             status, body = self.get(request)
  47             if excepted_status == httplib.OK:
  48                 getattr(self, excepted_response)(body)
  49             else:
  50                 self.assertIn(excepted_response, body)
  51         return wrap
  52     return _get
  53
  54
  55 def update(excepted_status, excepted_response):
  56     def _update(update_request):
  57         @functools.wraps(update_request)
  58         def wrap(self):
  59             request, resource = update_request(self)
  60             status, body = self.update(request, resource)
  61             if excepted_status == httplib.OK:
  62                 getattr(self, excepted_response)(request, body)
  63             else:
  64                 self.assertIn(excepted_response, body)
  65         return wrap
  66     return _update
  67
  68
  69 def delete(excepted_status, excepted_response):
  70     def _delete(delete_request):
  71         @functools.wraps(delete_request)
  72         def wrap(self):
  73             request = delete_request(self)
  74             if isinstance(request, tuple):
  75                 status, body = self.delete(request[0], *(request[1]))
  76             else:
  77                 status, body = self.delete(request)
  78             if excepted_status == httplib.OK:
  79                 getattr(self, excepted_response)(body)
  80             else:
  81                 self.assertIn(excepted_response, body)
  82         return wrap
  83     return _delete
  84
  85
  86 def query(excepted_status, excepted_response, number=0):
  87     def _query(get_request):
  88         @functools.wraps(get_request)
  89         def wrap(self):
  90             request = get_request(self)
  91             status, body = self.query(request)
  92             if excepted_status == httplib.OK:
  93                 getattr(self, excepted_response)(body, number)
  94             else:
  95                 self.assertIn(excepted_response, body)
  96         return wrap
  97     return _query