csit/libraries/RequestsLibrary.py

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