]> git.scottworley.com Git - paperdoorknob/blame_incremental - testing/fakeserver.py
FakeFetcher for faster tests
[paperdoorknob] / testing / fakeserver.py
... / ...
CommitLineData
1from typing import Any, Self
2import threading
3from http.server import BaseHTTPRequestHandler, HTTPServer
4
5
6class 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 with open("testdata/this-is-glowfic.html", "rb") as f:
20 body = f.read(9999)
21 self.send_response(self._response_code())
22 self.send_header("Content-type", "text/html")
23 self.send_header("Content-Length", str(len(body)))
24 self.end_headers()
25 self.wfile.write(body)
26 self._notify_server()
27
28
29class 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