]>
Commit | Line | Data |
---|---|---|
705973e7 SW |
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 | ||
75c5665e | 8 | from io import StringIO |
705973e7 | 9 | import unittest |
75c5665e | 10 | |
705973e7 | 11 | from requests import HTTPError |
75c5665e | 12 | |
705973e7 SW |
13 | from testing.fakeserver import FakeGlowficServer |
14 | from fetch import CachingFetcher, DirectFetcher | |
15 | ||
16 | TIMEOUT = 8 | |
17 | ||
18 | ||
19 | class TestFetch(unittest.TestCase): | |
20 | def setUp(self) -> None: | |
21 | self._server = self.enterContext(FakeGlowficServer()) | |
22 | self._port = self._server.port() | |
23 | ||
24 | def testDirectFetch(self) -> None: | |
25 | with DirectFetcher(TIMEOUT) as f: | |
26 | f.fetch(f"http://localhost:{self._port}") | |
27 | self.assertEqual(self._server.request_count(), 1) | |
28 | f.fetch(f"http://localhost:{self._port}") | |
29 | self.assertEqual(self._server.request_count(), 2) | |
30 | ||
31 | def testFetchCaching(self) -> None: | |
32 | with CachingFetcher("testcache", TIMEOUT) as f: | |
33 | f.fetch(f"http://localhost:{self._port}") | |
34 | self.assertEqual(self._server.request_count(), 1) | |
35 | f.fetch(f"http://localhost:{self._port}") | |
36 | self.assertEqual(self._server.request_count(), 1) | |
37 | ||
38 | def testFetchPersistentCaching(self) -> None: | |
39 | with CachingFetcher("testpersistentcache", TIMEOUT) as f: | |
40 | f.fetch(f"http://localhost:{self._port}") | |
41 | self.assertEqual(self._server.request_count(), 1) | |
42 | with CachingFetcher("testpersistentcache", TIMEOUT) as f: | |
43 | f.fetch(f"http://localhost:{self._port}") | |
44 | self.assertEqual(self._server.request_count(), 1) | |
45 | ||
75c5665e SW |
46 | def testCacheHitRateReport(self) -> None: |
47 | buf = StringIO() | |
47e94008 SW |
48 | |
49 | def log(msg: str) -> None: | |
50 | print(msg, file=buf) | |
51 | with CachingFetcher("testcachehitratereportwithcl", TIMEOUT, log) as f: | |
75c5665e SW |
52 | for _ in range(7): |
53 | f.fetch(f"http://localhost:{self._port}") | |
54 | self.assertEqual("Fetch cache hits: 6 (85.7%)\n", buf.getvalue()) | |
55 | ||
705973e7 SW |
56 | def testFetchErrors(self) -> None: |
57 | with DirectFetcher(TIMEOUT) as f: | |
58 | with self.assertRaises(HTTPError): | |
59 | f.fetch(f"http://localhost:{self._port}/not_found") | |
60 | with self.assertRaises(HTTPError): | |
61 | f.fetch(f"http://localhost:{self._port}/server_error") | |
62 | with CachingFetcher("testerrorscache", TIMEOUT) as f: | |
63 | with self.assertRaises(HTTPError): | |
64 | f.fetch(f"http://localhost:{self._port}/not_found") | |
65 | with self.assertRaises(HTTPError): | |
66 | f.fetch(f"http://localhost:{self._port}/server_error") | |
67 | ||
68 | ||
69 | if __name__ == '__main__': | |
70 | unittest.main() |