Mirroring/vcrpy
1
0
Fork 0
mirror of https://github.com/kevin1024/vcrpy.git synced 2025-12-09 17:15:35 +00:00
Code Releases Activity

Merge remote-tracking branch 'upstream/master'

Browse Source
This commit is contained in:
Allisson Azevedo
2018-02-17 11:51:04 -03:00
parent 0cab15658f 2156adb841
commit ad153bd733
21 changed files with 116 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
tests/integration/test_ignore.py
View File

@@ -26,9 +26,9 @@ 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))
urlopen('http://localhost:{}/'.format(httpbin.port))
assert len(cass) == 0
urlopen('http://httpbin.org:{0}/'.format(httpbin.port))
urlopen('http://httpbin.org:{}/'.format(httpbin.port))
assert len(cass) == 1
@@ -39,9 +39,9 @@ def test_ignore_httpbin(tmpdir, httpbin):
cass_file,
ignore_hosts=['httpbin.org']
) as cass:
urlopen('http://httpbin.org:{0}/'.format(httpbin.port))
urlopen('http://httpbin.org:{}/'.format(httpbin.port))
assert len(cass) == 0
urlopen('http://localhost:{0}/'.format(httpbin.port))
urlopen('http://localhost:{}/'.format(httpbin.port))
assert len(cass) == 1
@@ -53,8 +53,8 @@ def test_ignore_localhost_and_httpbin(tmpdir, httpbin):
ignore_hosts=['httpbin.org'],
ignore_localhost=True
) as cass:
urlopen('http://httpbin.org:{0}'.format(httpbin.port))
urlopen('http://localhost:{0}'.format(httpbin.port))
urlopen('http://httpbin.org:{}'.format(httpbin.port))
urlopen('http://localhost:{}'.format(httpbin.port))
assert len(cass) == 0
@@ -62,12 +62,12 @@ 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))
urlopen('http://localhost:{}'.format(httpbin.port))
assert len(cass) == 0
urlopen('http://httpbin.org:{0}'.format(httpbin.port))
urlopen('http://httpbin.org:{}'.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))
urlopen('http://localhost:{}'.format(httpbin.port))
urlopen('http://httpbin.org:{}'.format(httpbin.port))
assert len(cass) == 1

23
tests/integration/test_register_persister.py
View File

@@ -10,10 +10,25 @@ import vcr
from vcr.persisters.filesystem import FilesystemPersister
class CustomFilesystemPersister(object):
'''Behaves just like default FilesystemPersister but adds .test extension
to the cassette file'''
@staticmethod
def load_cassette(cassette_path, serializer):
cassette_path += '.test'
return FilesystemPersister.load_cassette(cassette_path, serializer)
@staticmethod
def save_cassette(cassette_path, cassette_dict, serializer):
cassette_path += '.test'
FilesystemPersister.save_cassette(cassette_path, cassette_dict,
serializer)
def test_save_cassette_with_custom_persister(tmpdir, httpbin):
'''Ensure you can save a cassette using custom persister'''
my_vcr = vcr.VCR()
my_vcr.register_persister(FilesystemPersister)
my_vcr.register_persister(CustomFilesystemPersister)
# Check to make sure directory doesnt exist
assert not os.path.exists(str(tmpdir.join('nonexistent')))
@@ -23,7 +38,7 @@ def test_save_cassette_with_custom_persister(tmpdir, httpbin):
urlopen(httpbin.url).read()
# Callback should have made the file and the directory
assert os.path.exists(str(tmpdir.join('nonexistent', 'cassette.yml')))
assert os.path.exists(str(tmpdir.join('nonexistent', 'cassette.yml.test')))
def test_load_cassette_with_custom_persister(tmpdir, httpbin):
@@ -31,9 +46,9 @@ def test_load_cassette_with_custom_persister(tmpdir, httpbin):
Ensure you can load a cassette using custom persister
'''
my_vcr = vcr.VCR()
my_vcr.register_persister(FilesystemPersister)
my_vcr.register_persister(CustomFilesystemPersister)
test_fixture = str(tmpdir.join('synopsis.json'))
test_fixture = str(tmpdir.join('synopsis.json.test'))
with my_vcr.use_cassette(test_fixture, serializer='json'):
response = urlopen(httpbin.url).read()

11
tests/integration/test_requests.py
View File

@@ -1,6 +1,8 @@
# -*- coding: utf-8 -*-
'''Test requests' interaction with vcr'''
import platform
import pytest
import sys
import vcr
from assertions import assert_cassette_empty, assert_is_json
@@ -115,6 +117,9 @@ def test_post_chunked_binary(tmpdir, httpbin):
@pytest.mark.xfail('sys.version_info >= (3, 6)', strict=True, raises=ConnectionError)
@pytest.mark.xfail((3, 5) < sys.version_info < (3, 6) and
platform.python_implementation() == 'CPython',
reason='Fails on CPython 3.5')
def test_post_chunked_binary_secure(tmpdir, httpbin_secure):
'''Ensure that we can send chunked binary without breaking while trying to concatenate bytes with str.'''
data1 = iter([b'data', b'to', b'send'])
@@ -249,10 +254,8 @@ def test_nested_cassettes_with_session_created_before_nesting(httpbin_both, tmpd
def test_post_file(tmpdir, httpbin_both):
'''Ensure that we handle posting a file.'''
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:
original_response = requests.post(url, f).content
with vcr.use_cassette(str(tmpdir.join('post_file.yaml'))) as cass, open('tox.ini') as f:
original_response = requests.post(url, f).content
# This also tests that we do the right thing with matching the body when they are files.
with vcr.use_cassette(str(tmpdir.join('post_file.yaml')),

4
tests/integration/test_urllib3.py
View File

@@ -56,7 +56,7 @@ def test_body(tmpdir, httpbin_both, 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))
headers = urllib3.util.make_headers(basic_auth='{}:{}'.format(*auth))
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)
@@ -70,7 +70,7 @@ def test_auth(tmpdir, httpbin_both, 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))
headers = urllib3.util.make_headers(basic_auth='{}:{}'.format(*auth))
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

