git.scottworley.com Git - paperdoorknob/blob - fetch_test.py

   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()