diff --git a/runtests.sh b/runtests.sh new file mode 100755 index 0000000..6779517 --- /dev/null +++ b/runtests.sh @@ -0,0 +1,3 @@ +#!/bin/bash + +REQUESTS_CA_BUNDLE=`python -m pytest_httpbin.certs` py.test $1 diff --git a/setup.py b/setup.py index 08c2fb4..90fb47a 100644 --- a/setup.py +++ b/setup.py @@ -64,8 +64,7 @@ setup( install_requires=install_requires, extras_require=extras_require, license='MIT', - tests_require=['pytest', 'mock', 'pytest-localserver'], - cmdclass={'test': PyTest}, + tests_require=['pytest', 'mock', 'pytest-localserver', 'pytest-httpbin'], classifiers=[ 'Development Status :: 4 - Beta', 'Environment :: Console', diff --git a/tests/integration/test_basic.py b/tests/integration/test_basic.py index 9d4d6d9..29b1854 100644 --- a/tests/integration/test_basic.py +++ b/tests/integration/test_basic.py @@ -9,74 +9,63 @@ from six.moves.urllib.request import urlopen import vcr -def test_nonexistent_directory(tmpdir): +def test_nonexistent_directory(tmpdir, httpbin): '''If we load a cassette in a nonexistent directory, it can save ok''' # Check to make sure directory doesnt exist assert not os.path.exists(str(tmpdir.join('nonexistent'))) # Run VCR to create dir and cassette file with vcr.use_cassette(str(tmpdir.join('nonexistent', 'cassette.yml'))): - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() # This should have made the file and the directory assert os.path.exists(str(tmpdir.join('nonexistent', 'cassette.yml'))) -def test_unpatch(tmpdir): +def test_unpatch(tmpdir, httpbin): '''Ensure that our cassette gets unpatched when we're done''' with vcr.use_cassette(str(tmpdir.join('unpatch.yaml'))) as cass: - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() # Make the same request, and assert that we haven't served any more # requests out of cache - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() assert cass.play_count == 0 -def test_basic_use(tmpdir): - ''' - Copied from the docs - ''' - with vcr.use_cassette('fixtures/vcr_cassettes/synopsis.yaml'): - response = urlopen( - 'http://www.iana.org/domains/reserved' - ).read() - assert b'Example domains' in response - - -def test_basic_json_use(tmpdir): +def test_basic_json_use(tmpdir, httpbin): ''' Ensure you can load a json serialized cassette ''' - test_fixture = 'fixtures/vcr_cassettes/synopsis.json' + test_fixture = str(tmpdir.join('synopsis.json')) with vcr.use_cassette(test_fixture, serializer='json'): - response = urlopen('http://httpbin.org/').read() + response = urlopen(httpbin.url).read() assert b'difficult sometimes' in response -def test_patched_content(tmpdir): +def test_patched_content(tmpdir, httpbin): ''' Ensure that what you pull from a cassette is what came from the request ''' with vcr.use_cassette(str(tmpdir.join('synopsis.yaml'))) as cass: - response = urlopen('http://httpbin.org/').read() + response = urlopen(httpbin.url).read() assert cass.play_count == 0 with vcr.use_cassette(str(tmpdir.join('synopsis.yaml'))) as cass: - response2 = urlopen('http://httpbin.org/').read() + response2 = urlopen(httpbin.url).read() assert cass.play_count == 1 cass._save(force=True) with vcr.use_cassette(str(tmpdir.join('synopsis.yaml'))) as cass: - response3 = urlopen('http://httpbin.org/').read() + response3 = urlopen(httpbin.url).read() assert cass.play_count == 1 assert response == response2 assert response2 == response3 -def test_patched_content_json(tmpdir): +def test_patched_content_json(tmpdir, httpbin): ''' Ensure that what you pull from a json cassette is what came from the request @@ -85,16 +74,16 @@ def test_patched_content_json(tmpdir): testfile = str(tmpdir.join('synopsis.json')) with vcr.use_cassette(testfile) as cass: - response = urlopen('http://httpbin.org/').read() + response = urlopen(httpbin.url).read() assert cass.play_count == 0 with vcr.use_cassette(testfile) as cass: - response2 = urlopen('http://httpbin.org/').read() + response2 = urlopen(httpbin.url).read() assert cass.play_count == 1 cass._save(force=True) with vcr.use_cassette(testfile) as cass: - response3 = urlopen('http://httpbin.org/').read() + response3 = urlopen(httpbin.url).read() assert cass.play_count == 1 assert response == response2 diff --git a/tests/integration/test_config.py b/tests/integration/test_config.py index d58e32d..1ce07be 100644 --- a/tests/integration/test_config.py +++ b/tests/integration/test_config.py @@ -5,46 +5,46 @@ import vcr from six.moves.urllib.request import urlopen -def test_set_serializer_default_config(tmpdir): +def test_set_serializer_default_config(tmpdir, httpbin): my_vcr = vcr.VCR(serializer='json') with my_vcr.use_cassette(str(tmpdir.join('test.json'))): assert my_vcr.serializer == 'json' - urlopen('http://httpbin.org/get') + urlopen(httpbin.url + '/get') with open(str(tmpdir.join('test.json'))) as f: assert json.loads(f.read()) -def test_default_set_cassette_library_dir(tmpdir): +def test_default_set_cassette_library_dir(tmpdir, httpbin): my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join('subdir'))) with my_vcr.use_cassette('test.json'): - urlopen('http://httpbin.org/get') + urlopen(httpbin.url + '/get') assert os.path.exists(str(tmpdir.join('subdir').join('test.json'))) -def test_override_set_cassette_library_dir(tmpdir): +def test_override_set_cassette_library_dir(tmpdir, httpbin): my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join('subdir'))) cld = str(tmpdir.join('subdir2')) with my_vcr.use_cassette('test.json', cassette_library_dir=cld): - urlopen('http://httpbin.org/get') + urlopen(httpbin.url + '/get') assert os.path.exists(str(tmpdir.join('subdir2').join('test.json'))) assert not os.path.exists(str(tmpdir.join('subdir').join('test.json'))) -def test_override_match_on(tmpdir): +def test_override_match_on(tmpdir, httpbin): my_vcr = vcr.VCR(match_on=['method']) with my_vcr.use_cassette(str(tmpdir.join('test.json'))): - urlopen('http://httpbin.org/') + urlopen(httpbin.url) with my_vcr.use_cassette(str(tmpdir.join('test.json'))) as cass: - urlopen('http://httpbin.org/get') + urlopen(httpbin.url + '/get') assert len(cass) == 1 assert cass.play_count == 1 diff --git a/tests/integration/test_disksaver.py b/tests/integration/test_disksaver.py index 0f853b9..ffca2fc 100644 --- a/tests/integration/test_disksaver.py +++ b/tests/integration/test_disksaver.py @@ -10,19 +10,19 @@ from six.moves.urllib.request import urlopen import vcr -def test_disk_saver_nowrite(tmpdir): +def test_disk_saver_nowrite(tmpdir, httpbin): ''' Ensure that when you close a cassette without changing it it doesn't rewrite the file ''' fname = str(tmpdir.join('synopsis.yaml')) with vcr.use_cassette(fname) as cass: - urlopen('http://www.iana.org/domains/reserved').read() + urlopen(httpbin.url).read() assert cass.play_count == 0 last_mod = os.path.getmtime(fname) with vcr.use_cassette(fname) as cass: - urlopen('http://www.iana.org/domains/reserved').read() + urlopen(httpbin.url).read() assert cass.play_count == 1 assert cass.dirty is False last_mod2 = os.path.getmtime(fname) @@ -30,14 +30,14 @@ def test_disk_saver_nowrite(tmpdir): assert last_mod == last_mod2 -def test_disk_saver_write(tmpdir): +def test_disk_saver_write(tmpdir, httpbin): ''' Ensure that when you close a cassette after changing it it does rewrite the file ''' fname = str(tmpdir.join('synopsis.yaml')) with vcr.use_cassette(fname) as cass: - urlopen('http://www.iana.org/domains/reserved').read() + urlopen(httpbin.url).read() assert cass.play_count == 0 last_mod = os.path.getmtime(fname) @@ -46,8 +46,8 @@ def test_disk_saver_write(tmpdir): time.sleep(1) with vcr.use_cassette(fname, record_mode='any') as cass: - urlopen('http://www.iana.org/domains/reserved').read() - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() + urlopen(httpbin.url + '/get').read() assert cass.play_count == 1 assert cass.dirty last_mod2 = os.path.getmtime(fname) diff --git a/tests/integration/test_filter.py b/tests/integration/test_filter.py index a5dc3a6..ccdc94f 100644 --- a/tests/integration/test_filter.py +++ b/tests/integration/test_filter.py @@ -21,8 +21,8 @@ def _find_header(cassette, header): return any(header in request.headers for request in cassette.requests) -def test_filter_basic_auth(tmpdir): - url = 'http://httpbin.org/basic-auth/user/passwd' +def test_filter_basic_auth(tmpdir, httpbin): + url = httpbin.url + '/basic-auth/user/passwd' cass_file = str(tmpdir.join('basic_auth_filter.yaml')) my_vcr = vcr.VCR(match_on=['uri', 'method', 'headers']) # 2 requests, one with auth failure and one with auth success @@ -44,8 +44,8 @@ def test_filter_basic_auth(tmpdir): assert len(cass) == 2 -def test_filter_querystring(tmpdir): - url = 'http://httpbin.org/?foo=bar' +def test_filter_querystring(tmpdir, httpbin): + url = httpbin.url + '/?foo=bar' cass_file = str(tmpdir.join('filter_qs.yaml')) with vcr.use_cassette(cass_file, filter_query_parameters=['foo']): urlopen(url) @@ -54,8 +54,8 @@ def test_filter_querystring(tmpdir): assert 'foo' not in cass.requests[0].url -def test_filter_post_data(tmpdir): - url = 'http://httpbin.org/post' +def test_filter_post_data(tmpdir, httpbin): + url = httpbin.url + '/post' data = urlencode({'id': 'secret', 'foo': 'bar'}).encode('utf-8') cass_file = str(tmpdir.join('filter_pd.yaml')) with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']): @@ -64,9 +64,9 @@ def test_filter_post_data(tmpdir): assert b'id=secret' not in cass.requests[0].body -def test_filter_json_post_data(tmpdir): +def test_filter_json_post_data(tmpdir, httpbin): data = json.dumps({'id': 'secret', 'foo': 'bar'}).encode('utf-8') - request = Request('http://httpbin.org/post', data=data) + request = Request(httpbin.url + '/post', data=data) request.add_header('Content-Type', 'application/json') cass_file = str(tmpdir.join('filter_jpd.yaml')) @@ -76,8 +76,8 @@ def test_filter_json_post_data(tmpdir): assert b'"id": "secret"' not in cass.requests[0].body -def test_filter_callback(tmpdir): - url = 'http://httpbin.org/get' +def test_filter_callback(tmpdir, httpbin): + url = httpbin.url + '/get' cass_file = str(tmpdir.join('basic_auth_filter.yaml')) def before_record_cb(request): @@ -96,8 +96,8 @@ def test_filter_callback(tmpdir): assert len(cass) == 0 -def test_decompress_gzip(tmpdir): - url = 'http://httpbin.org/gzip' +def test_decompress_gzip(tmpdir, httpbin): + url = httpbin.url + '/gzip' request = Request(url, headers={'Accept-Encoding': ['gzip, deflate']}) cass_file = str(tmpdir.join('gzip_response.yaml')) with vcr.use_cassette(cass_file, decode_compressed_response=True): @@ -108,8 +108,8 @@ def test_decompress_gzip(tmpdir): assert_is_json(decoded_response) -def test_decompress_deflate(tmpdir): - url = 'http://httpbin.org/deflate' +def test_decompress_deflate(tmpdir, httpbin): + url = httpbin.url + '/deflate' request = Request(url, headers={'Accept-Encoding': ['gzip, deflate']}) cass_file = str(tmpdir.join('deflate_response.yaml')) with vcr.use_cassette(cass_file, decode_compressed_response=True): @@ -120,9 +120,9 @@ def test_decompress_deflate(tmpdir): assert_is_json(decoded_response) -def test_decompress_regular(tmpdir): +def test_decompress_regular(tmpdir, httpbin): """Test that it doesn't try to decompress content that isn't compressed""" - url = 'http://httpbin.org/get' + url = httpbin.url + '/get' cass_file = str(tmpdir.join('noncompressed_response.yaml')) with vcr.use_cassette(cass_file, decode_compressed_response=True): urlopen(url) diff --git a/tests/integration/test_httplib2.py b/tests/integration/test_httplib2.py index eaa5d75..05a25d3 100644 --- a/tests/integration/test_httplib2.py +++ b/tests/integration/test_httplib2.py @@ -4,6 +4,7 @@ # External imports from six.moves.urllib_parse import urlencode import pytest +import pytest_httpbin.certs # Internal imports import vcr @@ -13,141 +14,141 @@ from assertions import assert_cassette_has_one_response httplib2 = pytest.importorskip("httplib2") -@pytest.fixture(params=["https", "http"]) -def scheme(request): +def http(): """ - Fixture that returns both http and https + Returns an httplib2 HTTP instance + with the certificate replaced by the httpbin one. """ - return request.param + return httplib2.Http(ca_certs=pytest_httpbin.certs.where()) -def test_response_code(scheme, tmpdir): +def test_response_code(tmpdir, httpbin_both): '''Ensure we can read a response code from a fetch''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) code = resp.status with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) assert code == resp.status -def test_random_body(scheme, tmpdir): +def test_random_body(httpbin_both, tmpdir): '''Ensure we can read the content, and that it's served from cache''' - url = scheme + '://httpbin.org/bytes/1024' + url = httpbin_both.url + '/bytes/1024' with vcr.use_cassette(str(tmpdir.join('body.yaml'))): - _, content = httplib2.Http().request(url) + _, content = http().request(url) body = content with vcr.use_cassette(str(tmpdir.join('body.yaml'))): - _, content = httplib2.Http().request(url) + _, content = http().request(url) assert body == content -def test_response_headers(scheme, tmpdir): +def test_response_headers(tmpdir, httpbin_both): '''Ensure we can get information from the response''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) headers = resp.items() with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) assert set(headers) == set(resp.items()) -def test_effective_url(scheme, tmpdir): +def test_effective_url(tmpdir, httpbin_both): '''Ensure that the effective_url is captured''' - url = scheme + '://httpbin.org/redirect-to?url=/html' + url = httpbin_both.url + '/redirect-to?url=/html' with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) effective_url = resp['content-location'] - assert effective_url == scheme + '://httpbin.org/html' + assert effective_url == httpbin_both + '/html' with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) assert effective_url == resp['content-location'] -def test_multiple_requests(scheme, tmpdir): +def test_multiple_requests(tmpdir, httpbin_both): '''Ensure that we can cache multiple requests''' urls = [ - scheme + '://httpbin.org/', - scheme + '://httpbin.org/', - scheme + '://httpbin.org/get', - scheme + '://httpbin.org/bytes/1024' + httpbin_both.url, + httpbin_both.url, + httpbin_both.url + '/get', + httpbin_both.url + '/bytes/1024', ] with vcr.use_cassette(str(tmpdir.join('multiple.yaml'))) as cass: - [httplib2.Http().request(url) for url in urls] + [http().request(url) for url in urls] assert len(cass) == len(urls) -def test_get_data(scheme, tmpdir): +def test_get_data(tmpdir, httpbin_both): '''Ensure that it works with query data''' data = urlencode({'some': 1, 'data': 'here'}) - url = scheme + '://httpbin.org/get?' + data + url = httpbin_both.url + '/get?' + data with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))): - _, res1 = httplib2.Http().request(url) + _, res1 = http().request(url) with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))): - _, res2 = httplib2.Http().request(url) + _, res2 = http().request(url) assert res1 == res2 -def test_post_data(scheme, tmpdir): +def test_post_data(tmpdir, httpbin_both): '''Ensure that it works when posting data''' data = urlencode({'some': 1, 'data': 'here'}) - url = scheme + '://httpbin.org/post' + url = httpbin_both.url + '/post' with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))): - _, res1 = httplib2.Http().request(url, "POST", data) + _, res1 = http().request(url, "POST", data) with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass: - _, res2 = httplib2.Http().request(url, "POST", data) + _, res2 = http().request(url, "POST", data) assert res1 == res2 assert_cassette_has_one_response(cass) -def test_post_unicode_data(scheme, tmpdir): +def test_post_unicode_data(tmpdir, httpbin_both): '''Ensure that it works when posting unicode data''' data = urlencode({'snowman': u'☃'.encode('utf-8')}) - url = scheme + '://httpbin.org/post' + url = httpbin_both.url + '/post' with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))): - _, res1 = httplib2.Http().request(url, "POST", data) + _, res1 = http().request(url, "POST", data) with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass: - _, res2 = httplib2.Http().request(url, "POST", data) + _, res2 = http().request(url, "POST", data) assert res1 == res2 assert_cassette_has_one_response(cass) -def test_cross_scheme(tmpdir): +def test_cross_scheme(tmpdir, httpbin, httpbin_secure): '''Ensure that requests between schemes are treated separately''' # First fetch a url under https, and then again under https and then # ensure that we haven't served anything out of cache, and we have two # requests / response pairs in the cassette with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass: - httplib2.Http().request('https://httpbin.org/') - httplib2.Http().request('http://httpbin.org/') + http().request(httpbin_secure.url) + http().request(httpbin.url) assert len(cass) == 2 assert cass.play_count == 0 -def test_decorator(scheme, tmpdir): +def test_decorator(tmpdir, httpbin_both): '''Test the decorator version of VCR.py''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url @vcr.use_cassette(str(tmpdir.join('atts.yaml'))) def inner1(): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) return resp['status'] @vcr.use_cassette(str(tmpdir.join('atts.yaml'))) def inner2(): - resp, _ = httplib2.Http().request(url) + resp, _ = http().request(url) return resp['status'] assert inner1() == inner2() diff --git a/tests/integration/test_ignore.py b/tests/integration/test_ignore.py index d32682f..164edb8 100644 --- a/tests/integration/test_ignore.py +++ b/tests/integration/test_ignore.py @@ -1,54 +1,73 @@ from six.moves.urllib.request import urlopen - +import socket +from contextlib import contextmanager import vcr -def test_ignore_localhost(tmpdir, httpserver): - httpserver.serve_content('Hello!') - cass_file = str(tmpdir.join('filter_qs.yaml')) - with vcr.use_cassette(cass_file, ignore_localhost=True) as cass: - urlopen(httpserver.url) - assert len(cass) == 0 - urlopen('http://httpbin.org') - assert len(cass) == 1 +@contextmanager +def overridden_dns(overrides): + """ + Monkeypatch socket.getaddrinfo() to override DNS lookups (name will resolve + to address) + """ + real_getaddrinfo = socket.getaddrinfo + + def fake_getaddrinfo(*args, **kwargs): + if args[0] in overrides: + address = overrides[args[0]] + return [(2, 1, 6, '', (address, args[1]))] + return real_getaddrinfo(*args, **kwargs) + socket.getaddrinfo = fake_getaddrinfo + yield + socket.getaddrinfo = real_getaddrinfo -def test_ignore_httpbin(tmpdir, httpserver): - httpserver.serve_content('Hello!') - cass_file = str(tmpdir.join('filter_qs.yaml')) - with vcr.use_cassette( - cass_file, - ignore_hosts=['httpbin.org'] - ) as cass: - urlopen('http://httpbin.org') - assert len(cass) == 0 - urlopen(httpserver.url) - assert len(cass) == 1 +def test_ignore_localhost(tmpdir, httpbin): + with overridden_dns({'httpbin.org': '127.0.0.1'}): + cass_file = str(tmpdir.join('filter_qs.yaml')) + with vcr.use_cassette(cass_file, ignore_localhost=True) as cass: + urlopen('http://localhost:{0}/'.format(httpbin.port)) + assert len(cass) == 0 + urlopen('http://httpbin.org:{0}/'.format(httpbin.port)) + assert len(cass) == 1 -def test_ignore_localhost_and_httpbin(tmpdir, httpserver): - httpserver.serve_content('Hello!') - cass_file = str(tmpdir.join('filter_qs.yaml')) - with vcr.use_cassette( - cass_file, - ignore_hosts=['httpbin.org'], - ignore_localhost=True - ) as cass: - urlopen('http://httpbin.org') - urlopen(httpserver.url) - assert len(cass) == 0 +def test_ignore_httpbin(tmpdir, httpbin): + with overridden_dns({'httpbin.org': '127.0.0.1'}): + cass_file = str(tmpdir.join('filter_qs.yaml')) + with vcr.use_cassette( + cass_file, + ignore_hosts=['httpbin.org'] + ) as cass: + urlopen('http://httpbin.org:{0}/'.format(httpbin.port)) + assert len(cass) == 0 + urlopen('http://localhost:{0}/'.format(httpbin.port)) + assert len(cass) == 1 -def test_ignore_localhost_twice(tmpdir, httpserver): - httpserver.serve_content('Hello!') - cass_file = str(tmpdir.join('filter_qs.yaml')) - with vcr.use_cassette(cass_file, ignore_localhost=True) as cass: - urlopen(httpserver.url) - assert len(cass) == 0 - urlopen('http://httpbin.org') - assert len(cass) == 1 - with vcr.use_cassette(cass_file, ignore_localhost=True) as cass: - assert len(cass) == 1 - urlopen(httpserver.url) - urlopen('http://httpbin.org') - assert len(cass) == 1 +def test_ignore_localhost_and_httpbin(tmpdir, httpbin): + with overridden_dns({'httpbin.org': '127.0.0.1'}): + cass_file = str(tmpdir.join('filter_qs.yaml')) + with vcr.use_cassette( + cass_file, + ignore_hosts=['httpbin.org'], + ignore_localhost=True + ) as cass: + urlopen('http://httpbin.org:{0}'.format(httpbin.port)) + urlopen('http://localhost:{0}'.format(httpbin.port)) + assert len(cass) == 0 + + +def test_ignore_localhost_twice(tmpdir, httpbin): + with overridden_dns({'httpbin.org': '127.0.0.1'}): + cass_file = str(tmpdir.join('filter_qs.yaml')) + with vcr.use_cassette(cass_file, ignore_localhost=True) as cass: + urlopen('http://localhost:{0}'.format(httpbin.port)) + assert len(cass) == 0 + urlopen('http://httpbin.org:{0}'.format(httpbin.port)) + assert len(cass) == 1 + with vcr.use_cassette(cass_file, ignore_localhost=True) as cass: + assert len(cass) == 1 + urlopen('http://localhost:{0}'.format(httpbin.port)) + urlopen('http://httpbin.org:{0}'.format(httpbin.port)) + assert len(cass) == 1 diff --git a/tests/integration/test_matchers.py b/tests/integration/test_matchers.py index 396bb40..44cdeb5 100644 --- a/tests/integration/test_matchers.py +++ b/tests/integration/test_matchers.py @@ -6,15 +6,21 @@ from six.moves.urllib.request import urlopen DEFAULT_URI = 'http://httpbin.org/get?p1=q1&p2=q2' # base uri for testing +def _replace_httpbin(uri, httpbin, httpbin_secure): + return uri.replace('http://httpbin.org', httpbin.url).replace('https://httpbin.org', httpbin_secure.url) + + @pytest.fixture -def cassette(tmpdir): +def cassette(tmpdir, httpbin, httpbin_secure): """ Helper fixture used to prepare the cassete returns path to the recorded cassette """ + default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure) + cassette_path = str(tmpdir.join('test.yml')) with vcr.use_cassette(cassette_path, record_mode='all'): - urlopen(DEFAULT_URI) + urlopen(default_uri) return cassette_path @@ -28,9 +34,6 @@ def cassette(tmpdir): ('host', 'https://httpbin.org/post?a=b', 'http://google.com/get?p1=q1&p2=q2'), - ('port', - 'https://google.com:80/post?a=b', - 'http://httpbin.org:5000/get?p1=q1&p2=q2'), ('path', 'https://google.com/get?a=b', 'http://httpbin.org/post?p1=q1&p2=q2'), @@ -38,10 +41,15 @@ def cassette(tmpdir): 'https://google.com/get?p2=q2&p1=q1', 'http://httpbin.org/get?p1=q1&a=b') ]) -def test_matchers(cassette, matcher, matching_uri, not_matching_uri): +def test_matchers(httpbin, httpbin_secure, cassette, matcher, matching_uri, not_matching_uri): + + matching_uri = _replace_httpbin(matching_uri, httpbin, httpbin_secure) + not_matching_uri = _replace_httpbin(not_matching_uri, httpbin, httpbin_secure) + default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure) + # play cassette with default uri with vcr.use_cassette(cassette, match_on=[matcher]) as cass: - urlopen(DEFAULT_URI) + urlopen(default_uri) assert cass.play_count == 1 # play cassette with matching on uri @@ -55,7 +63,9 @@ def test_matchers(cassette, matcher, matching_uri, not_matching_uri): urlopen(not_matching_uri) -def test_method_matcher(cassette): +def test_method_matcher(cassette, httpbin, httpbin_secure): + default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure) + # play cassette with matching on method with vcr.use_cassette(cassette, match_on=['method']) as cass: urlopen('https://google.com/get?a=b') @@ -65,7 +75,7 @@ def test_method_matcher(cassette): with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException): with vcr.use_cassette(cassette, match_on=['method']) as cass: # is a POST request - urlopen(DEFAULT_URI, data=b'') + urlopen(default_uri, data=b'') @pytest.mark.parametrize("uri", [ @@ -73,7 +83,10 @@ def test_method_matcher(cassette): 'http://httpbin.org/get?p2=q2&p1=q1', 'http://httpbin.org/get?p2=q2&p1=q1', ]) -def test_default_matcher_matches(cassette, uri): +def test_default_matcher_matches(cassette, uri, httpbin, httpbin_secure): + + uri = _replace_httpbin(uri, httpbin, httpbin_secure) + with vcr.use_cassette(cassette) as cass: urlopen(uri) assert cass.play_count == 1 @@ -82,18 +95,19 @@ def test_default_matcher_matches(cassette, uri): @pytest.mark.parametrize("uri", [ 'https://httpbin.org/get?p1=q1&p2=q2', 'http://google.com/get?p1=q1&p2=q2', - 'http://httpbin.org:5000/get?p1=q1&p2=q2', 'http://httpbin.org/post?p1=q1&p2=q2', 'http://httpbin.org/get?p1=q1&a=b' ]) -def test_default_matcher_does_not_match(cassette, uri): +def test_default_matcher_does_not_match(cassette, uri, httpbin, httpbin_secure): + uri = _replace_httpbin(uri, httpbin, httpbin_secure) with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException): with vcr.use_cassette(cassette): urlopen(uri) -def test_default_matcher_does_not_match_on_method(cassette): +def test_default_matcher_does_not_match_on_method(cassette, httpbin, httpbin_secure): + default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure) with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException): with vcr.use_cassette(cassette): # is a POST request - urlopen(DEFAULT_URI, data=b'') + urlopen(default_uri, data=b'') diff --git a/tests/integration/test_multiple.py b/tests/integration/test_multiple.py index db0f2da..fbcedda 100644 --- a/tests/integration/test_multiple.py +++ b/tests/integration/test_multiple.py @@ -3,18 +3,18 @@ import vcr from six.moves.urllib.request import urlopen -def test_making_extra_request_raises_exception(tmpdir): +def test_making_extra_request_raises_exception(tmpdir, httpbin): # make two requests in the first request that are considered # identical (since the match is based on method) with vcr.use_cassette(str(tmpdir.join('test.json')), match_on=['method']): - urlopen('http://httpbin.org/status/200') - urlopen('http://httpbin.org/status/201') + urlopen(httpbin.url + '/status/200') + urlopen(httpbin.url + '/status/201') # Now, try to make three requests. The first two should return the # correct status codes in order, and the third should raise an # exception. with vcr.use_cassette(str(tmpdir.join('test.json')), match_on=['method']): - assert urlopen('http://httpbin.org/status/200').getcode() == 200 - assert urlopen('http://httpbin.org/status/201').getcode() == 201 + assert urlopen(httpbin.url + '/status/200').getcode() == 200 + assert urlopen(httpbin.url + '/status/201').getcode() == 201 with pytest.raises(Exception): - urlopen('http://httpbin.org/status/200') + urlopen(httpbin.url + '/status/200') diff --git a/tests/integration/test_record_mode.py b/tests/integration/test_record_mode.py index 5d9b603..f697bcc 100644 --- a/tests/integration/test_record_mode.py +++ b/tests/integration/test_record_mode.py @@ -3,62 +3,62 @@ import vcr from six.moves.urllib.request import urlopen -def test_once_record_mode(tmpdir): +def test_once_record_mode(tmpdir, httpbin): testfile = str(tmpdir.join('recordmode.yml')) with vcr.use_cassette(testfile, record_mode="once"): # cassette file doesn't exist, so create. - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() with vcr.use_cassette(testfile, record_mode="once"): # make the same request again - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() # the first time, it's played from the cassette. # but, try to access something else from the same cassette, and an # exception is raised. with pytest.raises(Exception): - urlopen('http://httpbin.org/get').read() + urlopen(httpbin.url + '/get').read() -def test_once_record_mode_two_times(tmpdir): +def test_once_record_mode_two_times(tmpdir, httpbin): testfile = str(tmpdir.join('recordmode.yml')) with vcr.use_cassette(testfile, record_mode="once"): # get two of the same file - urlopen('http://httpbin.org/').read() - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() + urlopen(httpbin.url).read() with vcr.use_cassette(testfile, record_mode="once"): # do it again - urlopen('http://httpbin.org/').read() - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() + urlopen(httpbin.url).read() -def test_once_mode_three_times(tmpdir): +def test_once_mode_three_times(tmpdir, httpbin): testfile = str(tmpdir.join('recordmode.yml')) with vcr.use_cassette(testfile, record_mode="once"): # get three of the same file - urlopen('http://httpbin.org/').read() - urlopen('http://httpbin.org/').read() - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() + urlopen(httpbin.url).read() + urlopen(httpbin.url).read() -def test_new_episodes_record_mode(tmpdir): +def test_new_episodes_record_mode(tmpdir, httpbin): testfile = str(tmpdir.join('recordmode.yml')) with vcr.use_cassette(testfile, record_mode="new_episodes"): # cassette file doesn't exist, so create. - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() with vcr.use_cassette(testfile, record_mode="new_episodes") as cass: # make the same request again - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() # all responses have been played assert cass.all_played # in the "new_episodes" record mode, we can add more requests to # a cassette without repurcussions. - urlopen('http://httpbin.org/get').read() + urlopen(httpbin.url + '/get').read() # one of the responses has been played assert cass.play_count == 1 @@ -71,9 +71,9 @@ def test_new_episodes_record_mode(tmpdir): assert len(cass.responses) == 2 -def test_new_episodes_record_mode_two_times(tmpdir): +def test_new_episodes_record_mode_two_times(tmpdir, httpbin): testfile = str(tmpdir.join('recordmode.yml')) - url = 'http://httpbin.org/bytes/1024' + url = httpbin.url + '/bytes/1024' with vcr.use_cassette(testfile, record_mode="new_episodes"): # cassette file doesn't exist, so create. original_first_response = urlopen(url).read() @@ -96,12 +96,12 @@ def test_new_episodes_record_mode_two_times(tmpdir): urlopen(url).read() -def test_all_record_mode(tmpdir): +def test_all_record_mode(tmpdir, httpbin): testfile = str(tmpdir.join('recordmode.yml')) with vcr.use_cassette(testfile, record_mode="all"): # cassette file doesn't exist, so create. - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() with vcr.use_cassette(testfile, record_mode="all") as cass: # make the same request again @@ -109,7 +109,7 @@ def test_all_record_mode(tmpdir): # in the "all" record mode, we can add more requests to # a cassette without repurcussions. - urlopen('http://httpbin.org/get').read() + urlopen(httpbin.url + '/get').read() # The cassette was never actually played, even though it existed. # that's because, in "all" mode, the requests all go directly to @@ -117,26 +117,26 @@ def test_all_record_mode(tmpdir): assert cass.play_count == 0 -def test_none_record_mode(tmpdir): +def test_none_record_mode(tmpdir, httpbin): # Cassette file doesn't exist, yet we are trying to make a request. # raise hell. testfile = str(tmpdir.join('recordmode.yml')) with vcr.use_cassette(testfile, record_mode="none"): with pytest.raises(Exception): - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() -def test_none_record_mode_with_existing_cassette(tmpdir): +def test_none_record_mode_with_existing_cassette(tmpdir, httpbin): # create a cassette file testfile = str(tmpdir.join('recordmode.yml')) with vcr.use_cassette(testfile, record_mode="all"): - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() # play from cassette file with vcr.use_cassette(testfile, record_mode="none") as cass: - urlopen('http://httpbin.org/').read() + urlopen(httpbin.url).read() assert cass.play_count == 1 # but if I try to hit the net, raise an exception. with pytest.raises(Exception): - urlopen('http://httpbin.org/get').read() + urlopen(httpbin.url + '/get').read() diff --git a/tests/integration/test_register_matcher.py b/tests/integration/test_register_matcher.py index ed83afd..ed189f8 100644 --- a/tests/integration/test_register_matcher.py +++ b/tests/integration/test_register_matcher.py @@ -10,27 +10,27 @@ def false_matcher(r1, r2): return False -def test_registered_true_matcher(tmpdir): +def test_registered_true_matcher(tmpdir, httpbin): my_vcr = vcr.VCR() my_vcr.register_matcher('true', true_matcher) testfile = str(tmpdir.join('test.yml')) with my_vcr.use_cassette(testfile, match_on=['true']): # These 2 different urls are stored as the same request - urlopen('http://httpbin.org/') - urlopen('https://httpbin.org/get') + urlopen(httpbin.url) + urlopen(httpbin.url + '/get') with my_vcr.use_cassette(testfile, match_on=['true']): # I can get the response twice even though I only asked for it once - urlopen('http://httpbin.org/get') - urlopen('https://httpbin.org/get') + urlopen(httpbin.url + '/get') + urlopen(httpbin.url + '/get') -def test_registered_false_matcher(tmpdir): +def test_registered_false_matcher(tmpdir, httpbin): my_vcr = vcr.VCR() my_vcr.register_matcher('false', false_matcher) testfile = str(tmpdir.join('test.yml')) with my_vcr.use_cassette(testfile, match_on=['false']) as cass: # These 2 different urls are stored as different requests - urlopen('http://httpbin.org/') - urlopen('https://httpbin.org/get') + urlopen(httpbin.url) + urlopen(httpbin.url + '/get') assert len(cass) == 2 diff --git a/tests/integration/test_request.py b/tests/integration/test_request.py index cbb6242..c2e034b 100644 --- a/tests/integration/test_request.py +++ b/tests/integration/test_request.py @@ -2,21 +2,18 @@ import vcr from six.moves.urllib.request import urlopen -def test_recorded_request_uri_with_redirected_request(tmpdir): +def test_recorded_request_uri_with_redirected_request(tmpdir, httpbin): with vcr.use_cassette(str(tmpdir.join('test.yml'))) as cass: assert len(cass) == 0 - urlopen('http://httpbin.org/redirect/3') - assert cass.requests[0].uri == 'http://httpbin.org/redirect/3' - assert cass.requests[3].uri == 'http://httpbin.org/get' + urlopen(httpbin.url + '/redirect/3') + assert cass.requests[0].uri == httpbin.url + '/redirect/3' + assert cass.requests[3].uri == httpbin.url + '/get' assert len(cass) == 4 -def test_records_multiple_header_values(tmpdir, httpserver): - httpserver.serve_content('Hello!', headers=[('foo', 'bar'), ('foo', 'baz')]) - +def test_records_multiple_header_values(tmpdir, httpbin): with vcr.use_cassette(str(tmpdir.join('test.yml'))) as cass: assert len(cass) == 0 - - urlopen(httpserver.url) + urlopen(httpbin.url + '/response-headers?foo=bar&foo=baz') assert len(cass) == 1 assert cass.responses[0]['headers']['foo'] == ['bar', 'baz'] diff --git a/tests/integration/test_requests.py b/tests/integration/test_requests.py index 117e92e..84a992b 100644 --- a/tests/integration/test_requests.py +++ b/tests/integration/test_requests.py @@ -8,15 +8,9 @@ from assertions import assert_cassette_empty, assert_is_json requests = pytest.importorskip("requests") -@pytest.fixture(params=["https", "http"]) -def scheme(request): - '''Fixture that returns both http and https.''' - return request.param - - -def test_status_code(scheme, tmpdir): +def test_status_code(httpbin_both, tmpdir): '''Ensure that we can read the status code''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url + '/' with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): status_code = requests.get(url).status_code @@ -24,9 +18,9 @@ def test_status_code(scheme, tmpdir): assert status_code == requests.get(url).status_code -def test_headers(scheme, tmpdir): +def test_headers(httpbin_both, tmpdir): '''Ensure that we can read the headers back''' - url = scheme + '://httpbin.org/' + url = httpbin_both + '/' with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): headers = requests.get(url).headers @@ -34,9 +28,9 @@ def test_headers(scheme, tmpdir): assert headers == requests.get(url).headers -def test_body(tmpdir, scheme): +def test_body(tmpdir, httpbin_both): '''Ensure the responses are all identical enough''' - url = scheme + '://httpbin.org/bytes/1024' + url = httpbin_both + '/bytes/1024' with vcr.use_cassette(str(tmpdir.join('body.yaml'))): content = requests.get(url).content @@ -44,21 +38,21 @@ def test_body(tmpdir, scheme): assert content == requests.get(url).content -def test_effective_url(scheme, tmpdir): +def test_effective_url(tmpdir, httpbin_both): '''Ensure that the effective_url is captured''' - url = scheme + '://httpbin.org/redirect-to?url=/html' + url = httpbin_both.url + '/redirect-to?url=/html' with vcr.use_cassette(str(tmpdir.join('url.yaml'))): effective_url = requests.get(url).url - assert effective_url == scheme + '://httpbin.org/html' + assert effective_url == httpbin_both.url + '/html' with vcr.use_cassette(str(tmpdir.join('url.yaml'))): assert effective_url == requests.get(url).url -def test_auth(tmpdir, scheme): +def test_auth(tmpdir, httpbin_both): '''Ensure that we can handle basic auth''' auth = ('user', 'passwd') - url = scheme + '://httpbin.org/basic-auth/user/passwd' + url = httpbin_both + '/basic-auth/user/passwd' with vcr.use_cassette(str(tmpdir.join('auth.yaml'))): one = requests.get(url, auth=auth) @@ -68,10 +62,10 @@ def test_auth(tmpdir, scheme): assert one.status_code == two.status_code -def test_auth_failed(tmpdir, scheme): +def test_auth_failed(tmpdir, httpbin_both): '''Ensure that we can save failed auth statuses''' auth = ('user', 'wrongwrongwrong') - url = scheme + '://httpbin.org/basic-auth/user/passwd' + url = httpbin_both + '/basic-auth/user/passwd' with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass: # Ensure that this is empty to begin with assert_cassette_empty(cass) @@ -81,10 +75,10 @@ def test_auth_failed(tmpdir, scheme): assert one.status_code == two.status_code == 401 -def test_post(tmpdir, scheme): +def test_post(tmpdir, httpbin_both): '''Ensure that we can post and cache the results''' data = {'key1': 'value1', 'key2': 'value2'} - url = scheme + '://httpbin.org/post' + url = httpbin_both + '/post' with vcr.use_cassette(str(tmpdir.join('requests.yaml'))): req1 = requests.post(url, data).content @@ -94,11 +88,11 @@ def test_post(tmpdir, scheme): assert req1 == req2 -def test_post_chunked_binary(tmpdir, scheme): +def test_post_chunked_binary(tmpdir, httpbin_both): '''Ensure that we can send chunked binary without breaking while trying to concatenate bytes with str.''' data1 = iter([b'data', b'to', b'send']) data2 = iter([b'data', b'to', b'send']) - url = scheme + '://httpbin.org/post' + url = httpbin_both.url + '/post' with vcr.use_cassette(str(tmpdir.join('requests.yaml'))): req1 = requests.post(url, data1).content print(req1) @@ -109,9 +103,9 @@ def test_post_chunked_binary(tmpdir, scheme): assert req1 == req2 -def test_redirects(tmpdir, scheme): +def test_redirects(tmpdir, httpbin_both): '''Ensure that we can handle redirects''' - url = scheme + '://httpbin.org/redirect-to?url=bytes/1024' + url = httpbin_both + '/redirect-to?url=bytes/1024' with vcr.use_cassette(str(tmpdir.join('requests.yaml'))): content = requests.get(url).content @@ -123,24 +117,24 @@ def test_redirects(tmpdir, scheme): assert cass.play_count == 2 -def test_cross_scheme(tmpdir, scheme): +def test_cross_scheme(tmpdir, httpbin_secure, httpbin): '''Ensure that requests between schemes are treated separately''' # First fetch a url under http, and then again under https and then # ensure that we haven't served anything out of cache, and we have two # requests / response pairs in the cassette with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass: - requests.get('https://httpbin.org/') - requests.get('http://httpbin.org/') + requests.get(httpbin_secure + '/') + requests.get(httpbin + '/') assert cass.play_count == 0 assert len(cass) == 2 -def test_gzip(tmpdir, scheme): +def test_gzip(tmpdir, httpbin_both): ''' Ensure that requests (actually urllib3) is able to automatically decompress the response body ''' - url = scheme + '://httpbin.org/gzip' + url = httpbin_both + '/gzip' response = requests.get(url) with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))): @@ -151,7 +145,7 @@ def test_gzip(tmpdir, scheme): assert_is_json(response.content) -def test_session_and_connection_close(tmpdir, scheme): +def test_session_and_connection_close(tmpdir, httpbin): ''' This tests the issue in https://github.com/kevin1024/vcrpy/issues/48 @@ -162,29 +156,29 @@ def test_session_and_connection_close(tmpdir, scheme): with vcr.use_cassette(str(tmpdir.join('session_connection_closed.yaml'))): session = requests.session() - session.get('http://httpbin.org/get', headers={'Connection': 'close'}) - session.get('http://httpbin.org/get', headers={'Connection': 'close'}) + session.get(httpbin + '/get', headers={'Connection': 'close'}) + session.get(httpbin + '/get', headers={'Connection': 'close'}) -def test_https_with_cert_validation_disabled(tmpdir): +def test_https_with_cert_validation_disabled(tmpdir, httpbin_secure): with vcr.use_cassette(str(tmpdir.join('cert_validation_disabled.yaml'))): - requests.get('https://httpbin.org', verify=False) + requests.get(httpbin_secure.url, verify=False) -def test_session_can_make_requests_after_requests_unpatched(tmpdir): +def test_session_can_make_requests_after_requests_unpatched(tmpdir, httpbin): with vcr.use_cassette(str(tmpdir.join('test_session_after_unpatched.yaml'))): session = requests.session() - session.get('http://httpbin.org/get') + session.get(httpbin + '/get') with vcr.use_cassette(str(tmpdir.join('test_session_after_unpatched.yaml'))): session = requests.session() - session.get('http://httpbin.org/get') + session.get(httpbin + '/get') - session.get('http://httpbin.org/status/200') + session.get(httpbin + '/status/200') -def test_session_created_before_use_cassette_is_patched(tmpdir, scheme): - url = scheme + '://httpbin.org/bytes/1024' +def test_session_created_before_use_cassette_is_patched(tmpdir, httpbin_both): + url = httpbin_both + '/bytes/1024' # Record arbitrary, random data to the cassette with vcr.use_cassette(str(tmpdir.join('session_created_outside.yaml'))): session = requests.session() @@ -193,20 +187,20 @@ def test_session_created_before_use_cassette_is_patched(tmpdir, scheme): # Create a session outside of any cassette context manager session = requests.session() # Make a request to make sure that a connectionpool is instantiated - session.get(scheme + '://httpbin.org/get') + session.get(httpbin_both + '/get') with vcr.use_cassette(str(tmpdir.join('session_created_outside.yaml'))): # These should only be the same if the patching succeeded. assert session.get(url).content == body -def test_nested_cassettes_with_session_created_before_nesting(scheme, tmpdir): +def test_nested_cassettes_with_session_created_before_nesting(httpbin_both, tmpdir): ''' This tests ensures that a session that was created while one cassette was active is patched to the use the responses of a second cassette when it is enabled. ''' - url = scheme + '://httpbin.org/bytes/1024' + url = httpbin_both + '/bytes/1024' with vcr.use_cassette(str(tmpdir.join('first_nested.yaml'))): session = requests.session() first_body = session.get(url).content @@ -222,12 +216,12 @@ def test_nested_cassettes_with_session_created_before_nesting(scheme, tmpdir): assert session.get(url).content == third_body # Make sure that the session can now get content normally. - session.get('http://www.reddit.com') + assert 'User-agent' in session.get(httpbin_both.url + '/robots.txt').text -def test_post_file(tmpdir, scheme): +def test_post_file(tmpdir, httpbin_both): '''Ensure that we handle posting a file.''' - url = scheme + '://httpbin.org/post' + url = httpbin_both + '/post' with vcr.use_cassette(str(tmpdir.join('post_file.yaml'))) as cass: # Don't use 2.7+ only style ',' separated with here because we support python 2.6 with open('tox.ini') as f: @@ -244,7 +238,7 @@ def test_post_file(tmpdir, scheme): assert original_response == new_response -def test_filter_post_params(tmpdir, scheme): +def test_filter_post_params(tmpdir, httpbin_both): ''' This tests the issue in https://github.com/kevin1024/vcrpy/issues/158 @@ -252,7 +246,7 @@ def test_filter_post_params(tmpdir, scheme): with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']) as cass: assert b'id=secret' not in cass.requests[0].body ''' - url = scheme + '://httpbin.org/post' + url = httpbin_both.url + '/post' cass_loc = str(tmpdir.join('filter_post_params.yaml')) with vcr.use_cassette(cass_loc, filter_post_data_parameters=['key']) as cass: requests.post(url, data={'key': 'value'}) diff --git a/tests/integration/test_stubs.py b/tests/integration/test_stubs.py index 2b069ea..aa40004 100644 --- a/tests/integration/test_stubs.py +++ b/tests/integration/test_stubs.py @@ -2,45 +2,45 @@ import vcr import six.moves.http_client as httplib -def _headers_are_case_insensitive(): - conn = httplib.HTTPConnection('httpbin.org') +def _headers_are_case_insensitive(host, port): + conn = httplib.HTTPConnection(host, port) conn.request('GET', "/cookies/set?k1=v1") r1 = conn.getresponse() cookie_data1 = r1.getheader('set-cookie') - conn = httplib.HTTPConnection('httpbin.org') + conn = httplib.HTTPConnection(host, port) conn.request('GET', "/cookies/set?k1=v1") r2 = conn.getresponse() cookie_data2 = r2.getheader('Set-Cookie') return cookie_data1 == cookie_data2 -def test_case_insensitivity(tmpdir): +def test_case_insensitivity(tmpdir, httpbin): testfile = str(tmpdir.join('case_insensitivity.yml')) # check if headers are case insensitive outside of vcrpy - outside = _headers_are_case_insensitive() + host, port = httpbin.host, httpbin.port + outside = _headers_are_case_insensitive(host, port) with vcr.use_cassette(testfile): # check if headers are case insensitive inside of vcrpy - inside = _headers_are_case_insensitive() + inside = _headers_are_case_insensitive(host, port) # check if headers are case insensitive after vcrpy deserializes headers - inside2 = _headers_are_case_insensitive() + inside2 = _headers_are_case_insensitive(host, port) # behavior should be the same both inside and outside assert outside == inside == inside2 -def _multiple_header_value(httpserver): - conn = httplib.HTTPConnection('%s:%s' % httpserver.server_address) - conn.request('GET', "/") +def _multiple_header_value(httpbin): + conn = httplib.HTTPConnection(httpbin.host, httpbin.port) + conn.request('GET', "/response-headers?foo=bar&foo=baz") r = conn.getresponse() return r.getheader('foo') -def test_multiple_headers(tmpdir, httpserver): +def test_multiple_headers(tmpdir, httpbin): testfile = str(tmpdir.join('multiple_headers.yaml')) - httpserver.serve_content('Hello!', headers=[('foo', 'bar'), ('foo', 'baz')]) - outside = _multiple_header_value(httpserver) + outside = _multiple_header_value(httpbin) with vcr.use_cassette(testfile): - inside = _multiple_header_value(httpserver) + inside = _multiple_header_value(httpbin) assert outside == inside diff --git a/tests/integration/test_urllib2.py b/tests/integration/test_urllib2.py index be1adc4..8a633ba 100644 --- a/tests/integration/test_urllib2.py +++ b/tests/integration/test_urllib2.py @@ -1,9 +1,9 @@ # -*- coding: utf-8 -*- '''Integration tests with urllib2''' -import pytest from six.moves.urllib.request import urlopen from six.moves.urllib_parse import urlencode +import pytest_httpbin.certs # Internal imports import vcr @@ -11,135 +11,136 @@ import vcr from assertions import assert_cassette_has_one_response -@pytest.fixture(params=["https", "http"]) -def scheme(request): - """ - Fixture that returns both http and https - """ - return request.param +def urlopen_with_cafile(*args, **kwargs): + kwargs['cafile'] = pytest_httpbin.certs.where() + try: + return urlopen(*args, **kwargs) + except TypeError: + # python2/pypi don't let us override this + del kwargs['cafile'] + return urlopen(*args, **kwargs) -def test_response_code(scheme, tmpdir): +def test_response_code(httpbin_both, tmpdir): '''Ensure we can read a response code from a fetch''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): - code = urlopen(url).getcode() + code = urlopen_with_cafile(url).getcode() with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): - assert code == urlopen(url).getcode() + assert code == urlopen_with_cafile(url).getcode() -def test_random_body(scheme, tmpdir): +def test_random_body(httpbin_both, tmpdir): '''Ensure we can read the content, and that it's served from cache''' - url = scheme + '://httpbin.org/bytes/1024' + url = httpbin_both.url + '/bytes/1024' with vcr.use_cassette(str(tmpdir.join('body.yaml'))): - body = urlopen(url).read() + body = urlopen_with_cafile(url).read() with vcr.use_cassette(str(tmpdir.join('body.yaml'))): - assert body == urlopen(url).read() + assert body == urlopen_with_cafile(url).read() -def test_response_headers(scheme, tmpdir): +def test_response_headers(httpbin_both, tmpdir): '''Ensure we can get information from the response''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - open1 = urlopen(url).info().items() + open1 = urlopen_with_cafile(url).info().items() with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - open2 = urlopen(url).info().items() + open2 = urlopen_with_cafile(url).info().items() assert sorted(open1) == sorted(open2) -def test_effective_url(scheme, tmpdir): +def test_effective_url(httpbin_both, tmpdir): '''Ensure that the effective_url is captured''' - url = scheme + '://httpbin.org/redirect-to?url=/html' + url = httpbin_both.url + '/redirect-to?url=/html' with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - effective_url = urlopen(url).geturl() - assert effective_url == scheme + '://httpbin.org/html' + effective_url = urlopen_with_cafile(url).geturl() + assert effective_url == httpbin_both.url + '/html' with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): - assert effective_url == urlopen(url).geturl() + assert effective_url == urlopen_with_cafile(url).geturl() -def test_multiple_requests(scheme, tmpdir): +def test_multiple_requests(httpbin_both, tmpdir): '''Ensure that we can cache multiple requests''' urls = [ - scheme + '://httpbin.org/', - scheme + '://httpbin.org/', - scheme + '://httpbin.org/get', - scheme + '://httpbin.org/bytes/1024' + httpbin_both.url, + httpbin_both.url, + httpbin_both.url + '/get', + httpbin_both.url + '/bytes/1024', ] with vcr.use_cassette(str(tmpdir.join('multiple.yaml'))) as cass: - [urlopen(url) for url in urls] + [urlopen_with_cafile(url) for url in urls] assert len(cass) == len(urls) -def test_get_data(scheme, tmpdir): +def test_get_data(httpbin_both, tmpdir): '''Ensure that it works with query data''' data = urlencode({'some': 1, 'data': 'here'}) - url = scheme + '://httpbin.org/get?' + data + url = httpbin_both.url + '/get?' + data with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))): - res1 = urlopen(url).read() + res1 = urlopen_with_cafile(url).read() with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))): - res2 = urlopen(url).read() - + res2 = urlopen_with_cafile(url).read() assert res1 == res2 -def test_post_data(scheme, tmpdir): +def test_post_data(httpbin_both, tmpdir): '''Ensure that it works when posting data''' data = urlencode({'some': 1, 'data': 'here'}).encode('utf-8') - url = scheme + '://httpbin.org/post' + url = httpbin_both.url + '/post' with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))): - res1 = urlopen(url, data).read() + res1 = urlopen_with_cafile(url, data).read() with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass: - res2 = urlopen(url, data).read() + res2 = urlopen_with_cafile(url, data).read() assert len(cass) == 1 assert res1 == res2 assert_cassette_has_one_response(cass) -def test_post_unicode_data(scheme, tmpdir): +def test_post_unicode_data(httpbin_both, tmpdir): '''Ensure that it works when posting unicode data''' data = urlencode({'snowman': u'☃'.encode('utf-8')}).encode('utf-8') - url = scheme + '://httpbin.org/post' + url = httpbin_both.url + '/post' with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))): - res1 = urlopen(url, data).read() + res1 = urlopen_with_cafile(url, data).read() with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass: - res2 = urlopen(url, data).read() + res2 = urlopen_with_cafile(url, data).read() assert len(cass) == 1 assert res1 == res2 assert_cassette_has_one_response(cass) -def test_cross_scheme(tmpdir): +def test_cross_scheme(tmpdir, httpbin_secure, httpbin): '''Ensure that requests between schemes are treated separately''' # First fetch a url under https, and then again under https and then # ensure that we haven't served anything out of cache, and we have two # requests / response pairs in the cassette with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass: - urlopen('https://httpbin.org/') - urlopen('http://httpbin.org/') + urlopen_with_cafile(httpbin_secure.url) + urlopen_with_cafile(httpbin.url) assert len(cass) == 2 assert cass.play_count == 0 -def test_decorator(scheme, tmpdir): +def test_decorator(httpbin_both, tmpdir): '''Test the decorator version of VCR.py''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url @vcr.use_cassette(str(tmpdir.join('atts.yaml'))) def inner1(): - return urlopen(url).getcode() + return urlopen_with_cafile(url).getcode() @vcr.use_cassette(str(tmpdir.join('atts.yaml'))) def inner2(): - return urlopen(url).getcode() + return urlopen_with_cafile(url).getcode() assert inner1() == inner2() diff --git a/tests/integration/test_urllib3.py b/tests/integration/test_urllib3.py index 2a3665c..0c02c40 100644 --- a/tests/integration/test_urllib3.py +++ b/tests/integration/test_urllib3.py @@ -3,25 +3,17 @@ # coding=utf-8 import pytest +import pytest_httpbin import vcr from assertions import assert_cassette_empty, assert_is_json -certifi = pytest.importorskip("certifi") urllib3 = pytest.importorskip("urllib3") -@pytest.fixture(params=["https", "http"]) -def scheme(request): - """ - Fixture that returns both http and https - """ - return request.param - - @pytest.fixture(scope='module') def verify_pool_mgr(): return urllib3.PoolManager( cert_reqs='CERT_REQUIRED', # Force certificate check. - ca_certs=certifi.where() + ca_certs=pytest_httpbin.certs.where() ) @@ -30,9 +22,9 @@ def pool_mgr(): return urllib3.PoolManager() -def test_status_code(scheme, tmpdir, verify_pool_mgr): +def test_status_code(httpbin_both, tmpdir, verify_pool_mgr): '''Ensure that we can read the status code''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url with vcr.use_cassette(str(tmpdir.join('atts.yaml'))): status_code = verify_pool_mgr.request('GET', url).status @@ -40,9 +32,9 @@ def test_status_code(scheme, tmpdir, verify_pool_mgr): assert status_code == verify_pool_mgr.request('GET', url).status -def test_headers(scheme, tmpdir, verify_pool_mgr): +def test_headers(tmpdir, httpbin_both, verify_pool_mgr): '''Ensure that we can read the headers back''' - url = scheme + '://httpbin.org/' + url = httpbin_both.url with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): headers = verify_pool_mgr.request('GET', url).headers @@ -50,9 +42,9 @@ def test_headers(scheme, tmpdir, verify_pool_mgr): assert headers == verify_pool_mgr.request('GET', url).headers -def test_body(tmpdir, scheme, verify_pool_mgr): +def test_body(tmpdir, httpbin_both, verify_pool_mgr): '''Ensure the responses are all identical enough''' - url = scheme + '://httpbin.org/bytes/1024' + url = httpbin_both.url + '/bytes/1024' with vcr.use_cassette(str(tmpdir.join('body.yaml'))): content = verify_pool_mgr.request('GET', url).data @@ -60,11 +52,11 @@ def test_body(tmpdir, scheme, verify_pool_mgr): assert content == verify_pool_mgr.request('GET', url).data -def test_auth(tmpdir, scheme, verify_pool_mgr): +def test_auth(tmpdir, httpbin_both, verify_pool_mgr): '''Ensure that we can handle basic auth''' auth = ('user', 'passwd') headers = urllib3.util.make_headers(basic_auth='{0}:{1}'.format(*auth)) - url = scheme + '://httpbin.org/basic-auth/user/passwd' + url = httpbin_both.url + '/basic-auth/user/passwd' with vcr.use_cassette(str(tmpdir.join('auth.yaml'))): one = verify_pool_mgr.request('GET', url, headers=headers) @@ -74,11 +66,11 @@ def test_auth(tmpdir, scheme, verify_pool_mgr): assert one.status == two.status -def test_auth_failed(tmpdir, scheme, verify_pool_mgr): +def test_auth_failed(tmpdir, httpbin_both, verify_pool_mgr): '''Ensure that we can save failed auth statuses''' auth = ('user', 'wrongwrongwrong') headers = urllib3.util.make_headers(basic_auth='{0}:{1}'.format(*auth)) - url = scheme + '://httpbin.org/basic-auth/user/passwd' + url = httpbin_both.url + '/basic-auth/user/passwd' with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass: # Ensure that this is empty to begin with assert_cassette_empty(cass) @@ -88,10 +80,10 @@ def test_auth_failed(tmpdir, scheme, verify_pool_mgr): assert one.status == two.status == 401 -def test_post(tmpdir, scheme, verify_pool_mgr): +def test_post(tmpdir, httpbin_both, verify_pool_mgr): '''Ensure that we can post and cache the results''' data = {'key1': 'value1', 'key2': 'value2'} - url = scheme + '://httpbin.org/post' + url = httpbin_both.url + '/post' with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))): req1 = verify_pool_mgr.request('POST', url, data).data @@ -101,9 +93,9 @@ def test_post(tmpdir, scheme, verify_pool_mgr): assert req1 == req2 -def test_redirects(tmpdir, scheme, verify_pool_mgr): +def test_redirects(tmpdir, httpbin_both, verify_pool_mgr): '''Ensure that we can handle redirects''' - url = scheme + '://httpbin.org/redirect-to?url=bytes/1024' + url = httpbin_both.url + '/redirect-to?url=bytes/1024' with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))): content = verify_pool_mgr.request('GET', url).data @@ -115,24 +107,24 @@ def test_redirects(tmpdir, scheme, verify_pool_mgr): assert cass.play_count == 2 -def test_cross_scheme(tmpdir, scheme, verify_pool_mgr): +def test_cross_scheme(tmpdir, httpbin, httpbin_secure, verify_pool_mgr): '''Ensure that requests between schemes are treated separately''' # First fetch a url under http, and then again under https and then # ensure that we haven't served anything out of cache, and we have two # requests / response pairs in the cassette with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass: - verify_pool_mgr.request('GET', 'https://httpbin.org/') - verify_pool_mgr.request('GET', 'http://httpbin.org/') + verify_pool_mgr.request('GET', httpbin_secure.url) + verify_pool_mgr.request('GET', httpbin.url) assert cass.play_count == 0 assert len(cass) == 2 -def test_gzip(tmpdir, scheme, verify_pool_mgr): +def test_gzip(tmpdir, httpbin_both, verify_pool_mgr): ''' Ensure that requests (actually urllib3) is able to automatically decompress the response body ''' - url = scheme + '://httpbin.org/gzip' + url = httpbin_both.url + '/gzip' response = verify_pool_mgr.request('GET', url) with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))): @@ -143,6 +135,6 @@ def test_gzip(tmpdir, scheme, verify_pool_mgr): assert_is_json(response.data) -def test_https_with_cert_validation_disabled(tmpdir, pool_mgr): +def test_https_with_cert_validation_disabled(tmpdir, httpbin_secure, pool_mgr): with vcr.use_cassette(str(tmpdir.join('cert_validation_disabled.yaml'))): - pool_mgr.request('GET', 'https://httpbin.org') + pool_mgr.request('GET', httpbin_secure.url) diff --git a/tests/integration/test_wild.py b/tests/integration/test_wild.py index e8feeb5..9b69456 100644 --- a/tests/integration/test_wild.py +++ b/tests/integration/test_wild.py @@ -24,14 +24,14 @@ def test_domain_redirect(): assert len(cass) == 2 -def test_flickr_multipart_upload(): +def test_flickr_multipart_upload(httpbin, tmpdir): """ The python-flickr-api project does a multipart upload that confuses vcrpy """ def _pretend_to_be_flickr_library(): content_type, body = "text/plain", "HELLO WORLD" - h = httplib.HTTPConnection("httpbin.org") + h = httplib.HTTPConnection(httpbin.host, httpbin.port) headers = { "Content-Type": content_type, "content-length": str(len(body)) @@ -44,11 +44,12 @@ def test_flickr_multipart_upload(): return data - with vcr.use_cassette('fixtures/vcr_cassettes/flickr.yaml') as cass: + testfile = str(tmpdir.join('flickr.yml')) + with vcr.use_cassette(testfile) as cass: _pretend_to_be_flickr_library() assert len(cass) == 1 - with vcr.use_cassette('fixtures/vcr_cassettes/flickr.yaml') as cass: + with vcr.use_cassette(testfile) as cass: assert len(cass) == 1 _pretend_to_be_flickr_library() assert cass.play_count == 1 @@ -61,13 +62,13 @@ def test_flickr_should_respond_with_200(tmpdir): assert r.status_code == 200 -def test_cookies(tmpdir): +def test_cookies(tmpdir, httpbin): testfile = str(tmpdir.join('cookies.yml')) with vcr.use_cassette(testfile): s = requests.Session() - s.get("http://httpbin.org/cookies/set?k1=v1&k2=v2") + s.get(httpbin.url + "/cookies/set?k1=v1&k2=v2") - r2 = s.get("http://httpbin.org/cookies") + r2 = s.get(httpbin.url + "/cookies") assert len(r2.json()['cookies']) == 2 diff --git a/tox.ini b/tox.ini index 2bf91a0..f129486 100644 --- a/tox.ini +++ b/tox.ini @@ -11,11 +11,11 @@ deps = flake8 [testenv] commands = - py.test {posargs} + ./runtests.sh {posargs} deps = mock pytest - pytest-localserver + pytest-httpbin PyYAML requests1: requests==1.2.3 requests27: requests==2.7.0 diff --git a/vcr/stubs/httplib2_stubs.py b/vcr/stubs/httplib2_stubs.py index 04b508b..b08f357 100644 --- a/vcr/stubs/httplib2_stubs.py +++ b/vcr/stubs/httplib2_stubs.py @@ -41,6 +41,7 @@ class VCRHTTPSConnectionWithTimeout(VCRHTTPSConnection, 'strict', 'timeout', 'source_address', + 'ca_certs', )) unknown_keys = set(kwargs.keys()) - safe_keys safe_kwargs = kwargs.copy()