]>
Commit | Line | Data |
---|---|---|
1 | # paperdoorknob: Print glowfic | |
2 | # | |
3 | # This program is free software: you can redistribute it and/or modify it | |
4 | # under the terms of the GNU General Public License as published by the | |
5 | # Free Software Foundation, version 3. | |
6 | ||
7 | ||
8 | import unittest | |
9 | import threading | |
10 | from http.server import BaseHTTPRequestHandler, HTTPServer | |
11 | import requests | |
12 | import requests_cache | |
13 | import paperdoorknob | |
14 | ||
15 | TIMEOUT = 8 | |
16 | ||
17 | ||
18 | class FakeGlowficHTTPRequestHandler(BaseHTTPRequestHandler): | |
19 | ||
20 | def _notify_test(self) -> None: | |
21 | raise NotImplementedError() | |
22 | ||
23 | def _response_code(self) -> int: | |
24 | if self.path == "/not_found": | |
25 | return 404 | |
26 | if self.path == "/server_error": | |
27 | return 500 | |
28 | return 200 | |
29 | ||
30 | def do_GET(self) -> None: | |
31 | body = b'<html><body>This is glowfic</body></html>' | |
32 | self.send_response(self._response_code()) | |
33 | self.send_header("Content-type", "text/html") | |
34 | self.send_header("Content-Length", str(len(body))) | |
35 | self.end_headers() | |
36 | self.wfile.write(body) | |
37 | self._notify_test() | |
38 | ||
39 | ||
40 | class TestFetch(unittest.TestCase): | |
41 | def _port(self) -> int: | |
42 | port = self._web_server.socket.getsockname()[1] | |
43 | assert isinstance(port, int) | |
44 | return port | |
45 | ||
46 | def _count_request(self) -> None: | |
47 | self._request_counter += 1 | |
48 | ||
49 | def setUp(self) -> None: | |
50 | self._request_counter = 0 | |
51 | handler = type("Handler", (FakeGlowficHTTPRequestHandler,), { | |
52 | '_notify_test': lambda _: self._count_request()}) | |
53 | self._web_server = HTTPServer(('', 0), handler) | |
54 | self._thread = threading.Thread(target=self._web_server.serve_forever) | |
55 | self._thread.start() | |
56 | ||
57 | def tearDown(self) -> None: | |
58 | self._web_server.shutdown() | |
59 | self._thread.join() | |
60 | self._web_server.server_close() | |
61 | ||
62 | def testFetch(self) -> None: | |
63 | with requests.session() as s: | |
64 | paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT) | |
65 | self.assertEqual(self._request_counter, 1) | |
66 | paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT) | |
67 | self.assertEqual(self._request_counter, 2) | |
68 | ||
69 | def testFetchCaching(self) -> None: | |
70 | with requests_cache.CachedSession() as s: | |
71 | paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT) | |
72 | self.assertEqual(self._request_counter, 1) | |
73 | paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT) | |
74 | self.assertEqual(self._request_counter, 1) | |
75 | ||
76 | def testFetchPersistentCaching(self) -> None: | |
77 | with requests_cache.CachedSession() as s: | |
78 | paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT) | |
79 | self.assertEqual(self._request_counter, 1) | |
80 | with requests_cache.CachedSession() as s: | |
81 | paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT) | |
82 | self.assertEqual(self._request_counter, 1) | |
83 | ||
84 | def testFetchErrors(self) -> None: | |
85 | with requests.session() as s: | |
86 | with self.assertRaises(requests.HTTPError): | |
87 | paperdoorknob.fetch( | |
88 | f"http://localhost:{self._port()}/not_found", s, TIMEOUT) | |
89 | with self.assertRaises(requests.HTTPError): | |
90 | paperdoorknob.fetch( | |
91 | f"http://localhost:{self._port()}/server_error", s, TIMEOUT) | |
92 | ||
93 | ||
94 | if __name__ == '__main__': | |
95 | unittest.main() |