Mirroring/vcrpy
1
0
Fork 0
mirror of https://github.com/kevin1024/vcrpy.git synced 2025-12-08 16:53:23 +00:00
Code Releases Activity

Merge pull request #138 from aisch/patch-and-test-urllib3

Browse Source 
update urllib3 patch/stub to be same as used for requests and add tests
This commit is contained in:
Ivan 'Goat' Malison
2015-03-23 17:47:42 -07:00
parent 84c45b2742 8930c97ff7
commit ec6f27bbad
7 changed files with 206 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.travis.yml
View File

@@ -12,6 +12,9 @@ env:
- WITH_LIB="requests1.x"
- WITH_LIB="httplib2"
- WITH_LIB="boto"
- WITH_LIB="urllib31.7"
- WITH_LIB="urllib31.9"
- WITH_LIB="urllib31.10"
matrix:
allow_failures:
- env: WITH_LIB="boto"
@@ -37,4 +40,7 @@ install:
- if [ $WITH_LIB = "requests2.5" ] ; then pip install requests==2.5.0; fi
- if [ $WITH_LIB = "httplib2" ] ; then pip install httplib2; fi
- if [ $WITH_LIB = "boto" ] ; then pip install boto; fi
- if [ $WITH_LIB = "urllib31.7" ] ; then pip install certifi urllib3==1.7.1; fi
- if [ $WITH_LIB = "urllib31.9" ] ; then pip install certifi urllib3==1.9.1; fi
- if [ $WITH_LIB = "urllib31.10" ] ; then pip install certifi urllib3==1.10.2; fi
script: python setup.py test

7
tests/integration/test_requests.py
View File

@@ -2,14 +2,9 @@
# coding=utf-8
import os
import pytest
import vcr
from assertions import (
assert_cassette_empty,
assert_cassette_has_one_response,
assert_is_json
)
from assertions import assert_cassette_empty, assert_is_json
requests = pytest.importorskip("requests")

5
tests/integration/test_urllib2.py
View File

@@ -1,9 +1,6 @@
'''Integration tests with urllib2'''
# coding=utf-8
# External imports
import os
import pytest
from six.moves.urllib.request import urlopen
from six.moves.urllib_parse import urlencode
@@ -11,7 +8,7 @@ from six.moves.urllib_parse import urlencode
# Internal imports
import vcr
from assertions import assert_cassette_empty, assert_cassette_has_one_response
from assertions import assert_cassette_has_one_response
@pytest.fixture(params=["https", "http"])

148
tests/integration/test_urllib3.py Normal file
View File

