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
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
|
import asyncio
from collections.abc import Sequence
import json
from typing import Mapping, Optional, Tuple
import pytest
import pook
from pook.exceptions import PookNoMatches
class StandardTests:
is_async: bool = False
loop: asyncio.AbstractEventLoop
async def amake_request(
self,
method: str,
url: str,
content: Optional[bytes] = None,
headers: Optional[Sequence[tuple[str, str]]] = None,
) -> Tuple[int, Optional[bytes], Mapping[str, str]]:
raise NotImplementedError(
"Sub-classes for async transports must implement `amake_request`"
)
def make_request(
self,
method: str,
url: str,
content: Optional[bytes] = None,
headers: Optional[Sequence[tuple[str, str]]] = None,
) -> Tuple[int, Optional[bytes], Mapping[str, str]]:
if self.is_async:
return self.loop.run_until_complete(
self.amake_request(method, url, content, headers)
)
raise NotImplementedError("Sub-classes must implement `make_request`")
@pytest.fixture(autouse=True, scope="class")
def _loop(self, request):
if self.is_async:
request.cls.loop = asyncio.new_event_loop()
yield
request.cls.loop.close()
else:
yield
@pytest.fixture
def url_404(self, httpbin):
"""404 httpbin URL.
Useful in tests if pook is configured to reply 200, and the status is checked.
If pook does not match the request (and if that was the intended behaviour)
then the 404 status code makes that obvious!"""
return f"{httpbin.url}/status/404"
@pytest.fixture
def url_500(self, httpbin):
return f"{httpbin.url}/status/500"
@pytest.fixture
def url_401(self, httpbin):
return httpbin + "/status/401"
@pytest.mark.pook
def test_activate_deactivate(self, url_404):
"""Deactivating pook allows requests to go through."""
pook.get(url_404).reply(200).body("hello from pook")
status, body, *_ = self.make_request("GET", url_404)
assert status == 200
assert body == b"hello from pook"
pook.disable()
status, body, *_ = self.make_request("GET", url_404)
assert status == 404
@pytest.mark.pook(allow_pending_mocks=True)
def test_network_mode(self, url_404, url_500):
"""Enabling network mode allows requests to pass through even if no mock is matched."""
pook.get(url_404).reply(200).body("hello from pook")
pook.enable_network()
# Avoid matching the mocks
status, *_ = self.make_request("POST", url_500)
assert status == 500
mocked_status, mocked_body, *_ = self.make_request("GET", url_404)
assert mocked_status == 200
assert mocked_body == b"hello from pook"
@pytest.mark.pook(allow_pending_mocks=True)
def test_network_mode_hostname(self, url_401):
example_com = "http://example.com"
pook.get(example_com).header("x-pook", "1").reply(200).body("hello from pook")
# httpbin runs on loopback
pook.enable_network("127.0.0.1")
httpbin_status, *_ = self.make_request("GET", url_401)
# network is enabled for httpbin hostname so it goes through
assert httpbin_status == 401
with pytest.raises(PookNoMatches):
# Make the request without query params to avoid matching the mock
# which should raise a no match exception, as network mode is not enabled
# for example.com hostname
self.make_request("GET", example_com)
# this matches the mock on the header, so gets 200 with the hello from pook body
example_status, body, *_ = self.make_request(
"GET", example_com, headers=[("x-pook", "1")]
)
assert example_status == 200
assert body == b"hello from pook"
@pytest.mark.pook(allow_pending_mocks=True)
def test_multiple_network_filters(self, url_401):
"""When multiple network filters are added, only one is required to match for the
request to be allowed through the network."""
def has_x_header(request: pook.Request):
return request.headers.get("x-pook") == "x"
def has_y_header(request: pook.Request):
return request.headers.get("x-pook") == "y"
pook.enable_network()
pook.use_network_filter(has_x_header, has_y_header)
pook.get(url_401).header("x-pook", "z").reply(200).body("hello from pook")
# Network filter matches, so request is allowed despite not matching a mock
x_status, *_ = self.make_request("GET", url_401, headers=[("x-pook", "x")])
assert x_status == 401
# Network filter matches, so request is allowed despite not matching a mock
y_status, *_ = self.make_request("GET", url_401, headers=[("x-pook", "y")])
assert y_status == 401
# Mock matches, so the response is mocked
z_status, z_body, *_ = self.make_request(
"GET", url_401, headers=[("x-pook", "z")]
)
assert z_status == 200
assert z_body == b"hello from pook"
@pytest.mark.pook
def test_json_request(self, url_404):
"""JSON request bodies are correctly matched."""
json_request = {"hello": "json-request"}
pook.get(url_404).json(json_request).reply(200).body("hello from pook")
status, body, *_ = self.make_request(
"GET", url_404, json.dumps(json_request).encode()
)
assert status == 200
assert body == b"hello from pook"
@pytest.mark.pook
def test_json_response(self, url_404):
"""JSON responses are correctly mocked."""
json_response = {"hello": "json-request"}
pook.get(url_404).reply(200).json(json_response)
status, body, *_ = self.make_request("GET", url_404)
assert status == 200
assert body
assert json.loads(body) == json_response
@pytest.mark.pook
def test_json_request_and_response(self, url_404):
"""JSON requests and responses do not interfere with each other."""
json_request = {"id": "123abc"}
json_response = {"title": "123abc title"}
pook.get(url_404).json(json_request).reply(200).json(json_response)
status, body, *_ = self.make_request(
"GET", url_404, content=json.dumps(json_request).encode()
)
assert status == 200
assert body
assert json.loads(body) == json_response
@pytest.mark.pook
def test_header_sent(self, url_404):
"""Sent headers can be matched."""
headers = [("x-hello", "from pook")]
pook.get(url_404).header("x-hello", "from pook").reply(200).body(
"hello from pook"
)
status, body, _ = self.make_request("GET", url_404, headers=headers)
assert status == 200
assert body == b"hello from pook"
@pytest.mark.pook
def test_mocked_response_headers(self, url_404):
"""Mocked response headers are appropriately returned."""
pook.get(url_404).reply(200).header("x-hello", "from pook")
status, _, headers = self.make_request("GET", url_404)
assert status == 200
assert headers["x-hello"] == "from pook"
@pytest.mark.pook
def test_mutli_value_headers(self, url_404):
"""Multi-value headers can be matched."""
match_headers = [("x-hello", "from pook"), ("x-hello", "another time")]
pook.get(url_404).header("x-hello", "from pook, another time").reply(200)
status, *_ = self.make_request("GET", url_404, headers=match_headers)
assert status == 200
@pytest.mark.pook
def test_mutli_value_response_headers(self, url_404):
"""Multi-value response headers can be mocked."""
pook.get(url_404).reply(200).header("x-hello", "from pook").header(
"x-hello", "another time"
)
status, _, headers = self.make_request("GET", url_404)
assert status == 200
assert headers["x-hello"] == "from pook, another time"
@pytest.mark.pook(allow_pending_mocks=True)
def test_unmatched_headers_none_sent(self, url_404):
"""Header matching will run, but not match, on requests that send no headers."""
pook.get(url_404).header("x-hello", "from pook").reply(200)
with pytest.raises(PookNoMatches):
self.make_request("GET", url_404)
|
