| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871 |
- #!/usr/bin/env python
- # -*- coding: utf-8 -*-
- """Tests for Requests."""
- from __future__ import division
- import json
- import os
- import unittest
- import pickle
- import requests
- import pytest
- from requests.auth import HTTPDigestAuth
- from requests.adapters import HTTPAdapter
- from requests.compat import str, cookielib, getproxies, urljoin, urlparse
- from requests.cookies import cookiejar_from_dict
- from requests.exceptions import InvalidURL, MissingSchema
- from requests.structures import CaseInsensitiveDict
- try:
- import StringIO
- except ImportError:
- import io as StringIO
- HTTPBIN = os.environ.get('HTTPBIN_URL', 'http://httpbin.org/')
- # Issue #1483: Make sure the URL always has a trailing slash
- HTTPBIN = HTTPBIN.rstrip('/') + '/'
- def httpbin(*suffix):
- """Returns url for HTTPBIN resource."""
- return urljoin(HTTPBIN, '/'.join(suffix))
- class RequestsTestCase(unittest.TestCase):
- _multiprocess_can_split_ = True
- def setUp(self):
- """Create simple data set with headers."""
- pass
- def tearDown(self):
- """Teardown."""
- 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(MissingSchema, requests.get, 'hiwpefhipowhefopw')
- self.assertRaises(InvalidURL, requests.get, 'http://')
- def test_basic_building(self):
- req = requests.Request()
- req.url = 'http://kennethreitz.org/'
- req.data = {'life': '42'}
- pr = req.prepare()
- assert pr.url == req.url
- assert pr.body == 'life=42'
- def test_no_content_length(self):
- get_req = requests.Request('GET', httpbin('get')).prepare()
- self.assertTrue('Content-Length' not in get_req.headers)
- head_req = requests.Request('HEAD', httpbin('head')).prepare()
- self.assertTrue('Content-Length' not in head_req.headers)
- def test_path_is_not_double_encoded(self):
- request = requests.Request('GET', "http://0.0.0.0/get/test case").prepare()
- self.assertEqual(request.path_url, "/get/test%20case")
- def test_params_are_added_before_fragment(self):
- request = requests.Request('GET',
- "http://example.com/path#fragment", params={"a": "b"}).prepare()
- self.assertEqual(request.url,
- "http://example.com/path?a=b#fragment")
- request = requests.Request('GET',
- "http://example.com/path?key=value#fragment", params={"a": "b"}).prepare()
- self.assertEqual(request.url,
- "http://example.com/path?key=value&a=b#fragment")
- def test_mixed_case_scheme_acceptable(self):
- s = requests.Session()
- s.proxies = getproxies()
- parts = urlparse(httpbin('get'))
- schemes = ['http://', 'HTTP://', 'hTTp://', 'HttP://',
- 'https://', 'HTTPS://', 'hTTps://', 'HttPs://']
- for scheme in schemes:
- url = scheme + parts.netloc + parts.path
- r = requests.Request('GET', url)
- r = s.send(r.prepare())
- self.assertEqual(r.status_code, 200,
- "failed for scheme %s" % scheme)
- def test_HTTP_200_OK_GET_ALTERNATIVE(self):
- r = requests.Request('GET', httpbin('get'))
- s = requests.Session()
- s.proxies = getproxies()
- r = s.send(r.prepare())
- 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_ALLOW_REDIRECT_POST(self):
- # r = requests.post(httpbin('status', '302'), data={'some': 'data'})
- # 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)
- self.assertTrue(heads['User-agent'] in r.text)
- 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_set_cookie_on_301(self):
- s = requests.session()
- url = httpbin('cookies/set?foo=bar')
- r = s.get(url)
- self.assertTrue(s.cookies['foo'] == 'bar')
- def test_cookie_sent_on_redirect(self):
- s = requests.session()
- s.get(httpbin('cookies/set?foo=bar'))
- r = s.get(httpbin('redirect/1')) # redirects to httpbin('get')
- self.assertTrue("Cookie" in r.json()["headers"])
- def test_cookie_removed_on_expire(self):
- s = requests.session()
- s.get(httpbin('cookies/set?foo=bar'))
- self.assertTrue(s.cookies['foo'] == 'bar')
- s.get(
- httpbin('response-headers'),
- params={
- 'Set-Cookie':
- 'foo=deleted; expires=Thu, 01-Jan-1970 00:00:01 GMT'
- }
- )
- assert 'foo' not in s.cookies
- def test_cookie_quote_wrapped(self):
- s = requests.session()
- s.get(httpbin('cookies/set?foo="bar:baz"'))
- self.assertTrue(s.cookies['foo'] == '"bar:baz"')
- def test_request_cookie_overrides_session_cookie(self):
- s = requests.session()
- s.cookies['foo'] = 'bar'
- r = s.get(httpbin('cookies'), cookies={'foo': 'baz'})
- assert r.json()['cookies']['foo'] == 'baz'
- # Session cookie should not be modified
- assert s.cookies['foo'] == 'bar'
- def test_generic_cookiejar_works(self):
- cj = cookielib.CookieJar()
- cookiejar_from_dict({'foo': 'bar'}, cj)
- s = requests.session()
- s.cookies = cj
- r = s.get(httpbin('cookies'))
- # Make sure the cookie was sent
- assert r.json()['cookies']['foo'] == 'bar'
- # Make sure the session cj is still the custom one
- assert s.cookies is cj
- def test_requests_in_history_are_not_overridden(self):
- resp = requests.get(httpbin('redirect/3'))
- urls = [r.url for r in resp.history]
- req_urls = [r.request.url for r in resp.history]
- self.assertEquals(urls, req_urls)
- def test_user_agent_transfers(self):
- 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.text)
- 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.text)
- def test_HTTP_200_OK_HEAD(self):
- r = requests.head(httpbin('get'))
- 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_BASICAUTH_TUPLE_HTTP_200_OK_GET(self):
- auth = ('user', 'pass')
- url = httpbin('basic-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()
- s.auth = auth
- r = s.get(url)
- self.assertEqual(r.status_code, 200)
- def test_basicauth_with_netrc(self):
- auth = ('user', 'pass')
- wrong_auth = ('wronguser', 'wrongpass')
- url = httpbin('basic-auth', 'user', 'pass')
- def get_netrc_auth_mock(url):
- return auth
- requests.sessions.get_netrc_auth = get_netrc_auth_mock
- # Should use netrc and work.
- r = requests.get(url)
- self.assertEqual(r.status_code, 200)
- # Given auth should override and fail.
- r = requests.get(url, auth=wrong_auth)
- self.assertEqual(r.status_code, 401)
- s = requests.session()
- # Should use netrc and work.
- r = s.get(url)
- self.assertEqual(r.status_code, 200)
- # Given auth should override and fail.
- s.auth = wrong_auth
- r = s.get(url)
- self.assertEqual(r.status_code, 401)
- def test_DIGEST_HTTP_200_OK_GET(self):
- auth = HTTPDigestAuth('user', 'pass')
- url = httpbin('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()
- s.auth = HTTPDigestAuth('user', 'pass')
- r = s.get(url)
- self.assertEqual(r.status_code, 200)
- def test_DIGEST_AUTH_RETURNS_COOKIE(self):
- url = httpbin('digest-auth', 'auth', 'user', 'pass')
- auth = HTTPDigestAuth('user', 'pass')
- r = requests.get(url)
- assert r.cookies['fake'] == 'fake_value'
- r = requests.get(url, auth=auth)
- assert r.status_code == 200
- def test_DIGEST_AUTH_SETS_SESSION_COOKIES(self):
- url = httpbin('digest-auth', 'auth', 'user', 'pass')
- auth = HTTPDigestAuth('user', 'pass')
- s = requests.Session()
- s.get(url, auth=auth)
- assert s.cookies['fake'] == 'fake_value'
- def test_DIGEST_STREAM(self):
- auth = HTTPDigestAuth('user', 'pass')
- url = httpbin('digest-auth', 'auth', 'user', 'pass')
- r = requests.get(url, auth=auth, stream=True)
- self.assertNotEqual(r.raw.read(), b'')
- r = requests.get(url, auth=auth, stream=False)
- self.assertEqual(r.raw.read(), b'')
- def test_DIGESTAUTH_WRONG_HTTP_401_GET(self):
- auth = HTTPDigestAuth('user', 'wrongpass')
- url = httpbin('digest-auth', 'auth', 'user', 'pass')
- r = requests.get(url, auth=auth)
- self.assertEqual(r.status_code, 401)
- r = requests.get(url)
- self.assertEqual(r.status_code, 401)
- s = requests.session()
- s.auth = auth
- r = s.get(url)
- self.assertEqual(r.status_code, 401)
- def test_POSTBIN_GET_POST_FILES(self):
- url = httpbin('post')
- post1 = requests.post(url).raise_for_status()
- post1 = requests.post(url, data={'some': 'data'})
- self.assertEqual(post1.status_code, 200)
- with open('requirements.txt') as f:
- post2 = requests.post(url, files={'some': f})
- self.assertEqual(post2.status_code, 200)
- post4 = requests.post(url, data='[{"some": "json"}]')
- self.assertEqual(post4.status_code, 200)
- try:
- requests.post(url, files=['bad file data'])
- except ValueError:
- pass
- def test_POSTBIN_GET_POST_FILES_WITH_DATA(self):
- url = httpbin('post')
- post1 = requests.post(url).raise_for_status()
- post1 = requests.post(url, data={'some': 'data'})
- self.assertEqual(post1.status_code, 200)
- with open('requirements.txt') as f:
- post2 = requests.post(url, data={'some': 'data'}, files={'some': f})
- self.assertEqual(post2.status_code, 200)
- post4 = requests.post(url, data='[{"some": "json"}]')
- self.assertEqual(post4.status_code, 200)
- try:
- requests.post(url, files=['bad file data'])
- except ValueError:
- pass
- def test_conflicting_post_params(self):
- url = httpbin('post')
- with open('requirements.txt') as f:
- pytest.raises(ValueError, "requests.post(url, data='[{\"some\": \"data\"}]', files={'some': f})")
- pytest.raises(ValueError, "requests.post(url, data=u'[{\"some\": \"data\"}]', files={'some': f})")
- def test_request_ok_set(self):
- r = requests.get(httpbin('status', '404'))
- self.assertEqual(r.ok, False)
- def test_status_raising(self):
- r = requests.get(httpbin('status', '404'))
- self.assertRaises(requests.exceptions.HTTPError, r.raise_for_status)
- r = requests.get(httpbin('status', '500'))
- self.assertFalse(r.ok)
- def test_decompress_gzip(self):
- r = requests.get(httpbin('gzip'))
- r.content.decode('ascii')
- def test_unicode_get(self):
- url = httpbin('/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(httpbin('ø'), params={'foo': 'foo'})
- def test_unicode_header_name(self):
- requests.put(httpbin('put'), headers={str('Content-Type'): 'application/octet-stream'}, data='\xff') # compat.str is unicode.
- def test_urlencoded_get_query_multivalued_param(self):
- r = requests.get(httpbin('get'), params=dict(test=['foo', 'baz']))
- self.assertEqual(r.status_code, 200)
- self.assertEqual(r.url, httpbin('get?test=foo&test=baz'))
- def test_different_encodings_dont_break_post(self):
- r = requests.post(httpbin('post'),
- data={'stuff': json.dumps({'a': 123})},
- params={'blah': 'asdf1234'},
- files={'file': ('test_requests.py', open(__file__, 'rb'))})
- self.assertEqual(r.status_code, 200)
- def test_unicode_multipart_post(self):
- r = requests.post(httpbin('post'),
- data={'stuff': u'ëlïxr'},
- files={'file': ('test_requests.py', open(__file__, 'rb'))})
- self.assertEqual(r.status_code, 200)
- r = requests.post(httpbin('post'),
- data={'stuff': u'ëlïxr'.encode('utf-8')},
- files={'file': ('test_requests.py', open(__file__, 'rb'))})
- self.assertEqual(r.status_code, 200)
- r = requests.post(httpbin('post'),
- data={'stuff': 'elixr'},
- files={'file': ('test_requests.py', open(__file__, 'rb'))})
- self.assertEqual(r.status_code, 200)
- r = requests.post(httpbin('post'),
- data={'stuff': 'elixr'.encode('utf-8')},
- files={'file': ('test_requests.py', open(__file__, 'rb'))})
- self.assertEqual(r.status_code, 200)
- def test_unicode_multipart_post_fieldnames(self):
- filename = os.path.splitext(__file__)[0] + '.py'
- r = requests.Request(method='POST',
- url=httpbin('post'),
- data={'stuff'.encode('utf-8'): 'elixr'},
- files={'file': ('test_requests.py',
- open(filename, 'rb'))})
- prep = r.prepare()
- self.assertTrue(b'name="stuff"' in prep.body)
- self.assertFalse(b'name="b\'stuff\'"' in prep.body)
- def test_custom_content_type(self):
- r = requests.post(httpbin('post'),
- data={'stuff': json.dumps({'a': 123})},
- files={'file1': ('test_requests.py', open(__file__, 'rb')),
- 'file2': ('test_requests', open(__file__, 'rb'),
- 'text/py-content-type')})
- self.assertEqual(r.status_code, 200)
- self.assertTrue(b"text/py-content-type" in r.request.body)
- def test_hook_receives_request_arguments(self):
- def hook(resp, **kwargs):
- assert resp is not None
- assert kwargs != {}
- requests.Request('GET', HTTPBIN, hooks={'response': hook})
- def test_prepared_request_hook(self):
- def hook(resp, **kwargs):
- resp.hook_working = True
- return resp
- req = requests.Request('GET', HTTPBIN, hooks={'response': hook})
- prep = req.prepare()
- s = requests.Session()
- s.proxies = getproxies()
- resp = s.send(prep)
- self.assertTrue(hasattr(resp, 'hook_working'))
- def test_prepared_from_session(self):
- class DummyAuth(requests.auth.AuthBase):
- def __call__(self, r):
- r.headers['Dummy-Auth-Test'] = 'dummy-auth-test-ok'
- return r
- req = requests.Request('GET', httpbin('headers'))
- self.assertEqual(req.auth, None)
- s = requests.Session()
- s.auth = DummyAuth()
- prep = s.prepare_request(req)
- resp = s.send(prep)
- self.assertTrue(resp.json()['headers']['Dummy-Auth-Test'], 'dummy-auth-test-ok')
- def test_links(self):
- r = requests.Response()
- r.headers = {
- 'cache-control': 'public, max-age=60, s-maxage=60',
- 'connection': 'keep-alive',
- 'content-encoding': 'gzip',
- 'content-type': 'application/json; charset=utf-8',
- 'date': 'Sat, 26 Jan 2013 16:47:56 GMT',
- 'etag': '"6ff6a73c0e446c1f61614769e3ceb778"',
- 'last-modified': 'Sat, 26 Jan 2013 16:22:39 GMT',
- 'link': ('<https://api.github.com/users/kennethreitz/repos?'
- 'page=2&per_page=10>; rel="next", <https://api.github.'
- 'com/users/kennethreitz/repos?page=7&per_page=10>; '
- ' rel="last"'),
- 'server': 'GitHub.com',
- 'status': '200 OK',
- 'vary': 'Accept',
- 'x-content-type-options': 'nosniff',
- 'x-github-media-type': 'github.beta',
- 'x-ratelimit-limit': '60',
- 'x-ratelimit-remaining': '57'
- }
- self.assertEqual(r.links['next']['rel'], 'next')
- def test_cookie_parameters(self):
- key = 'some_cookie'
- value = 'some_value'
- secure = True
- domain = 'test.com'
- rest = {'HttpOnly': True}
- jar = requests.cookies.RequestsCookieJar()
- jar.set(key, value, secure=secure, domain=domain, rest=rest)
- self.assertEqual(len(jar), 1)
- self.assertTrue('some_cookie' in jar)
- cookie = list(jar)[0]
- self.assertEqual(cookie.secure, secure)
- self.assertEqual(cookie.domain, domain)
- self.assertEqual(cookie._rest['HttpOnly'], rest['HttpOnly'])
- def test_time_elapsed_blank(self):
- r = requests.get(httpbin('get'))
- td = r.elapsed
- total_seconds = ((td.microseconds + (td.seconds + td.days * 24 * 3600)
- * 10**6) / 10**6)
- self.assertTrue(total_seconds > 0.0)
- def test_response_is_iterable(self):
- r = requests.Response()
- io = StringIO.StringIO('abc')
- read_ = io.read
- def read_mock(amt, decode_content=None):
- return read_(amt)
- setattr(io, 'read', read_mock)
- r.raw = io
- self.assertTrue(next(iter(r)))
- io.close()
- def test_get_auth_from_url(self):
- url = 'http://user:pass@complex.url.com/path?query=yes'
- self.assertEqual(('user', 'pass'),
- requests.utils.get_auth_from_url(url))
- def test_cannot_send_unprepared_requests(self):
- r = requests.Request(url=HTTPBIN)
- self.assertRaises(ValueError, requests.Session().send, r)
- def test_http_error(self):
- error = requests.exceptions.HTTPError()
- self.assertEqual(error.response, None)
- response = requests.Response()
- error = requests.exceptions.HTTPError(response=response)
- self.assertEqual(error.response, response)
- error = requests.exceptions.HTTPError('message', response=response)
- self.assertEqual(str(error), 'message')
- self.assertEqual(error.response, response)
- def test_session_pickling(self):
- r = requests.Request('GET', httpbin('get'))
- s = requests.Session()
- s = pickle.loads(pickle.dumps(s))
- s.proxies = getproxies()
- r = s.send(r.prepare())
- self.assertEqual(r.status_code, 200)
- def test_fixes_1329(self):
- """
- Ensure that header updates are done case-insensitively.
- """
- s = requests.Session()
- s.headers.update({'ACCEPT': 'BOGUS'})
- s.headers.update({'accept': 'application/json'})
- r = s.get(httpbin('get'))
- headers = r.request.headers
- self.assertEqual(
- headers['accept'],
- 'application/json'
- )
- self.assertEqual(
- headers['Accept'],
- 'application/json'
- )
- self.assertEqual(
- headers['ACCEPT'],
- 'application/json'
- )
- def test_uppercase_scheme_redirect(self):
- parts = urlparse(httpbin('html'))
- url = "HTTP://" + parts.netloc + parts.path
- r = requests.get(httpbin('redirect-to'), params={'url': url})
- self.assertEqual(r.status_code, 200)
- self.assertEqual(r.url.lower(), url.lower())
- def test_transport_adapter_ordering(self):
- s = requests.Session()
- order = ['https://', 'http://']
- self.assertEqual(order, list(s.adapters))
- s.mount('http://git', HTTPAdapter())
- s.mount('http://github', HTTPAdapter())
- s.mount('http://github.com', HTTPAdapter())
- s.mount('http://github.com/about/', HTTPAdapter())
- order = [
- 'http://github.com/about/',
- 'http://github.com',
- 'http://github',
- 'http://git',
- 'https://',
- 'http://',
- ]
- self.assertEqual(order, list(s.adapters))
- s.mount('http://gittip', HTTPAdapter())
- s.mount('http://gittip.com', HTTPAdapter())
- s.mount('http://gittip.com/about/', HTTPAdapter())
- order = [
- 'http://github.com/about/',
- 'http://gittip.com/about/',
- 'http://github.com',
- 'http://gittip.com',
- 'http://github',
- 'http://gittip',
- 'http://git',
- 'https://',
- 'http://',
- ]
- self.assertEqual(order, list(s.adapters))
- s2 = requests.Session()
- s2.adapters = {'http://': HTTPAdapter()}
- s2.mount('https://', HTTPAdapter())
- self.assertTrue('http://' in s2.adapters)
- self.assertTrue('https://' in s2.adapters)
- def test_header_remove_is_case_insensitive(self):
- # From issue #1321
- s = requests.Session()
- s.headers['foo'] = 'bar'
- r = s.get(httpbin('get'), headers={'FOO': None})
- assert 'foo' not in r.request.headers
- def test_params_are_merged_case_sensitive(self):
- s = requests.Session()
- s.params['foo'] = 'bar'
- r = s.get(httpbin('get'), params={'FOO': 'bar'})
- assert r.json()['args'] == {'foo': 'bar', 'FOO': 'bar'}
- def test_long_authinfo_in_url(self):
- url = 'http://{0}:{1}@{2}:9000/path?query#frag'.format(
- 'E8A3BE87-9E3F-4620-8858-95478E385B5B',
- 'EA770032-DA4D-4D84-8CE9-29C6D910BF1E',
- 'exactly-------------sixty-----------three------------characters',
- )
- r = requests.Request('GET', url).prepare()
- self.assertEqual(r.url, url)
- def test_header_keys_are_native(self):
- headers = {u'unicode': 'blah', 'byte'.encode('ascii'): 'blah'}
- r = requests.Request('GET', httpbin('get'), headers=headers)
- p = r.prepare()
- # This is testing that they are builtin strings. A bit weird, but there
- # we go.
- self.assertTrue('unicode' in p.headers.keys())
- self.assertTrue('byte' in p.headers.keys())
- def test_can_send_nonstring_objects_with_files(self):
- data = {'a': 0.0}
- files = {'b': 'foo'}
- r = requests.Request('POST', httpbin('post'), data=data, files=files)
- p = r.prepare()
- self.assertTrue('multipart/form-data' in p.headers['Content-Type'])
- class TestContentEncodingDetection(unittest.TestCase):
- def test_none(self):
- encodings = requests.utils.get_encodings_from_content('')
- self.assertEqual(len(encodings), 0)
- def test_html_charset(self):
- """HTML5 meta charset attribute"""
- content = '<meta charset="UTF-8">'
- encodings = requests.utils.get_encodings_from_content(content)
- self.assertEqual(len(encodings), 1)
- self.assertEqual(encodings[0], 'UTF-8')
- def test_html4_pragma(self):
- """HTML4 pragma directive"""
- content = '<meta http-equiv="Content-type" content="text/html;charset=UTF-8">'
- encodings = requests.utils.get_encodings_from_content(content)
- self.assertEqual(len(encodings), 1)
- self.assertEqual(encodings[0], 'UTF-8')
- def test_xhtml_pragma(self):
- """XHTML 1.x served with text/html MIME type"""
- content = '<meta http-equiv="Content-type" content="text/html;charset=UTF-8" />'
- encodings = requests.utils.get_encodings_from_content(content)
- self.assertEqual(len(encodings), 1)
- self.assertEqual(encodings[0], 'UTF-8')
- def test_xml(self):
- """XHTML 1.x served as XML"""
- content = '<?xml version="1.0" encoding="UTF-8"?>'
- encodings = requests.utils.get_encodings_from_content(content)
- self.assertEqual(len(encodings), 1)
- self.assertEqual(encodings[0], 'UTF-8')
- def test_precedence(self):
- content = '''
- <?xml version="1.0" encoding="XML"?>
- <meta charset="HTML5">
- <meta http-equiv="Content-type" content="text/html;charset=HTML4" />
- '''.strip()
- encodings = requests.utils.get_encodings_from_content(content)
- self.assertEqual(encodings, ['HTML5', 'HTML4', 'XML'])
- class TestCaseInsensitiveDict(unittest.TestCase):
- def test_mapping_init(self):
- cid = CaseInsensitiveDict({'Foo': 'foo','BAr': 'bar'})
- self.assertEqual(len(cid), 2)
- self.assertTrue('foo' in cid)
- self.assertTrue('bar' in cid)
- def test_iterable_init(self):
- cid = CaseInsensitiveDict([('Foo', 'foo'), ('BAr', 'bar')])
- self.assertEqual(len(cid), 2)
- self.assertTrue('foo' in cid)
- self.assertTrue('bar' in cid)
- def test_kwargs_init(self):
- cid = CaseInsensitiveDict(FOO='foo', BAr='bar')
- self.assertEqual(len(cid), 2)
- self.assertTrue('foo' in cid)
- self.assertTrue('bar' in cid)
- def test_docstring_example(self):
- cid = CaseInsensitiveDict()
- cid['Accept'] = 'application/json'
- self.assertEqual(cid['aCCEPT'], 'application/json')
- self.assertEqual(list(cid), ['Accept'])
- def test_len(self):
- cid = CaseInsensitiveDict({'a': 'a', 'b': 'b'})
- cid['A'] = 'a'
- self.assertEqual(len(cid), 2)
- def test_getitem(self):
- cid = CaseInsensitiveDict({'Spam': 'blueval'})
- self.assertEqual(cid['spam'], 'blueval')
- self.assertEqual(cid['SPAM'], 'blueval')
- def test_fixes_649(self):
- """__setitem__ should behave case-insensitively."""
- cid = CaseInsensitiveDict()
- cid['spam'] = 'oneval'
- cid['Spam'] = 'twoval'
- cid['sPAM'] = 'redval'
- cid['SPAM'] = 'blueval'
- self.assertEqual(cid['spam'], 'blueval')
- self.assertEqual(cid['SPAM'], 'blueval')
- self.assertEqual(list(cid.keys()), ['SPAM'])
- def test_delitem(self):
- cid = CaseInsensitiveDict()
- cid['Spam'] = 'someval'
- del cid['sPam']
- self.assertFalse('spam' in cid)
- self.assertEqual(len(cid), 0)
- def test_contains(self):
- cid = CaseInsensitiveDict()
- cid['Spam'] = 'someval'
- self.assertTrue('Spam' in cid)
- self.assertTrue('spam' in cid)
- self.assertTrue('SPAM' in cid)
- self.assertTrue('sPam' in cid)
- self.assertFalse('notspam' in cid)
- def test_get(self):
- cid = CaseInsensitiveDict()
- cid['spam'] = 'oneval'
- cid['SPAM'] = 'blueval'
- self.assertEqual(cid.get('spam'), 'blueval')
- self.assertEqual(cid.get('SPAM'), 'blueval')
- self.assertEqual(cid.get('sPam'), 'blueval')
- self.assertEqual(cid.get('notspam', 'default'), 'default')
- def test_update(self):
- cid = CaseInsensitiveDict()
- cid['spam'] = 'blueval'
- cid.update({'sPam': 'notblueval'})
- self.assertEqual(cid['spam'], 'notblueval')
- cid = CaseInsensitiveDict({'Foo': 'foo','BAr': 'bar'})
- cid.update({'fOO': 'anotherfoo', 'bAR': 'anotherbar'})
- self.assertEqual(len(cid), 2)
- self.assertEqual(cid['foo'], 'anotherfoo')
- self.assertEqual(cid['bar'], 'anotherbar')
- def test_update_retains_unchanged(self):
- cid = CaseInsensitiveDict({'foo': 'foo', 'bar': 'bar'})
- cid.update({'foo': 'newfoo'})
- self.assertEquals(cid['bar'], 'bar')
- def test_iter(self):
- cid = CaseInsensitiveDict({'Spam': 'spam', 'Eggs': 'eggs'})
- keys = frozenset(['Spam', 'Eggs'])
- self.assertEqual(frozenset(iter(cid)), keys)
- def test_equality(self):
- cid = CaseInsensitiveDict({'SPAM': 'blueval', 'Eggs': 'redval'})
- othercid = CaseInsensitiveDict({'spam': 'blueval', 'eggs': 'redval'})
- self.assertEqual(cid, othercid)
- del othercid['spam']
- self.assertNotEqual(cid, othercid)
- self.assertEqual(cid, {'spam': 'blueval', 'eggs': 'redval'})
- def test_setdefault(self):
- cid = CaseInsensitiveDict({'Spam': 'blueval'})
- self.assertEqual(
- cid.setdefault('spam', 'notblueval'),
- 'blueval'
- )
- self.assertEqual(
- cid.setdefault('notspam', 'notblueval'),
- 'notblueval'
- )
- def test_lower_items(self):
- cid = CaseInsensitiveDict({
- 'Accept': 'application/json',
- 'user-Agent': 'requests',
- })
- keyset = frozenset(lowerkey for lowerkey, v in cid.lower_items())
- lowerkeyset = frozenset(['accept', 'user-agent'])
- self.assertEqual(keyset, lowerkeyset)
- def test_preserve_key_case(self):
- cid = CaseInsensitiveDict({
- 'Accept': 'application/json',
- 'user-Agent': 'requests',
- })
- keyset = frozenset(['Accept', 'user-Agent'])
- self.assertEqual(frozenset(i[0] for i in cid.items()), keyset)
- self.assertEqual(frozenset(cid.keys()), keyset)
- self.assertEqual(frozenset(cid), keyset)
- def test_preserve_last_key_case(self):
- cid = CaseInsensitiveDict({
- 'Accept': 'application/json',
- 'user-Agent': 'requests',
- })
- cid.update({'ACCEPT': 'application/json'})
- cid['USER-AGENT'] = 'requests'
- keyset = frozenset(['ACCEPT', 'USER-AGENT'])
- self.assertEqual(frozenset(i[0] for i in cid.items()), keyset)
- self.assertEqual(frozenset(cid.keys()), keyset)
- self.assertEqual(frozenset(cid), keyset)
- if __name__ == '__main__':
- unittest.main()
|