test_requests.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. """Tests for Requests."""
  4. from __future__ import division
  5. import json
  6. import os
  7. import unittest
  8. import pickle
  9. import requests
  10. import pytest
  11. from requests.auth import HTTPDigestAuth
  12. from requests.adapters import HTTPAdapter
  13. from requests.compat import str, cookielib, getproxies, urljoin, urlparse
  14. from requests.cookies import cookiejar_from_dict
  15. from requests.exceptions import InvalidURL, MissingSchema
  16. from requests.structures import CaseInsensitiveDict
  17. try:
  18. import StringIO
  19. except ImportError:
  20. import io as StringIO
  21. HTTPBIN = os.environ.get('HTTPBIN_URL', 'http://httpbin.org/')
  22. # Issue #1483: Make sure the URL always has a trailing slash
  23. HTTPBIN = HTTPBIN.rstrip('/') + '/'
  24. def httpbin(*suffix):
  25. """Returns url for HTTPBIN resource."""
  26. return urljoin(HTTPBIN, '/'.join(suffix))
  27. class RequestsTestCase(unittest.TestCase):
  28. _multiprocess_can_split_ = True
  29. def setUp(self):
  30. """Create simple data set with headers."""
  31. pass
  32. def tearDown(self):
  33. """Teardown."""
  34. pass
  35. def test_entry_points(self):
  36. requests.session
  37. requests.session().get
  38. requests.session().head
  39. requests.get
  40. requests.head
  41. requests.put
  42. requests.patch
  43. requests.post
  44. def test_invalid_url(self):
  45. self.assertRaises(MissingSchema, requests.get, 'hiwpefhipowhefopw')
  46. self.assertRaises(InvalidURL, requests.get, 'http://')
  47. def test_basic_building(self):
  48. req = requests.Request()
  49. req.url = 'http://kennethreitz.org/'
  50. req.data = {'life': '42'}
  51. pr = req.prepare()
  52. assert pr.url == req.url
  53. assert pr.body == 'life=42'
  54. def test_no_content_length(self):
  55. get_req = requests.Request('GET', httpbin('get')).prepare()
  56. self.assertTrue('Content-Length' not in get_req.headers)
  57. head_req = requests.Request('HEAD', httpbin('head')).prepare()
  58. self.assertTrue('Content-Length' not in head_req.headers)
  59. def test_path_is_not_double_encoded(self):
  60. request = requests.Request('GET', "http://0.0.0.0/get/test case").prepare()
  61. self.assertEqual(request.path_url, "/get/test%20case")
  62. def test_params_are_added_before_fragment(self):
  63. request = requests.Request('GET',
  64. "http://example.com/path#fragment", params={"a": "b"}).prepare()
  65. self.assertEqual(request.url,
  66. "http://example.com/path?a=b#fragment")
  67. request = requests.Request('GET',
  68. "http://example.com/path?key=value#fragment", params={"a": "b"}).prepare()
  69. self.assertEqual(request.url,
  70. "http://example.com/path?key=value&a=b#fragment")
  71. def test_mixed_case_scheme_acceptable(self):
  72. s = requests.Session()
  73. s.proxies = getproxies()
  74. parts = urlparse(httpbin('get'))
  75. schemes = ['http://', 'HTTP://', 'hTTp://', 'HttP://',
  76. 'https://', 'HTTPS://', 'hTTps://', 'HttPs://']
  77. for scheme in schemes:
  78. url = scheme + parts.netloc + parts.path
  79. r = requests.Request('GET', url)
  80. r = s.send(r.prepare())
  81. self.assertEqual(r.status_code, 200,
  82. "failed for scheme %s" % scheme)
  83. def test_HTTP_200_OK_GET_ALTERNATIVE(self):
  84. r = requests.Request('GET', httpbin('get'))
  85. s = requests.Session()
  86. s.proxies = getproxies()
  87. r = s.send(r.prepare())
  88. self.assertEqual(r.status_code, 200)
  89. def test_HTTP_302_ALLOW_REDIRECT_GET(self):
  90. r = requests.get(httpbin('redirect', '1'))
  91. self.assertEqual(r.status_code, 200)
  92. # def test_HTTP_302_ALLOW_REDIRECT_POST(self):
  93. # r = requests.post(httpbin('status', '302'), data={'some': 'data'})
  94. # self.assertEqual(r.status_code, 200)
  95. def test_HTTP_200_OK_GET_WITH_PARAMS(self):
  96. heads = {'User-agent': 'Mozilla/5.0'}
  97. r = requests.get(httpbin('user-agent'), headers=heads)
  98. self.assertTrue(heads['User-agent'] in r.text)
  99. self.assertEqual(r.status_code, 200)
  100. def test_HTTP_200_OK_GET_WITH_MIXED_PARAMS(self):
  101. heads = {'User-agent': 'Mozilla/5.0'}
  102. r = requests.get(httpbin('get') + '?test=true', params={'q': 'test'}, headers=heads)
  103. self.assertEqual(r.status_code, 200)
  104. def test_set_cookie_on_301(self):
  105. s = requests.session()
  106. url = httpbin('cookies/set?foo=bar')
  107. r = s.get(url)
  108. self.assertTrue(s.cookies['foo'] == 'bar')
  109. def test_cookie_sent_on_redirect(self):
  110. s = requests.session()
  111. s.get(httpbin('cookies/set?foo=bar'))
  112. r = s.get(httpbin('redirect/1')) # redirects to httpbin('get')
  113. self.assertTrue("Cookie" in r.json()["headers"])
  114. def test_cookie_removed_on_expire(self):
  115. s = requests.session()
  116. s.get(httpbin('cookies/set?foo=bar'))
  117. self.assertTrue(s.cookies['foo'] == 'bar')
  118. s.get(
  119. httpbin('response-headers'),
  120. params={
  121. 'Set-Cookie':
  122. 'foo=deleted; expires=Thu, 01-Jan-1970 00:00:01 GMT'
  123. }
  124. )
  125. assert 'foo' not in s.cookies
  126. def test_cookie_quote_wrapped(self):
  127. s = requests.session()
  128. s.get(httpbin('cookies/set?foo="bar:baz"'))
  129. self.assertTrue(s.cookies['foo'] == '"bar:baz"')
  130. def test_request_cookie_overrides_session_cookie(self):
  131. s = requests.session()
  132. s.cookies['foo'] = 'bar'
  133. r = s.get(httpbin('cookies'), cookies={'foo': 'baz'})
  134. assert r.json()['cookies']['foo'] == 'baz'
  135. # Session cookie should not be modified
  136. assert s.cookies['foo'] == 'bar'
  137. def test_generic_cookiejar_works(self):
  138. cj = cookielib.CookieJar()
  139. cookiejar_from_dict({'foo': 'bar'}, cj)
  140. s = requests.session()
  141. s.cookies = cj
  142. r = s.get(httpbin('cookies'))
  143. # Make sure the cookie was sent
  144. assert r.json()['cookies']['foo'] == 'bar'
  145. # Make sure the session cj is still the custom one
  146. assert s.cookies is cj
  147. def test_requests_in_history_are_not_overridden(self):
  148. resp = requests.get(httpbin('redirect/3'))
  149. urls = [r.url for r in resp.history]
  150. req_urls = [r.request.url for r in resp.history]
  151. self.assertEquals(urls, req_urls)
  152. def test_user_agent_transfers(self):
  153. heads = {
  154. 'User-agent': 'Mozilla/5.0 (github.com/kennethreitz/requests)'
  155. }
  156. r = requests.get(httpbin('user-agent'), headers=heads)
  157. self.assertTrue(heads['User-agent'] in r.text)
  158. heads = {
  159. 'user-agent': 'Mozilla/5.0 (github.com/kennethreitz/requests)'
  160. }
  161. r = requests.get(httpbin('user-agent'), headers=heads)
  162. self.assertTrue(heads['user-agent'] in r.text)
  163. def test_HTTP_200_OK_HEAD(self):
  164. r = requests.head(httpbin('get'))
  165. self.assertEqual(r.status_code, 200)
  166. def test_HTTP_200_OK_PUT(self):
  167. r = requests.put(httpbin('put'))
  168. self.assertEqual(r.status_code, 200)
  169. def test_BASICAUTH_TUPLE_HTTP_200_OK_GET(self):
  170. auth = ('user', 'pass')
  171. url = httpbin('basic-auth', 'user', 'pass')
  172. r = requests.get(url, auth=auth)
  173. self.assertEqual(r.status_code, 200)
  174. r = requests.get(url)
  175. self.assertEqual(r.status_code, 401)
  176. s = requests.session()
  177. s.auth = auth
  178. r = s.get(url)
  179. self.assertEqual(r.status_code, 200)
  180. def test_basicauth_with_netrc(self):
  181. auth = ('user', 'pass')
  182. wrong_auth = ('wronguser', 'wrongpass')
  183. url = httpbin('basic-auth', 'user', 'pass')
  184. def get_netrc_auth_mock(url):
  185. return auth
  186. requests.sessions.get_netrc_auth = get_netrc_auth_mock
  187. # Should use netrc and work.
  188. r = requests.get(url)
  189. self.assertEqual(r.status_code, 200)
  190. # Given auth should override and fail.
  191. r = requests.get(url, auth=wrong_auth)
  192. self.assertEqual(r.status_code, 401)
  193. s = requests.session()
  194. # Should use netrc and work.
  195. r = s.get(url)
  196. self.assertEqual(r.status_code, 200)
  197. # Given auth should override and fail.
  198. s.auth = wrong_auth
  199. r = s.get(url)
  200. self.assertEqual(r.status_code, 401)
  201. def test_DIGEST_HTTP_200_OK_GET(self):
  202. auth = HTTPDigestAuth('user', 'pass')
  203. url = httpbin('digest-auth', 'auth', 'user', 'pass')
  204. r = requests.get(url, auth=auth)
  205. self.assertEqual(r.status_code, 200)
  206. r = requests.get(url)
  207. self.assertEqual(r.status_code, 401)
  208. s = requests.session()
  209. s.auth = HTTPDigestAuth('user', 'pass')
  210. r = s.get(url)
  211. self.assertEqual(r.status_code, 200)
  212. def test_DIGEST_AUTH_RETURNS_COOKIE(self):
  213. url = httpbin('digest-auth', 'auth', 'user', 'pass')
  214. auth = HTTPDigestAuth('user', 'pass')
  215. r = requests.get(url)
  216. assert r.cookies['fake'] == 'fake_value'
  217. r = requests.get(url, auth=auth)
  218. assert r.status_code == 200
  219. def test_DIGEST_AUTH_SETS_SESSION_COOKIES(self):
  220. url = httpbin('digest-auth', 'auth', 'user', 'pass')
  221. auth = HTTPDigestAuth('user', 'pass')
  222. s = requests.Session()
  223. s.get(url, auth=auth)
  224. assert s.cookies['fake'] == 'fake_value'
  225. def test_DIGEST_STREAM(self):
  226. auth = HTTPDigestAuth('user', 'pass')
  227. url = httpbin('digest-auth', 'auth', 'user', 'pass')
  228. r = requests.get(url, auth=auth, stream=True)
  229. self.assertNotEqual(r.raw.read(), b'')
  230. r = requests.get(url, auth=auth, stream=False)
  231. self.assertEqual(r.raw.read(), b'')
  232. def test_DIGESTAUTH_WRONG_HTTP_401_GET(self):
  233. auth = HTTPDigestAuth('user', 'wrongpass')
  234. url = httpbin('digest-auth', 'auth', 'user', 'pass')
  235. r = requests.get(url, auth=auth)
  236. self.assertEqual(r.status_code, 401)
  237. r = requests.get(url)
  238. self.assertEqual(r.status_code, 401)
  239. s = requests.session()
  240. s.auth = auth
  241. r = s.get(url)
  242. self.assertEqual(r.status_code, 401)
  243. def test_POSTBIN_GET_POST_FILES(self):
  244. url = httpbin('post')
  245. post1 = requests.post(url).raise_for_status()
  246. post1 = requests.post(url, data={'some': 'data'})
  247. self.assertEqual(post1.status_code, 200)
  248. with open('requirements.txt') as f:
  249. post2 = requests.post(url, files={'some': f})
  250. self.assertEqual(post2.status_code, 200)
  251. post4 = requests.post(url, data='[{"some": "json"}]')
  252. self.assertEqual(post4.status_code, 200)
  253. try:
  254. requests.post(url, files=['bad file data'])
  255. except ValueError:
  256. pass
  257. def test_POSTBIN_GET_POST_FILES_WITH_DATA(self):
  258. url = httpbin('post')
  259. post1 = requests.post(url).raise_for_status()
  260. post1 = requests.post(url, data={'some': 'data'})
  261. self.assertEqual(post1.status_code, 200)
  262. with open('requirements.txt') as f:
  263. post2 = requests.post(url, data={'some': 'data'}, files={'some': f})
  264. self.assertEqual(post2.status_code, 200)
  265. post4 = requests.post(url, data='[{"some": "json"}]')
  266. self.assertEqual(post4.status_code, 200)
  267. try:
  268. requests.post(url, files=['bad file data'])
  269. except ValueError:
  270. pass
  271. def test_conflicting_post_params(self):
  272. url = httpbin('post')
  273. with open('requirements.txt') as f:
  274. pytest.raises(ValueError, "requests.post(url, data='[{\"some\": \"data\"}]', files={'some': f})")
  275. pytest.raises(ValueError, "requests.post(url, data=u'[{\"some\": \"data\"}]', files={'some': f})")
  276. def test_request_ok_set(self):
  277. r = requests.get(httpbin('status', '404'))
  278. self.assertEqual(r.ok, False)
  279. def test_status_raising(self):
  280. r = requests.get(httpbin('status', '404'))
  281. self.assertRaises(requests.exceptions.HTTPError, r.raise_for_status)
  282. r = requests.get(httpbin('status', '500'))
  283. self.assertFalse(r.ok)
  284. def test_decompress_gzip(self):
  285. r = requests.get(httpbin('gzip'))
  286. r.content.decode('ascii')
  287. def test_unicode_get(self):
  288. url = httpbin('/get')
  289. requests.get(url, params={'foo': 'føø'})
  290. requests.get(url, params={'føø': 'føø'})
  291. requests.get(url, params={'føø': 'føø'})
  292. requests.get(url, params={'foo': 'foo'})
  293. requests.get(httpbin('ø'), params={'foo': 'foo'})
  294. def test_unicode_header_name(self):
  295. requests.put(httpbin('put'), headers={str('Content-Type'): 'application/octet-stream'}, data='\xff') # compat.str is unicode.
  296. def test_urlencoded_get_query_multivalued_param(self):
  297. r = requests.get(httpbin('get'), params=dict(test=['foo', 'baz']))
  298. self.assertEqual(r.status_code, 200)
  299. self.assertEqual(r.url, httpbin('get?test=foo&test=baz'))
  300. def test_different_encodings_dont_break_post(self):
  301. r = requests.post(httpbin('post'),
  302. data={'stuff': json.dumps({'a': 123})},
  303. params={'blah': 'asdf1234'},
  304. files={'file': ('test_requests.py', open(__file__, 'rb'))})
  305. self.assertEqual(r.status_code, 200)
  306. def test_unicode_multipart_post(self):
  307. r = requests.post(httpbin('post'),
  308. data={'stuff': u'ëlïxr'},
  309. files={'file': ('test_requests.py', open(__file__, 'rb'))})
  310. self.assertEqual(r.status_code, 200)
  311. r = requests.post(httpbin('post'),
  312. data={'stuff': u'ëlïxr'.encode('utf-8')},
  313. files={'file': ('test_requests.py', open(__file__, 'rb'))})
  314. self.assertEqual(r.status_code, 200)
  315. r = requests.post(httpbin('post'),
  316. data={'stuff': 'elixr'},
  317. files={'file': ('test_requests.py', open(__file__, 'rb'))})
  318. self.assertEqual(r.status_code, 200)
  319. r = requests.post(httpbin('post'),
  320. data={'stuff': 'elixr'.encode('utf-8')},
  321. files={'file': ('test_requests.py', open(__file__, 'rb'))})
  322. self.assertEqual(r.status_code, 200)
  323. def test_unicode_multipart_post_fieldnames(self):
  324. filename = os.path.splitext(__file__)[0] + '.py'
  325. r = requests.Request(method='POST',
  326. url=httpbin('post'),
  327. data={'stuff'.encode('utf-8'): 'elixr'},
  328. files={'file': ('test_requests.py',
  329. open(filename, 'rb'))})
  330. prep = r.prepare()
  331. self.assertTrue(b'name="stuff"' in prep.body)
  332. self.assertFalse(b'name="b\'stuff\'"' in prep.body)
  333. def test_custom_content_type(self):
  334. r = requests.post(httpbin('post'),
  335. data={'stuff': json.dumps({'a': 123})},
  336. files={'file1': ('test_requests.py', open(__file__, 'rb')),
  337. 'file2': ('test_requests', open(__file__, 'rb'),
  338. 'text/py-content-type')})
  339. self.assertEqual(r.status_code, 200)
  340. self.assertTrue(b"text/py-content-type" in r.request.body)
  341. def test_hook_receives_request_arguments(self):
  342. def hook(resp, **kwargs):
  343. assert resp is not None
  344. assert kwargs != {}
  345. requests.Request('GET', HTTPBIN, hooks={'response': hook})
  346. def test_prepared_request_hook(self):
  347. def hook(resp, **kwargs):
  348. resp.hook_working = True
  349. return resp
  350. req = requests.Request('GET', HTTPBIN, hooks={'response': hook})
  351. prep = req.prepare()
  352. s = requests.Session()
  353. s.proxies = getproxies()
  354. resp = s.send(prep)
  355. self.assertTrue(hasattr(resp, 'hook_working'))
  356. def test_prepared_from_session(self):
  357. class DummyAuth(requests.auth.AuthBase):
  358. def __call__(self, r):
  359. r.headers['Dummy-Auth-Test'] = 'dummy-auth-test-ok'
  360. return r
  361. req = requests.Request('GET', httpbin('headers'))
  362. self.assertEqual(req.auth, None)
  363. s = requests.Session()
  364. s.auth = DummyAuth()
  365. prep = s.prepare_request(req)
  366. resp = s.send(prep)
  367. self.assertTrue(resp.json()['headers']['Dummy-Auth-Test'], 'dummy-auth-test-ok')
  368. def test_links(self):
  369. r = requests.Response()
  370. r.headers = {
  371. 'cache-control': 'public, max-age=60, s-maxage=60',
  372. 'connection': 'keep-alive',
  373. 'content-encoding': 'gzip',
  374. 'content-type': 'application/json; charset=utf-8',
  375. 'date': 'Sat, 26 Jan 2013 16:47:56 GMT',
  376. 'etag': '"6ff6a73c0e446c1f61614769e3ceb778"',
  377. 'last-modified': 'Sat, 26 Jan 2013 16:22:39 GMT',
  378. 'link': ('<https://api.github.com/users/kennethreitz/repos?'
  379. 'page=2&per_page=10>; rel="next", <https://api.github.'
  380. 'com/users/kennethreitz/repos?page=7&per_page=10>; '
  381. ' rel="last"'),
  382. 'server': 'GitHub.com',
  383. 'status': '200 OK',
  384. 'vary': 'Accept',
  385. 'x-content-type-options': 'nosniff',
  386. 'x-github-media-type': 'github.beta',
  387. 'x-ratelimit-limit': '60',
  388. 'x-ratelimit-remaining': '57'
  389. }
  390. self.assertEqual(r.links['next']['rel'], 'next')
  391. def test_cookie_parameters(self):
  392. key = 'some_cookie'
  393. value = 'some_value'
  394. secure = True
  395. domain = 'test.com'
  396. rest = {'HttpOnly': True}
  397. jar = requests.cookies.RequestsCookieJar()
  398. jar.set(key, value, secure=secure, domain=domain, rest=rest)
  399. self.assertEqual(len(jar), 1)
  400. self.assertTrue('some_cookie' in jar)
  401. cookie = list(jar)[0]
  402. self.assertEqual(cookie.secure, secure)
  403. self.assertEqual(cookie.domain, domain)
  404. self.assertEqual(cookie._rest['HttpOnly'], rest['HttpOnly'])
  405. def test_time_elapsed_blank(self):
  406. r = requests.get(httpbin('get'))
  407. td = r.elapsed
  408. total_seconds = ((td.microseconds + (td.seconds + td.days * 24 * 3600)
  409. * 10**6) / 10**6)
  410. self.assertTrue(total_seconds > 0.0)
  411. def test_response_is_iterable(self):
  412. r = requests.Response()
  413. io = StringIO.StringIO('abc')
  414. read_ = io.read
  415. def read_mock(amt, decode_content=None):
  416. return read_(amt)
  417. setattr(io, 'read', read_mock)
  418. r.raw = io
  419. self.assertTrue(next(iter(r)))
  420. io.close()
  421. def test_get_auth_from_url(self):
  422. url = 'http://user:pass@complex.url.com/path?query=yes'
  423. self.assertEqual(('user', 'pass'),
  424. requests.utils.get_auth_from_url(url))
  425. def test_cannot_send_unprepared_requests(self):
  426. r = requests.Request(url=HTTPBIN)
  427. self.assertRaises(ValueError, requests.Session().send, r)
  428. def test_http_error(self):
  429. error = requests.exceptions.HTTPError()
  430. self.assertEqual(error.response, None)
  431. response = requests.Response()
  432. error = requests.exceptions.HTTPError(response=response)
  433. self.assertEqual(error.response, response)
  434. error = requests.exceptions.HTTPError('message', response=response)
  435. self.assertEqual(str(error), 'message')
  436. self.assertEqual(error.response, response)
  437. def test_session_pickling(self):
  438. r = requests.Request('GET', httpbin('get'))
  439. s = requests.Session()
  440. s = pickle.loads(pickle.dumps(s))
  441. s.proxies = getproxies()
  442. r = s.send(r.prepare())
  443. self.assertEqual(r.status_code, 200)
  444. def test_fixes_1329(self):
  445. """
  446. Ensure that header updates are done case-insensitively.
  447. """
  448. s = requests.Session()
  449. s.headers.update({'ACCEPT': 'BOGUS'})
  450. s.headers.update({'accept': 'application/json'})
  451. r = s.get(httpbin('get'))
  452. headers = r.request.headers
  453. self.assertEqual(
  454. headers['accept'],
  455. 'application/json'
  456. )
  457. self.assertEqual(
  458. headers['Accept'],
  459. 'application/json'
  460. )
  461. self.assertEqual(
  462. headers['ACCEPT'],
  463. 'application/json'
  464. )
  465. def test_uppercase_scheme_redirect(self):
  466. parts = urlparse(httpbin('html'))
  467. url = "HTTP://" + parts.netloc + parts.path
  468. r = requests.get(httpbin('redirect-to'), params={'url': url})
  469. self.assertEqual(r.status_code, 200)
  470. self.assertEqual(r.url.lower(), url.lower())
  471. def test_transport_adapter_ordering(self):
  472. s = requests.Session()
  473. order = ['https://', 'http://']
  474. self.assertEqual(order, list(s.adapters))
  475. s.mount('http://git', HTTPAdapter())
  476. s.mount('http://github', HTTPAdapter())
  477. s.mount('http://github.com', HTTPAdapter())
  478. s.mount('http://github.com/about/', HTTPAdapter())
  479. order = [
  480. 'http://github.com/about/',
  481. 'http://github.com',
  482. 'http://github',
  483. 'http://git',
  484. 'https://',
  485. 'http://',
  486. ]
  487. self.assertEqual(order, list(s.adapters))
  488. s.mount('http://gittip', HTTPAdapter())
  489. s.mount('http://gittip.com', HTTPAdapter())
  490. s.mount('http://gittip.com/about/', HTTPAdapter())
  491. order = [
  492. 'http://github.com/about/',
  493. 'http://gittip.com/about/',
  494. 'http://github.com',
  495. 'http://gittip.com',
  496. 'http://github',
  497. 'http://gittip',
  498. 'http://git',
  499. 'https://',
  500. 'http://',
  501. ]
  502. self.assertEqual(order, list(s.adapters))
  503. s2 = requests.Session()
  504. s2.adapters = {'http://': HTTPAdapter()}
  505. s2.mount('https://', HTTPAdapter())
  506. self.assertTrue('http://' in s2.adapters)
  507. self.assertTrue('https://' in s2.adapters)
  508. def test_header_remove_is_case_insensitive(self):
  509. # From issue #1321
  510. s = requests.Session()
  511. s.headers['foo'] = 'bar'
  512. r = s.get(httpbin('get'), headers={'FOO': None})
  513. assert 'foo' not in r.request.headers
  514. def test_params_are_merged_case_sensitive(self):
  515. s = requests.Session()
  516. s.params['foo'] = 'bar'
  517. r = s.get(httpbin('get'), params={'FOO': 'bar'})
  518. assert r.json()['args'] == {'foo': 'bar', 'FOO': 'bar'}
  519. def test_long_authinfo_in_url(self):
  520. url = 'http://{0}:{1}@{2}:9000/path?query#frag'.format(
  521. 'E8A3BE87-9E3F-4620-8858-95478E385B5B',
  522. 'EA770032-DA4D-4D84-8CE9-29C6D910BF1E',
  523. 'exactly-------------sixty-----------three------------characters',
  524. )
  525. r = requests.Request('GET', url).prepare()
  526. self.assertEqual(r.url, url)
  527. def test_header_keys_are_native(self):
  528. headers = {u'unicode': 'blah', 'byte'.encode('ascii'): 'blah'}
  529. r = requests.Request('GET', httpbin('get'), headers=headers)
  530. p = r.prepare()
  531. # This is testing that they are builtin strings. A bit weird, but there
  532. # we go.
  533. self.assertTrue('unicode' in p.headers.keys())
  534. self.assertTrue('byte' in p.headers.keys())
  535. def test_can_send_nonstring_objects_with_files(self):
  536. data = {'a': 0.0}
  537. files = {'b': 'foo'}
  538. r = requests.Request('POST', httpbin('post'), data=data, files=files)
  539. p = r.prepare()
  540. self.assertTrue('multipart/form-data' in p.headers['Content-Type'])
  541. class TestContentEncodingDetection(unittest.TestCase):
  542. def test_none(self):
  543. encodings = requests.utils.get_encodings_from_content('')
  544. self.assertEqual(len(encodings), 0)
  545. def test_html_charset(self):
  546. """HTML5 meta charset attribute"""
  547. content = '<meta charset="UTF-8">'
  548. encodings = requests.utils.get_encodings_from_content(content)
  549. self.assertEqual(len(encodings), 1)
  550. self.assertEqual(encodings[0], 'UTF-8')
  551. def test_html4_pragma(self):
  552. """HTML4 pragma directive"""
  553. content = '<meta http-equiv="Content-type" content="text/html;charset=UTF-8">'
  554. encodings = requests.utils.get_encodings_from_content(content)
  555. self.assertEqual(len(encodings), 1)
  556. self.assertEqual(encodings[0], 'UTF-8')
  557. def test_xhtml_pragma(self):
  558. """XHTML 1.x served with text/html MIME type"""
  559. content = '<meta http-equiv="Content-type" content="text/html;charset=UTF-8" />'
  560. encodings = requests.utils.get_encodings_from_content(content)
  561. self.assertEqual(len(encodings), 1)
  562. self.assertEqual(encodings[0], 'UTF-8')
  563. def test_xml(self):
  564. """XHTML 1.x served as XML"""
  565. content = '<?xml version="1.0" encoding="UTF-8"?>'
  566. encodings = requests.utils.get_encodings_from_content(content)
  567. self.assertEqual(len(encodings), 1)
  568. self.assertEqual(encodings[0], 'UTF-8')
  569. def test_precedence(self):
  570. content = '''
  571. <?xml version="1.0" encoding="XML"?>
  572. <meta charset="HTML5">
  573. <meta http-equiv="Content-type" content="text/html;charset=HTML4" />
  574. '''.strip()
  575. encodings = requests.utils.get_encodings_from_content(content)
  576. self.assertEqual(encodings, ['HTML5', 'HTML4', 'XML'])
  577. class TestCaseInsensitiveDict(unittest.TestCase):
  578. def test_mapping_init(self):
  579. cid = CaseInsensitiveDict({'Foo': 'foo','BAr': 'bar'})
  580. self.assertEqual(len(cid), 2)
  581. self.assertTrue('foo' in cid)
  582. self.assertTrue('bar' in cid)
  583. def test_iterable_init(self):
  584. cid = CaseInsensitiveDict([('Foo', 'foo'), ('BAr', 'bar')])
  585. self.assertEqual(len(cid), 2)
  586. self.assertTrue('foo' in cid)
  587. self.assertTrue('bar' in cid)
  588. def test_kwargs_init(self):
  589. cid = CaseInsensitiveDict(FOO='foo', BAr='bar')
  590. self.assertEqual(len(cid), 2)
  591. self.assertTrue('foo' in cid)
  592. self.assertTrue('bar' in cid)
  593. def test_docstring_example(self):
  594. cid = CaseInsensitiveDict()
  595. cid['Accept'] = 'application/json'
  596. self.assertEqual(cid['aCCEPT'], 'application/json')
  597. self.assertEqual(list(cid), ['Accept'])
  598. def test_len(self):
  599. cid = CaseInsensitiveDict({'a': 'a', 'b': 'b'})
  600. cid['A'] = 'a'
  601. self.assertEqual(len(cid), 2)
  602. def test_getitem(self):
  603. cid = CaseInsensitiveDict({'Spam': 'blueval'})
  604. self.assertEqual(cid['spam'], 'blueval')
  605. self.assertEqual(cid['SPAM'], 'blueval')
  606. def test_fixes_649(self):
  607. """__setitem__ should behave case-insensitively."""
  608. cid = CaseInsensitiveDict()
  609. cid['spam'] = 'oneval'
  610. cid['Spam'] = 'twoval'
  611. cid['sPAM'] = 'redval'
  612. cid['SPAM'] = 'blueval'
  613. self.assertEqual(cid['spam'], 'blueval')
  614. self.assertEqual(cid['SPAM'], 'blueval')
  615. self.assertEqual(list(cid.keys()), ['SPAM'])
  616. def test_delitem(self):
  617. cid = CaseInsensitiveDict()
  618. cid['Spam'] = 'someval'
  619. del cid['sPam']
  620. self.assertFalse('spam' in cid)
  621. self.assertEqual(len(cid), 0)
  622. def test_contains(self):
  623. cid = CaseInsensitiveDict()
  624. cid['Spam'] = 'someval'
  625. self.assertTrue('Spam' in cid)
  626. self.assertTrue('spam' in cid)
  627. self.assertTrue('SPAM' in cid)
  628. self.assertTrue('sPam' in cid)
  629. self.assertFalse('notspam' in cid)
  630. def test_get(self):
  631. cid = CaseInsensitiveDict()
  632. cid['spam'] = 'oneval'
  633. cid['SPAM'] = 'blueval'
  634. self.assertEqual(cid.get('spam'), 'blueval')
  635. self.assertEqual(cid.get('SPAM'), 'blueval')
  636. self.assertEqual(cid.get('sPam'), 'blueval')
  637. self.assertEqual(cid.get('notspam', 'default'), 'default')
  638. def test_update(self):
  639. cid = CaseInsensitiveDict()
  640. cid['spam'] = 'blueval'
  641. cid.update({'sPam': 'notblueval'})
  642. self.assertEqual(cid['spam'], 'notblueval')
  643. cid = CaseInsensitiveDict({'Foo': 'foo','BAr': 'bar'})
  644. cid.update({'fOO': 'anotherfoo', 'bAR': 'anotherbar'})
  645. self.assertEqual(len(cid), 2)
  646. self.assertEqual(cid['foo'], 'anotherfoo')
  647. self.assertEqual(cid['bar'], 'anotherbar')
  648. def test_update_retains_unchanged(self):
  649. cid = CaseInsensitiveDict({'foo': 'foo', 'bar': 'bar'})
  650. cid.update({'foo': 'newfoo'})
  651. self.assertEquals(cid['bar'], 'bar')
  652. def test_iter(self):
  653. cid = CaseInsensitiveDict({'Spam': 'spam', 'Eggs': 'eggs'})
  654. keys = frozenset(['Spam', 'Eggs'])
  655. self.assertEqual(frozenset(iter(cid)), keys)
  656. def test_equality(self):
  657. cid = CaseInsensitiveDict({'SPAM': 'blueval', 'Eggs': 'redval'})
  658. othercid = CaseInsensitiveDict({'spam': 'blueval', 'eggs': 'redval'})
  659. self.assertEqual(cid, othercid)
  660. del othercid['spam']
  661. self.assertNotEqual(cid, othercid)
  662. self.assertEqual(cid, {'spam': 'blueval', 'eggs': 'redval'})
  663. def test_setdefault(self):
  664. cid = CaseInsensitiveDict({'Spam': 'blueval'})
  665. self.assertEqual(
  666. cid.setdefault('spam', 'notblueval'),
  667. 'blueval'
  668. )
  669. self.assertEqual(
  670. cid.setdefault('notspam', 'notblueval'),
  671. 'notblueval'
  672. )
  673. def test_lower_items(self):
  674. cid = CaseInsensitiveDict({
  675. 'Accept': 'application/json',
  676. 'user-Agent': 'requests',
  677. })
  678. keyset = frozenset(lowerkey for lowerkey, v in cid.lower_items())
  679. lowerkeyset = frozenset(['accept', 'user-agent'])
  680. self.assertEqual(keyset, lowerkeyset)
  681. def test_preserve_key_case(self):
  682. cid = CaseInsensitiveDict({
  683. 'Accept': 'application/json',
  684. 'user-Agent': 'requests',
  685. })
  686. keyset = frozenset(['Accept', 'user-Agent'])
  687. self.assertEqual(frozenset(i[0] for i in cid.items()), keyset)
  688. self.assertEqual(frozenset(cid.keys()), keyset)
  689. self.assertEqual(frozenset(cid), keyset)
  690. def test_preserve_last_key_case(self):
  691. cid = CaseInsensitiveDict({
  692. 'Accept': 'application/json',
  693. 'user-Agent': 'requests',
  694. })
  695. cid.update({'ACCEPT': 'application/json'})
  696. cid['USER-AGENT'] = 'requests'
  697. keyset = frozenset(['ACCEPT', 'USER-AGENT'])
  698. self.assertEqual(frozenset(i[0] for i in cid.items()), keyset)
  699. self.assertEqual(frozenset(cid.keys()), keyset)
  700. self.assertEqual(frozenset(cid), keyset)
  701. if __name__ == '__main__':
  702. unittest.main()