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
|
# SPDX-FileCopyrightText: 2015 Eric Larson
#
# SPDX-License-Identifier: Apache-2.0
from unittest.mock import Mock
import msgpack
import requests
from cachecontrol.serialize import Serializer
class TestSerializer:
def setup_method(self):
self.serializer = Serializer()
self.response_data = {
"response": {
# Encode the body as bytes b/c it will eventually be
# converted back into a BytesIO object.
"body": b"Hello World",
"headers": {
"Content-Type": "text/plain",
"Expires": "87654",
"Cache-Control": "public",
},
"status": 200,
"version": 11,
"reason": "",
"strict": True,
"decode_content": True,
}
}
def test_load_by_version_v0(self):
data = b"cc=0,somedata"
req = Mock()
resp = self.serializer.loads(req, data)
assert resp is None
def test_load_by_version_v1(self):
data = b"cc=1,somedata"
req = Mock()
resp = self.serializer.loads(req, data)
assert resp is None
def test_load_by_version_v2(self):
data = b"cc=2,somedata"
req = Mock()
resp = self.serializer.loads(req, data)
assert resp is None
def test_load_by_version_v3(self):
data = b"cc=3,somedata"
req = Mock()
resp = self.serializer.loads(req, data)
assert resp is None
def test_read_version_v4(self):
req = Mock()
resp = self.serializer._loads_v4(req, msgpack.dumps(self.response_data))
# We have to decode our urllib3 data back into a unicode string.
assert resp.data == b"Hello World"
def test_read_latest_version_streamable(self, url):
original_resp = requests.get(url, stream=True)
req = original_resp.request
resp = self.serializer.loads(
req, self.serializer.dumps(req, original_resp.raw, original_resp.content)
)
assert resp.read()
def test_read_latest_version(self, url):
original_resp = requests.get(url)
data = original_resp.content
req = original_resp.request
resp = self.serializer.loads(
req, self.serializer.dumps(req, original_resp.raw, data)
)
assert resp.read() == data
def test_no_vary_header(self, url):
original_resp = requests.get(url)
data = original_resp.content
req = original_resp.request
# We make sure our response has a Vary header and that the
# request doesn't have the header.
original_resp.raw.headers["vary"] = "Foo"
assert self.serializer.loads(
req, self.serializer.dumps(req, original_resp.raw, data)
)
def test_no_body_creates_response_file_handle_on_dumps(self, url):
original_resp = requests.get(url, stream=True)
data = None
req = original_resp.request
assert self.serializer.loads(
req, self.serializer.dumps(req, original_resp.raw, data)
)
# By passing in data=None it will force a read of the file
# handle. Reading it again proves we're resetting the internal
# file handle with a buffer.
assert original_resp.raw.read()
def test_no_incomplete_read_on_dumps(self, url):
resp = requests.get(url + "fixed_length", stream=True)
self.serializer.dumps(resp.request, resp.raw)
assert resp.content == b"0123456789"
|
