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 branch 'master' into unicode-match-on-body

Browse Source
This commit is contained in:
Martin Valgur
2019-07-01 22:09:57 +03:00
committed by GitHub
parent 302ea35d9a d07915ccf6
commit c6e7cb12c6
36 changed files with 804 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
tests/integration/aiohttp_utils.py
View File

@@ -2,23 +2,42 @@
import asyncio
import aiohttp
from aiohttp.test_utils import TestClient
@asyncio.coroutine
def aiohttp_request(loop, method, url, output='text', encoding='utf-8', content_type=None, **kwargs):
async def aiohttp_request(loop, method, url, output='text', encoding='utf-8', content_type=None, **kwargs):
session = aiohttp.ClientSession(loop=loop)
response_ctx = session.request(method, url, **kwargs)
response = yield from response_ctx.__aenter__()
response = await response_ctx.__aenter__()
if output == 'text':
content = yield from response.text()
content = await response.text()
elif output == 'json':
content_type = content_type or 'application/json'
content = yield from response.json(encoding=encoding, content_type=content_type)
content = await response.json(encoding=encoding, content_type=content_type)
elif output == 'raw':
content = yield from response.read()
content = await response.read()
elif output == 'stream':
content = await response.content.read()
response_ctx._resp.close()
yield from session.close()
await session.close()
return response, content
def aiohttp_app():
async def hello(request):
return aiohttp.web.Response(text='hello')
async def json(request):
return aiohttp.web.json_response({})
async def json_empty_body(request):
return aiohttp.web.json_response()
app = aiohttp.web.Application()
app.router.add_get('/', hello)
app.router.add_get('/json', json)
app.router.add_get('/json/empty', json_empty_body)
return app

62
tests/integration/test_aiohttp.py
View File

