From 0b475dd65487f2c14efa6f917657a0f1d19bae02 Mon Sep 17 00:00:00 2001 From: Kenneth Reitz Date: Sat, 1 Oct 2011 20:51:41 -0400 Subject: [PATCH] Test improvements #180 --- requests/packages/urllib3/connectionpool.py | 1 + requests/packages/urllib3/poolmanager.py | 4 +- tests/integration/test_requests.py | 530 -------------------- tests/unit/test_requests_api.py | 209 -------- 4 files changed, 4 insertions(+), 740 deletions(-) delete mode 100755 tests/integration/test_requests.py delete mode 100755 tests/unit/test_requests_api.py diff --git a/requests/packages/urllib3/connectionpool.py b/requests/packages/urllib3/connectionpool.py index 36ecfb5..8f74b1f 100644 --- a/requests/packages/urllib3/connectionpool.py +++ b/requests/packages/urllib3/connectionpool.py @@ -38,6 +38,7 @@ log = logging.getLogger(__name__) _Default = object() + ## Connection objects (extension of httplib) class VerifiedHTTPSConnection(HTTPSConnection): diff --git a/requests/packages/urllib3/poolmanager.py b/requests/packages/urllib3/poolmanager.py index 8d915ad..acf1c5a 100644 --- a/requests/packages/urllib3/poolmanager.py +++ b/requests/packages/urllib3/poolmanager.py @@ -71,7 +71,9 @@ class PoolManager(object): port = port or port_by_scheme.get(scheme, 80) - return self.connection_from_host(host, port=port, scheme=scheme) + r = self.connection_from_host(host, port=port, scheme=scheme) + print r.__dict__ + return r def urlopen(self, method, url, **kw): "Same as HTTP(S)ConnectionPool.urlopen, ``url`` must be absolute." diff --git a/tests/integration/test_requests.py b/tests/integration/test_requests.py deleted file mode 100755 index 406d23c..0000000 --- a/tests/integration/test_requests.py +++ /dev/null @@ -1,530 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -from __future__ import with_statement - -import unittest -import cookielib - -try: - import omnijson as json -except ImportError: - import json - -import requests - -from requests.sessions import Session -from requests.utils import curl_from_request - - -HTTPBIN_URL = 'http://httpbin.org/' -HTTPSBIN_URL = 'https://httpbin.ep.io/' - -# HTTPBIN_URL = 'http://staging.httpbin.org/' -# HTTPSBIN_URL = 'https://httpbin-staging.ep.io/' - - -def httpbin(*suffix): - """Returns url for HTTPBIN resource.""" - - return HTTPBIN_URL + '/'.join(suffix) - - -def httpsbin(*suffix): - """Returns url for HTTPSBIN resource.""" - - return HTTPSBIN_URL + '/'.join(suffix) - - -SERVICES = (httpbin, httpsbin) - - - -class RequestsTestSuite(unittest.TestCase): - """Requests test cases.""" - - - def setUp(self): - pass - - - def tearDown(self): - """Teardown.""" - pass - - - def test_invalid_url(self): - self.assertRaises(ValueError, requests.get, 'hiwpefhipowhefopw') - - - def test_HTTP_200_OK_GET(self): - r = requests.get(httpbin('/')) - 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) - - 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) - - - def test_HTTP_200_OK_GET_WITH_PARAMS(self): - heads = {'User-agent': 'Mozilla/5.0'} - - r = requests.get(httpbin('user-agent'), headers=heads) - - assert heads['User-agent'] in r.content - self.assertEqual(r.status_code, 200) - - - 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) - self.assertEqual(r.status_code, 200) - - - def test_user_agent_transfers(self): - """Issue XX""" - - heads = { - 'User-agent': - 'Mozilla/5.0 (github.com/kennethreitz/requests)' - } - - r = requests.get(httpbin('user-agent'), headers=heads); - self.assertTrue(heads['User-agent'] in r.content) - - heads = { - 'user-agent': - 'Mozilla/5.0 (github.com/kennethreitz/requests)' - } - - r = requests.get(httpbin('user-agent'), headers=heads); - self.assertTrue(heads['user-agent'] in r.content) - - - 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('/')) - self.assertEqual(r.status_code, 200) - - - def test_HTTP_200_OK_PUT(self): - r = requests.put(httpbin('put')) - 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): - - for service in SERVICES: - - auth = ('user', 'pass') - 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, 200) - - - def test_POSTBIN_GET_POST_FILES(self): - - for service in SERVICES: - - url = service('post') - post = requests.post(url).raise_for_status() - - post = requests.post(url, data={'some': 'data'}) - self.assertEqual(post.status_code, 200) - - post2 = requests.post(url, files={'some': open('test_requests.py')}) - self.assertEqual(post2.status_code, 200) - - post3 = requests.post(url, data='[{"some": "json"}]') - self.assertEqual(post3.status_code, 200) - - - def test_POSTBIN_GET_POST_FILES_WITH_PARAMS(self): - - for service in SERVICES: - - url = service('post') - post = requests.post(url, - files={'some': open('test_requests.py')}, - data={'some': 'data'}) - - self.assertEqual(post.status_code, 200) - - - def test_POSTBIN_GET_POST_FILES_WITH_HEADERS(self): - - for service in SERVICES: - - url = service('post') - - post2 = requests.post(url, - files={'some': open('test_requests.py')}, - headers = {'User-Agent': 'requests-tests'}) - - self.assertEqual(post2.status_code, 200) - - - def test_nonzero_evaluation(self): - - for service in SERVICES: - - r = requests.get(service('status', '500')) - self.assertEqual(bool(r), False) - - r = requests.get(service('/')) - self.assertEqual(bool(r), True) - - - def test_request_ok_set(self): - - for service in SERVICES: - - r = requests.get(service('status', '404')) - self.assertEqual(r.ok, False) - - - def test_status_raising(self): - r = requests.get(httpbin('status', '404')) - self.assertRaises(requests.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')) - r.content.decode('ascii') - - - def test_unicode_get(self): - - for service in SERVICES: - - url = service('/') - - requests.get(url, params={'foo': u'føø'}) - requests.get(url, params={u'føø': u'føø'}) - requests.get(url, params={'føø': 'føø'}) - requests.get(url, params={'foo': u'foo'}) - requests.get(service('ø'), params={'foo': u'foo'}) - - - def test_httpauth_recursion(self): - - http_auth = ('user', 'BADpass') - - for service in SERVICES: - r = requests.get(service('basic-auth', 'user', 'pass'), auth=http_auth) - 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: - - r = requests.post(service('post'), data=dict(test='fooaowpeuf')) - - self.assertEquals(r.status_code, 200) - self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post')) - - rbody = json.loads(r.content) - - self.assertEquals(rbody.get('form'), dict(test='fooaowpeuf')) - self.assertEquals(rbody.get('data'), '') - - - def test_nonurlencoded_post_data(self): - - for service in SERVICES: - - r = requests.post(service('post'), data='fooaowpeuf') - - self.assertEquals(r.status_code, 200) - self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post')) - - 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('data'), 'fooaowpeuf') - - - def test_urlencoded_post_querystring(self): - - for service in SERVICES: - - r = requests.post(service('post'), params=dict(test='fooaowpeuf')) - - self.assertEquals(r.status_code, 200) - self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post?test=fooaowpeuf')) - - rbody = json.loads(r.content) - self.assertEquals(rbody.get('form'), {}) # No form supplied - 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: - - r = requests.post( - service('post'), - params=dict(test='fooaowpeuf'), - data=dict(test2="foobar")) - - self.assertEquals(r.status_code, 200) - self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post?test=fooaowpeuf')) - - rbody = json.loads(r.content) - self.assertEquals(rbody.get('form'), dict(test2='foobar')) - self.assertEquals(rbody.get('data'), '') - - - def test_nonurlencoded_post_query_and_data(self): - - for service in SERVICES: - - r = requests.post(service('post'), - params='fooaowpeuf', 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('data'), 'foobar') - - - def test_idna(self): - r = requests.get(u'http://➡.ws/httpbin') - assert 'httpbin' in r.url - - - def test_urlencoded_get_query_multivalued_param(self): - - for service in SERVICES: - - r = requests.get(service('get'), params=dict(test=['foo','baz'])) - self.assertEquals(r.status_code, 200) - self.assertEquals(r.url, service('get?test=foo&test=baz')) - - - def test_urlencoded_post_querystring_multivalued(self): - - for service in SERVICES: - - r = requests.post(service('post'), params=dict(test=['foo','baz'])) - self.assertEquals(r.status_code, 200) - self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post?test=foo&test=baz')) - - rbody = json.loads(r.content) - self.assertEquals(rbody.get('form'), {}) # No form supplied - self.assertEquals(rbody.get('data'), '') - - - def test_urlencoded_post_query_multivalued_and_data(self): - - for service in SERVICES: - - r = requests.post( - service('post'), - params=dict(test=['foo','baz']), - data=dict(test2="foobar",test3=['foo','baz'])) - - self.assertEquals(r.status_code, 200) - self.assertEquals(r.headers['content-type'], 'application/json') - self.assertEquals(r.url, service('post?test=foo&test=baz')) - rbody = json.loads(r.content) - self.assertEquals(rbody.get('form'), dict(test2='foobar',test3='foo')) - self.assertEquals(rbody.get('data'), '') - - - def test_redirect_history(self): - - for service in SERVICES: - - r = requests.get(service('redirect', '3')) - self.assertEquals(r.status_code, 200) - self.assertEquals(len(r.history), 3) - - - def test_relative_redirect_history(self): - - for service in SERVICES: - - r = requests.get(service('relative-redirect', '3')) - self.assertEquals(r.status_code, 200) - self.assertEquals(len(r.history), 3) - - - 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('/')) - self.assertEqual(r.status_code, 200) - - - def test_session_persistent_headers(self): - - heads = {'User-agent': 'Mozilla/5.0'} - - s = 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')) - - assert heads['User-agent'] in r2.content - self.assertEqual(r2.status_code, 200) - - - - def test_curl_HTTP_OK_GET(self): - curl_str = 'curl -L -X GET -H "Accept-Encoding:gzip" -H "User-Agent:python-requests.org" "http://httpbin.org//"' - r = requests.get(httpbin('/')) - self.assertEqual(curl_from_request(r.request), curl_str) - - - def test_curl_HTTP_OK_GET_WITH_PARAMS(self): - curl_str = 'curl -L -X GET -H "Accept-Encoding:gzip" -H "User-agent:Mozilla/5.0" "http://httpbin.org/user-agent"' - - heads = {'User-agent': 'Mozilla/5.0'} - r = requests.get(httpbin('user-agent'), headers=heads) - self.assertEqual(curl_from_request(r.request), curl_str) - - - def test_curl_HTTP_OK_HEAD(self): - curl_str ='curl -L -I -H "Accept-Encoding:gzip" -H "User-Agent:python-requests.org" "http://httpbin.org//"' - r = requests.head(httpbin('/')) - self.assertEqual(curl_from_request(r.request), curl_str) - - - def test_curl_HTTP_OK_PATCH(self): - curl_str = 'curl -L -X PATCH -H "Accept-Encoding:gzip" -H "User-Agent:python-requests.org" "http://httpbin.org/patch"' - r = requests.patch(httpbin('patch')) - self.assertEqual(curl_from_request(r.request), curl_str) - - - def test_curl_AUTH_HTTPS_OK_GET(self): - curl_str = 'curl -L -u "user:pass" -X GET -H "Accept-Encoding:gzip" -H "User-Agent:python-requests.org" "https://httpbin.ep.io/basic-auth/user/pass"' - auth = ('user', 'pass') - r = requests.get(httpsbin('basic-auth', 'user', 'pass'), auth=auth) - self.assertEqual(curl_from_request(r.request), curl_str) - - - def test_curl_POSTBIN_GET_POST_FILES(self): - curl_str = 'curl -L -X POST -H "Accept-Encoding:gzip" -H "User-Agent:python-requests.org" -d "some=data" "http://httpbin.org/post"' - post = requests.post(httpbin('post'), data={'some': 'data'}) - self.assertEqual(curl_from_request(post.request), curl_str) - - curl_str = 'curl -L -X POST -H "Accept-Encoding:gzip" -H "User-Agent:python-requests.org" -F "some=@test_requests.py" "https://httpbin.ep.io/post"' - post2 = requests.post(httpsbin('post'), files={'some': open('test_requests.py')}) - self.assertEqual(curl_from_request(post2.request), curl_str) - - curl_str = 'curl -L -X POST -H "Accept-Encoding:gzip" -H "User-Agent:python-requests.org" -d \'[{"some": "json"}]\' "http://httpbin.org/post"' - post3 = requests.post(httpbin('post'), data='[{"some": "json"}]') - self.assertEqual(curl_from_request(post3.request), curl_str) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/unit/test_requests_api.py b/tests/unit/test_requests_api.py deleted file mode 100755 index ae63ab8..0000000 --- a/tests/unit/test_requests_api.py +++ /dev/null @@ -1,209 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -import unittest -import mock -import sys -import os -sys.path.append(os.getcwd()) - -try: - import omnijson as json -except ImportError: - import json - -import requests -from requests.models import Response - -class RequestsAPIUnitTests(unittest.TestCase): - """Requests API unit test cases.""" - - def setUp(self): - pass - - - def tearDown(self): - """Teardown.""" - pass - - - @mock.patch('requests.api.dispatch_hook') - @mock.patch('requests.api.Request') - @mock.patch('requests.api.cookiejar_from_dict') - def test_request(self, mock_cjar, mock_request, mock_hook): - args = dict( - method = None, - url = None, - data = None, - params = None, - headers = None, - cookiejar = None, - files = None, - auth = None, - timeout = 1, - allow_redirects = None, - proxies = None, - ) - hooks = {'args': args, 'pre_request': mock_request, - 'post_request': mock_request, 'response': 'response'} - sideeffect = lambda x,y,z: hooks[x] - mock_cjar.return_value = None - mock_request.send = mock.Mock(return_value={}) - mock_request.response = "response" - mock_hook.side_effect = sideeffect - - r = requests.request('get','http://google.com') - - - mock_cjar.assert_called_once_with({}) - mock_hook.assert_called__with('args', None, args) - mock_request.assert_called_once_with(**args) - mock_hook.assert_called__with('pre_request', None, mock_request) - mock_request.send.assert_called_once_with() - mock_hook.assert_called__with('post_request', None, mock_request) - mock_hook.assert_called__with('response', None, mock_request) - self.assertEqual(r, "response") - - - - @mock.patch('requests.api.request') - def test_http_get(self, mock_request): - mock_request.return_value = Response() - requests.get('http://google.com') - mock_request.assert_called_once_with('get', 'http://google.com', - allow_redirects= True) - - @mock.patch('requests.api.request') - def test_http_get_with_kwargs(self, mock_request): - mock_request.return_value = Response() - requests.get('http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - mock_request.assert_called_once_with('get', 'http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - - @mock.patch('requests.api.request') - def test_http_head(self, mock_request): - mock_request.return_value = Response() - requests.head('http://google.com') - mock_request.assert_called_once_with('head', 'http://google.com', - allow_redirects= True) - - @mock.patch('requests.api.request') - def test_http_head_with_kwargs(self, mock_request): - mock_request.return_value = Response() - requests.head('http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - mock_request.assert_called_once_with('head', 'http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - - @mock.patch('requests.api.request') - def test_http_post(self, mock_request): - mock_request.return_value = Response() - requests.post('http://google.com', {}) - mock_request.assert_called_once_with('post', 'http://google.com', - data= {}) - - @mock.patch('requests.api.request') - def test_http_post_with_kwargs(self, mock_request): - mock_request.return_value = Response() - requests.post('http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - mock_request.assert_called_once_with('post', 'http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - - - @mock.patch('requests.api.request') - def test_http_put(self, mock_request): - mock_request.return_value = Response() - requests.put('http://google.com', {}) - mock_request.assert_called_once_with('put', 'http://google.com', - data= {}) - - @mock.patch('requests.api.request') - def test_http_put_with_kwargs(self, mock_request): - mock_request.return_value = Response() - requests.put('http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - mock_request.assert_called_once_with('put', 'http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - - - @mock.patch('requests.api.request') - def test_http_patch(self, mock_request): - mock_request.return_value = Response() - requests.patch('http://google.com', {}) - mock_request.assert_called_once_with('patch', 'http://google.com', - data= {}) - - @mock.patch('requests.api.request') - def test_http_patch_with_kwargs(self, mock_request): - mock_request.return_value = Response() - requests.patch('http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - mock_request.assert_called_once_with('patch', 'http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - - @mock.patch('requests.api.request') - def test_http_delete(self, mock_request): - mock_request.return_value = Response() - requests.delete('http://google.com') - mock_request.assert_called_once_with('delete', 'http://google.com') - - @mock.patch('requests.api.request') - def test_http_delete_with_kwargs(self, mock_request): - mock_request.return_value = Response() - requests.delete('http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - mock_request.assert_called_once_with('delete', 'http://google.com', - params="params", data="data", headers="headers", - cookies="cookies", - files="files", auth="auth", timeout="timeout", - allow_redirects=False, - proxies="proxies", hooks="hooks") - - -if __name__ == '__main__': - unittest.main() -- 2.34.1