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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
"""Fixtures that will be automatically picked up by pytest
Short description:
* The ``mock_session`` fixture uses pre-configured mock requests, and should be used for unit tests.
* The ``tempfile_session`` fixture makes real HTTP requests, and should be used for integration test.
Note: The protocol ``http(s)+mock://`` helps :py:class:`requests_mock.Adapter` play nicely with
:py:class:`requests.PreparedRequest`. More info here:
https://requests-mock.readthedocs.io/en/latest/adapter.html
"""
import os
import pytest
from functools import wraps
from logging import basicConfig, getLogger
from tempfile import NamedTemporaryFile
import requests
from requests_mock import ANY as ANY_METHOD
from requests_mock import Adapter
from timeout_decorator import timeout
import requests_cache
from requests_cache.session import ALL_METHODS, CachedSession
CACHE_NAME = 'pytest_cache'
# Allow running longer stress tests with an environment variable
STRESS_TEST_MULTIPLIER = int(os.getenv('STRESS_TEST_MULTIPLIER', '1'))
N_THREADS = 2 * STRESS_TEST_MULTIPLIER
N_ITERATIONS = 4 * STRESS_TEST_MULTIPLIER
HTTPBIN_METHODS = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE']
HTTPBIN_FORMATS = [
'brotli',
'deflate',
'deny',
'encoding/utf8',
'gzip',
'html',
'image/jpeg',
'image/png',
'image/svg',
'image/webp',
'json',
'robots.txt',
'xml',
]
MOCKED_URL = 'http+mock://requests-cache.com/text'
MOCKED_URL_HTTPS = 'https+mock://requests-cache.com/text'
MOCKED_URL_JSON = 'http+mock://requests-cache.com/json'
MOCKED_URL_REDIRECT = 'http+mock://requests-cache.com/redirect'
MOCKED_URL_REDIRECT_TARGET = 'http+mock://requests-cache.com/redirect_target'
MOCK_PROTOCOLS = ['mock://', 'http+mock://', 'https+mock://']
AWS_OPTIONS = {
'endpoint_url': 'http://localhost:8000',
'region_name': 'us-east-1',
'aws_access_key_id': 'placeholder',
'aws_secret_access_key': 'placeholder',
}
# Configure logging to show log output when tests fail (or with pytest -s)
basicConfig(level='INFO')
# getLogger('requests_cache').setLevel('DEBUG')
logger = getLogger(__name__)
def httpbin(path):
"""Get the url for either a local or remote httpbin instance"""
base_url = os.getenv('HTTPBIN_URL', 'http://localhost:80').rstrip('/')
return f'{base_url}/{path}'
try:
import pytest_httpbin # noqa: F401
USE_PYTEST_HTTPBIN = os.getenv('USE_PYTEST_HTTPBIN', '').lower() == 'true'
except ImportError:
USE_PYTEST_HTTPBIN = False
@pytest.fixture(scope='session', autouse=USE_PYTEST_HTTPBIN)
def httpbin_wrapper(httpbin):
"""Allow pytest-httpbin to be used instead of the httpbin Docker container. This fixture does
not need to be used manually. It will be autoused if both:
* pytest-httpbin is installed
* The environment variable USE_PYTEST_HTTPBIN is set to 'true'
"""
logger.info('Using pytest-httpin for integration tests')
os.environ['HTTPBIN_URL'] = httpbin.url
return httpbin
@pytest.fixture(scope='function')
def mock_session() -> CachedSession:
"""Fixture for combining requests-cache with requests-mock. This will behave the same as a
CachedSession, except it will make mock requests for ``mock://`` URLs, if it hasn't been cached
already.
For example, ``mock_session.get(MOCKED_URL)`` will return a mock response on the first call,
and a cached mock response on the second call. Additional mock responses can be added via
``mock_session.mock_adapter.register_uri()``.
This uses a temporary SQLite db stored in ``/tmp``, which will be removed after the fixture has
exited.
"""
with NamedTemporaryFile(suffix='.db') as temp:
session = CachedSession(
db_path=temp.name,
backend='sqlite',
allowable_methods=ALL_METHODS,
suppress_warnings=True,
)
yield mount_mock_adapter(session)
@pytest.fixture(scope='function')
def tempfile_session() -> CachedSession:
"""Get a CachedSession using a temporary SQLite db"""
with NamedTemporaryFile(suffix='.db') as temp:
session = CachedSession(
cache_name=temp.name,
backend='sqlite',
allowable_methods=ALL_METHODS,
suppress_warnings=True,
)
yield session
@pytest.fixture(scope='function')
def installed_session() -> CachedSession:
"""Get a CachedSession using a temporary SQLite db, with global patching.
Installs cache before test and uninstalls after.
"""
with NamedTemporaryFile(suffix='.db') as temp:
requests_cache.install_cache(
cache_name=temp.name,
backend='sqlite',
allowable_methods=ALL_METHODS,
suppress_warnings=True,
)
yield requests.Session()
requests_cache.uninstall_cache()
def mount_mock_adapter(session: CachedSession) -> CachedSession:
adapter = get_mock_adapter()
for protocol in MOCK_PROTOCOLS:
session.mount(protocol, adapter)
session.mock_adapter = adapter
return session
def get_mock_adapter() -> Adapter:
"""Get a requests-mock Adapter with some URLs mocked by default"""
adapter = Adapter()
adapter.register_uri(
ANY_METHOD,
MOCKED_URL,
headers={'Content-Type': 'text/plain'},
text='mock response',
status_code=200,
)
adapter.register_uri(
ANY_METHOD,
MOCKED_URL_HTTPS,
headers={'Content-Type': 'text/plain'},
text='mock https response',
status_code=200,
)
adapter.register_uri(
ANY_METHOD,
MOCKED_URL_JSON,
headers={'Content-Type': 'application/json'},
json={'message': 'mock json response'},
status_code=200,
)
adapter.register_uri(
ANY_METHOD,
MOCKED_URL_REDIRECT,
headers={'Content-Type': 'text/plain', 'Location': MOCKED_URL_REDIRECT_TARGET},
text='mock redirect response',
status_code=302,
)
adapter.register_uri(
ANY_METHOD,
MOCKED_URL_REDIRECT_TARGET,
headers={'Content-Type': 'text/plain'},
text='mock redirected response',
status_code=200,
)
return adapter
def fail_if_no_connection(func) -> bool:
"""Decorator for testing a backend connection. This will intentionally cause a test failure if
the wrapped function doesn't have dependencies installed, doesn't connect after a short timeout,
or raises any exceptions.
This allows us to fail quickly for backends that aren't set up, rather than hanging for an
extended period of time.
"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
timeout(1.0)(func)(*args, **kwargs)
except Exception as e:
logger.error(e)
pytest.fail('Could not connect to backend')
return wrapper
|
