1# -*- coding: utf-8 -*- 2'''Integration tests with httplib2''' 3 4import sys 5 6from six.moves.urllib_parse import urlencode 7import pytest 8import pytest_httpbin.certs 9 10import vcr 11 12from assertions import assert_cassette_has_one_response 13 14httplib2 = pytest.importorskip("httplib2") 15 16 17def http(): 18 """ 19 Returns an httplib2 HTTP instance 20 with the certificate replaced by the httpbin one. 21 """ 22 kwargs = { 23 'ca_certs': pytest_httpbin.certs.where() 24 } 25 if sys.version_info[:2] == (3, 7): 26 kwargs['disable_ssl_certificate_validation'] = True 27 return httplib2.Http(**kwargs) 28 29 30def test_response_code(tmpdir, httpbin_both): 31 '''Ensure we can read a response code from a fetch''' 32 url = httpbin_both.url 33 with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): 34 resp, _ = http().request(url) 35 code = resp.status 36 37 with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): 38 resp, _ = http().request(url) 39 assert code == resp.status 40 41 42def test_random_body(httpbin_both, tmpdir): 43 '''Ensure we can read the content, and that it's served from cache''' 44 url = httpbin_both.url + '/bytes/1024' 45 with vcr.use_cassette(str(tmpdir.join('body.yaml'))): 46 _, content = http().request(url) 47 body = content 48 49 with vcr.use_cassette(str(tmpdir.join('body.yaml'))): 50 _, content = http().request(url) 51 assert body == content 52 53 54def test_response_headers(tmpdir, httpbin_both): 55 '''Ensure we can get information from the response''' 56 url = httpbin_both.url 57 with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): 58 resp, _ = http().request(url) 59 headers = resp.items() 60 61 with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): 62 resp, _ = http().request(url) 63 assert set(headers) == set(resp.items()) 64 65 66def test_effective_url(tmpdir, httpbin_both): 67 '''Ensure that the effective_url is captured''' 68 url = httpbin_both.url + '/redirect-to?url=/html' 69 with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): 70 resp, _ = http().request(url) 71 effective_url = resp['content-location'] 72 assert effective_url == httpbin_both + '/html' 73 74 with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): 75 resp, _ = http().request(url) 76 assert effective_url == resp['content-location'] 77 78 79def test_multiple_requests(tmpdir, httpbin_both): 80 '''Ensure that we can cache multiple requests''' 81 urls = [ 82 httpbin_both.url, 83 httpbin_both.url, 84 httpbin_both.url + '/get', 85 httpbin_both.url + '/bytes/1024', 86 ] 87 with vcr.use_cassette(str(tmpdir.join('multiple.yaml'))) as cass: 88 [http().request(url) for url in urls] 89 assert len(cass) == len(urls) 90 91 92def test_get_data(tmpdir, httpbin_both): 93 '''Ensure that it works with query data''' 94 data = urlencode({'some': 1, 'data': 'here'}) 95 url = httpbin_both.url + '/get?' + data 96 with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))): 97 _, res1 = http().request(url) 98 99 with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))): 100 _, res2 = http().request(url) 101 102 assert res1 == res2 103 104 105def test_post_data(tmpdir, httpbin_both): 106 '''Ensure that it works when posting data''' 107 data = urlencode({'some': 1, 'data': 'here'}) 108 url = httpbin_both.url + '/post' 109 with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))): 110 _, res1 = http().request(url, "POST", data) 111 112 with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass: 113 _, res2 = http().request(url, "POST", data) 114 115 assert res1 == res2 116 assert_cassette_has_one_response(cass) 117 118 119def test_post_unicode_data(tmpdir, httpbin_both): 120 '''Ensure that it works when posting unicode data''' 121 data = urlencode({'snowman': u'☃'.encode('utf-8')}) 122 url = httpbin_both.url + '/post' 123 with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))): 124 _, res1 = http().request(url, "POST", data) 125 126 with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass: 127 _, res2 = http().request(url, "POST", data) 128 129 assert res1 == res2 130 assert_cassette_has_one_response(cass) 131 132 133def test_cross_scheme(tmpdir, httpbin, httpbin_secure): 134 '''Ensure that requests between schemes are treated separately''' 135 # First fetch a url under https, and then again under https and then 136 # ensure that we haven't served anything out of cache, and we have two 137 # requests / response pairs in the cassette 138 with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass: 139 http().request(httpbin_secure.url) 140 http().request(httpbin.url) 141 assert len(cass) == 2 142 assert cass.play_count == 0 143 144 145def test_decorator(tmpdir, httpbin_both): 146 '''Test the decorator version of VCR.py''' 147 url = httpbin_both.url 148 149 @vcr.use_cassette(str(tmpdir.join('atts.yaml'))) 150 def inner1(): 151 resp, _ = http().request(url) 152 return resp['status'] 153 154 @vcr.use_cassette(str(tmpdir.join('atts.yaml'))) 155 def inner2(): 156 resp, _ = http().request(url) 157 return resp['status'] 158 159 assert inner1() == inner2() 160