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
151
152
153
|
"""BaseCache tests that use mocked responses only"""
from datetime import datetime, timedelta
from unittest.mock import patch
import pytest
from requests_cache import CachedResponse
from requests_cache.backends import BaseCache, SQLiteDict
from tests.conftest import MOCKED_URL, MOCKED_URL_HTTPS, MOCKED_URL_JSON, MOCKED_URL_REDIRECT
YESTERDAY = datetime.utcnow() - timedelta(days=1)
class TimeBomb:
"""Class that will raise an error when unpickled"""
def __init__(self):
self.foo = 'bar'
def __setstate__(self, value):
raise ValueError('Invalid response!')
def test_urls__with_invalid_response(mock_session):
responses = [mock_session.get(url) for url in [MOCKED_URL, MOCKED_URL_JSON, MOCKED_URL_HTTPS]]
responses[2] = AttributeError
with patch.object(SQLiteDict, '__getitem__', side_effect=responses):
expected_urls = [MOCKED_URL, MOCKED_URL_JSON]
assert set(mock_session.cache.urls) == set(expected_urls)
# The invalid response should be skipped, but remain in the cache for now
assert len(mock_session.cache.responses.keys()) == 3
def test_keys(mock_session):
for url in [MOCKED_URL, MOCKED_URL_JSON, MOCKED_URL_REDIRECT]:
mock_session.get(url)
all_keys = set(mock_session.cache.responses.keys()) | set(mock_session.cache.redirects.keys())
assert set(mock_session.cache.keys()) == all_keys
def test_update(mock_session):
src_cache = BaseCache()
for i in range(20):
src_cache.responses[f'key_{i}'] = f'value_{i}'
src_cache.redirects[f'key_{i}'] = f'value_{i}'
mock_session.cache.update(src_cache)
assert len(mock_session.cache.responses) == 20
assert len(mock_session.cache.redirects) == 20
def test_values(mock_session):
for url in [MOCKED_URL, MOCKED_URL_JSON, MOCKED_URL_HTTPS]:
mock_session.get(url)
responses = list(mock_session.cache.values())
assert len(responses) == 3
assert all([isinstance(response, CachedResponse) for response in responses])
@pytest.mark.parametrize('check_expiry, expected_count', [(True, 1), (False, 2)])
def test_values__with_invalid_responses(check_expiry, expected_count, mock_session):
"""values() should always exclude invalid responses, and optionally exclude expired responses"""
responses = [mock_session.get(url) for url in [MOCKED_URL, MOCKED_URL_JSON, MOCKED_URL_HTTPS]]
responses[1] = AttributeError
responses[2] = CachedResponse(expires=YESTERDAY, url='test')
with patch.object(SQLiteDict, '__getitem__', side_effect=responses):
values = mock_session.cache.values(check_expiry=check_expiry)
assert len(list(values)) == expected_count
# The invalid response should be skipped, but remain in the cache for now
assert len(mock_session.cache.responses.keys()) == 3
@pytest.mark.parametrize('check_expiry, expected_count', [(True, 2), (False, 3)])
def test_response_count(check_expiry, expected_count, mock_session):
"""response_count() should always exclude invalid responses, and optionally exclude expired
and invalid responses"""
mock_session.get(MOCKED_URL)
mock_session.get(MOCKED_URL_JSON)
mock_session.cache.responses['expired_response'] = CachedResponse(expires=YESTERDAY)
mock_session.cache.responses['invalid_response'] = TimeBomb()
assert mock_session.cache.response_count(check_expiry=check_expiry) == expected_count
def test_clear(mock_session):
mock_session.get(MOCKED_URL)
mock_session.get(MOCKED_URL_REDIRECT)
mock_session.cache.clear()
assert not mock_session.cache.has_url(MOCKED_URL)
assert not mock_session.cache.has_url(MOCKED_URL_REDIRECT)
def test_has_url(mock_session):
mock_session.get(MOCKED_URL)
assert mock_session.cache.has_url(MOCKED_URL)
assert not mock_session.cache.has_url(MOCKED_URL_REDIRECT)
def test_has_url__request_args(mock_session):
mock_session.get(MOCKED_URL, params={'foo': 'bar'})
assert mock_session.cache.has_url(MOCKED_URL, params={'foo': 'bar'})
assert not mock_session.cache.has_url(MOCKED_URL)
def test_delete_url(mock_session):
mock_session.get(MOCKED_URL)
mock_session.cache.delete_url(MOCKED_URL)
assert not mock_session.cache.has_url(MOCKED_URL)
def test_delete_url__request_args(mock_session):
mock_session.get(MOCKED_URL, params={'foo': 'bar'})
mock_session.cache.delete_url(MOCKED_URL, params={'foo': 'bar'})
assert not mock_session.cache.has_url(MOCKED_URL, params={'foo': 'bar'})
def test_delete_url__nonexistent_response(mock_session):
"""Deleting a response that was either already deleted (or never added) should fail silently"""
mock_session.cache.delete_url(MOCKED_URL)
mock_session.get(MOCKED_URL)
mock_session.cache.delete_url(MOCKED_URL)
assert not mock_session.cache.has_url(MOCKED_URL)
mock_session.cache.delete_url(MOCKED_URL) # Should fail silently
def test_delete_url__redirect(mock_session):
mock_session.get(MOCKED_URL_REDIRECT)
assert mock_session.cache.has_url(MOCKED_URL_REDIRECT)
mock_session.cache.delete_url(MOCKED_URL_REDIRECT)
assert not mock_session.cache.has_url(MOCKED_URL_REDIRECT)
def test_delete_urls(mock_session):
urls = [MOCKED_URL, MOCKED_URL_JSON, MOCKED_URL_REDIRECT]
for url in urls:
mock_session.get(url)
mock_session.cache.delete_urls(urls)
for url in urls:
assert not mock_session.cache.has_url(MOCKED_URL_REDIRECT)
def test_save_response_manual(mock_session):
response = mock_session.get(MOCKED_URL)
mock_session.cache.clear()
mock_session.cache.save_response(response)
|