@@ -5,7 +5,7 @@ asyncio = pytest.importorskip("asyncio")
aiohttp = pytest.importorskip("aiohttp")
import vcr # noqa: E402
from .aiohttp_utils import aiohttp_request # noqa: E402
from .aiohttp_utils import aiohttp_app, aiohttp_request # noqa: E402
def run_in_loop(fn):
@@ -93,6 +93,18 @@ def test_binary(tmpdir, scheme):
assert cassette.play_count == 1
def test_stream(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
with vcr.use_cassette(str(tmpdir.join('stream.yaml'))):
resp, body = get(url, output='raw') # Do not use stream here, as the stream is exhausted by vcr
with vcr.use_cassette(str(tmpdir.join('stream.yaml'))) as cassette:
cassette_resp, cassette_body = get(url, output='stream')
assert cassette_body == body
assert cassette.play_count == 1
def test_post(tmpdir, scheme):
data = {'key1': 'value1', 'key2': 'value2'}
url = scheme + '://httpbin.org/post'
@@ -154,3 +166,51 @@ def test_params_on_url(tmpdir, scheme):
assert request.url == url
assert cassette_response_json == response_json
assert cassette.play_count == 1
def test_aiohttp_test_client(aiohttp_client, tmpdir):
loop = asyncio.get_event_loop()
app = aiohttp_app()
url = '/'
client = loop.run_until_complete(aiohttp_client(app))
with vcr.use_cassette(str(tmpdir.join('get.yaml'))):
response = loop.run_until_complete(client.get(url))
assert response.status == 200
response_text = loop.run_until_complete(response.text())
assert response_text == 'hello'
response_text = loop.run_until_complete(response.text(errors='replace'))
assert response_text == 'hello'
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
response = loop.run_until_complete(client.get(url))
request = cassette.requests[0]
assert request.url == str(client.make_url(url))
response_text = loop.run_until_complete(response.text())
assert response_text == 'hello'
assert cassette.play_count == 1
def test_aiohttp_test_client_json(aiohttp_client, tmpdir):
loop = asyncio.get_event_loop()
app = aiohttp_app()
url = '/json/empty'
client = loop.run_until_complete(aiohttp_client(app))
with vcr.use_cassette(str(tmpdir.join('get.yaml'))):
response = loop.run_until_complete(client.get(url))
assert response.status == 200
response_json = loop.run_until_complete(response.json())
assert response_json is None
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
response = loop.run_until_complete(client.get(url))
request = cassette.requests[0]
assert request.url == str(client.make_url(url))
response_json = loop.run_until_complete(response.json())
assert response_json is None
assert cassette.play_count == 1

11
tests/integration/test_httplib2.py
View File

@@ -1,12 +1,12 @@
# -*- coding: utf-8 -*-
'''Integration tests with httplib2'''
# External imports
import sys
from six.moves.urllib_parse import urlencode
import pytest
import pytest_httpbin.certs
# Internal imports
import vcr
from assertions import assert_cassette_has_one_response
@@ -19,7 +19,12 @@ def http():
Returns an httplib2 HTTP instance
with the certificate replaced by the httpbin one.
"""
return httplib2.Http(ca_certs=pytest_httpbin.certs.where())
kwargs = {
'ca_certs': pytest_httpbin.certs.where()
}
if sys.version_info[:2] in [(2, 7), (3, 7)]:
kwargs['disable_ssl_certificate_validation'] = True
return httplib2.Http(**kwargs)
def test_response_code(tmpdir, httpbin_both):

60
tests/integration/test_proxy.py Normal file
View File

@@ -0,0 +1,60 @@
# -*- coding: utf-8 -*-
'''Test using a proxy.'''
# External imports
import multiprocessing
import pytest
from six.moves import socketserver, SimpleHTTPServer
from six.moves.urllib.request import urlopen
# Internal imports
import vcr
# Conditional imports
requests = pytest.importorskip("requests")
class Proxy(SimpleHTTPServer.SimpleHTTPRequestHandler):
'''
Simple proxy server.
(Inspired by: http://effbot.org/librarybook/simplehttpserver.htm).
'''
def do_GET(self):
upstream_response = urlopen(self.path)
try:
status = upstream_response.status
headers = upstream_response.headers.items()
except AttributeError:
# In Python 2 the response is an addinfourl instance.
status = upstream_response.code
headers = upstream_response.info().items()
self.send_response(status, upstream_response.msg)
for header in headers:
self.send_header(*header)
self.end_headers()
self.copyfile(upstream_response, self.wfile)
@pytest.yield_fixture(scope='session')
def proxy_server():
httpd = socketserver.ThreadingTCPServer(('', 0), Proxy)
proxy_process = multiprocessing.Process(
target=httpd.serve_forever,
)
proxy_process.start()
yield 'http://{}:{}'.format(*httpd.server_address)
proxy_process.terminate()
def test_use_proxy(tmpdir, httpbin, proxy_server):
'''Ensure that it works with a proxy.'''
with vcr.use_cassette(str(tmpdir.join('proxy.yaml'))):
response = requests.get(httpbin.url, proxies={'http': proxy_server})
with vcr.use_cassette(str(tmpdir.join('proxy.yaml'))) as cassette:
cassette_response = requests.get(httpbin.url, proxies={'http': proxy_server})
assert cassette_response.headers == response.headers
assert cassette.play_count == 1

8
tests/integration/test_requests.py
View File

@@ -116,10 +116,10 @@ def test_post_chunked_binary(tmpdir, httpbin):
assert req1 == req2
@pytest.mark.xskip('sys.version_info >= (3, 6)', strict=True, raises=ConnectionError)
@pytest.mark.xskip((3, 5) < sys.version_info < (3, 6) and
platform.python_implementation() == 'CPython',
reason='Fails on CPython 3.5')
@pytest.mark.skipif('sys.version_info >= (3, 6)', strict=True, raises=ConnectionError)
@pytest.mark.skipif((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'])

5
tests/integration/test_urllib2.py
View File

@@ -1,6 +1,7 @@
# -*- coding: utf-8 -*-
'''Integration tests with urllib2'''
import ssl
from six.moves.urllib.request import urlopen
from six.moves.urllib_parse import urlencode
import pytest_httpbin.certs
@@ -12,7 +13,9 @@ from assertions import assert_cassette_has_one_response
def urlopen_with_cafile(*args, **kwargs):
kwargs['cafile'] = pytest_httpbin.certs.where()
context = ssl.create_default_context(cafile=pytest_httpbin.certs.where())
context.check_hostname = False
kwargs['context'] = context
try:
return urlopen(*args, **kwargs)
except TypeError:

2
tests/integration/test_urllib3.py
View File

@@ -20,7 +20,7 @@ def verify_pool_mgr():
@pytest.fixture(scope='module')
def pool_mgr():
return urllib3.PoolManager()
return urllib3.PoolManager(cert_reqs='CERT_NONE')
def test_status_code(httpbin_both, tmpdir, verify_pool_mgr):

2
tests/integration/test_wild.py
View File

@@ -59,7 +59,7 @@ def test_flickr_multipart_upload(httpbin, tmpdir):
def test_flickr_should_respond_with_200(tmpdir):
testfile = str(tmpdir.join('flickr.yml'))
with vcr.use_cassette(testfile):
r = requests.post("http://api.flickr.com/services/upload")
r = requests.post("https://api.flickr.com/services/upload", verify=False)
assert r.status_code == 200

59
tests/unit/test_cassettes.py
View File

@@ -133,6 +133,17 @@ def test_cassette_all_played():
assert a.all_played
@mock.patch('vcr.cassette.requests_match', _mock_requests_match)
def test_cassette_rewound():
a = Cassette('test')
a.append('foo', 'bar')
a.play_response('foo')
assert a.all_played
a.rewind()
assert not a.all_played
def test_before_record_response():
before_record_response = mock.Mock(return_value='mutated')
cassette = Cassette('test', before_record_response=before_record_response)
@@ -306,3 +317,51 @@ def test_use_as_decorator_on_generator():
yield 2
assert list(test_function()) == [1, 2]
@mock.patch("vcr.cassette.get_matchers_results")
def test_find_requests_with_most_matches_one_similar_request(mock_get_matchers_results):
mock_get_matchers_results.side_effect = [
(["method"], [("path", "failed : path"), ("query", "failed : query")]),
(["method", "path"], [("query", "failed : query")]),
([], [("method", "failed : method"), ("path", "failed : path"), ("query", "failed : query")]),
]
cassette = Cassette("test")
for request in range(1, 4):
cassette.append(request, 'response')
result = cassette.find_requests_with_most_matches("fake request")
assert result == [(2, ["method", "path"], [("query", "failed : query")])]
@mock.patch("vcr.cassette.get_matchers_results")
def test_find_requests_with_most_matches_no_similar_requests(mock_get_matchers_results):
mock_get_matchers_results.side_effect = [
([], [("path", "failed : path"), ("query", "failed : query")]),
([], [("path", "failed : path"), ("query", "failed : query")]),
([], [("path", "failed : path"), ("query", "failed : query")]),
]
cassette = Cassette("test")
for request in range(1, 4):
cassette.append(request, 'response')
result = cassette.find_requests_with_most_matches("fake request")
assert result == []
@mock.patch("vcr.cassette.get_matchers_results")
def test_find_requests_with_most_matches_many_similar_requests(mock_get_matchers_results):
mock_get_matchers_results.side_effect = [
(["method", "path"], [("query", "failed : query")]),
(["method"], [("path", "failed : path"), ("query", "failed : query")]),
(["method", "path"], [("query", "failed : query")]),
]
cassette = Cassette("test")
for request in range(1, 4):
cassette.append(request, 'response')
result = cassette.find_requests_with_most_matches("fake request")
assert result == [
(1, ["method", "path"], [("query", "failed : query")]),
(3, ["method", "path"], [("query", "failed : query")])
]

160
tests/unit/test_matchers.py
View File

@@ -1,4 +1,5 @@
import itertools
from vcr.compat import mock
import pytest
@@ -21,20 +22,22 @@ REQUESTS = {
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 {k1, k2}:
assert not matched
expecting_assertion_error = matcher_name in {k1, k2}
if expecting_assertion_error:
with pytest.raises(AssertionError):
matcher(REQUESTS[k1], REQUESTS[k2])
else:
assert matched
assert matcher(REQUESTS[k1], REQUESTS[k2]) is None
def test_uri_matcher():
for k1, k2 in itertools.permutations(REQUESTS, 2):
matched = matchers.uri(REQUESTS[k1], REQUESTS[k2])
if {k1, k2} != {'base', 'method'}:
assert not matched
expecting_assertion_error = {k1, k2} != {"base", "method"}
if expecting_assertion_error:
with pytest.raises(AssertionError):
matchers.uri(REQUESTS[k1], REQUESTS[k2])
else:
assert matched
assert matchers.uri(REQUESTS[k1], REQUESTS[k2]) is None
req1_body = (b"<?xml version='1.0'?><methodCall><methodName>test</methodName>"
@@ -107,7 +110,7 @@ req2_body = (b"<?xml version='1.0'?><methodCall><methodName>test</methodName>"
)
])
def test_body_matcher_does_match(r1, r2):
assert matchers.body(r1, r2)
assert matchers.body(r1, r2) is None
@pytest.mark.parametrize("r1, r2", [
@@ -135,25 +138,132 @@ def test_body_matcher_does_match(r1, r2):
)
])
def test_body_match_does_not_match(r1, r2):
assert not matchers.body(r1, r2)
with pytest.raises(AssertionError):
matchers.body(r1, r2)
def test_query_matcher():
req1 = request.Request('GET', 'http://host.com/?a=b&c=d', '', {})
req2 = request.Request('GET', 'http://host.com/?c=d&a=b', '', {})
assert matchers.query(req1, req2)
req1 = request.Request("GET", "http://host.com/?a=b&c=d", "", {})
req2 = request.Request("GET", "http://host.com/?c=d&a=b", "", {})
assert matchers.query(req1, req2) is None
req1 = request.Request('GET', 'http://host.com/?a=b&a=b&c=d', '', {})
req2 = request.Request('GET', 'http://host.com/?a=b&c=d&a=b', '', {})
req3 = request.Request('GET', 'http://host.com/?c=d&a=b&a=b', '', {})
assert matchers.query(req1, req2)
assert matchers.query(req1, req3)
req1 = request.Request("GET", "http://host.com/?a=b&a=b&c=d", "", {})
req2 = request.Request("GET", "http://host.com/?a=b&c=d&a=b", "", {})
req3 = request.Request("GET", "http://host.com/?c=d&a=b&a=b", "", {})
assert matchers.query(req1, req2) is None
assert matchers.query(req1, req3) is None
def test_metchers():
assert_matcher('method')
assert_matcher('scheme')
assert_matcher('host')
assert_matcher('port')
assert_matcher('path')
assert_matcher('query')
def test_matchers():
assert_matcher("method")
assert_matcher("scheme")
assert_matcher("host")
assert_matcher("port")
assert_matcher("path")
assert_matcher("query")
def test_evaluate_matcher_does_match():
def bool_matcher(r1, r2):
return True
def assertion_matcher(r1, r2):
assert 1 == 1
r1, r2 = None, None
for matcher in [bool_matcher, assertion_matcher]:
match, assertion_msg = matchers._evaluate_matcher(matcher, r1, r2)
assert match is True
assert assertion_msg is None
def test_evaluate_matcher_does_not_match():
def bool_matcher(r1, r2):
return False
def assertion_matcher(r1, r2):
# This is like the "assert" statement preventing pytest to recompile it
raise AssertionError()
r1, r2 = None, None
for matcher in [bool_matcher, assertion_matcher]:
match, assertion_msg = matchers._evaluate_matcher(matcher, r1, r2)
assert match is False
assert not assertion_msg
def test_evaluate_matcher_does_not_match_with_assert_message():
def assertion_matcher(r1, r2):
# This is like the "assert" statement preventing pytest to recompile it
raise AssertionError("Failing matcher")
r1, r2 = None, None
match, assertion_msg = matchers._evaluate_matcher(assertion_matcher, r1, r2)
assert match is False
assert assertion_msg == "Failing matcher"
def test_get_assertion_message():
assert matchers.get_assertion_message(None) == ""
assert matchers.get_assertion_message("") == ""
def test_get_assertion_message_with_details():
assertion_msg = "q1=1 != q2=1"
expected = (
"--------------- DETAILS ---------------\n"
"{}\n"
"----------------------------------------\n".format(assertion_msg)
)
assert matchers.get_assertion_message(assertion_msg) == expected
@pytest.mark.parametrize(
"r1, r2, expected_successes, expected_failures",
[
(
request.Request("GET", "http://host.com/p?a=b", "", {}),
request.Request("GET", "http://host.com/p?a=b", "", {}),
["method", "path"],
[],
),
(
request.Request("GET", "http://host.com/p?a=b", "", {}),
request.Request("POST", "http://host.com/p?a=b", "", {}),
["path"],
["method"],
),
(
request.Request("GET", "http://host.com/p?a=b", "", {}),
request.Request("POST", "http://host.com/path?a=b", "", {}),
[],
["method", "path"],
),
],
)
def test_get_matchers_results(r1, r2, expected_successes, expected_failures):
successes, failures = matchers.get_matchers_results(
r1, r2, [matchers.method, matchers.path]
)
assert successes == expected_successes
assert len(failures) == len(expected_failures)
for i, expected_failure in enumerate(expected_failures):
assert failures[i][0] == expected_failure
assert failures[i][1] is not None
@mock.patch("vcr.matchers.get_matchers_results")
@pytest.mark.parametrize(
"successes, failures, expected_match",
[
(["method", "path"], [], True),
(["method"], ["path"], False),
([], ["method", "path"], False),
],
)
def test_requests_match(mock_get_matchers_results, successes, failures, expected_match):
mock_get_matchers_results.return_value = (successes, failures)
r1 = request.Request("GET", "http://host.com/p?a=b", "", {})
r2 = request.Request("GET", "http://host.com/p?a=b", "", {})
match = matchers.requests_match(r1, r2, [matchers.method, matchers.path])
assert match is expected_match

10
tests/unit/test_migration.py
View File

@@ -5,6 +5,12 @@ import yaml
import vcr.migration
# Use the libYAML versions if possible
try:
from yaml import CLoader as Loader
except ImportError:
from yaml import Loader
def test_try_migrate_with_json(tmpdir):
cassette = tmpdir.join('cassette.json').strpath
@@ -22,9 +28,9 @@ def test_try_migrate_with_yaml(tmpdir):
shutil.copy('tests/fixtures/migration/old_cassette.yaml', cassette)
assert vcr.migration.try_migrate(cassette)
with open('tests/fixtures/migration/new_cassette.yaml', 'r') as f:
expected_yaml = yaml.load(f)
expected_yaml = yaml.load(f, Loader=Loader)
with open(cassette, 'r') as f:
actual_yaml = yaml.load(f)
actual_yaml = yaml.load(f, Loader=Loader)
assert actual_yaml == expected_yaml

25
tests/unit/test_request.py
View File

@@ -3,9 +3,13 @@ import pytest
from vcr.request import Request, HeadersDict
def test_str():
req = Request('GET', 'http://www.google.com/', '', {})
assert str(req) == '<Request (GET) http://www.google.com/>'
@pytest.mark.parametrize("method, uri, expected_str", [
('GET', 'http://www.google.com/', '<Request (GET) http://www.google.com/>'),
('OPTIONS', '*', '<Request (OPTIONS) *>'),
('CONNECT', 'host.some.where:1234', '<Request (CONNECT) host.some.where:1234>')
])
def test_str(method, uri, expected_str):
assert str(Request(method, uri, '', {})) == expected_str
def test_headers():
@@ -29,18 +33,21 @@ def test_add_header_deprecated():
('https://go.com/', 443),
('https://go.com:443/', 443),
('https://go.com:3000/', 3000),
('*', None)
])
def test_port(uri, expected_port):
req = Request('GET', uri, '', {})
assert req.port == expected_port
def test_uri():
req = Request('GET', 'http://go.com/', '', {})
assert req.uri == 'http://go.com/'
req = Request('GET', 'http://go.com:80/', '', {})
assert req.uri == 'http://go.com:80/'
@pytest.mark.parametrize("method, uri", [
('GET', 'http://go.com/'),
('GET', 'http://go.com:80/'),
('CONNECT', 'localhost:1234'),
('OPTIONS', '*')
])
def test_uri(method, uri):
assert Request(method, uri, '', {}).uri == uri
def test_HeadersDict():

54
tests/unit/test_response.py
View File

@@ -1,4 +1,9 @@
# coding: UTF-8
import io
import unittest
import six
from vcr.stubs import VCRHTTPResponse
@@ -66,3 +71,52 @@ def test_response_headers_should_have_correct_values():
assert response.headers.get('content-length') == "10806"
assert response.headers.get('date') == "Fri, 24 Oct 2014 18:35:37 GMT"
@unittest.skipIf(six.PY2, "Regression test for Python3 only")
def test_response_parses_correctly_and_fp_attribute_error_is_not_thrown():
"""
Regression test for https://github.com/kevin1024/vcrpy/issues/440
:return:
"""
recorded_response = {
"status": {
"message": "OK",
"code": 200
},
"headers": {
"content-length": ["0"],
"server": ["gunicorn/18.0"],
"connection": ["Close"],
"access-control-allow-credentials": ["true"],
"date": ["Fri, 24 Oct 2014 18:35:37 GMT"],
"access-control-allow-origin": ["*"],
"content-type": ["text/html; charset=utf-8"],
},
"body": {
"string": b"\nPMID- 19416910\nOWN - NLM\nSTAT- MEDLINE\nDA - 20090513\nDCOM- "
b"20090622\nLR - "
b"20141209\nIS - 1091-6490 (Electronic)\nIS - 0027-8424 (Linking)\nVI - "
b"106\nIP - "
b"19\nDP - 2009 May 12\nTI - Genetic dissection of histone deacetylase "
b"requirement in "
b"tumor cells.\nPG - 7751-5\nLID - 10.1073/pnas.0903139106 [doi]\nAB - "
b"Histone "
b"deacetylase inhibitors (HDACi) represent a new group of drugs currently\n "
b" being "
b"tested in a wide variety of clinical applications. They are especially\n "
b" effective "
b"in preclinical models of cancer where they show antiproliferative\n "
b"action in many "
b"different types of cancer cells. Recently, the first HDACi was\n "
b"approved for the "
b"treatment of cutaneous T cell lymphomas. Most HDACi currently in\n "
b"clinical "
}
}
vcr_response = VCRHTTPResponse(recorded_response)
handle = io.TextIOWrapper(io.BufferedReader(vcr_response), encoding='utf-8')
handle = iter(handle)
articles = [line for line in handle]
assert len(articles) > 1