mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-08 16:53:23 +00:00
support python 3.4
This commit is contained in:
@@ -96,9 +96,10 @@ def test_filter_callback(tmpdir):
|
||||
|
||||
def test_decompress_gzip(tmpdir):
|
||||
url = 'http://httpbin.org/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):
|
||||
urlopen(url)
|
||||
urlopen(request)
|
||||
with vcr.use_cassette(cass_file) as cass:
|
||||
decoded_response = urlopen(url).read()
|
||||
assert_cassette_has_one_response(cass)
|
||||
@@ -107,9 +108,10 @@ def test_decompress_gzip(tmpdir):
|
||||
|
||||
def test_decompress_deflate(tmpdir):
|
||||
url = 'http://httpbin.org/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):
|
||||
urlopen(url)
|
||||
urlopen(request)
|
||||
with vcr.use_cassette(cass_file) as cass:
|
||||
decoded_response = urlopen(url).read()
|
||||
assert_cassette_has_one_response(cass)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from six import BytesIO
|
||||
from vcr.filters import (
|
||||
remove_headers, replace_headers,
|
||||
remove_query_parameters, replace_query_parameters,
|
||||
@@ -6,7 +7,6 @@ from vcr.filters import (
|
||||
)
|
||||
from vcr.compat import mock
|
||||
from vcr.request import Request
|
||||
import cStringIO
|
||||
import gzip
|
||||
import json
|
||||
import zlib
|
||||
@@ -225,7 +225,7 @@ def test_decode_response_uncompressed():
|
||||
|
||||
|
||||
def test_decode_response_deflate():
|
||||
body = 'deflate message'
|
||||
body = b'deflate message'
|
||||
deflate_response = {
|
||||
'body': {'string': zlib.compress(body)},
|
||||
'headers': {
|
||||
@@ -242,14 +242,17 @@ def test_decode_response_deflate():
|
||||
}
|
||||
decoded_response = decode_response(deflate_response)
|
||||
assert decoded_response['body']['string'] == body
|
||||
assert decoded_response['content-length'] == len(body)
|
||||
assert decoded_response['headers']['content-length'] == [str(len(body))]
|
||||
|
||||
|
||||
def test_decode_response_gzip():
|
||||
body = 'gzip message'
|
||||
buf = cStringIO.StringIO()
|
||||
with gzip.GzipFile('a', fileobj=buf, mode='wb') as f:
|
||||
f.write(body)
|
||||
body = b'gzip message'
|
||||
|
||||
buf = BytesIO()
|
||||
f = gzip.GzipFile('a', fileobj=buf, mode='wb')
|
||||
f.write(body)
|
||||
f.close()
|
||||
|
||||
compressed_body = buf.getvalue()
|
||||
buf.close()
|
||||
gzip_response = {
|
||||
@@ -268,4 +271,4 @@ def test_decode_response_gzip():
|
||||
}
|
||||
decoded_response = decode_response(gzip_response)
|
||||
assert decoded_response['body']['string'] == body
|
||||
assert decoded_response['content-length'] == len(body)
|
||||
assert decoded_response['headers']['content-length'] == [str(len(body))]
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
from six import BytesIO, text_type
|
||||
from six.moves.urllib.parse import urlparse, urlencode, urlunparse
|
||||
import copy
|
||||
import json
|
||||
import zlib
|
||||
|
||||
from .util import CaseInsensitiveDict
|
||||
|
||||
|
||||
def replace_headers(request, replacements):
|
||||
"""
|
||||
@@ -130,8 +133,8 @@ def decode_response(response):
|
||||
2. delete the content-encoding header
|
||||
3. update content-length header to decompressed length
|
||||
"""
|
||||
def is_compressed(response):
|
||||
encoding = response['headers'].get('content-encoding', [])
|
||||
def is_compressed(headers):
|
||||
encoding = headers.get('content-encoding', [])
|
||||
return encoding and encoding[0] in ('gzip', 'deflate')
|
||||
|
||||
def decompress_body(body, encoding):
|
||||
@@ -143,15 +146,16 @@ def decode_response(response):
|
||||
else: # encoding == 'deflate'
|
||||
return zlib.decompress(body)
|
||||
|
||||
if is_compressed(response):
|
||||
response = response.copy()
|
||||
encoding = response['headers']['content-encoding'][0]
|
||||
|
||||
response['headers']['content-encoding'].remove(encoding)
|
||||
if not response['headers']['content-encoding']:
|
||||
del response['headers']['content-encoding']
|
||||
headers = CaseInsensitiveDict(response['headers'])
|
||||
if is_compressed(headers):
|
||||
response = copy.deepcopy(response)
|
||||
encoding = headers['content-encoding'][0]
|
||||
headers['content-encoding'].remove(encoding)
|
||||
if not headers['content-encoding']:
|
||||
del headers['content-encoding']
|
||||
|
||||
new_body = decompress_body(response['body']['string'], encoding)
|
||||
response['body']['string'] = new_body
|
||||
response['content-length'] = len(new_body)
|
||||
headers['content-length'] = [str(len(new_body))]
|
||||
response['headers'] = dict(headers)
|
||||
return response
|
||||
|
||||
Reference in New Issue
Block a user