test/tools/CSIT_Test/base/restlib.py

   1 """
   2 CSIT test tools.
   3 Authors: Denghui Huang@IBM, Baohua Yang@IBM
   4 Updated: 2013-11-06
   5 """
   6 import json
   7
   8 import requests
   9
  10
  11 # Global variables
  12 DEFAULT_CONTROLLER_IP = '127.0.0.1'
  13 # DEFAULT_CONTROLLER_IP = '9.186.105.113' #just for temp test
  14 DEFAULT_PORT = '8080'
  15 DEFAULT_PREFIX = 'http://' + DEFAULT_CONTROLLER_IP + ':' + DEFAULT_PORT
  16 DEFAULT_CONTAINER = 'default'
  17 DEFAULT_USER = 'admin'
  18 DEFAULT_PWD = 'admin'
  19 MODULES_DIR = 'modules'
  20 TIMEOUTS = 2
  21
  22 '''
  23 Send a POST request.
  24 '''
  25
  26
  27 def do_post_request(url, content_type, payload=None, user=DEFAULT_USER, password=DEFAULT_PWD):
  28     data = payload
  29     headers = {}
  30     if content_type == 'json':
  31         headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
  32         if payload is not None:
  33             data = json.dumps(payload)
  34     elif content_type == 'xml':
  35         headers = {'Content-type': 'application/xml', 'Accept': 'application/xml'}
  36     else:
  37         print 'unsupported content-type'
  38     try:
  39         r = requests.post(url, data, headers=headers, auth=(user, password), timeout=TIMEOUTS)
  40         r.raise_for_status()
  41     except(requests.exceptions.HTTPError, requests.exceptions.Timeout):
  42         return 400
  43     else:
  44         return r.status_code
  45
  46
  47 def do_get_request_with_status_code(url, content_type, user=DEFAULT_USER, password=DEFAULT_PWD):
  48     '''
  49     Send a GET request.
  50     @return The status code.
  51     '''
  52     r = None
  53     try:
  54         r = requests.get(url, auth=(user, password), timeout=TIMEOUTS)
  55         r.raise_for_status()
  56     except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
  57         print e
  58         return r.status_code
  59     finally:
  60         return r.status_code
  61
  62
  63 def do_put_request(url, content_type, payload=None, user=DEFAULT_USER, password=DEFAULT_PWD):
  64     '''
  65     Send a PUT request.
  66     @return The status code.
  67     '''
  68     data = payload
  69     headers = {}
  70     if content_type == 'json':
  71         headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
  72         if payload is not None:
  73             data = json.dumps(payload)
  74     elif content_type == 'xml':
  75         headers = {'Content-type': 'application/xml', 'Accept': 'application/xml'}
  76     else:
  77         print 'unsupported content-type'
  78     try:
  79         r = requests.put(url, data, headers=headers, auth=(user, password), timeout=TIMEOUTS)
  80         r.raise_for_status()
  81     except(requests.exceptions.HTTPError, requests.exceptions.Timeout):
  82         return 400
  83     else:
  84         return r.status_code
  85
  86
  87 def do_delete_request(url, user=DEFAULT_USER, password=DEFAULT_PWD):
  88     '''
  89     Send a DELETE request.
  90     @return The status code.
  91     '''
  92     r = None
  93     try:
  94         r = requests.delete(url, auth=(user, password), timeout=TIMEOUTS)
  95         r.raise_for_status()
  96     except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
  97         print e
  98     finally:
  99         if r:
 100             return r.status_code
 101
 102
 103 def convert_result_to_list(result):
 104     '''
 105     Convert the result content to list.
 106     '''
 107     list2 = []
 108     # print result
 109     content = result.values()
 110     for list1 in content:
 111         list2 = [dict1.values() for dict1 in list1]
 112         # print list2
 113     list3 = []
 114     for list4 in list2:
 115         for element in list4:
 116             list3.append(element)
 117             # print list3
 118     return list3
 119
 120
 121 def do_get_request_with_response_content(url, content_type, user=DEFAULT_USER, password=DEFAULT_PWD,
 122                                          convert_to_list=False):
 123     '''
 124     Send a GET request and get the response.
 125     @return response content as list.
 126     '''
 127     try:
 128         r = requests.get(url, auth=(user, password), timeout=TIMEOUTS)
 129         r.raise_for_status()
 130     except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
 131         print e
 132         return None
 133     else:
 134         if r is not None:
 135             if content_type == 'json':
 136                 content = r.json()
 137                 return convert_result_to_list(content) if convert_to_list else content
 138             elif content_type == 'xml':  # TODO: add parser to xml
 139                 return None
 140
 141
 142 if __name__ == '__main__':
 143     # example
 144     # Note: in json body, all field name and value (if it is string type) must be enclosed in double quotes.
 145     # This constraint maybe cause by json parser.
 146     body = {"status": "Success", "dstNodeConnector": "OF|1@OF|00:00:00:00:00:00:00:01", "name": "link3",
 147             "srcNodeConnector": "OF|1@OF|00:00:00:00:00:00:00:03"}
 148     url = 'http://127.0.0.1:8080/controller/nb/v2/topology/default/userLink/link3'
 149     content_type = 'json'
 150     print do_put_request(url, content_type, body)