]>
Commit | Line | Data |
---|---|---|
1 | from typing import Any, Self | |
2 | import threading | |
3 | from http.server import BaseHTTPRequestHandler, HTTPServer | |
4 | ||
5 | ||
6 | class FakeGlowficHTTPRequestHandler(BaseHTTPRequestHandler): | |
7 | ||
8 | def _notify_server(self) -> None: | |
9 | raise NotImplementedError() | |
10 | ||
11 | def _response_code(self) -> int: | |
12 | if self.path == "/not_found": | |
13 | return 404 | |
14 | if self.path == "/server_error": | |
15 | return 500 | |
16 | return 200 | |
17 | ||
18 | def do_GET(self) -> None: | |
19 | self._notify_server() | |
20 | with open("testdata/this-is-glowfic.html", "rb") as f: | |
21 | body = f.read(9999) | |
22 | self.send_response(self._response_code()) | |
23 | self.send_header("Content-type", "text/html") | |
24 | self.send_header("Content-Length", str(len(body))) | |
25 | self.end_headers() | |
26 | self.wfile.write(body) | |
27 | ||
28 | ||
29 | class FakeGlowficServer(): | |
30 | def __init__(self) -> None: | |
31 | self._request_counter = 0 | |
32 | handler = type("Handler", (FakeGlowficHTTPRequestHandler,), { | |
33 | '_notify_server': lambda _: self._count_request()}) | |
34 | self._web_server = HTTPServer(('', 0), handler) | |
35 | self._thread = threading.Thread(target=self._web_server.serve_forever) | |
36 | ||
37 | def __enter__(self) -> Self: | |
38 | self._thread.start() | |
39 | return self | |
40 | ||
41 | def __exit__(self, *_: Any) -> None: | |
42 | self._web_server.shutdown() | |
43 | self._thread.join() | |
44 | self._web_server.server_close() | |
45 | ||
46 | def _count_request(self) -> None: | |
47 | self._request_counter += 1 | |
48 | ||
49 | def port(self) -> int: | |
50 | p = self._web_server.socket.getsockname()[1] | |
51 | assert isinstance(p, int) | |
52 | return p | |
53 | ||
54 | def request_count(self) -> int: | |
55 | return self._request_counter |