]>
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 | with CachingFetcher("testcachehitratereportwithcl", TIMEOUT, buf) as f: | |
49 | for _ in range(7): | |
50 | f.fetch(f"http://localhost:{self._port}") | |
51 | self.assertEqual("Fetch cache hits: 6 (85.7%)\n", buf.getvalue()) | |
52 | ||
53 | def testFetchErrors(self) -> None: | |
54 | with DirectFetcher(TIMEOUT) as f: | |
55 | with self.assertRaises(HTTPError): | |
56 | f.fetch(f"http://localhost:{self._port}/not_found") | |
57 | with self.assertRaises(HTTPError): | |
58 | f.fetch(f"http://localhost:{self._port}/server_error") | |
59 | with CachingFetcher("testerrorscache", TIMEOUT) as f: | |
60 | with self.assertRaises(HTTPError): | |
61 | f.fetch(f"http://localhost:{self._port}/not_found") | |
62 | with self.assertRaises(HTTPError): | |
63 | f.fetch(f"http://localhost:{self._port}/server_error") | |
64 | ||
65 | ||
66 | if __name__ == '__main__': | |
67 | unittest.main() |