Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Newer
Older
100644 93 lines (78 sloc) 3.417 kB
c4678df @peterbe initial commit
authored
1 from urllib import urlencode
2 import Cookie
3 from tornado.httpclient import HTTPRequest
4 from tornado import escape
5
8af533f @peterbe fix for utf-8 encoding POST data
authored
6 __version__ = '1.3'
c4678df @peterbe initial commit
authored
7
8 class LoginError(Exception):
9 pass
10
11 class HTTPClientMixin(object):
12
13 def get(self, url, data=None, headers=None, follow_redirects=False):
14 if data is not None:
15 if isinstance(data, dict):
16 data = urlencode(data, True)
17 if '?' in url:
18 url += '&%s' % data
19 else:
20 url += '?%s' % data
21 return self._fetch(url, 'GET', headers=headers,
22 follow_redirects=follow_redirects)
23
24 def post(self, url, data, headers=None, follow_redirects=False):
25 if data is not None:
26 if isinstance(data, dict):
8af533f @peterbe fix for utf-8 encoding POST data
authored
27 for key, value in data.items():
28 if isinstance(value, unicode):
29 data[key] = value.encode('utf-8')
c4678df @peterbe initial commit
authored
30 data = urlencode(data, True)
31 return self._fetch(url, 'POST', data, headers,
32 follow_redirects=follow_redirects)
33
34 def _fetch(self, url, method, data=None, headers=None, follow_redirects=True):
35 full_url = self.get_url(url)
36 request = HTTPRequest(full_url, follow_redirects=follow_redirects,
37 headers=headers, method=method, body=data)
38 self.http_client.fetch(request, self.stop)
39 return self.wait()
40
41
42 class TestClient(HTTPClientMixin):
43 def __init__(self, testcase):
44 self.testcase = testcase
45 self.cookies = Cookie.SimpleCookie()
46
47 def _render_cookie_back(self):
48 return ''.join(['%s=%s;' %(x, morsel.value)
49 for (x, morsel)
50 in self.cookies.items()])
51
52 def get(self, url, data=None, headers=None, follow_redirects=False):
53 if self.cookies:
54 if headers is None:
55 headers = dict()
56 headers['Cookie'] = self._render_cookie_back()
57 response = self.testcase.get(url, data=data, headers=headers,
58 follow_redirects=follow_redirects)
59
60 self._update_cookies(response.headers)
61 return response
62
63 def post(self, url, data, headers=None, follow_redirects=False):
64 if self.cookies:
65 if headers is None:
66 headers = dict()
67 headers['Cookie'] = self._render_cookie_back()
68 response = self.testcase.post(url, data=data, headers=headers,
69 follow_redirects=follow_redirects)
70 self._update_cookies(response.headers)
71 return response
72
73 def _update_cookies(self, headers):
74 try:
75 sc = headers['Set-Cookie']
0b9ad13 @bootandy cookie can now handle more than 1 value being set
bootandy authored
76 cookies = escape.native_str(sc)
77cc919 @peterbe tidying up the additional update-cookie stuff from bootandy
authored
77 self.cookies.update(Cookie.SimpleCookie(cookies))
0b9ad13 @bootandy cookie can now handle more than 1 value being set
bootandy authored
78 while True:
79 self.cookies.update(Cookie.SimpleCookie(cookies))
77cc919 @peterbe tidying up the additional update-cookie stuff from bootandy
authored
80 if ',' not in cookies:
0b9ad13 @bootandy cookie can now handle more than 1 value being set
bootandy authored
81 break
77cc919 @peterbe tidying up the additional update-cookie stuff from bootandy
authored
82 cookies = cookies[cookies.find(',') + 1:]
c4678df @peterbe initial commit
authored
83 except KeyError:
84 return
85
86 def login(self, email, password, url='/auth/login/'):
87 data = dict(email=email, password=password)
88 response = self.post(url, data, follow_redirects=False)
89 if response.code != 302:
90 raise LoginError(response.body)
91 if 'Error' in response.body:
92 raise LoginError(response.body)
Something went wrong with that request. Please try again.