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

fix test_aiohttp

Browse Source
This commit is contained in:
Luiz Menezes
2016-08-10 17:42:13 -03:00
parent ce14de8251
commit 609d8e35be
2 changed files with 12 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
tests/integration/test_aiohttp.py
View File

@@ -1,30 +1,27 @@
import pytest
asyncio = pytest.importorskip("boto3")
aiohttp = pytest.importorskip("aiohttp")
import asyncio # NOQA
import sys # NOQA
import aiohttp # NOQA
import pytest # NOQA
import vcr # NOQA
@asyncio.coroutine
def request(session, method, url, as_text, **kwargs):
response = yield from session.request(method, url, **kwargs)
return response, (yield from response.text()) if as_text else (yield from response.json())
from .utils import aiohttp_request # NOQA
def get(url, as_text=True, **kwargs):
loop = asyncio.get_event_loop()
with aiohttp.ClientSession() as session:
task = loop.create_task(request(session, 'GET', url, as_text, **kwargs))
task = loop.create_task(aiohttp_request(session, 'GET', url, as_text, **kwargs))
return loop.run_until_complete(task)
def post(url, as_text=True, **kwargs):
loop = asyncio.get_event_loop()
with aiohttp.ClientSession() as session:
task = loop.create_task(request(session, 'POST', url, as_text, **kwargs))
task = loop.create_task(aiohttp_request(session, 'POST', url, as_text, **kwargs))
return loop.run_until_complete(task)