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