1import contextlib 2 3import pytest 4asyncio = pytest.importorskip("asyncio") 5aiohttp = pytest.importorskip("aiohttp") 6 7import vcr # noqa: E402 8from .aiohttp_utils import aiohttp_app, aiohttp_request # noqa: E402 9 10 11def run_in_loop(fn): 12 with contextlib.closing(asyncio.new_event_loop()) as loop: 13 asyncio.set_event_loop(loop) 14 task = loop.create_task(fn(loop)) 15 return loop.run_until_complete(task) 16 17 18def request(method, url, output='text', **kwargs): 19 def run(loop): 20 return aiohttp_request(loop, method, url, output=output, **kwargs) 21 22 return run_in_loop(run) 23 24 25def get(url, output='text', **kwargs): 26 return request('GET', url, output=output, **kwargs) 27 28 29def post(url, output='text', **kwargs): 30 return request('POST', url, output='text', **kwargs) 31 32 33@pytest.fixture(params=["https", "http"]) 34def scheme(request): 35 '''Fixture that returns both http and https.''' 36 return request.param 37 38 39def test_status(tmpdir, scheme): 40 url = scheme + '://httpbin.org' 41 with vcr.use_cassette(str(tmpdir.join('status.yaml'))): 42 response, _ = get(url) 43 44 with vcr.use_cassette(str(tmpdir.join('status.yaml'))) as cassette: 45 cassette_response, _ = get(url) 46 assert cassette_response.status == response.status 47 assert cassette.play_count == 1 48 49 50def test_headers(tmpdir, scheme): 51 url = scheme + '://httpbin.org' 52 with vcr.use_cassette(str(tmpdir.join('headers.yaml'))): 53 response, _ = get(url) 54 55 with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cassette: 56 cassette_response, _ = get(url) 57 assert cassette_response.headers == response.headers 58 assert cassette.play_count == 1 59 60 61def test_text(tmpdir, scheme): 62 url = scheme + '://httpbin.org' 63 with vcr.use_cassette(str(tmpdir.join('text.yaml'))): 64 _, response_text = get(url) 65 66 with vcr.use_cassette(str(tmpdir.join('text.yaml'))) as cassette: 67 _, cassette_response_text = get(url) 68 assert cassette_response_text == response_text 69 assert cassette.play_count == 1 70 71 72def test_json(tmpdir, scheme): 73 url = scheme + '://httpbin.org/get' 74 headers = {'Content-Type': 'application/json'} 75 76 with vcr.use_cassette(str(tmpdir.join('json.yaml'))): 77 _, response_json = get(url, output='json', headers=headers) 78 79 with vcr.use_cassette(str(tmpdir.join('json.yaml'))) as cassette: 80 _, cassette_response_json = get(url, output='json', headers=headers) 81 assert cassette_response_json == response_json 82 assert cassette.play_count == 1 83 84 85def test_binary(tmpdir, scheme): 86 url = scheme + '://httpbin.org/image/png' 87 with vcr.use_cassette(str(tmpdir.join('binary.yaml'))): 88 _, response_binary = get(url, output='raw') 89 90 with vcr.use_cassette(str(tmpdir.join('binary.yaml'))) as cassette: 91 _, cassette_response_binary = get(url, output='raw') 92 assert cassette_response_binary == response_binary 93 assert cassette.play_count == 1 94 95 96def test_post(tmpdir, scheme): 97 data = {'key1': 'value1', 'key2': 'value2'} 98 url = scheme + '://httpbin.org/post' 99 with vcr.use_cassette(str(tmpdir.join('post.yaml'))): 100 _, response_json = post(url, data=data) 101 102 with vcr.use_cassette(str(tmpdir.join('post.yaml'))) as cassette: 103 _, cassette_response_json = post(url, data=data) 104 assert cassette_response_json == response_json 105 assert cassette.play_count == 1 106 107 108def test_params(tmpdir, scheme): 109 url = scheme + '://httpbin.org/get' 110 headers = {'Content-Type': 'application/json'} 111 params = {'a': 1, 'b': False, 'c': 'c'} 112 113 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 114 _, response_json = get(url, output='json', params=params, headers=headers) 115 116 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 117 _, cassette_response_json = get(url, output='json', params=params, headers=headers) 118 assert cassette_response_json == response_json 119 assert cassette.play_count == 1 120 121 122def test_params_same_url_distinct_params(tmpdir, scheme): 123 url = scheme + '://httpbin.org/get' 124 headers = {'Content-Type': 'application/json'} 125 params = {'a': 1, 'b': False, 'c': 'c'} 126 127 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 128 _, response_json = get(url, output='json', params=params, headers=headers) 129 130 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 131 _, cassette_response_json = get(url, output='json', params=params, headers=headers) 132 assert cassette_response_json == response_json 133 assert cassette.play_count == 1 134 135 other_params = {'other': 'params'} 136 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 137 response, cassette_response_text = get(url, output='text', params=other_params) 138 assert 'No match for the request' in cassette_response_text 139 assert response.status == 599 140 141 142def test_params_on_url(tmpdir, scheme): 143 url = scheme + '://httpbin.org/get?a=1&b=foo' 144 headers = {'Content-Type': 'application/json'} 145 146 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 147 _, response_json = get(url, output='json', headers=headers) 148 request = cassette.requests[0] 149 assert request.url == url 150 151 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 152 _, cassette_response_json = get(url, output='json', headers=headers) 153 request = cassette.requests[0] 154 assert request.url == url 155 assert cassette_response_json == response_json 156 assert cassette.play_count == 1 157 158 159def test_aiohttp_test_client(aiohttp_client, tmpdir): 160 loop = asyncio.get_event_loop() 161 app = aiohttp_app() 162 url = '/' 163 client = loop.run_until_complete(aiohttp_client(app)) 164 165 with vcr.use_cassette(str(tmpdir.join('get.yaml'))): 166 response = loop.run_until_complete(client.get(url)) 167 168 assert response.status == 200 169 response_text = loop.run_until_complete(response.text()) 170 assert response_text == 'hello' 171 172 with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette: 173 response = loop.run_until_complete(client.get(url)) 174 175 request = cassette.requests[0] 176 assert request.url == str(client.make_url(url)) 177 response_text = loop.run_until_complete(response.text()) 178 assert response_text == 'hello' 179 assert cassette.play_count == 1 180