From 1c0a691ebf468d42b7c0d6b0e9daf0b2ff82cc20 Mon Sep 17 00:00:00 2001 From: SVN-Git Migration Date: Thu, 8 Oct 2015 13:41:19 -0700 Subject: Imported Upstream version 0.8.2 --- test_requests.py | 295 +++++++++++++++++++++++++++++++++---------------------- 1 file changed, 176 insertions(+), 119 deletions(-) (limited to 'test_requests.py') diff --git a/test_requests.py b/test_requests.py index 83b827a..61953a3 100755 --- a/test_requests.py +++ b/test_requests.py @@ -3,24 +3,25 @@ from __future__ import with_statement +import time +import os import unittest -import cookielib + +import requests +import envoy +from requests import HTTPError try: import omnijson as json except ImportError: import json -import requests - -from requests.sessions import Session +# TODO: Detect an open port. +PORT = os.environ.get('HTTPBIN_PORT', '7077') -HTTPBIN_URL = 'http://httpbin.ep.io/' -HTTPSBIN_URL = 'https://httpbin.ep.io/' - -# HTTPBIN_URL = 'http://staging.httpbin.org/' -# HTTPSBIN_URL = 'https://httpbin-staging.ep.io/' +HTTPBIN_URL = 'http://0.0.0.0:%s/' % (PORT) +# HTTPBIN_URL = 'http://127.0.0.1:8000/' def httpbin(*suffix): @@ -29,15 +30,9 @@ def httpbin(*suffix): return HTTPBIN_URL + '/'.join(suffix) -def httpsbin(*suffix): - """Returns url for HTTPSBIN resource.""" - - return HTTPSBIN_URL + '/'.join(suffix) - - -SERVICES = (httpbin, httpsbin) - +SERVICES = (httpbin, ) +_httpbin = False class RequestsTestSuite(unittest.TestCase): """Requests test cases.""" @@ -46,33 +41,52 @@ class RequestsTestSuite(unittest.TestCase): _multiprocess_can_split_ = True def setUp(self): - pass + + global _httpbin + + if not _httpbin: + + c = envoy.connect('gunicorn httpbin:app --bind=0.0.0.0:%s' % (PORT)) + + self.httpbin = c + _httpbin = True + time.sleep(1) + def tearDown(self): """Teardown.""" + # self.httpbin.kill() pass + def test_entry_points(self): + + requests.session + requests.session().get + requests.session().head + requests.get + requests.head + requests.put + requests.patch + requests.post + + + def test_invalid_url(self): self.assertRaises(ValueError, requests.get, 'hiwpefhipowhefopw') - def test_HTTP_200_OK_GET(self): - r = requests.get(httpbin('/')) + r = requests.get(httpbin('/get')) self.assertEqual(r.status_code, 200) def test_HTTP_302_ALLOW_REDIRECT_GET(self): - r = requests.get(httpbin('redirect', '1')) - self.assertEqual(r.status_code, 200) + r = requests.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) - self.assertEqual(r.status_code, 302) - - def test_HTTPS_200_OK_GET(self): - r = requests.get(httpsbin('/')) - self.assertEqual(r.status_code, 200) + r = requests.get(httpbin('redirect', '1'), allow_redirects=False) + self.assertEqual(r.status_code, 302) def test_HTTP_200_OK_GET_WITH_PARAMS(self): @@ -112,12 +126,7 @@ class RequestsTestSuite(unittest.TestCase): def test_HTTP_200_OK_HEAD(self): - r = requests.head(httpbin('/')) - self.assertEqual(r.status_code, 200) - - - def test_HTTPS_200_OK_HEAD(self): - r = requests.head(httpsbin('/')) + r = requests.head(httpbin('/get')) self.assertEqual(r.status_code, 200) @@ -126,22 +135,12 @@ class RequestsTestSuite(unittest.TestCase): self.assertEqual(r.status_code, 200) - def test_HTTPS_200_OK_PUT(self): - r = requests.put(httpsbin('put')) - self.assertEqual(r.status_code, 200) - - def test_HTTP_200_OK_PATCH(self): r = requests.patch(httpbin('patch')) self.assertEqual(r.status_code, 200) - def test_HTTPS_200_OK_PATCH(self): - r = requests.patch(httpsbin('patch')) - self.assertEqual(r.status_code, 200) - - - def test_AUTH_HTTP_200_OK_GET(self): + def test_BASICAUTH_HTTP_200_OK_GET(self): for service in SERVICES: @@ -149,14 +148,35 @@ class RequestsTestSuite(unittest.TestCase): url = service('basic-auth', 'user', 'pass') r = requests.get(url, auth=auth) - # print r.__dict__ self.assertEqual(r.status_code, 200) - r = requests.get(url) + self.assertEqual(r.status_code, 401) + + + s = requests.session(auth=auth) + r = s.get(url) self.assertEqual(r.status_code, 200) + def test_DIGESTAUTH_HTTP_200_OK_GET(self): + + for service in SERVICES: + + auth = ('digest', 'user', 'pass') + url = service('digest-auth', 'auth', 'user', 'pass') + + r = requests.get(url, auth=auth) + self.assertEqual(r.status_code, 200) + + r = requests.get(url) + self.assertEqual(r.status_code, 401) + + + s = requests.session(auth=auth) + r = s.get(url) + self.assertEqual(r.status_code, 200) + def test_POSTBIN_GET_POST_FILES(self): for service in SERVICES: @@ -206,7 +226,7 @@ class RequestsTestSuite(unittest.TestCase): r = requests.get(service('status', '500')) self.assertEqual(bool(r), False) - r = requests.get(service('/')) + r = requests.get(service('/get')) self.assertEqual(bool(r), True) @@ -215,38 +235,20 @@ class RequestsTestSuite(unittest.TestCase): for service in SERVICES: r = requests.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')) - self.assertRaises(requests.HTTPError, r.raise_for_status) + self.assertRaises(HTTPError, r.raise_for_status) r = requests.get(httpbin('status', '200')) self.assertFalse(r.error) r.raise_for_status() - def test_cookie_jar(self): - - jar = cookielib.CookieJar() - self.assertFalse(jar) - - url = httpbin('cookies', 'set', 'requests_cookie', 'awesome') - r = requests.get(url, cookies=jar) - self.assertTrue(jar) - - cookie_found = False - for cookie in jar: - if cookie.name == 'requests_cookie': - self.assertEquals(cookie.value, 'awesome') - cookie_found = True - self.assertTrue(cookie_found) - - r = requests.get(httpbin('cookies'), cookies=jar) - self.assertTrue('awesome' in r.content) - - def test_decompress_gzip(self): r = requests.get(httpbin('gzip')) @@ -257,7 +259,7 @@ class RequestsTestSuite(unittest.TestCase): for service in SERVICES: - url = service('/') + url = service('/get') requests.get(url, params={'foo': u'føø'}) requests.get(url, params={u'føø': u'føø'}) @@ -275,19 +277,6 @@ class RequestsTestSuite(unittest.TestCase): self.assertEquals(r.status_code, 401) - def test_settings(self): - - def test(): - r = requests.get(httpbin('')) - r.raise_for_status() - - with requests.settings(timeout=0.0000001): - self.assertRaises(requests.Timeout, test) - - with requests.settings(timeout=100): - requests.get(httpbin('')) - - def test_urlencoded_post_data(self): for service in SERVICES: @@ -317,7 +306,7 @@ class RequestsTestSuite(unittest.TestCase): rbody = json.loads(r.content) # Body wasn't valid url encoded data, so the server returns None as # "form" and the raw body as "data". - self.assertEquals(rbody.get('form'), None) + self.assertEquals(rbody.get('form'), {}) self.assertEquals(rbody.get('data'), 'fooaowpeuf') @@ -336,21 +325,6 @@ class RequestsTestSuite(unittest.TestCase): self.assertEquals(rbody.get('data'), '') - def test_nonurlencoded_post_querystring(self): - - for service in SERVICES: - - r = requests.post(service('post'), params='fooaowpeuf') - - self.assertEquals(r.status_code, 200) - self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post?fooaowpeuf')) - - rbody = json.loads(r.content) - self.assertEquals(rbody.get('form'), {}) # No form supplied - self.assertEquals(rbody.get('data'), '') - - def test_urlencoded_post_query_and_data(self): for service in SERVICES: @@ -369,26 +343,24 @@ class RequestsTestSuite(unittest.TestCase): self.assertEquals(rbody.get('data'), '') - def test_nonurlencoded_post_query_and_data(self): + def test_nonurlencoded_postdata(self): for service in SERVICES: - r = requests.post(service('post'), - params='fooaowpeuf', data="foobar") + r = requests.post(service('post'), data="foobar") self.assertEquals(r.status_code, 200) self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post?fooaowpeuf')) rbody = json.loads(r.content) - self.assertEquals(rbody.get('form'), None) + self.assertEquals(rbody.get('form'), {}) self.assertEquals(rbody.get('data'), 'foobar') - def test_idna(self): - r = requests.get(u'http://➡.ws/httpbin') - assert 'httpbin' in r.url + # def test_idna(self): + # r = requests.get(u'http://➡.ws/httpbin') + # assert 'httpbin' in r.url def test_urlencoded_get_query_multivalued_param(self): @@ -469,15 +441,8 @@ class RequestsTestSuite(unittest.TestCase): def test_session_HTTP_200_OK_GET(self): - s = Session() - r = s.get(httpbin('/')) - self.assertEqual(r.status_code, 200) - - - def test_session_HTTPS_200_OK_GET(self): - - s = Session() - r = s.get(httpsbin('/')) + s = requests.session() + r = s.get(httpbin('/get')) self.assertEqual(r.status_code, 200) @@ -485,17 +450,109 @@ class RequestsTestSuite(unittest.TestCase): heads = {'User-agent': 'Mozilla/5.0'} - s = Session() + s = requests.session() s.headers = heads + # Make 2 requests from Session object, should send header both times r1 = s.get(httpbin('user-agent')) - assert heads['User-agent'] in r1.content - r2 = s.get(httpbin('user-agent')) + r2 = s.get(httpbin('user-agent')) assert heads['User-agent'] in r2.content + + new_heads = {'User-agent': 'blah'} + r3 = s.get(httpbin('user-agent'), headers=new_heads) + assert new_heads['User-agent'] in r3.content + self.assertEqual(r2.status_code, 200) + def test_session_persistent_cookies(self): + + s = requests.session() + + # Internally dispatched cookies are sent. + _c = {'kenneth': 'reitz', 'bessie': 'monke'} + r = s.get(httpbin('cookies'), cookies=_c) + r = s.get(httpbin('cookies')) + + # Those cookies persist transparently. + c = json.loads(r.content).get('cookies') + assert c == _c + + # Double check. + r = s.get(httpbin('cookies'), cookies={}) + c = json.loads(r.content).get('cookies') + assert c == _c + + # Remove a cookie by setting it's value to None. + r = s.get(httpbin('cookies'), cookies={'bessie': None}) + c = json.loads(r.content).get('cookies') + del _c['bessie'] + assert c == _c + + # Test session-level cookies. + s = requests.session(cookies=_c) + r = s.get(httpbin('cookies')) + c = json.loads(r.content).get('cookies') + assert c == _c + + # Have the server set a cookie. + r = s.get(httpbin('cookies', 'set', 'k', 'v'), allow_redirects=True) + c = json.loads(r.content).get('cookies') + + assert 'k' in c + + # And server-set cookie persistience. + r = s.get(httpbin('cookies')) + c = json.loads(r.content).get('cookies') + + assert 'k' in c + + + + def test_session_persistent_params(self): + + params = {'a': 'a_test'} + + s = requests.session() + s.params = params + + # Make 2 requests from Session object, should send header both times + r1 = s.get(httpbin('get')) + assert params['a'] in r1.content + + + params2 = {'b': 'b_test'} + + r2 = s.get(httpbin('get'), params=params2) + assert params['a'] in r2.content + assert params2['b'] in r2.content + + + params3 = {'b': 'b_test', 'a': None, 'c': 'c_test'} + + r3 = s.get(httpbin('get'), params=params3) + + assert not params['a'] in r3.content + assert params3['b'] in r3.content + assert params3['c'] in r3.content + + def test_invalid_content(self): + # WARNING: if you're using a terrible DNS provider (comcast), + # this will fail. + try: + hah = 'http://somedomainthatclearlydoesntexistg.com' + r = requests.get(hah, allow_redirects=False) + except requests.ConnectionError: + pass # \o/ + else: + assert False + + + config = {'safe_mode': True} + r = requests.get(hah, allow_redirects=False, config=config) + assert r.content == None + if __name__ == '__main__': unittest.main() -- cgit v1.2.3