kennethreitz/requests
1
0
Fork 0
mirror of https://github.com/kennethreitz/requests.git synced 2026-06-05 22:50:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Standard tests from RequestsTestSuite run using async api

Browse Source
This commit is contained in:
Piotr Dobrogost
2012-01-24 21:56:59 +01:00
parent 50298bbf97
commit c430b60e38
2 changed files with 148 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_requests.py
Executable → Regular
+90 -86
View File
@@ -14,6 +14,7 @@ import requests
from requests.compat import str, bytes, StringIO
# import envoy
from requests import HTTPError
from requests import get, post, head, put
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
@@ -36,7 +37,7 @@ SERVICES = (httpbin, )
_httpbin = False
class RequestsTestSuite(unittest.TestCase):
class TestSetup(object):
"""Requests test cases."""
# It goes to eleven.
@@ -52,6 +53,9 @@ class RequestsTestSuite(unittest.TestCase):
_httpbin = True
class RequestsTestSuite(TestSetup, unittest.TestCase):
"""Requests test cases."""
def test_entry_points(self):
requests.session
@@ -66,30 +70,30 @@ class RequestsTestSuite(unittest.TestCase):
def test_invalid_url(self):
self.assertRaises(ValueError, requests.get, 'hiwpefhipowhefopw')
self.assertRaises(ValueError, get, 'hiwpefhipowhefopw')
def test_HTTP_200_OK_GET(self):
r = requests.get(httpbin('/get'))
r = get(httpbin('/get'))
self.assertEqual(r.status_code, 200)
def test_response_sent(self):
r = requests.get(httpbin('/get'))
r = get(httpbin('/get'))
self.assertTrue(r.request.sent)
def test_HTTP_302_ALLOW_REDIRECT_GET(self):
r = requests.get(httpbin('redirect', '1'))
r = get(httpbin('redirect', '1'))
self.assertEqual(r.status_code, 200)
def test_HTTP_302_GET(self):
r = requests.get(httpbin('redirect', '1'), allow_redirects=False)
r = get(httpbin('redirect', '1'), allow_redirects=False)
self.assertEqual(r.status_code, 302)
def test_HTTP_200_OK_GET_WITH_PARAMS(self):
heads = {'User-agent': 'Mozilla/5.0'}
r = requests.get(httpbin('user-agent'), headers=heads)
r = get(httpbin('user-agent'), headers=heads)
assert heads['User-agent'] in r.text
self.assertEqual(r.status_code, 200)
@@ -98,7 +102,7 @@ class RequestsTestSuite(unittest.TestCase):
def test_HTTP_200_OK_GET_WITH_MIXED_PARAMS(self):
heads = {'User-agent': 'Mozilla/5.0'}
r = requests.get(httpbin('get') + '?test=true', params={'q': 'test'}, headers=heads)
r = get(httpbin('get') + '?test=true', params={'q': 'test'}, headers=heads)
self.assertEqual(r.status_code, 200)
@@ -110,7 +114,7 @@ class RequestsTestSuite(unittest.TestCase):
'Mozilla/5.0 (github.com/kennethreitz/requests)'
}
r = requests.get(httpbin('user-agent'), headers=heads);
r = get(httpbin('user-agent'), headers=heads);
self.assertTrue(heads['User-agent'] in r.text)
heads = {
@@ -118,17 +122,17 @@ class RequestsTestSuite(unittest.TestCase):
'Mozilla/5.0 (github.com/kennethreitz/requests)'
}
r = requests.get(httpbin('user-agent'), headers=heads);
r = get(httpbin('user-agent'), headers=heads);
self.assertTrue(heads['user-agent'] in r.text)
def test_HTTP_200_OK_HEAD(self):
r = requests.head(httpbin('/get'))
r = head(httpbin('/get'))
self.assertEqual(r.status_code, 200)
def test_HTTP_200_OK_PUT(self):
r = requests.put(httpbin('put'))
r = put(httpbin('put'))
self.assertEqual(r.status_code, 200)
@@ -139,15 +143,15 @@ class RequestsTestSuite(unittest.TestCase):
auth = ('user', 'pass')
url = service('basic-auth', 'user', 'pass')
r = requests.get(url, auth=auth)
r = get(url, auth=auth)
self.assertEqual(r.status_code, 200)
r = requests.get(url)
r = get(url)
self.assertEqual(r.status_code, 401)
s = requests.session(auth=auth)
r = s.get(url)
r = get(url, session=s)
self.assertEqual(r.status_code, 200)
@@ -158,19 +162,19 @@ class RequestsTestSuite(unittest.TestCase):
auth = HTTPBasicAuth('user', 'pass')
url = service('basic-auth', 'user', 'pass')
r = requests.get(url, auth=auth)
r = get(url, auth=auth)
self.assertEqual(r.status_code, 200)
auth = ('user', 'pass')
r = requests.get(url, auth=auth)
r = get(url, auth=auth)
self.assertEqual(r.status_code, 200)
r = requests.get(url)
r = get(url)
self.assertEqual(r.status_code, 401)
s = requests.session(auth=auth)
r = s.get(url)
r = get(url, session=s)
self.assertEqual(r.status_code, 200)
@@ -181,15 +185,15 @@ class RequestsTestSuite(unittest.TestCase):
auth = HTTPDigestAuth('user', 'pass')
url = service('digest-auth', 'auth', 'user', 'pass')
r = requests.get(url, auth=auth)
r = get(url, auth=auth)
self.assertEqual(r.status_code, 200)
r = requests.get(url)
r = get(url)
self.assertEqual(r.status_code, 401)
s = requests.session(auth=auth)
r = s.get(url)
r = get(url, session=s)
self.assertEqual(r.status_code, 200)
def test_POSTBIN_GET_POST_FILES(self):
@@ -197,15 +201,15 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
url = service('post')
post = requests.post(url).raise_for_status()
post1 = post(url).raise_for_status()
post = requests.post(url, data={'some': 'data'})
self.assertEqual(post.status_code, 200)
post1 = post(url, data={'some': 'data'})
self.assertEqual(post1.status_code, 200)
post2 = requests.post(url, files={'some': open('test_requests.py')})
post2 = post(url, files={'some': open('test_requests.py')})
self.assertEqual(post2.status_code, 200)
post3 = requests.post(url, data='[{"some": "json"}]')
post3 = post(url, data='[{"some": "json"}]')
self.assertEqual(post3.status_code, 200)
@@ -214,11 +218,11 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
url = service('post')
post = requests.post(url,
files={'some': open('test_requests.py')},
data={'some': 'data'})
post1 = post(url,
files={'some': open('test_requests.py')},
data={'some': 'data'})
self.assertEqual(post.status_code, 200)
self.assertEqual(post1.status_code, 200)
def test_POSTBIN_GET_POST_FILES_WITH_HEADERS(self):
@@ -227,7 +231,7 @@ class RequestsTestSuite(unittest.TestCase):
url = service('post')
post2 = requests.post(url,
post2 = post(url,
files={'some': open('test_requests.py')},
headers = {'User-Agent': 'requests-tests'})
@@ -238,10 +242,10 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.get(service('status', '500'))
r = get(service('status', '500'))
self.assertEqual(bool(r), False)
r = requests.get(service('/get'))
r = get(service('/get'))
self.assertEqual(bool(r), True)
@@ -249,17 +253,17 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.get(service('status', '404'))
r = get(service('status', '404'))
# print r.status_code
# r.raise_for_status()
self.assertEqual(r.ok, False)
def test_status_raising(self):
r = requests.get(httpbin('status', '404'))
r = get(httpbin('status', '404'))
self.assertRaises(HTTPError, r.raise_for_status)
r = requests.get(httpbin('status', '200'))
r = get(httpbin('status', '200'))
self.assertFalse(r.error)
r.raise_for_status()
@@ -268,15 +272,15 @@ class RequestsTestSuite(unittest.TestCase):
config = {'danger_mode': True}
args = [httpbin('status', '404')]
kwargs = dict(config=config)
self.assertRaises(HTTPError, requests.get, *args, **kwargs)
self.assertRaises(HTTPError, get, *args, **kwargs)
r = requests.get(httpbin('status', '200'))
r = get(httpbin('status', '200'))
self.assertEqual(r.status_code, 200)
def test_decompress_gzip(self):
r = requests.get(httpbin('gzip'))
r = get(httpbin('gzip'))
r.content.decode('ascii')
def test_response_has_unicode_url(self):
@@ -285,7 +289,7 @@ class RequestsTestSuite(unittest.TestCase):
url = service('get')
response = requests.get(url)
response = get(url)
assert isinstance(response.url, str)
@@ -296,11 +300,11 @@ class RequestsTestSuite(unittest.TestCase):
url = service('/get')
requests.get(url, params={'foo': 'føø'})
requests.get(url, params={'føø': 'føø'})
requests.get(url, params={'føø': 'føø'})
requests.get(url, params={'foo': 'foo'})
requests.get(service('ø'), params={'foo': 'foo'})
get(url, params={'foo': 'føø'})
get(url, params={'føø': 'føø'})
get(url, params={'føø': 'føø'})
get(url, params={'foo': 'foo'})
get(service('ø'), params={'foo': 'foo'})
def test_httpauth_recursion(self):
@@ -308,7 +312,7 @@ class RequestsTestSuite(unittest.TestCase):
http_auth = HTTPBasicAuth('user', 'BADpass')
for service in SERVICES:
r = requests.get(service('basic-auth', 'user', 'pass'), auth=http_auth)
r = get(service('basic-auth', 'user', 'pass'), auth=http_auth)
self.assertEqual(r.status_code, 401)
@@ -316,7 +320,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.post(service('post'), data=dict(test='fooaowpeuf'))
r = post(service('post'), data=dict(test='fooaowpeuf'))
self.assertEqual(r.status_code, 200)
self.assertEqual(r.headers['content-type'], 'application/json')
@@ -332,7 +336,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.post(service('post'), data='fooaowpeuf')
r = post(service('post'), data='fooaowpeuf')
self.assertEqual(r.status_code, 200)
self.assertEqual(r.headers['content-type'], 'application/json')
@@ -350,7 +354,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.post(service('post'), params=dict(test='fooaowpeuf'))
r = post(service('post'), params=dict(test='fooaowpeuf'))
self.assertEqual(r.status_code, 200)
self.assertEqual(r.headers['content-type'], 'application/json')
@@ -365,7 +369,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.post(
r = post(
service('post'),
params=dict(test='fooaowpeuf'),
data=dict(test2="foobar"))
@@ -383,7 +387,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.post(service('post'), data="foobar")
r = post(service('post'), data="foobar")
self.assertEqual(r.status_code, 200)
self.assertEqual(r.headers['content-type'], 'application/json')
@@ -395,7 +399,7 @@ class RequestsTestSuite(unittest.TestCase):
# def test_idna(self):
# r = requests.get(u'http://➡.ws/httpbin')
# r = get(u'http://➡.ws/httpbin')
# assert 'httpbin' in r.url
@@ -403,7 +407,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.get(service('get'), params=dict(test=['foo','baz']))
r = get(service('get'), params=dict(test=['foo','baz']))
self.assertEqual(r.status_code, 200)
self.assertEqual(r.url, service('get?test=foo&test=baz'))
@@ -412,7 +416,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.post(service('post'), params=dict(test=['foo','baz']))
r = post(service('post'), params=dict(test=['foo','baz']))
self.assertEqual(r.status_code, 200)
self.assertEqual(r.headers['content-type'], 'application/json')
self.assertEqual(r.url, service('post?test=foo&test=baz'))
@@ -426,7 +430,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.post(
r = post(
service('post'),
params=dict(test=['foo','baz']),
data=dict(test2="foobar",test3=['foo','baz']))
@@ -447,7 +451,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.get(service('redirect', '3'), allow_redirects=False)
r = get(service('redirect', '3'), allow_redirects=False)
self.assertEqual(r.status_code, 302)
self.assertEqual(len(r.history), 0)
@@ -456,7 +460,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.head(service('redirect', '3'), allow_redirects=False)
r = head(service('redirect', '3'), allow_redirects=False)
self.assertEqual(r.status_code, 302)
self.assertEqual(len(r.history), 0)
@@ -465,7 +469,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.get(service('redirect', '3'))
r = get(service('redirect', '3'))
self.assertEqual(r.status_code, 200)
self.assertEqual(len(r.history), 3)
@@ -474,7 +478,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
r = requests.get(service('relative-redirect', '3'))
r = get(service('relative-redirect', '3'))
self.assertEqual(r.status_code, 200)
self.assertEqual(len(r.history), 3)
@@ -482,7 +486,7 @@ class RequestsTestSuite(unittest.TestCase):
def test_session_HTTP_200_OK_GET(self):
s = requests.session()
r = s.get(httpbin('/get'))
r = get(httpbin('/get'), session=s)
self.assertEqual(r.status_code, 200)
@@ -494,14 +498,14 @@ class RequestsTestSuite(unittest.TestCase):
s.headers = heads
# Make 2 requests from Session object, should send header both times
r1 = s.get(httpbin('user-agent'))
r1 = get(httpbin('user-agent'), session=s)
assert heads['User-agent'] in r1.text
r2 = s.get(httpbin('user-agent'))
r2 = get(httpbin('user-agent'), session=s)
assert heads['User-agent'] in r2.text
new_heads = {'User-agent': 'blah'}
r3 = s.get(httpbin('user-agent'), headers=new_heads)
r3 = get(httpbin('user-agent'), headers=new_heads, session=s)
assert new_heads['User-agent'] in r3.text
self.assertEqual(r2.status_code, 200)
@@ -521,7 +525,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
url = service('headers')
response = requests.get(
response = get(
url = url,
hooks = {
'args': add_foo_header
@@ -555,7 +559,7 @@ class RequestsTestSuite(unittest.TestCase):
for service in SERVICES:
url = service('headers')
response = requests.get(
response = get(
url = url,
hooks = {
'args': [add_foo_header, add_bar_header]
@@ -571,38 +575,38 @@ class RequestsTestSuite(unittest.TestCase):
# Internally dispatched cookies are sent.
_c = {'kenneth': 'reitz', 'bessie': 'monke'}
r = s.get(httpbin('cookies'), cookies=_c)
r = s.get(httpbin('cookies'))
r = get(httpbin('cookies'), cookies=_c, session=s)
r = get(httpbin('cookies'), session=s)
# Those cookies persist transparently.
c = json.loads(r.text).get('cookies')
assert c == _c
# Double check.
r = s.get(httpbin('cookies'), cookies={})
r = get(httpbin('cookies'), cookies={}, session=s)
c = json.loads(r.text).get('cookies')
assert c == _c
# Remove a cookie by setting it's value to None.
r = s.get(httpbin('cookies'), cookies={'bessie': None})
r = get(httpbin('cookies'), cookies={'bessie': None}, session=s)
c = json.loads(r.text).get('cookies')
del _c['bessie']
assert c == _c
# Test session-level cookies.
s = requests.session(cookies=_c)
r = s.get(httpbin('cookies'))
r = get(httpbin('cookies'), session=s)
c = json.loads(r.text).get('cookies')
assert c == _c
# Have the server set a cookie.
r = s.get(httpbin('cookies', 'set', 'k', 'v'), allow_redirects=True)
r = get(httpbin('cookies', 'set', 'k', 'v'), allow_redirects=True, session=s)
c = json.loads(r.text).get('cookies')
assert 'k' in c
# And server-set cookie persistience.
r = s.get(httpbin('cookies'))
r = get(httpbin('cookies'), session=s)
c = json.loads(r.text).get('cookies')
assert 'k' in c
@@ -617,20 +621,20 @@ class RequestsTestSuite(unittest.TestCase):
s.params = params
# Make 2 requests from Session object, should send header both times
r1 = s.get(httpbin('get'))
r1 = get(httpbin('get'), session=s)
assert params['a'] in r1.text
params2 = {'b': 'b_test'}
r2 = s.get(httpbin('get'), params=params2)
r2 = get(httpbin('get'), params=params2, session=s)
assert params['a'] in r2.text
assert params2['b'] in r2.text
params3 = {'b': 'b_test', 'a': None, 'c': 'c_test'}
r3 = s.get(httpbin('get'), params=params3)
r3 = get(httpbin('get'), params=params3, session=s)
assert not params['a'] in r3.text
assert params3['b'] in r3.text
@@ -641,7 +645,7 @@ class RequestsTestSuite(unittest.TestCase):
# this will fail.
try:
hah = 'http://somedomainthatclearlydoesntexistg.com'
r = requests.get(hah, allow_redirects=False)
r = get(hah, allow_redirects=False)
except requests.ConnectionError:
pass # \o/
else:
@@ -649,17 +653,17 @@ class RequestsTestSuite(unittest.TestCase):
config = {'safe_mode': True}
r = requests.get(hah, allow_redirects=False, config=config)
r = get(hah, allow_redirects=False, config=config)
assert r.content == None
def test_cached_response(self):
r1 = requests.get(httpbin('get'), prefetch=False)
r1 = get(httpbin('get'), prefetch=False)
assert not r1._content
assert r1.content
assert r1.text
r2 = requests.get(httpbin('get'), prefetch=True)
r2 = get(httpbin('get'), prefetch=True)
assert r2._content
assert r2.content
assert r2.text
@@ -669,7 +673,7 @@ class RequestsTestSuite(unittest.TestCase):
lines = (0, 2, 10, 100)
for i in lines:
r = requests.get(httpbin('stream', str(i)), prefetch=False)
r = get(httpbin('stream', str(i)), prefetch=False)
lines = list(r.iter_lines())
len_lines = len(lines)
@@ -683,7 +687,7 @@ class RequestsTestSuite(unittest.TestCase):
)
# Make a request and monkey-patch its contents
r = requests.get(httpbin('get'))
r = get(httpbin('get'))
r.raw = StringIO(quote)
# Make sure iter_lines doesn't chop the trailing bit
@@ -696,22 +700,22 @@ class RequestsTestSuite(unittest.TestCase):
# Safe mode creates empty responses for failed requests.
# Iterating on these responses should produce empty sequences
r = safe.get('http://_/')
r = get('http://_/', session=safe)
self.assertEqual(list(r.iter_lines()), [])
assert isinstance(r.error, requests.exceptions.ConnectionError)
r = safe.get('http://_/')
r = get('http://_/', session=safe)
self.assertEqual(list(r.iter_content()), [])
assert isinstance(r.error, requests.exceptions.ConnectionError)
# When not in safe mode, should raise Timeout exception
self.assertRaises(
requests.exceptions.Timeout,
requests.get,
get,
httpbin('stream', '1000'), timeout=0.0001)
# In safe mode, should return a blank response
r = requests.get(httpbin('stream', '1000'), timeout=0.0001,
r = get(httpbin('stream', '1000'), timeout=0.0001,
config=dict(safe_mode=True))
assert r.content is None
assert isinstance(r.error, requests.exceptions.Timeout)
test_requests_async.py
+58
View File
@@ -0,0 +1,58 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import unittest
from requests import async
import envoy
sys.path.append('.')
from test_requests import httpbin, RequestsTestSuite, SERVICES
class RequestsTestSuiteUsingAsyncApi(RequestsTestSuite):
"""Requests async test cases."""
def patched(f):
"""Automatically send request after creation."""
def wrapped(*args, **kwargs):
request = f(*args, **kwargs)
return async.map([request])[0]
return wrapped
# Patched requests.api functions.
global request
request = patched(async.request)
global delete, get, head, options, patch, post, put
delete = patched(async.delete)
get = patched(async.get)
head = patched(async.head)
options = patched(async.options)
patch = patched(async.patch)
post = patched(async.post)
put = patched(async.put)
def test_entry_points(self):
async.request
async.delete
async.get
async.head
async.options
async.patch
async.post
async.put
async.map
async.send
if __name__ == '__main__':
unittest.main()