]>
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 | from io import StringIO | |
9 | import unittest | |
10 | ||
11 | from requests import HTTPError | |
12 | ||
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 | ||
46 | def testCacheHitRateReport(self) -> None: | |
47 | buf = StringIO() | |
48 | ||
49 | def log(msg: str) -> None: | |
50 | print(msg, file=buf) | |
51 | with CachingFetcher("testcachehitratereportwithcl", TIMEOUT, log) as f: | |
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 | ||
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() |