@@ -0,0 +1,148 @@
'''Integration tests with urllib3'''
# coding=utf-8
import pytest
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()
)
@pytest.fixture(scope='module')
def pool_mgr():
return urllib3.PoolManager()
def test_status_code(scheme, tmpdir, verify_pool_mgr):
'''Ensure that we can read the status code'''
url = scheme + '://httpbin.org/'
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
status_code = verify_pool_mgr.request('GET', url).status
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
assert status_code == verify_pool_mgr.request('GET', url).status
def test_headers(scheme, tmpdir, verify_pool_mgr):
'''Ensure that we can read the headers back'''
url = scheme + '://httpbin.org/'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
headers = verify_pool_mgr.request('GET', url).headers
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
assert headers == verify_pool_mgr.request('GET', url).headers
def test_body(tmpdir, scheme, verify_pool_mgr):
'''Ensure the responses are all identical enough'''
url = scheme + '://httpbin.org/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
content = verify_pool_mgr.request('GET', url).data
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
assert content == verify_pool_mgr.request('GET', url).data
def test_auth(tmpdir, scheme, 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'
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
one = verify_pool_mgr.request('GET', url, headers=headers)
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
two = verify_pool_mgr.request('GET', url, headers=headers)
assert one.data == two.data
assert one.status == two.status
def test_auth_failed(tmpdir, scheme, 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'
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
# Ensure that this is empty to begin with
assert_cassette_empty(cass)
one = verify_pool_mgr.request('GET', url, headers=headers)
two = verify_pool_mgr.request('GET', url, headers=headers)
assert one.data == two.data
assert one.status == two.status == 401
def test_post(tmpdir, scheme, verify_pool_mgr):
'''Ensure that we can post and cache the results'''
data = {'key1': 'value1', 'key2': 'value2'}
url = scheme + '://httpbin.org/post'
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))):
req1 = verify_pool_mgr.request('POST', url, data).data
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))):
req2 = verify_pool_mgr.request('POST', url, data).data
assert req1 == req2
def test_redirects(tmpdir, scheme, verify_pool_mgr):
'''Ensure that we can handle redirects'''
url = scheme + '://httpbin.org/redirect-to?url=bytes/1024'
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))):
content = verify_pool_mgr.request('GET', url).data
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))) as cass:
assert content == verify_pool_mgr.request('GET', url).data
# Ensure that we've now cached *two* responses. One for the redirect
# and one for the final fetch
assert len(cass) == 2
assert cass.play_count == 2
def test_cross_scheme(tmpdir, scheme, 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/')
assert cass.play_count == 0
assert len(cass) == 2
def test_gzip(tmpdir, scheme, verify_pool_mgr):
'''
Ensure that requests (actually urllib3) is able to automatically decompress
the response body
'''
url = scheme + '://httpbin.org/gzip'
response = verify_pool_mgr.request('GET', url)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
response = verify_pool_mgr.request('GET', url)
assert_is_json(response.data)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
assert_is_json(response.data)
def test_https_with_cert_validation_disabled(tmpdir, pool_mgr):
with vcr.use_cassette(str(tmpdir.join('cert_validation_disabled.yaml'))):
pool_mgr.request('GET', 'https://httpbin.org')

4
tox.ini
View File

@@ -21,5 +21,7 @@ deps =
requests23: requests==2.3.0
requests22: requests==2.2.1
httplib2: httplib2
urllib3: urllib3==1.7.1
urllib317: urllib3==1.7.1
urllib319: urllib3==1.9.1
urllib3110: urllib3==1.10.2
boto: boto

79
vcr/patch.py
View File

@@ -138,39 +138,8 @@ class CassettePatcherBuilder(object):
import requests.packages.urllib3.connectionpool as cpool
except ImportError: # pragma: no cover
return ()
from .stubs.requests_stubs import VCRRequestsHTTPConnection, VCRRequestsHTTPSConnection
http_connection_remover = ConnectionRemover(
self._get_cassette_subclass(VCRRequestsHTTPConnection)
)
https_connection_remover = ConnectionRemover(
self._get_cassette_subclass(VCRRequestsHTTPSConnection)
)
mock_triples = (
(cpool, 'VerifiedHTTPSConnection', VCRRequestsHTTPSConnection),
(cpool, 'VerifiedHTTPSConnection', VCRRequestsHTTPSConnection),
(cpool, 'HTTPConnection', VCRRequestsHTTPConnection),
(cpool, 'HTTPSConnection', VCRRequestsHTTPSConnection),
(cpool, 'is_connection_dropped', mock.Mock(return_value=False)), # Needed on Windows only
(cpool.HTTPConnectionPool, 'ConnectionCls', VCRRequestsHTTPConnection),
(cpool.HTTPSConnectionPool, 'ConnectionCls', VCRRequestsHTTPSConnection),
)
# These handle making sure that sessions only use the
# connections of the appropriate type.
mock_triples += ((cpool.HTTPConnectionPool, '_get_conn',
self._patched_get_conn(cpool.HTTPConnectionPool,
lambda : cpool.HTTPConnection)),
(cpool.HTTPSConnectionPool, '_get_conn',
self._patched_get_conn(cpool.HTTPSConnectionPool,
lambda : cpool.HTTPSConnection)),
(cpool.HTTPConnectionPool, '_new_conn',
self._patched_new_conn(cpool.HTTPConnectionPool,
http_connection_remover)),
(cpool.HTTPSConnectionPool, '_new_conn',
self._patched_new_conn(cpool.HTTPSConnectionPool,
https_connection_remover)))
return itertools.chain(self._build_patchers_from_mock_triples(mock_triples),
(http_connection_remover, https_connection_remover))
from .stubs import requests_stubs
return self._urllib3_patchers(cpool, requests_stubs)
def _patched_get_conn(self, connection_pool_class, connection_class_getter):
get_conn = connection_pool_class._get_conn
@@ -193,17 +162,13 @@ class CassettePatcherBuilder(object):
return new_connection
return patched_new_conn
@_build_patchers_from_mock_triples_decorator
def _urllib3(self):
try:
import urllib3.connectionpool as cpool
except ImportError: # pragma: no cover
pass
else:
from .stubs.urllib3_stubs import VCRVerifiedHTTPSConnection
yield cpool, 'VerifiedHTTPSConnection', VCRVerifiedHTTPSConnection
yield cpool, 'HTTPConnection', VCRHTTPConnection
return ()
from .stubs import urllib3_stubs
return self._urllib3_patchers(cpool, urllib3_stubs)
@_build_patchers_from_mock_triples_decorator
def _httplib2(self):
@@ -229,6 +194,40 @@ class CassettePatcherBuilder(object):
else:
from .stubs.boto_stubs import VCRCertValidatingHTTPSConnection
yield cpool, 'CertValidatingHTTPSConnection', VCRCertValidatingHTTPSConnection
def _urllib3_patchers(self, cpool, stubs):
http_connection_remover = ConnectionRemover(
self._get_cassette_subclass(stubs.VCRRequestsHTTPConnection)
)
https_connection_remover = ConnectionRemover(
self._get_cassette_subclass(stubs.VCRRequestsHTTPSConnection)
)
mock_triples = (
(cpool, 'VerifiedHTTPSConnection', stubs.VCRRequestsHTTPSConnection),
(cpool, 'VerifiedHTTPSConnection', stubs.VCRRequestsHTTPSConnection),
(cpool, 'HTTPConnection', stubs.VCRRequestsHTTPConnection),
(cpool, 'HTTPSConnection', stubs.VCRRequestsHTTPSConnection),
(cpool, 'is_connection_dropped', mock.Mock(return_value=False)), # Needed on Windows only
(cpool.HTTPConnectionPool, 'ConnectionCls', stubs.VCRRequestsHTTPConnection),
(cpool.HTTPSConnectionPool, 'ConnectionCls', stubs.VCRRequestsHTTPSConnection),
)
# These handle making sure that sessions only use the
# connections of the appropriate type.
mock_triples += ((cpool.HTTPConnectionPool, '_get_conn',
self._patched_get_conn(cpool.HTTPConnectionPool,
lambda : cpool.HTTPConnection)),
(cpool.HTTPSConnectionPool, '_get_conn',
self._patched_get_conn(cpool.HTTPSConnectionPool,
lambda : cpool.HTTPSConnection)),
(cpool.HTTPConnectionPool, '_new_conn',
self._patched_new_conn(cpool.HTTPConnectionPool,
http_connection_remover)),
(cpool.HTTPSConnectionPool, '_new_conn',
self._patched_new_conn(cpool.HTTPSConnectionPool,
https_connection_remover)))
return itertools.chain(self._build_patchers_from_mock_triples(mock_triples),
(http_connection_remover, https_connection_remover))
class ConnectionRemover(object):

11
vcr/stubs/urllib3_stubs.py
View File

@@ -1,8 +1,13 @@
'''Stubs for urllib3'''
from urllib3.connectionpool import VerifiedHTTPSConnection
from ..stubs import VCRHTTPSConnection
from urllib3.connectionpool import HTTPConnection, VerifiedHTTPSConnection
from ..stubs import VCRHTTPConnection, VCRHTTPSConnection
# urllib3 defines its own HTTPConnection classes. It includes some polyfills
# for newer features missing in older pythons.
class VCRVerifiedHTTPSConnection(VCRHTTPSConnection, VerifiedHTTPSConnection):
class VCRRequestsHTTPConnection(VCRHTTPConnection, HTTPConnection):
_baseclass = HTTPConnection
class VCRRequestsHTTPSConnection(VCRHTTPSConnection, VerifiedHTTPSConnection):
_baseclass = VerifiedHTTPSConnection