27
tests/integration/test_wild.py
View File

@@ -1,5 +1,6 @@
import multiprocessing
import pytest
from six.moves import xmlrpc_client
from six.moves import xmlrpc_client, xmlrpc_server
requests = pytest.importorskip("requests")
@@ -80,13 +81,27 @@ def test_amazon_doctype(tmpdir):
assert 'html' in r.text
def test_xmlrpclib(tmpdir):
@pytest.yield_fixture(scope='session')
def rpc_server():
httpd = xmlrpc_server.SimpleXMLRPCServer(('', 0))
httpd.register_function(pow)
proxy_process = multiprocessing.Process(
target=httpd.serve_forever,
)
try:
proxy_process.start()
yield 'http://{}:{}'.format(*httpd.server_address)
finally:
proxy_process.terminate()
def test_xmlrpclib(tmpdir, rpc_server):
with vcr.use_cassette(str(tmpdir.join('xmlrpcvideo.yaml'))):
roundup_server = xmlrpc_client.ServerProxy('http://bugs.python.org/xmlrpc', allow_none=True)
original_schema = roundup_server.schema()
roundup_server = xmlrpc_client.ServerProxy(rpc_server, allow_none=True)
original_schema = roundup_server.pow(2, 4)
with vcr.use_cassette(str(tmpdir.join('xmlrpcvideo.yaml'))):
roundup_server = xmlrpc_client.ServerProxy('http://bugs.python.org/xmlrpc', allow_none=True)
second_schema = roundup_server.schema()
roundup_server = xmlrpc_client.ServerProxy(rpc_server, allow_none=True)
second_schema = roundup_server.pow(2, 4)
assert original_schema == second_schema

4
tests/unit/test_matchers.py
View File

@@ -22,7 +22,7 @@ def assert_matcher(matcher_name):
matcher = getattr(matchers, matcher_name)
for k1, k2 in itertools.permutations(REQUESTS, 2):
matched = matcher(REQUESTS[k1], REQUESTS[k2])
if matcher_name in set((k1, k2)):
if matcher_name in {k1, k2}:
assert not matched
else:
assert matched
@@ -31,7 +31,7 @@ def assert_matcher(matcher_name):
def test_uri_matcher():
for k1, k2 in itertools.permutations(REQUESTS, 2):
matched = matchers.uri(REQUESTS[k1], REQUESTS[k2])
if set((k1, k2)) != set(('base', 'method')):
if {k1, k2} != {'base', 'method'}:
assert not matched
else:
assert matched

4
tests/unit/test_vcr.py
View File

@@ -319,11 +319,11 @@ def test_additional_matchers():
@vcr.use_cassette
def function_defaults(cassette):
assert set(cassette._match_on) == set([vcr.matchers['uri']])
assert set(cassette._match_on) == {vcr.matchers['uri']}
@vcr.use_cassette(additional_matchers=('body',))
def function_additional(cassette):
assert set(cassette._match_on) == set([vcr.matchers['uri'], vcr.matchers['body']])
assert set(cassette._match_on) == {vcr.matchers['uri'], vcr.matchers['body']}
function_defaults()
function_additional()