4 from urllib import urlencode
8 from robot.libraries.BuiltIn import BuiltIn
11 class RequestsLibrary(object):
12 ROBOT_LIBRARY_SCOPE = 'Global'
15 self._cache = robot.utils.ConnectionCache('No sessions created')
16 self.builtin = BuiltIn()
18 def _utf8_urlencode(self, data):
19 if not type(data) is dict:
23 for k,v in data.iteritems():
24 utf8_data[k] = unicode(v).encode('utf-8')
25 return urlencode(utf8_data)
27 def create_session(self, alias, url, headers={}, cookies=None,
28 auth=None, timeout=None, proxies=None,
31 """ Create Session: create a HTTP session to a server
33 `url` Base url of the server
35 `alias` Robot Framework alias to identify the session
37 `headers` Dictionary of default headers
39 `auth` Dictionary of username & password for HTTP Basic Auth
41 `timeout` connection timeout
43 `proxies` proxy server url
45 `verify` set to True if Requests should verify the certificate
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
55 s.verify = self.builtin.convert_to_boolean(verify)
57 # cant pass these into the Session anymore
58 self.timeout = timeout
59 self.cookies = cookies
65 self._cache.register(session, alias=alias)
68 def delete_all_sessions(self):
69 """ Removes all the session objects """
71 self._cache.empty_cache()
73 def to_json(self, content):
74 """ Convert a string to a JSON object
76 `content` String content to convert into JSON
78 return json.loads(content)
81 def _get_url(self, session, uri):
82 ''' Helpere method to get the full url
86 slash = '' if uri.startswith('/') else '/'
87 url = "%s%s%s" %(session.url, slash, uri)
90 def get(self, alias, uri, headers=None):
91 """ Send a GET request on the session object found using the
94 `alias` that will be used to identify the Session object in the cache
96 `uri` to send the GET request to
98 `headers` a dictionary of headers to use with the request
101 session = self._cache.switch(alias)
102 resp = session.get(self._get_url(session, uri),
104 cookies=self.cookies, timeout=self.timeout)
106 # store the last response object
107 session.last_resp = resp
110 def post(self, alias, uri, data={}, headers=None, files={}):
111 """ Send a POST request on the session object found using the
114 `alias` that will be used to identify the Session object in the cache
116 `uri` to send the GET request to
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
122 `headers` a dictionary of headers to use with the request
124 `files` a dictionary of file names containing file data to POST to the server
127 session = self._cache.switch(alias)
128 data = self._utf8_urlencode(data)
130 resp = session.post(self._get_url(session, uri),
131 data=data, headers=headers,
133 cookies=self.cookies, timeout=self.timeout)
135 # store the last response object
136 session.last_resp = resp
137 self.builtin.log("Post response: " + resp.content, 'DEBUG')
140 def put(self, alias, uri, data=None, headers=None):
141 """ Send a PUT request on the session object found using the
144 `alias` that will be used to identify the Session object in the cache
146 `uri` to send the PUT request to
148 `headers` a dictionary of headers to use with the request
152 session = self._cache.switch(alias)
153 #data = json.dumps(self._utf8_urlencode(data))
154 data = json.dumps(data)
156 resp = session.put(self._get_url(session, uri),
157 data=data, headers=headers,
158 cookies=self.cookies, timeout=self.timeout)
160 self.builtin.log("PUT response: %s DEBUG" % resp.content)
162 # store the last response object
163 session.last_resp = resp
166 def delete(self, alias, uri, data=(), headers=None):
167 """ Send a DELETE request on the session object found using the
170 `alias` that will be used to identify the Session object in the cache
172 `uri` to send the DELETE request to
174 `headers` a dictionary of headers to use with the request
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)
184 # store the last response object
185 session.last_resp = resp
188 def head(self, alias, uri, headers=None):
189 """ Send a HEAD request on the session object found using the
192 `alias` that will be used to identify the Session object in the cache
194 `uri` to send the HEAD request to
196 `headers` a dictionary of headers to use with the request
200 session = self._cache.switch(alias)
201 resp = session.head(self._get_url(session, uri), headers=headers,
202 cookies=self.cookies, timeout=self.timeout)
204 # store the last response object
205 session.last_resp = resp