1# -*- coding: utf-8 -*-
2'''Integration tests with httplib2'''
3
4import sys
5
6from six.moves.urllib_parse import urlencode
7import pytest
8import pytest_httpbin.certs
9
10import vcr
11
12from assertions import assert_cassette_has_one_response
13
14httplib2 = pytest.importorskip("httplib2")
15
16
17def http():
18    """
19    Returns an httplib2 HTTP instance
20    with the certificate replaced by the httpbin one.
21    """
22    kwargs = {
23        'ca_certs': pytest_httpbin.certs.where()
24    }
25    if sys.version_info[:2] == (3, 7):
26        kwargs['disable_ssl_certificate_validation'] = True
27    return httplib2.Http(**kwargs)
28
29
30def test_response_code(tmpdir, httpbin_both):
31    '''Ensure we can read a response code from a fetch'''
32    url = httpbin_both.url
33    with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
34        resp, _ = http().request(url)
35        code = resp.status
36
37    with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
38        resp, _ = http().request(url)
39        assert code == resp.status
40
41
42def test_random_body(httpbin_both, tmpdir):
43    '''Ensure we can read the content, and that it's served from cache'''
44    url = httpbin_both.url + '/bytes/1024'
45    with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
46        _, content = http().request(url)
47        body = content
48
49    with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
50        _, content = http().request(url)
51        assert body == content
52
53
54def test_response_headers(tmpdir, httpbin_both):
55    '''Ensure we can get information from the response'''
56    url = httpbin_both.url
57    with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
58        resp, _ = http().request(url)
59        headers = resp.items()
60
61    with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
62        resp, _ = http().request(url)
63        assert set(headers) == set(resp.items())
64
65
66def test_effective_url(tmpdir, httpbin_both):
67    '''Ensure that the effective_url is captured'''
68    url = httpbin_both.url + '/redirect-to?url=/html'
69    with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
70        resp, _ = http().request(url)
71        effective_url = resp['content-location']
72        assert effective_url == httpbin_both + '/html'
73
74    with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
75        resp, _ = http().request(url)
76        assert effective_url == resp['content-location']
77
78
79def test_multiple_requests(tmpdir, httpbin_both):
80    '''Ensure that we can cache multiple requests'''
81    urls = [
82        httpbin_both.url,
83        httpbin_both.url,
84        httpbin_both.url + '/get',
85        httpbin_both.url + '/bytes/1024',
86    ]
87    with vcr.use_cassette(str(tmpdir.join('multiple.yaml'))) as cass:
88        [http().request(url) for url in urls]
89    assert len(cass) == len(urls)
90
91
92def test_get_data(tmpdir, httpbin_both):
93    '''Ensure that it works with query data'''
94    data = urlencode({'some': 1, 'data': 'here'})
95    url = httpbin_both.url + '/get?' + data
96    with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
97        _, res1 = http().request(url)
98
99    with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
100        _, res2 = http().request(url)
101
102    assert res1 == res2
103
104
105def test_post_data(tmpdir, httpbin_both):
106    '''Ensure that it works when posting data'''
107    data = urlencode({'some': 1, 'data': 'here'})
108    url = httpbin_both.url + '/post'
109    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
110        _, res1 = http().request(url, "POST", data)
111
112    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
113        _, res2 = http().request(url, "POST", data)
114
115    assert res1 == res2
116    assert_cassette_has_one_response(cass)
117
118
119def test_post_unicode_data(tmpdir, httpbin_both):
120    '''Ensure that it works when posting unicode data'''
121    data = urlencode({'snowman': u'☃'.encode('utf-8')})
122    url = httpbin_both.url + '/post'
123    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
124        _, res1 = http().request(url, "POST", data)
125
126    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
127        _, res2 = http().request(url, "POST", data)
128
129    assert res1 == res2
130    assert_cassette_has_one_response(cass)
131
132
133def test_cross_scheme(tmpdir, httpbin, httpbin_secure):
134    '''Ensure that requests between schemes are treated separately'''
135    # First fetch a url under https, and then again under https and then
136    # ensure that we haven't served anything out of cache, and we have two
137    # requests / response pairs in the cassette
138    with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass:
139        http().request(httpbin_secure.url)
140        http().request(httpbin.url)
141        assert len(cass) == 2
142        assert cass.play_count == 0
143
144
145def test_decorator(tmpdir, httpbin_both):
146    '''Test the decorator version of VCR.py'''
147    url = httpbin_both.url
148
149    @vcr.use_cassette(str(tmpdir.join('atts.yaml')))
150    def inner1():
151        resp, _ = http().request(url)
152        return resp['status']
153
154    @vcr.use_cassette(str(tmpdir.join('atts.yaml')))
155    def inner2():
156        resp, _ = http().request(url)
157        return resp['status']
158
159    assert inner1() == inner2()
160