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

Update aiohttp_stub to work with binary content

Browse Source
This commit is contained in:
Allisson Azevedo
2017-06-22 15:12:58 -03:00
parent 47ccddafee
commit c55d976277
3 changed files with 34 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
tests/integration/aiohttp_utils.py
View File

@@ -3,11 +3,13 @@ import aiohttp
@asyncio.coroutine
def aiohttp_request(loop, method, url, as_text, **kwargs):
def aiohttp_request(loop, method, url, output='text', **kwargs):
with aiohttp.ClientSession(loop=loop) as session:
response = yield from session.request(method, url, **kwargs) # NOQA: E999
if as_text:
if output == 'text':
content = yield from response.text() # NOQA: E999
else:
elif output == 'json':
content = yield from response.json() # NOQA: E999
elif output == 'raw':
content = yield from response.read() # NOQA: E999
return response, content

37
tests/integration/test_aiohttp.py
View File

@@ -22,19 +22,19 @@ def run_in_loop(fn):
return loop.run_until_complete(task)
def request(method, url, as_text=True, **kwargs):
def request(method, url, output='text', **kwargs):
def run(loop):
return aiohttp_request(loop, method, url, as_text, **kwargs)
return aiohttp_request(loop, method, url, output=output, **kwargs)
return run_in_loop(run)
def get(url, as_text=True, **kwargs):
return request('GET', url, as_text, **kwargs)
def get(url, output='text', **kwargs):
return request('GET', url, output=output, **kwargs)
def post(url, as_text=True, **kwargs):
return request('POST', url, as_text, **kwargs)
def post(url, output='text', **kwargs):
return request('POST', url, output='text', **kwargs)
@pytest.fixture(params=["https", "http"])
@@ -79,14 +79,25 @@ def test_text(tmpdir, scheme):
def test_json(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
with vcr.use_cassette(str(tmpdir.join('json.yaml'))):
_, response_json = get(url, as_text=False)
_, response_json = get(url, output='json')
with vcr.use_cassette(str(tmpdir.join('json.yaml'))) as cassette:
_, cassette_response_json = get(url, as_text=False)
_, cassette_response_json = get(url, output='json')
assert cassette_response_json == response_json
assert cassette.play_count == 1
def test_binary(tmpdir, scheme):
url = scheme + '://httpbin.org/image/png'
with vcr.use_cassette(str(tmpdir.join('binary.yaml'))):
_, response_binary = get(url, output='raw')
with vcr.use_cassette(str(tmpdir.join('binary.yaml'))) as cassette:
_, cassette_response_binary = get(url, output='raw')
assert cassette_response_binary == response_binary
assert cassette.play_count == 1
def test_post(tmpdir, scheme):
data = {'key1': 'value1', 'key2': 'value2'}
url = scheme + '://httpbin.org/post'
@@ -103,10 +114,10 @@ def test_params(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
params = {'a': 1, 'b': False, 'c': 'c'}
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, response_json = get(url, as_text=False, params=params)
_, response_json = get(url, output='json', params=params)
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, cassette_response_json = get(url, as_text=False, params=params)
_, cassette_response_json = get(url, output='json', params=params)
assert cassette_response_json == response_json
assert cassette.play_count == 1
@@ -115,15 +126,15 @@ def test_params_same_url_distinct_params(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
params = {'a': 1, 'b': False, 'c': 'c'}
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, response_json = get(url, as_text=False, params=params)
_, response_json = get(url, output='json', params=params)
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, cassette_response_json = get(url, as_text=False, params=params)
_, cassette_response_json = get(url, output='json', params=params)
assert cassette_response_json == response_json
assert cassette.play_count == 1
other_params = {'other': 'params'}
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
response, cassette_response_text = get(url, as_text=True, params=other_params)
response, cassette_response_text = get(url, output='text', params=other_params)
assert 'No match for the request' in cassette_response_text
assert response.status == 599

6
vcr/stubs/aiohttp_stubs/__init__.py
View File

@@ -21,6 +21,10 @@ class MockClientResponse(ClientResponse):
def text(self, encoding='utf-8'):
return self.content.decode(encoding)
@asyncio.coroutine
def read(self):
return self.content
@asyncio.coroutine
def release(self):
pass
@@ -71,7 +75,7 @@ def vcr_request(cassette, real_request):
'message': response.reason,
},
'headers': dict(response.headers),
'body': {'string': (yield from response.text())}, # NOQA: E999
'body': {'string': (yield from response.read())}, # NOQA: E999
'url': response.url,
}
cassette.append(vcr_request, vcr_response)