test/csit/libraries/RequestsLibrary.py

   1 import requests
   2 import json
   3
   4 from urllib import urlencode
   5
   6
   7 import robot
   8
   9 from robot.libraries.BuiltIn import BuiltIn
  10
  11
  12 class RequestsLibrary(object):
  13     ROBOT_LIBRARY_SCOPE = 'Global'
  14
  15     def __init__(self):
  16         self._cache = robot.utils.ConnectionCache('No sessions created')
  17         self.builtin = BuiltIn()
  18
  19     def _utf8_urlencode(self, data):
  20         if not type(data) is dict:
  21             return data
  22
  23         utf8_data = {}
  24         for k,v in data.iteritems():
  25             utf8_data[k] = unicode(v).encode('utf-8')
  26         return urlencode(utf8_data)
  27
  28     def create_session(self, alias, url, headers={}, cookies=None,
  29                        auth=None, timeout=None, proxies=None,
  30                        verify=False):
  31
  32         """ Create Session: create a HTTP session to a server
  33
  34         `url` Base url of the server
  35
  36         `alias` Robot Framework alias to identify the session
  37
  38         `headers` Dictionary of default headers
  39
  40         `auth` Dictionary of username & password for HTTP Basic Auth
  41
  42         `timeout` connection timeout
  43
  44         `proxies` proxy server url
  45
  46         `verify` set to True if Requests should verify the certificate
  47         """
  48
  49         self.builtin.log('Creating session: %s' % alias, 'DEBUG')
  50         auth = requests.auth.HTTPBasicAuth(*auth) if auth else None
  51         s = session = requests.Session()
  52         s.headers.update(headers)
  53         s.auth = auth if auth else s.auth
  54         s.proxies = proxies if proxies else  s.proxies
  55
  56         s.verify = self.builtin.convert_to_boolean(verify)
  57
  58         # cant pass these into the Session anymore
  59         self.timeout = timeout
  60         self.cookies = cookies
  61         self.verify = verify
  62
  63         # cant use hooks :(
  64         s.url = url
  65
  66         self._cache.register(session, alias=alias)
  67         return session
  68
  69     def delete_all_sessions(self):
  70         """ Removes all the session objects """
  71
  72         self._cache.empty_cache()
  73
  74     def to_json(self, content):
  75         """ Convert a string to a JSON object
  76
  77         `content` String content to convert into JSON
  78         """
  79         return json.loads(content)
  80
  81
  82     def _get_url(self, session, uri):
  83         ''' Helpere method to get the full url
  84         '''
  85         url = session.url
  86         if uri:
  87             slash = '' if uri.startswith('/') else '/'
  88             url = "%s%s%s" %(session.url, slash, uri)
  89         return url
  90
  91     def get(self, alias, uri, headers=None):
  92         """ Send a GET request on the session object found using the
  93             given `alias`
  94
  95         `alias` that will be used to identify the Session object in the cache
  96
  97         `uri` to send the GET request to
  98
  99         `headers` a dictionary of headers to use with the request
 100         """
 101
 102         session = self._cache.switch(alias)
 103         resp = session.get(self._get_url(session, uri),
 104                            headers=headers,
 105                            cookies=self.cookies, timeout=self.timeout)
 106
 107         # store the last response object
 108         session.last_resp = resp
 109         return resp
 110
 111     def post(self, alias, uri, data={}, headers=None, files={}):
 112         """ Send a POST request on the session object found using the
 113         given `alias`
 114
 115         `alias` that will be used to identify the Session object in the cache
 116
 117         `uri` to send the GET request to
 118
 119         `data` a dictionary of key-value pairs that will be urlencoded
 120                and sent as POST data
 121                or binary data that is sent as the raw body content
 122
 123         `headers` a dictionary of headers to use with the request
 124
 125         `files` a dictionary of file names containing file data to POST to the server
 126         """
 127
 128         session = self._cache.switch(alias)
 129         data = self._utf8_urlencode(data)
 130
 131         resp = session.post(self._get_url(session, uri),
 132                        data=data, headers=headers,
 133                        files=files,
 134                        cookies=self.cookies, timeout=self.timeout)
 135
 136         # store the last response object
 137         session.last_resp = resp
 138         self.builtin.log("Post response: " + resp.content, 'DEBUG')
 139         return resp
 140
 141     def postjson(self, alias, uri, data={}, headers=None, files={}):
 142         """ Send a POST request on the session object found using the
 143         given `alias`
 144
 145         `alias` that will be used to identify the Session object in the cache
 146
 147         `uri` to send the GET request to
 148
 149         `data` a dictionary of key-value pairs that will be urlencoded
 150                and sent as POST data
 151                or binary data that is sent as the raw body content
 152
 153         `headers` a dictionary of headers to use with the request
 154
 155         `files` a dictionary of file names containing file data to POST to the server
 156         """
 157
 158         session = self._cache.switch(alias)
 159         data = json.dumps(data)
 160
 161         resp = session.post(self._get_url(session, uri),
 162                        data=data, headers=headers,
 163                        files=files,
 164                        cookies=self.cookies, timeout=self.timeout)
 165
 166         # store the last response object
 167         session.last_resp = resp
 168         self.builtin.log("Post response: " + resp.content, 'DEBUG')
 169         return resp
 170
 171     def put(self, alias, uri, data=None, headers=None):
 172         """ Send a PUT request on the session object found using the
 173         given `alias`
 174
 175         `alias` that will be used to identify the Session object in the cache
 176
 177         `uri` to send the PUT request to
 178
 179         `headers` a dictionary of headers to use with the request
 180
 181         """
 182
 183         session = self._cache.switch(alias)
 184         #data = self._utf8_urlencode(data)
 185         data = json.dumps(data)
 186
 187         resp = session.put(self._get_url(session, uri),
 188                     data=data, headers=headers,
 189                     cookies=self.cookies, timeout=self.timeout)
 190
 191         self.builtin.log("PUT response: %s DEBUG" % resp.content)
 192
 193         # store the last response object
 194         session.last_resp = resp
 195         return resp
 196
 197     def put_xml(self, alias, uri, data=None, headers=None):
 198         """ Send a PUT_xml request on the session object found using the
 199         given `alias`
 200
 201         `alias` that will be used to identify the Session object in the cache
 202
 203         `uri` to send the PUT_xml request to
 204
 205         `headers` a dictionary of headers to use with the request
 206
 207         """
 208
 209         session = self._cache.switch(alias)
 210         data = self._utf8_urlencode(data)
 211         #data = json.dumps(data)
 212
 213         resp = session.put(self._get_url(session, uri),
 214                     data=data, headers=headers,
 215                     cookies=self.cookies, timeout=self.timeout)
 216
 217         self.builtin.log("PUT response: %s DEBUG" % resp.content)
 218
 219         # store the last response object
 220         session.last_resp = resp
 221         return resp
 222
 223     def delete(self, alias, uri, data=(), headers=None):
 224         """ Send a DELETE request on the session object found using the
 225         given `alias`
 226
 227         `alias` that will be used to identify the Session object in the cache
 228
 229         `uri` to send the DELETE request to
 230
 231         `headers` a dictionary of headers to use with the request
 232
 233         """
 234
 235         session = self._cache.switch(alias)
 236         args = "?%s" % urlencode(data) if data else ''
 237         resp = session.delete("%s%s" % (self._get_url(session, uri), args),
 238                             headers=headers, cookies=self.cookies,
 239                             timeout=self.timeout)
 240
 241         # store the last response object
 242         session.last_resp = resp
 243         return resp
 244
 245
 246     def head(self, alias, uri, headers=None):
 247         """ Send a HEAD request on the session object found using the
 248         given `alias`
 249
 250         `alias` that will be used to identify the Session object in the cache
 251
 252         `uri` to send the HEAD request to
 253
 254         `headers` a dictionary of headers to use with the request
 255
 256         """
 257
 258         session = self._cache.switch(alias)
 259         resp = session.head(self._get_url(session, uri), headers=headers,
 260                            cookies=self.cookies, timeout=self.timeout)
 261
 262         # store the last response object
 263         session.last_resp = resp
 264         return resp