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
|
import pytest
from datetime import datetime, timedelta
from io import BytesIO
from time import sleep
from urllib3.response import HTTPResponse
from requests_cache import CachedHTTPResponse, CachedResponse
from tests.conftest import MOCKED_URL
def test_basic_attrs(mock_session):
response = CachedResponse(mock_session.get(MOCKED_URL))
assert response.from_cache is True
assert response.url == MOCKED_URL
assert response.status_code == 200
assert response.reason is None
assert response.encoding == 'ISO-8859-1'
assert response.headers['Content-Type'] == 'text/plain'
assert response.text == 'mock response'
assert response.created_at is not None
assert response.expires is None
assert response.is_expired is False
@pytest.mark.parametrize(
'expire_after, is_expired',
[
(None, False),
(datetime.utcnow() + timedelta(days=1), False),
(datetime.utcnow() - timedelta(days=1), True),
],
)
def test_expiration(expire_after, is_expired, mock_session):
response = CachedResponse(mock_session.get(MOCKED_URL), expire_after)
assert response.from_cache is True
assert response.is_expired == is_expired
def test_history(mock_session):
original_response = mock_session.get(MOCKED_URL)
original_response.history = [mock_session.get(MOCKED_URL)] * 3
response = CachedResponse(original_response)
assert len(response.history) == 3
assert all([isinstance(r, CachedResponse) for r in response.history])
def test_raw_response__read(mock_session):
response = CachedResponse(mock_session.get(MOCKED_URL))
assert isinstance(response.raw, CachedHTTPResponse)
assert response.raw.read(10) == b'mock respo'
assert response.raw.read(None) == b'nse'
assert response.raw.read(1) == b''
assert response.raw._fp.closed is True
def test_raw_response__close(mock_session):
response = CachedResponse(mock_session.get(MOCKED_URL))
response.close()
assert response.raw._fp.closed is True
def test_raw_response__reset(mock_session):
response = CachedResponse(mock_session.get(MOCKED_URL))
response.raw.read(None)
assert response.raw.read(1) == b''
assert response.raw._fp.closed is True
response.reset()
assert response.raw.read(None) == b'mock response'
def test_raw_response__stream(mock_session):
response = CachedResponse(mock_session.get(MOCKED_URL))
data = b''
for chunk in response.raw.stream(1):
data += chunk
assert data == b'mock response'
assert response.raw._fp.closed
def test_raw_response__iterator(mock_session):
# Set up mock response with streamed content
url = f'{MOCKED_URL}/stream'
mock_raw_response = HTTPResponse(
body=BytesIO(b'mock response'),
status=200,
request_method='GET',
decode_content=False,
preload_content=False,
)
mock_session.mock_adapter.register_uri(
'GET',
url,
status_code=200,
raw=mock_raw_response,
)
# Expect the same chunks of data from the original response and subsequent cached responses
last_request_chunks = None
for i in range(3):
response = mock_session.get(url, stream=True)
chunks = list(response.iter_lines())
if i == 0:
assert response.from_cache is False
else:
assert response.from_cache is True
assert chunks == last_request_chunks
last_request_chunks = chunks
def test_revalidate__extend_expiration(mock_session):
# Start with an expired response
response = CachedResponse(
mock_session.get(MOCKED_URL),
expire_after=datetime.utcnow() - timedelta(seconds=0.01),
)
assert response.is_expired is True
# Set expiration in the future and revalidate
is_expired = response.revalidate(datetime.utcnow() + timedelta(seconds=0.01))
assert is_expired is response.is_expired is False
sleep(0.1)
assert response.is_expired is True
def test_revalidate__shorten_expiration(mock_session):
# Start with a non-expired response
response = CachedResponse(
mock_session.get(MOCKED_URL),
expire_after=datetime.utcnow() + timedelta(seconds=1),
)
assert response.is_expired is False
# Set expiration in the past and revalidate
is_expired = response.revalidate(datetime.utcnow() - timedelta(seconds=1))
assert is_expired is response.is_expired is True
|
