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