test/csit/libraries/RequestsLibrary.py

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