Repository navigation
Expand file tree
/
Copy pathtest_blocking_httpserver.py
More file actions
155 lines (106 loc) · 4.71 KB
/
Copy pathtest_blocking_httpserver.py
File metadata and controls
155 lines (106 loc) · 4.71 KB
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
from contextlib import contextmanager
from copy import deepcopy
from multiprocessing.pool import ThreadPool
from urllib.parse import urlparse
import pytest
import requests
from pytest_httpserver import BlockingHTTPServer
from pytest_httpserver import CaseInsensitiveHeaderValueMatcher
@contextmanager
def when_a_request_is_being_sent_to_the_server(request):
with ThreadPool(1) as pool:
yield pool.apply_async(requests.request, kwds=request)
def then_the_server_gets_the_request(server, request):
request = deepcopy(request)
replace_url_with_uri(request)
return server.assert_request(**request)
def replace_url_with_uri(request):
request["uri"] = get_uri(request["url"])
del request["url"]
def get_uri(url):
url = urlparse(url)
return "?".join(item for item in [url.path, url.query] if item)
def when_the_server_responds_to(client_connection, response):
client_connection.respond_with_json(response)
def then_the_response_is_got_from(server_connection, response):
assert server_connection.get(timeout=9).json() == response
@pytest.fixture
def httpserver():
server = BlockingHTTPServer(timeout=1)
server.start()
yield server
server.clear()
if server.is_running():
server.stop()
def test_behave_workflow(httpserver: BlockingHTTPServer):
request = dict(
method="GET",
url=httpserver.url_for("/my/path"),
)
with when_a_request_is_being_sent_to_the_server(request) as server_connection:
client_connection = then_the_server_gets_the_request(httpserver, request)
response = {"foo": "bar"}
when_the_server_responds_to(client_connection, response)
then_the_response_is_got_from(server_connection, response)
def test_form_request(httpserver: BlockingHTTPServer):
request = dict(method="POST", url=httpserver.url_for("/form"), data={"foo": "bar"})
with when_a_request_is_being_sent_to_the_server(request) as server_connection:
handler = httpserver.assert_request("/form", method="POST", data_form={"foo": "bar"})
handler.respond_with_data("ok")
response = server_connection.get(timeout=9)
assert response.status_code == 200
assert response.text == "ok"
httpserver.check()
def test_case_insensitive_header_matcher(httpserver: BlockingHTTPServer):
request = dict(
method="GET",
url=httpserver.url_for("/"),
headers={"Authorization": 'Digest a="1", b="2"'},
)
with when_a_request_is_being_sent_to_the_server(request) as server_connection:
client_connection = httpserver.assert_request(
uri="/",
headers={"authorization": 'Digest b="2", a="1"'},
header_value_matcher=CaseInsensitiveHeaderValueMatcher(),
)
client_connection.respond_with_data("OK")
assert server_connection.get(timeout=9).text == "OK"
def test_raises_assertion_error_when_request_does_not_match(httpserver: BlockingHTTPServer):
request = dict(
method="GET",
url=httpserver.url_for("/my/path"),
)
with when_a_request_is_being_sent_to_the_server(request):
with pytest.raises(AssertionError) as exc:
httpserver.assert_request(uri="/not/my/path/")
assert "/not/my/path/" in str(exc)
assert "does not match" in str(exc)
def test_raises_assertion_error_when_request_was_not_sent(httpserver: BlockingHTTPServer):
with pytest.raises(AssertionError) as exc:
httpserver.assert_request(uri="/my/path/", timeout=1)
assert "/my/path/" in str(exc)
assert "timed out" in str(exc)
def test_ignores_when_request_is_not_asserted(httpserver: BlockingHTTPServer):
request = dict(
method="GET",
url=httpserver.url_for("/my/path"),
)
with when_a_request_is_being_sent_to_the_server(request) as server_connection:
assert (
server_connection.get(timeout=9).text == "No handler found for request "
f"<Request 'http://localhost:{httpserver.port}/my/path' [GET]> with data b''."
)
def test_raises_assertion_error_when_request_was_not_responded(httpserver: BlockingHTTPServer):
request = dict(
method="GET",
url=httpserver.url_for("/my/path"),
)
with when_a_request_is_being_sent_to_the_server(request):
then_the_server_gets_the_request(httpserver, request)
httpserver.stop() # waiting for timeout of waiting for the response
with pytest.raises(AssertionError) as exc:
httpserver.check_assertions()
assert "/my/path" in str(exc)
assert "no response" in str(exc).lower()
def test_repr(httpserver: BlockingHTTPServer):
assert repr(httpserver) == f"<BlockingHTTPServer host={httpserver.host} port={httpserver.port}>"