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
|
import collections
import email.message
import enum
import random
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, HTTPServer
from test.utils.wildcard import EQ_WILDCARD
from threading import Thread
from typing import (
Dict,
Iterable,
Iterator,
List,
NamedTuple,
Optional,
Tuple,
Type,
TypeVar,
Union,
)
from urllib.parse import ParseResult
__all__: List[str] = []
HeadersT = Union[Dict[str, List[str]], Iterable[Tuple[str, str]]]
PathQueryT = Dict[str, List[str]]
def header_items(headers: HeadersT) -> Iterable[Tuple[str, str]]:
if isinstance(headers, collections.abc.Mapping):
for header, value in headers.items():
if isinstance(value, list):
for item in value:
yield header, item
else:
yield from headers
def apply_headers_to(headers: HeadersT, handler: BaseHTTPRequestHandler) -> None:
for header, value in header_items(headers):
handler.send_header(header, value)
# handler.end_headers()
class MethodName(str, enum.Enum):
CONNECT = enum.auto()
DELETE = enum.auto()
GET = enum.auto()
HEAD = enum.auto()
OPTIONS = enum.auto()
PATCH = enum.auto()
POST = enum.auto()
PUT = enum.auto()
TRACE = enum.auto()
class MockHTTPRequest(NamedTuple):
method: MethodName
path: str
parsed_path: ParseResult
path_query: PathQueryT
headers: email.message.Message
body: Optional[bytes]
MOCK_HTTP_REQUEST_WILDCARD = MockHTTPRequest(
EQ_WILDCARD, EQ_WILDCARD, EQ_WILDCARD, EQ_WILDCARD, EQ_WILDCARD, EQ_WILDCARD
)
"""
This object should be equal to any `MockHTTPRequest` object.
"""
class MockHTTPResponse(NamedTuple):
status_code: int
reason_phrase: str
body: bytes
headers: HeadersT
def get_random_ip(ip_prefix: Optional[List[str]] = None) -> str:
if ip_prefix is None:
parts = ["127"]
for _ in range(4 - len(parts)):
parts.append(f"{random.randint(0, 255)}")
return ".".join(parts)
@contextmanager
def ctx_http_handler(
handler: Type[BaseHTTPRequestHandler], host: Optional[str] = "127.0.0.1"
) -> Iterator[HTTPServer]:
host = get_random_ip() if host is None else host
server = HTTPServer((host, 0), handler)
with ctx_http_server(server) as server:
yield server
HTTPServerT = TypeVar("HTTPServerT", bound=HTTPServer)
@contextmanager
def ctx_http_server(server: HTTPServerT) -> Iterator[HTTPServerT]:
server_thread = Thread(target=server.serve_forever)
server_thread.daemon = True
server_thread.start()
yield server
server.shutdown()
server.socket.close()
server_thread.join()
|
