From: Kenneth Reitz Date: Mon, 10 Oct 2011 05:27:05 +0000 (-0400) Subject: lost in merge X-Git-Tag: v0.8.0~94^2~18 X-Git-Url: http://review.tizen.org/git/?a=commitdiff_plain;h=e9f94e698d3cb1e84d383dbc5a7393e55140b542;p=services%2Fpython-requests.git lost in merge --- diff --git a/tests/api_tests.py b/tests/api_tests.py new file mode 100755 index 0000000..ae63ab8 --- /dev/null +++ b/tests/api_tests.py @@ -0,0 +1,209 @@ +#!/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() diff --git a/tests/integration_tests.py b/tests/integration_tests.py new file mode 100755 index 0000000..4cfb627 --- /dev/null +++ b/tests/integration_tests.py @@ -0,0 +1,528 @@ +#!/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.herokuapp.com/' + +# 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.""" + + # It goes to eleven. + _multiprocess_can_split_ = True + + def setUp(self): + pass + + def tearDown(self): + 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()