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 #73 from mshytikov/feature/new-matchers

Browse Source 
Feature/new matchers
This commit is contained in:
Kevin McCarthy
2014-05-03 15:11:02 -10:00
parent 5aa2fb017f 78f6ce46b5
commit adf127c073
29 changed files with 808 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
README.md
View File

@@ -73,7 +73,7 @@ my_vcr = vcr.VCR(
serializer = 'json',
cassette_library_dir = 'fixtures/cassettes',
record_mode = 'once',
match_on = ['url', 'method'],
match_on = ['uri', 'method'],
)
with my_vcr.use_cassette('test.json'):
@@ -92,20 +92,26 @@ Note: Per-cassette overrides take precedence over the global config.
## Request matching
Request matching is configurable and allows you to change which requests VCR
considers identical. The default behavior is `['url', method']` which means
that requests with both the same URL and method (ie POST or GET) are considered
considers identical. The default behavior is
`['method', 'scheme', 'host', 'port', 'path', 'query']` which means that
requests with both the same URL and method (ie POST or GET) are considered
identical.
This can be configured by changing the `match_on` setting.
The following options are available :
* method (for example, POST or GET)
* url (the full URL, including the protocol)
* host (the hostname of the server receiving the request)
* path (excluding the hostname)
* body (the entire request body)
* headers (the headers of the request)
* method (for example, POST or GET)
* uri (the full URI.)
* host (the hostname of the server receiving the request)
* port (the port of the server receiving the request)
* path (the path of the request)
* query (the query string of the request)
* body (the entire request body)
* headers (the headers of the request)
Backwards compatible matchers:
* url (the `uri` alias)
If these options don't work for you, you can also register your own request
matcher. This is described in the Advanced section of this README.
@@ -163,8 +169,8 @@ with vcr.use_cassette('fixtures/vcr_cassettes/synopsis.yaml') as cass:
response = urllib2.urlopen('http://www.zombo.com/').read()
# cass should have 1 request inside it
assert len(cass) == 1
# the request url should have been http://www.zombo.com/
assert cass.requests[0].url == 'http://www.zombo.com/'
# the request uri should have been http://www.zombo.com/
assert cass.requests[0].uri == 'http://www.zombo.com/'
```
The `Cassette` object exposes the following properties which I consider part of
@@ -177,17 +183,23 @@ the API. The fields are as follows:
back
* `responses_of(request)`: Access the responses that match a given request
The `Request` object has the following properties
The `Request` object has the following properties:
* `url`: The full url of the request, including the protocol. Example:
"http://www.google.com/"
* `path`: The path of the request. For example "/" or "/home.html"
* `uri`: The full uri of the request. Example: "https://google.com/?q=vcrpy"
* `scheme`: The scheme used to make the request (http or https)
* `host`: The host of the request, for example "www.google.com"
* `port`: The port the request was made on
* `path`: The path of the request. For example "/" or "/home.html"
* `query`: The parsed query string of the request. Sorted list of name, value pairs.
* `method` : The method used to make the request, for example "GET" or "POST"
* `protocol`: The protocol used to make the request (http or https)
* `body`: The body of the request, usually empty except for POST / PUT / etc
Backwards compatible properties:
* `url`: The `uri` alias
* `protocol`: The `scheme` alias
## Register your own serializer
Don't like JSON or YAML? That's OK, VCR.py can serialize to any format you
@@ -239,7 +251,7 @@ Finally, register your method with VCR to use your new request matcher.
import vcr
def jurassic_matcher(r1, r2):
return r1.url == r2.url and 'JURASSIC PARK' in r1.body
return r1.uri == r2.uri and 'JURASSIC PARK' in r1.body
my_vcr = vcr.VCR()
my_vcr.register_matcher('jurassic', jurassic_matcher)
@@ -364,6 +376,25 @@ If you set the loglevel to DEBUG, you will also get information about which
matchers didn't match. This can help you with debugging custom matchers.
## Upgrade
### New Cassette Format
The cassette format has changed in _VCR.py 1.x_, the _VCR.py 0.x_ cassettes cannot be
used with _VCR.py 1.x_.
The easiest way to upgrade is to simply delete your cassettes and re-record all of them
VCR.py also provides migration script that attempts to upgrade your 0.x cassettes to the
new 1.x format. To use it, run the following command:
```
python -m vcr.migration PATH
```
The PATH can be path to the directory with cassettes or cassette itself.
*Note*: Backup your cassettes files before migration.
The migration runs successfully and upgrades the file content only for the old formatted
cassettes.
## Changelog
* 1.0.0 (in development) - Add support for filtering sensitive data from
requests, bump supported Python3 version to 3.4, fix some bugs with Boto

54
setup.py
View File

@@ -13,42 +13,46 @@ class PyTest(TestCommand):
self.test_suite = True
def run_tests(self):
#import here, cause outside the eggs aren't loaded
# import here, cause outside the eggs aren't loaded
import pytest
errno = pytest.main(self.test_args)
sys.exit(errno)
setup(name='vcrpy',
version='0.7.0',
description="Automatically mock your HTTP interactions to simplify and speed up testing",
author='Kevin McCarthy',
author_email='me@kevinmccarthy.org',
url='https://github.com/kevin1024/vcrpy',
packages = [
setup(
name='vcrpy',
version='0.7.0',
description=(
"Automatically mock your HTTP interactions to simplify and "
"speed up testing"
),
author='Kevin McCarthy',
author_email='me@kevinmccarthy.org',
url='https://github.com/kevin1024/vcrpy',
packages=[
'vcr',
'vcr.stubs',
'vcr.compat',
'vcr.persisters',
'vcr.serializers',
],
package_dir={
],
package_dir={
'vcr': 'vcr',
'vcr.stubs': 'vcr/stubs',
'vcr.compat': 'vcr/compat',
'vcr.persisters': 'vcr/persisters',
},
install_requires=['PyYAML','contextdecorator','six'],
license='MIT',
tests_require=['pytest','mock'],
cmdclass={'test': PyTest},
classifiers=[
'Development Status :: 4 - Beta',
'Environment :: Console',
'Intended Audience :: Developers',
'Programming Language :: Python',
'Programming Language :: Python :: 3',
'Topic :: Software Development :: Testing',
'Topic :: Internet :: WWW/HTTP',
'License :: OSI Approved :: MIT License',
],
},
install_requires=['PyYAML', 'contextdecorator', 'six'],
license='MIT',
tests_require=['pytest', 'mock'],
cmdclass={'test': PyTest},
classifiers=[
'Development Status :: 4 - Beta',
'Environment :: Console',
'Intended Audience :: Developers',
'Programming Language :: Python',
'Programming Language :: Python :: 3',
'Topic :: Software Development :: Testing',
'Topic :: Internet :: WWW/HTTP',
'License :: OSI Approved :: MIT License',
],
)

31
tests/fixtures/migration/new_cassette.json vendored Normal file
View File

@@ -0,0 +1,31 @@
[
{
"request": {
"body": null,
"headers": {
"Accept": ["*/*"],
"Accept-Encoding": ["gzip, deflate, compress"],
"User-Agent": ["python-requests/2.2.1 CPython/2.6.1 Darwin/10.8.0"]
},
"method": "GET",
"uri": "http://httpbin.org/ip"
},
"response": {
"status": {
"message": "OK",
"code": 200
},
"headers": [
"Access-Control-Allow-Origin: *\r\n",
"Content-Type: application/json\r\n",
"Date: Mon, 21 Apr 2014 23:13:40 GMT\r\n",
"Server: gunicorn/0.17.4\r\n",
"Content-Length: 32\r\n",
"Connection: keep-alive\r\n"
],
"body": {
"string": "{\n \"origin\": \"217.122.164.194\"\n}"
}
}
}
]

15
tests/fixtures/migration/new_cassette.yaml vendored Normal file
View File

@@ -0,0 +1,15 @@
- request:
body: null
headers:
Accept: ['*/*']
Accept-Encoding: ['gzip, deflate, compress']
User-Agent: ['python-requests/2.2.1 CPython/2.6.1 Darwin/10.8.0']
method: GET
uri: http://httpbin.org/ip
response:
body: {string: !!python/unicode "{\n \"origin\": \"217.122.164.194\"\n}"}
headers: [!!python/unicode "Access-Control-Allow-Origin: *\r\n", !!python/unicode "Content-Type:
application/json\r\n", !!python/unicode "Date: Mon, 21 Apr 2014 23:06:09 GMT\r\n",
!!python/unicode "Server: gunicorn/0.17.4\r\n", !!python/unicode "Content-Length:
32\r\n", !!python/unicode "Connection: keep-alive\r\n"]
status: {code: 200, message: OK}

1
tests/fixtures/migration/not_cassette.txt vendored Normal file
View File

@@ -0,0 +1 @@
This is not a cassette

34
tests/fixtures/migration/old_cassette.json vendored Normal file
View File

@@ -0,0 +1,34 @@
[
{
"request": {
"body": null,
"protocol": "http",
"method": "GET",
"headers": {
"Accept-Encoding": "gzip, deflate, compress",
"Accept": "*/*",
"User-Agent": "python-requests/2.2.1 CPython/2.6.1 Darwin/10.8.0"
},
"host": "httpbin.org",
"path": "/ip",
"port": 80
},
"response": {
"status": {
"message": "OK",
"code": 200
},
"headers": [
"Access-Control-Allow-Origin: *\r\n",
"Content-Type: application/json\r\n",
"Date: Mon, 21 Apr 2014 23:13:40 GMT\r\n",
"Server: gunicorn/0.17.4\r\n",
"Content-Length: 32\r\n",
"Connection: keep-alive\r\n"
],
"body": {
"string": "{\n \"origin\": \"217.122.164.194\"\n}"
}
}
}
]

18
tests/fixtures/migration/old_cassette.yaml vendored Normal file
View File

@@ -0,0 +1,18 @@
- request: !!python/object:vcr.request.Request
body: null
headers: !!python/object/apply:__builtin__.frozenset
- - !!python/tuple [Accept-Encoding, 'gzip, deflate, compress']
- !!python/tuple [User-Agent, python-requests/2.2.1 CPython/2.6.1 Darwin/10.8.0]
- !!python/tuple [Accept, '*/*']
host: httpbin.org
method: GET
path: /ip
port: 80
protocol: http
response:
body: {string: !!python/unicode "{\n \"origin\": \"217.122.164.194\"\n}"}
headers: [!!python/unicode "Access-Control-Allow-Origin: *\r\n", !!python/unicode "Content-Type:
application/json\r\n", !!python/unicode "Date: Mon, 21 Apr 2014 23:06:09 GMT\r\n",
!!python/unicode "Server: gunicorn/0.17.4\r\n", !!python/unicode "Content-Length:
32\r\n", !!python/unicode "Connection: keep-alive\r\n"]
status: {code: 200, message: OK}

30
tests/fixtures/wild/domain_redirect.yaml vendored
View File

@@ -1,30 +1,24 @@
- request: !!python/object:vcr.request.Request
- request:
body: null
headers: !!python/object/apply:__builtin__.frozenset
- - !!python/tuple [Accept-Encoding, 'gzip, deflate, compress']
- !!python/tuple [User-Agent, vcrpy-test]
- !!python/tuple [Accept, '*/*']
host: seomoz.org
headers:
Accept: ['*/*']
Accept-Encoding: ['gzip, deflate, compress']
User-Agent: ['vcrpy-test']
method: GET
path: /
port: 80
protocol: http
uri: http://seomoz.org/
response:
body: {string: ''}
headers: ["Location: http://moz.com/\r\n", "Server: BigIP\r\n", "Connection: Keep-Alive\r\n",
"Content-Length: 0\r\n"]
status: {code: 301, message: Moved Permanently}
- request: !!python/object:vcr.request.Request
- request:
body: null
headers: !!python/object/apply:__builtin__.frozenset
- - !!python/tuple [Accept-Encoding, 'gzip, deflate, compress']
- !!python/tuple [User-Agent, vcrpy-test]
- !!python/tuple [Accept, '*/*']
host: moz.com
headers:
Accept: ['*/*']
Accept-Encoding: ['gzip, deflate, compress']
User-Agent: ['vcrpy-test']
method: GET
path: /
port: 80
protocol: http
uri: http://moz.com/
response:
body:
string: !!binary |

4
tests/integration/test_filter.py
View File

@@ -16,7 +16,7 @@ def _request_with_auth(url, username, password):
def _find_header(cassette, header):
for request in cassette.requests:
for k, v in request.headers:
for k in request.headers:
if header.lower() == k.lower():
return True
return False
@@ -25,7 +25,7 @@ def _find_header(cassette, header):
def test_filter_basic_auth(tmpdir):
url = 'http://httpbin.org/basic-auth/user/passwd'
cass_file = str(tmpdir.join('basic_auth_filter.yaml'))
my_vcr = vcr.VCR(match_on = ['url', 'method', 'headers'])
my_vcr = vcr.VCR(match_on=['uri', 'method', 'headers'])
# 2 requests, one with auth failure and one with auth success
with my_vcr.use_cassette(cass_file, filter_headers=['authorization']):
with pytest.raises(HTTPError):

99
tests/integration/test_matchers.py Normal file
View File

@@ -0,0 +1,99 @@
import vcr
import pytest
from six.moves.urllib.request import urlopen
DEFAULT_URI = 'http://httpbin.org/get?p1=q1&p2=q2' # base uri for testing
@pytest.fixture
def cassette(tmpdir):
"""
Helper fixture used to prepare the cassete
returns path to the recorded cassette
"""
cassette_path = str(tmpdir.join('test.yml'))
with vcr.use_cassette(cassette_path, record_mode='all'):
urlopen(DEFAULT_URI)
return cassette_path
@pytest.mark.parametrize("matcher, matching_uri, not_matching_uri", [
('uri',
'http://httpbin.org/get?p1=q1&p2=q2',
'http://httpbin.org/get?p2=q2&p1=q1'),
('scheme',
'http://google.com/post?a=b',
'https://httpbin.org/get?p1=q1&p2=q2'),
('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'),
('query',
'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):
# play cassette with default uri
with vcr.use_cassette(cassette, match_on=[matcher]) as cass:
urlopen(DEFAULT_URI)
assert cass.play_count == 1
# play cassette with matching on uri
with vcr.use_cassette(cassette, match_on=[matcher]) as cass:
urlopen(matching_uri)
assert cass.play_count == 1
# play cassette with not matching on uri, it should fail
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
with vcr.use_cassette(cassette, match_on=[matcher]) as cass:
urlopen(not_matching_uri)
def test_method_matcher(cassette):
# play cassette with matching on method
with vcr.use_cassette(cassette, match_on=['method']) as cass:
urlopen('https://google.com/get?a=b')
assert cass.play_count == 1
# should fail if method does not match
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'')
@pytest.mark.parametrize("uri", [
DEFAULT_URI,
'http://httpbin.org/get?p2=q2&p1=q1',
'http://httpbin.org/get?p2=q2&p1=q1',
])
def test_default_matcher_matches(cassette, uri):
with vcr.use_cassette(cassette) as cass:
urlopen(uri)
assert cass.play_count == 1
@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):
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
with vcr.use_cassette(cassette):
urlopen(uri)
def test_default_matcher_does_not_match_on_method(cassette):
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
with vcr.use_cassette(cassette):
# is a POST request
urlopen(DEFAULT_URI, data=b'')

4
tests/integration/test_register_matcher.py
View File

@@ -10,7 +10,7 @@ def false_matcher(r1, r2):
return False
def test_registered_serializer_true_matcher(tmpdir):
def test_registered_true_matcher(tmpdir):
my_vcr = vcr.VCR()
my_vcr.register_matcher('true', true_matcher)
testfile = str(tmpdir.join('test.yml'))
@@ -25,7 +25,7 @@ def test_registered_serializer_true_matcher(tmpdir):
urlopen('https://httpbin.org/get')
def test_registered_serializer_false_matcher(tmpdir):
def test_registered_false_matcher(tmpdir):
my_vcr = vcr.VCR()
my_vcr.register_matcher('false', false_matcher)
testfile = str(tmpdir.join('test.yml'))

6
tests/integration/test_request.py
View File

@@ -2,10 +2,10 @@ import vcr
from six.moves.urllib.request import urlopen
def test_recorded_request_url_with_redirected_request(tmpdir):
def test_recorded_request_uri_with_redirected_request(tmpdir):
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].url == 'http://httpbin.org/redirect/3'
assert cass.requests[3].url == 'http://httpbin.org/get'
assert cass.requests[0].uri == 'http://httpbin.org/redirect/3'
assert cass.requests[3].uri == 'http://httpbin.org/get'
assert len(cass) == 4

3
tests/unit/test_cassettes.py
View File

@@ -8,7 +8,8 @@ from vcr.errors import UnhandledHTTPRequestError
def test_cassette_load(tmpdir):
a_file = tmpdir.join('test_cassette.yml')
a_file.write(yaml.dump([
{'request': 'foo', 'response': 'bar'}
{'request': {'body': '', 'uri': 'foo', 'method': 'GET', 'headers': {}},
'response': 'bar'}
]))
a_cassette = Cassette.load(str(a_file))
assert len(a_cassette) == 1

31
tests/unit/test_filters.py
View File

@@ -1,28 +1,37 @@
import mock
from vcr.filters import _remove_headers, _remove_query_parameters
from vcr.request import Request
def test_remove_headers():
request = mock.Mock(headers=[('hello','goodbye'),('secret','header')])
assert _remove_headers(request, ['secret']).headers == frozenset([('hello','goodbye')])
headers = {'hello': ['goodbye'], 'secret': ['header']}
request = Request('GET', 'http://google.com', '', headers)
_remove_headers(request, ['secret'])
assert request.headers == {'hello': ['goodbye']}
def test_remove_headers_empty():
request = mock.Mock(headers=[('hello','goodbye'),('secret','header')])
assert _remove_headers(request, []).headers == frozenset([('hello','goodbye'),('secret','header')])
headers = {'hello': ['goodbye'], 'secret': ['header']}
request = Request('GET', 'http://google.com', '', headers)
_remove_headers(request, [])
assert request.headers == headers
def test_remove_query_parameters():
request = mock.Mock(url='http://g.com/?q=cowboys&w=1')
assert _remove_query_parameters(request, ['w']).path == '/?q=cowboys'
uri = 'http://g.com/?q=cowboys&w=1'
request = Request('GET', uri, '', {})
_remove_query_parameters(request, ['w'])
assert request.uri == 'http://g.com/?q=cowboys'
def test_remove_all_query_parameters():
request = mock.Mock(url='http://g.com/?q=cowboys&w=1')
assert _remove_query_parameters(request, ['w','q']).path == '/'
uri = 'http://g.com/?q=cowboys&w=1'
request = Request('GET', uri, '', {})
_remove_query_parameters(request, ['w', 'q'])
assert request.uri == 'http://g.com/'
def test_remove_nonexistent_query_parameters():
request = mock.Mock(url='http://g.com/')
assert _remove_query_parameters(request, ['w','q']).path == '/'
uri = 'http://g.com/'
request = Request('GET', uri, '', {})
_remove_query_parameters(request, ['w', 'q'])
assert request.uri == 'http://g.com/'

9
tests/unit/test_json_serializer.py
View File

@@ -4,8 +4,13 @@ from vcr.request import Request
def test_serialize_binary():
request = Request('http','localhost',80,'GET','/',{},{})
cassette = {'requests': [request], 'responses': [{'body':b'\x8c'}]}
request = Request(
method='GET',
uri='http://localhost/',
body='',
headers={},
)
cassette = {'requests': [request], 'responses': [{'body': b'\x8c'}]}
with pytest.raises(Exception) as e:
serialize(cassette)

56
tests/unit/test_matchers.py Normal file
View File

@@ -0,0 +1,56 @@
import itertools
from vcr import matchers
from vcr import request
# the dict contains requests with corresponding to its key difference
# with 'base' request.
REQUESTS = {
'base': request.Request('GET', 'http://host.com/p?a=b', '', {}),
'method': request.Request('POST', 'http://host.com/p?a=b', '', {}),
'scheme': request.Request('GET', 'https://host.com:80/p?a=b', '', {}),
'host': request.Request('GET', 'http://another-host.com/p?a=b', '', {}),
'port': request.Request('GET', 'http://host.com:90/p?a=b', '', {}),
'path': request.Request('GET', 'http://host.com/x?a=b', '', {}),
'query': request.Request('GET', 'http://host.com/p?c=d', '', {}),
}
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)):
assert not matched
else:
assert matched
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')):
assert not matched
else:
assert matched
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&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)
def test_metchers():
assert_matcher('method')
assert_matcher('scheme')
assert_matcher('host')
assert_matcher('port')
assert_matcher('path')
assert_matcher('query')

41
tests/unit/test_migration.py Normal file
View File

@@ -0,0 +1,41 @@
import filecmp
import json
import shutil
import yaml
import vcr.migration
def test_try_migrate_with_json(tmpdir):
cassette = tmpdir.join('cassette').strpath
shutil.copy('tests/fixtures/migration/old_cassette.json', cassette)
assert vcr.migration.try_migrate(cassette)
with open('tests/fixtures/migration/new_cassette.json', 'r') as f:
expected_json = json.load(f)
with open(cassette, 'r') as f:
actual_json = json.load(f)
assert actual_json == expected_json
def test_try_migrate_with_yaml(tmpdir):
cassette = tmpdir.join('cassette').strpath
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)
with open(cassette, 'r') as f:
actual_yaml = yaml.load(f)
assert actual_yaml == expected_yaml
def test_try_migrate_with_invalid_or_new_cassettes(tmpdir):
cassette = tmpdir.join('cassette').strpath
files = [
'tests/fixtures/migration/not_cassette.txt',
'tests/fixtures/migration/new_cassette.yaml',
'tests/fixtures/migration/new_cassette.json',
]
for file_path in files:
shutil.copy(file_path, cassette)
assert not vcr.migration.try_migrate(cassette)
assert filecmp.cmp(cassette, file_path) # shold not change file

24
tests/unit/test_persist.py Normal file
View File

@@ -0,0 +1,24 @@
import pytest
import vcr.persist
from vcr.serializers import jsonserializer, yamlserializer
@pytest.mark.parametrize("cassette_path, serializer", [
('tests/fixtures/migration/old_cassette.json', jsonserializer),
('tests/fixtures/migration/old_cassette.yaml', yamlserializer),
])
def test_load_cassette_with_old_cassettes(cassette_path, serializer):
with pytest.raises(ValueError) as excinfo:
vcr.persist.load_cassette(cassette_path, serializer)
assert "run the migration script" in excinfo.exconly()
@pytest.mark.parametrize("cassette_path, serializer", [
('tests/fixtures/migration/not_cassette.txt', jsonserializer),
('tests/fixtures/migration/not_cassette.txt', yamlserializer),
])
def test_load_cassette_with_invalid_cassettes(cassette_path, serializer):
with pytest.raises(Exception) as excinfo:
vcr.persist.load_cassette(cassette_path, serializer)
assert "run the migration script" not in excinfo.exconly()

47
tests/unit/test_request.py
View File

@@ -1,11 +1,44 @@
import pytest
from vcr.request import Request
def test_url():
req = Request('http', 'www.google.com', 80, 'GET', '/', '', {})
assert req.url == 'http://www.google.com/'
def test_str():
req = Request('http', 'www.google.com', 80, 'GET', '/', '', {})
str(req) == '<Request (GET) http://www.google.com>'
req = Request('GET', 'http://www.google.com/', '', {})
str(req) == '<Request (GET) http://www.google.com/>'
def test_headers():
headers = {'X-Header1': ['h1'], 'X-Header2': 'h2'}
req = Request('GET', 'http://go.com/', '', headers)
assert req.headers == {'X-Header1': ['h1'], 'X-Header2': ['h2']}
req.add_header('X-Header1', 'h11')
assert req.headers == {'X-Header1': ['h1', 'h11'], 'X-Header2': ['h2']}
def test_flat_headers_dict():
headers = {'X-Header1': ['h1', 'h11'], 'X-Header2': ['h2']}
req = Request('GET', 'http://go.com/', '', headers)
assert req.flat_headers_dict() == {'X-Header1': 'h1', 'X-Header2': 'h2'}
@pytest.mark.parametrize("uri, expected_port", [
('http://go.com/', 80),
('http://go.com:80/', 80),
('http://go.com:3000/', 3000),
('https://go.com/', 433),
('https://go.com:433/', 433),
('https://go.com:3000/', 3000),
])
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/'

23
vcr/cassette.py
View File

@@ -1,10 +1,9 @@
'''The container for recorded requests and responses'''
try:
from collections import Counter, OrderedDict
from collections import Counter
except ImportError:
from .compat.counter import Counter
from .compat.ordereddict import OrderedDict
from contextdecorator import ContextDecorator
@@ -13,7 +12,7 @@ from .patch import install, reset
from .persist import load_cassette, save_cassette
from .filters import filter_request
from .serializers import yamlserializer
from .matchers import requests_match, url, method
from .matchers import requests_match, uri, method
from .errors import UnhandledHTTPRequestError
@@ -31,7 +30,7 @@ class Cassette(ContextDecorator):
path,
serializer=yamlserializer,
record_mode='once',
match_on=[url, method],
match_on=[uri, method],
filter_headers=[],
filter_query_parameters=[],
before_record=None,
@@ -70,10 +69,10 @@ class Cassette(ContextDecorator):
def append(self, request, response):
'''Add a request, response pair to this cassette'''
request = filter_request(
request = request,
filter_headers = self._filter_headers,
filter_query_parameters = self._filter_query_parameters,
before_record = self._before_record
request=request,
filter_headers=self._filter_headers,
filter_query_parameters=self._filter_query_parameters,
before_record=self._before_record
)
if not request:
return
@@ -86,10 +85,10 @@ class Cassette(ContextDecorator):
the request.
"""
request = filter_request(
request = request,
filter_headers = self._filter_headers,
filter_query_parameters = self._filter_query_parameters,
before_record = self._before_record
request=request,
filter_headers=self._filter_headers,
filter_query_parameters=self._filter_query_parameters,
before_record=self._before_record
)
if not request:
return

27
vcr/config.py
View File

@@ -1,7 +1,7 @@
import os
from .cassette import Cassette
from .serializers import yamlserializer, jsonserializer
from .matchers import method, url, host, path, headers, body
from . import matchers
class VCR(object):
@@ -9,10 +9,17 @@ class VCR(object):
serializer='yaml',
cassette_library_dir=None,
record_mode="once",
match_on=['url', 'method'],
filter_headers=[],
filter_query_parameters=[],
before_record=None,
match_on=[
'method',
'scheme',
'host',
'port',
'path',
'query',
],
):
self.serializer = serializer
self.match_on = match_on
@@ -22,12 +29,16 @@ class VCR(object):
'json': jsonserializer,
}
self.matchers = {
'method': method,
'url': url,
'host': host,
'path': path,
'headers': headers,
'body': body,
'method': matchers.method,
'uri': matchers.uri,
'url': matchers.uri, # matcher for backwards compatibility
'scheme': matchers.scheme,
'host': matchers.host,
'port': matchers.port,
'path': matchers.path,
'query': matchers.query,
'headers': matchers.headers,
'body': matchers.body,
}
self.record_mode = record_mode
self.filter_headers = filter_headers

30
vcr/filters.py
View File

@@ -1,26 +1,26 @@
from six.moves.urllib.parse import urlparse, parse_qsl, urlunparse, urlencode
from six.moves.urllib.parse import urlparse, urlencode, urlunparse
import copy
def _remove_headers(request, headers_to_remove):
out = []
for k, v in request.headers:
if k.lower() not in [h.lower() for h in headers_to_remove]:
out.append((k, v))
request.headers = frozenset(out)
headers = copy.copy(request.headers)
headers_to_remove = [h.lower() for h in headers_to_remove]
keys = [k for k in headers if k.lower() in headers_to_remove]
if keys:
for k in keys:
headers.pop(k)
request.headers = headers
return request
def _remove_query_parameters(request, query_parameters_to_remove):
if not hasattr(request, 'path' or not query_parameters_to_remote):
return request
url = urlparse(request.url)
q = parse_qsl(url.query)
q = [(k, v) for k, v in q if k not in query_parameters_to_remove]
if q:
request.path = url.path + '?' + urlencode(q)
else:
request.path = url.path
query = request.query
new_query = [(k, v) for (k, v) in query
if k not in query_parameters_to_remove]
if len(new_query) != len(query):
uri_parts = list(urlparse(request.uri))
uri_parts[4] = urlencode(new_query)
request.uri = urlunparse(uri_parts)
return request

23
vcr/matchers.py
View File

@@ -1,22 +1,35 @@
import logging
log = logging.getLogger(__name__)
def method(r1, r2):
return r1.method == r2.method
def url(r1, r2):
return r1.url == r2.url
def uri(r1, r2):
return r1.uri == r2.uri
def host(r1, r2):
return r1.host == r2.host
def scheme(r1, r2):
return r1.scheme == r2.scheme
def port(r1, r2):
return r1.port == r2.port
def path(r1, r2):
return r1.path == r2.path
def query(r1, r2):
return r1.query == r2.query
def body(r1, r2):
return r1.body == r2.body
@@ -28,7 +41,11 @@ def headers(r1, r2):
def _log_matches(matches):
differences = [m for m in matches if not m[0]]
if differences:
log.debug('Requests differ according to the following matchers: {0}'.format(differences))
log.debug(
'Requests differ according to the following matchers: ' +
str(differences)
)
def requests_match(r1, r2, matchers):
matches = [(m(r1, r2), m) for m in matchers]

153
vcr/migration.py Normal file
View File

@@ -0,0 +1,153 @@
"""
Migration script for old 'yaml' and 'json' cassettes
.. warning:: Backup your cassettes files before migration.
It merges and deletes the request obsolete keys (protocol, host, port, path)
into new 'uri' key.
Usage::
python -m vcr.migration PATH
The PATH can be path to the directory with cassettes or cassette itself
"""
import json
import os
import shutil
import sys
import tempfile
import yaml
from .serializers import compat, yamlserializer
from . import request
# Use the libYAML versions if possible
try:
from yaml import CLoader as Loader
except ImportError:
from yaml import Loader
PARTS = [
'protocol',
'host',
'port',
'path',
]
def build_uri(**parts):
port = parts['port']
scheme = parts['protocol']
default_port = {'https': 433, 'http': 80}[scheme]
parts['port'] = ':{0}'.format(port) if port != default_port else ''
return "{protocol}://{host}{port}{path}".format(**parts)
def migrate_json(in_fp, out_fp):
data = json.load(in_fp)
for item in data:
req = item['request']
uri = dict((k, req.pop(k)) for k in PARTS)
req['uri'] = build_uri(**uri)
# convert headers to dict of lists
headers = req['headers']
for k in headers:
headers[k] = [headers[k]]
json.dump(data, out_fp, indent=4)
def _restore_frozenset():
"""
Restore __builtin__.frozenset for cassettes serialized in python2 but
deserialized in python3 and builtins.frozenset for cassettes serialized
in python3 and deserialized in python2
"""
if '__builtin__' not in sys.modules:
import builtins
sys.modules['__builtin__'] = builtins
if 'builtins' not in sys.modules:
sys.modules['builtins'] = sys.modules['__builtin__']
def _old_deserialize(cassette_string):
_restore_frozenset()
data = yaml.load(cassette_string, Loader=Loader)
requests = [r['request'] for r in data]
responses = [r['response'] for r in data]
responses = [compat.convert_to_bytes(r['response']) for r in data]
return requests, responses
def migrate_yml(in_fp, out_fp):
(requests, responses) = _old_deserialize(in_fp.read())
for req in requests:
if not isinstance(req, request.Request):
raise Exception("already migrated")
else:
req.uri = build_uri(
protocol=req.__dict__['protocol'],
host=req.__dict__['host'],
port=req.__dict__['port'],
path=req.__dict__['path'],
)
# convert headers to dict of lists
headers = req.headers
req.headers = {}
for key, value in headers:
req.add_header(key, value)
data = yamlserializer.serialize({
"requests": requests,
"responses": responses,
})
out_fp.write(data)
def migrate(file_path, migration_fn):
# because we assume that original files can be reverted
# we will try to copy the content. (os.rename not needed)
with tempfile.TemporaryFile(mode='w+') as out_fp:
with open(file_path, 'r') as in_fp:
migration_fn(in_fp, out_fp)
with open(file_path, 'w') as in_fp:
out_fp.seek(0)
shutil.copyfileobj(out_fp, in_fp)
def try_migrate(path):
try: # try to migrate as json
migrate(path, migrate_json)
except Exception: # probably the file is not a json
try: # let's try to migrate as yaml
migrate(path, migrate_yml)
except Exception: # oops probably the file is not a cassette
return False
return True
def main():
if len(sys.argv) != 2:
raise SystemExit("Please provide path to cassettes directory or file. "
"Usage: python -m vcr.migration PATH")
path = sys.argv[1]
if not os.path.isabs(path):
path = os.path.abspath(path)
files = [path]
if os.path.isdir(path):
files = (os.path.join(root, name)
for (root, dirs, files) in os.walk(path)
for name in files)
for file_path in files:
migrated = try_migrate(file_path)
status = 'OK' if migrated else 'FAIL'
sys.stderr.write("[{0}] {1}\n".format(status, file_path))
sys.stderr.write("Done.\n")
if __name__ == '__main__':
main()

24
vcr/persist.py
View File

@@ -1,9 +1,31 @@
import tempfile
from .persisters.filesystem import FilesystemPersister
from . import migration
def _check_for_old_cassette(cassette_content):
# crate tmp with cassete content and try to migrate it
with tempfile.NamedTemporaryFile(mode='w+') as f:
f.write(cassette_content)
f.seek(0)
if migration.try_migrate(f.name):
raise ValueError(
"Your cassette files were generated in an older version "
"of VCR. Delete your cassettes or run the migration script."
"See http://git.io/mHhLBg for more details."
)
def load_cassette(cassette_path, serializer):
with open(cassette_path) as f:
return serializer.deserialize(f.read())
cassette_content = f.read()
try:
cassette = serializer.deserialize(cassette_content)
except TypeError:
_check_for_old_cassette(cassette_content)
raise
return cassette
def save_cassette(cassette_path, cassette_dict, serializer):

77
vcr/request.py
View File

@@ -1,53 +1,68 @@
from six.moves.urllib.parse import urlparse, parse_qsl
class Request(object):
def __init__(self, protocol, host, port, method, path, body, headers):
self.protocol = protocol
self.host = host
self.port = port
def __init__(self, method, uri, body, headers):
self.method = method
self.path = path
self.uri = uri
self.body = body
# make headers a frozenset so it will be hashable
self.headers = frozenset(headers.items())
self.headers = {}
for key in headers:
self.add_header(key, headers[key])
def add_header(self, key, value):
tmp = dict(self.headers)
tmp[key] = value
self.headers = frozenset(tmp.iteritems())
value = list(value) if isinstance(value, (tuple, list)) else [value]
self.headers.setdefault(key, []).extend(value)
def flat_headers_dict(self):
return dict((key, self.headers[key][0]) for key in self.headers)
@property
def scheme(self):
return urlparse(self.uri).scheme
@property
def host(self):
return urlparse(self.uri).hostname
@property
def port(self):
parse_uri = urlparse(self.uri)
port = parse_uri.port
if port is None:
port = {'https': 433, 'http': 80}[parse_uri.scheme]
return port
@property
def path(self):
return urlparse(self.uri).path
@property
def query(self):
q = urlparse(self.uri).query
return sorted(parse_qsl(q))
# alias for backwards compatibility
@property
def url(self):
return "{0}://{1}{2}".format(self.protocol, self.host, self.path)
return self.uri
def __key(self):
return (
self.host,
self.port,
self.method,
self.path,
self.body,
self.headers
)
def __hash__(self):
return hash(self.__key())
def __eq__(self, other):
return hash(self) == hash(other)
# alias for backwards compatibility
@property
def protocol(self):
return self.scheme
def __str__(self):
return "<Request ({0}) {1}>".format(self.method, self.url)
return "<Request ({0}) {1}>".format(self.method, self.uri)
def __repr__(self):
return self.__str__()
def _to_dict(self):
return {
'protocol': self.protocol,
'host': self.host,
'port': self.port,
'method': self.method,
'path': self.path,
'uri': self.uri,
'body': self.body,
'headers': self.headers,
}

10
vcr/serializers/jsonserializer.py
View File

@@ -6,12 +6,6 @@ except ImportError:
import json
def _json_default(obj):
if isinstance(obj, frozenset):
return dict(obj)
return obj
def deserialize(cassette_string):
data = json.loads(cassette_string)
requests = [Request._from_dict(r['request']) for r in data]
@@ -28,8 +22,8 @@ def serialize(cassette_dict):
cassette_dict['responses']
)])
try:
return json.dumps(data, indent=4, default=_json_default)
except UnicodeDecodeError as e:
return json.dumps(data, indent=4)
except UnicodeDecodeError:
raise UnicodeDecodeError(
"Error serializing cassette to JSON. ",
"Does this HTTP interaction contain binary data? ",

23
vcr/serializers/yamlserializer.py
View File

@@ -1,5 +1,6 @@
import sys
import yaml
from vcr.request import Request
from . import compat
# Use the libYAML versions if possible
@@ -21,25 +22,9 @@ Deserializing: string (yaml converts from utf-8) -> bytestring
"""
def _restore_frozenset():
"""
Restore __builtin__.frozenset for cassettes serialized in python2 but
deserialized in python3 and builtins.frozenset for cassettes serialized
in python3 and deserialized in python2
"""
if '__builtin__' not in sys.modules:
import builtins
sys.modules['__builtin__'] = builtins
if 'builtins' not in sys.modules:
sys.modules['builtins'] = sys.modules['__builtin__']
def deserialize(cassette_string):
_restore_frozenset()
data = yaml.load(cassette_string, Loader=Loader)
requests = [r['request'] for r in data]
requests = [Request._from_dict(r['request']) for r in data]
responses = [r['response'] for r in data]
responses = [compat.convert_to_bytes(r['response']) for r in data]
return requests, responses
@@ -47,7 +32,7 @@ def deserialize(cassette_string):
def serialize(cassette_dict):
data = ([{
'request': request,
'request': request._to_dict(),
'response': response,
} for request, response in zip(
cassette_dict['requests'],

41
vcr/stubs/__init__.py
View File

@@ -119,14 +119,38 @@ class VCRConnection:
# A reference to the cassette that's currently being patched in
cassette = None
def _port_postfix(self):
"""
Returns empty string for the default port and ':port' otherwise
"""
port = self.real_connection.port
default_port = {'https': 433, 'http': 80}[self._protocol]
return ':{0}'.format(port) if port != default_port else ''
def _uri(self, url):
"""Returns request absolute URI"""
uri = "{0}://{1}{2}{3}".format(
self._protocol,
self.real_connection.host,
self._port_postfix(),
url,
)
return uri
def _url(self, uri):
"""Returns request selector url from absolute URI"""
prefix = "{0}://{1}{2}".format(
self._protocol,
self.real_connection.host,
self._port_postfix(),
)
return uri.replace(prefix, '', 1)
def request(self, method, url, body=None, headers=None):
'''Persist the request metadata in self._vcr_request'''
self._vcr_request = Request(
protocol=self._protocol,
host=self.real_connection.host,
port=self.real_connection.port,
method=method,
path=url,
uri=self._uri(url),
body=body,
headers=headers or {}
)
@@ -144,11 +168,8 @@ class VCRConnection:
of putheader() calls.
"""
self._vcr_request = Request(
protocol=self._protocol,
host=self.real_connection.host,
port=self.real_connection.port,
method=method,
path=url,
uri=self._uri(url),
body="",
headers={}
)
@@ -211,9 +232,9 @@ class VCRConnection:
)
self.real_connection.request(
method=self._vcr_request.method,
url=self._vcr_request.path,
url=self._url(self._vcr_request.uri),
body=self._vcr_request.body,
headers=dict(self._vcr_request.headers or {})
headers=self._vcr_request.flat_headers_dict(),
)
# get the response