Always have Thread.__init__ fetch the HTML

[paperdoorknob] / fetch_test.py

# paperdoorknob: Print glowfic
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, version 3.


from io import StringIO
import unittest

from requests import HTTPError

from testing.fakeserver import FakeGlowficServer
from fetch import CachingFetcher, DirectFetcher

TIMEOUT = 8


class TestFetch(unittest.TestCase):
    def setUp(self) -> None:
        self._server = self.enterContext(FakeGlowficServer())
        self._port = self._server.port()

    def testDirectFetch(self) -> None:
        with DirectFetcher(TIMEOUT) as f:
            f.fetch(f"http://localhost:{self._port}")
            self.assertEqual(self._server.request_count(), 1)
            f.fetch(f"http://localhost:{self._port}")
            self.assertEqual(self._server.request_count(), 2)

    def testFetchCaching(self) -> None:
        with CachingFetcher("testcache", TIMEOUT) as f:
            f.fetch(f"http://localhost:{self._port}")
            self.assertEqual(self._server.request_count(), 1)
            f.fetch(f"http://localhost:{self._port}")
            self.assertEqual(self._server.request_count(), 1)

    def testFetchPersistentCaching(self) -> None:
        with CachingFetcher("testpersistentcache", TIMEOUT) as f:
            f.fetch(f"http://localhost:{self._port}")
            self.assertEqual(self._server.request_count(), 1)
        with CachingFetcher("testpersistentcache", TIMEOUT) as f:
            f.fetch(f"http://localhost:{self._port}")
            self.assertEqual(self._server.request_count(), 1)

    def testCacheHitRateReport(self) -> None:
        buf = StringIO()

        def log(msg: str) -> None:
            print(msg, file=buf)
        with CachingFetcher("testcachehitratereportwithcl", TIMEOUT, log) as f:
            for _ in range(7):
                f.fetch(f"http://localhost:{self._port}")
        self.assertEqual("Fetch cache hits: 6 (85.7%)\n", buf.getvalue())

    def testFetchErrors(self) -> None:
        with DirectFetcher(TIMEOUT) as f:
            with self.assertRaises(HTTPError):
                f.fetch(f"http://localhost:{self._port}/not_found")
            with self.assertRaises(HTTPError):
                f.fetch(f"http://localhost:{self._port}/server_error")
        with CachingFetcher("testerrorscache", TIMEOUT) as f:
            with self.assertRaises(HTTPError):
                f.fetch(f"http://localhost:{self._port}/not_found")
            with self.assertRaises(HTTPError):
                f.fetch(f"http://localhost:{self._port}/server_error")


if __name__ == '__main__':
    unittest.main()