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     def _get_url(self, session, uri):
  82         """Helpere method to get the full url"""
  83         url = session.url
  84         if uri:
  85             slash = '' if uri.startswith('/') else '/'
  86             url = "%s%s%s" % (session.url, slash, uri)
  87         return url
  88
  89     def get(self, alias, uri, headers=None):
  90         """Send a GET request on the session object found using the given `alias`
  91
  92         `alias` that will be used to identify the Session object in the cache
  93
  94         `uri` to send the GET request to
  95
  96         `headers` a dictionary of headers to use with the request
  97         """
  98
  99         session = self._cache.switch(alias)
 100         resp = session.get(self._get_url(session, uri),
 101                            headers=headers,
 102                            cookies=self.cookies, timeout=self.timeout)
 103
 104         # store the last response object
 105         session.last_resp = resp
 106         return resp
 107
 108     def post(self, alias, uri, data={}, headers=None, files={}):
 109         """Send a POST request on the session object found using the given `alias`
 110
 111         `alias` that will be used to identify the Session object in the cache
 112
 113         `uri` to send the GET request to
 114
 115         `data` a dictionary of key-value pairs that will be urlencoded
 116                and sent as POST data
 117                or binary data that is sent as the raw body content
 118
 119         `headers` a dictionary of headers to use with the request
 120
 121         `files` a dictionary of file names containing file data to POST to the server
 122         """
 123
 124         session = self._cache.switch(alias)
 125         data = self._utf8_urlencode(data)
 126
 127         resp = session.post(self._get_url(session, uri),
 128                             data=data, headers=headers,
 129                             files=files,
 130                             cookies=self.cookies, timeout=self.timeout)
 131
 132         # store the last response object
 133         session.last_resp = resp
 134         self.builtin.log("Post response: " + resp.content, 'DEBUG')
 135         return resp
 136
 137     def postjson(self, alias, uri, data={}, headers=None, files={}):
 138         """Send a POST request on the session object found using the given `alias`
 139
 140         `alias` that will be used to identify the Session object in the cache
 141
 142         `uri` to send the GET request to
 143
 144         `data` a dictionary of key-value pairs that will be urlencoded
 145                and sent as POST data
 146                or binary data that is sent as the raw body content
 147
 148         `headers` a dictionary of headers to use with the request
 149
 150         `files` a dictionary of file names containing file data to POST to the server
 151         """
 152
 153         session = self._cache.switch(alias)
 154         data = json.dumps(data)
 155
 156         resp = session.post(self._get_url(session, uri),
 157                             data=data, headers=headers,
 158                             files=files,
 159                             cookies=self.cookies, timeout=self.timeout)
 160
 161         # store the last response object
 162         session.last_resp = resp
 163         self.builtin.log("Post response: " + resp.content, 'DEBUG')
 164         return resp
 165
 166     def put(self, alias, uri, data=None, headers=None):
 167         """Send a PUT request on the session object found using the given `alias`
 168
 169         `alias` that will be used to identify the Session object in the cache
 170
 171         `uri` to send the PUT request to
 172
 173         `headers` a dictionary of headers to use with the request
 174
 175         """
 176
 177         session = self._cache.switch(alias)
 178         # data = self._utf8_urlencode(data)
 179         data = json.dumps(data)
 180
 181         resp = session.put(self._get_url(session, uri),
 182                            data=data, headers=headers,
 183                            cookies=self.cookies, timeout=self.timeout)
 184
 185         self.builtin.log("PUT response: %s DEBUG" % resp.content)
 186
 187         # store the last response object
 188         session.last_resp = resp
 189         return resp
 190
 191     def put_xml(self, alias, uri, data=None, headers=None):
 192         """Send a PUT_xml request on the session object found using the given `alias`
 193
 194         `alias` that will be used to identify the Session object in the cache
 195
 196         `uri` to send the PUT_xml request to
 197
 198         `headers` a dictionary of headers to use with the request
 199
 200         """
 201
 202         session = self._cache.switch(alias)
 203         data = self._utf8_urlencode(data)
 204         # data = json.dumps(data)
 205
 206         resp = session.put(self._get_url(session, uri),
 207                            data=data, headers=headers,
 208                            cookies=self.cookies, timeout=self.timeout)
 209
 210         self.builtin.log("PUT response: %s DEBUG" % resp.content)
 211
 212         # store the last response object
 213         session.last_resp = resp
 214         return resp
 215
 216     def delete(self, alias, uri, data=(), headers=None):
 217         """Send a DELETE request on the session object found using the given `alias`
 218
 219         `alias` that will be used to identify the Session object in the cache
 220
 221         `uri` to send the DELETE request to
 222
 223         `headers` a dictionary of headers to use with the request
 224
 225         """
 226
 227         session = self._cache.switch(alias)
 228         args = "?%s" % urlencode(data) if data else ''
 229         resp = session.delete("%s%s" % (self._get_url(session, uri), args),
 230                               headers=headers, cookies=self.cookies,
 231                               timeout=self.timeout)
 232
 233         # store the last response object
 234         session.last_resp = resp
 235         return resp
 236
 237     def head(self, alias, uri, headers=None):
 238         """Send a HEAD request on the session object found using the given `alias`
 239
 240         `alias` that will be used to identify the Session object in the cache
 241
 242         `uri` to send the HEAD request to
 243
 244         `headers` a dictionary of headers to use with the request
 245
 246         """
 247
 248         session = self._cache.switch(alias)
 249         resp = session.head(self._get_url(session, uri), headers=headers,
 250                             cookies=self.cookies, timeout=self.timeout)
 251
 252         # store the last response object
 253         session.last_resp = resp
 254         return resp