git.scottworley.com Git - paperdoorknob/blame_incremental

... / ...

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	import unittest
	9	import threading
	10	from http.server import BaseHTTPRequestHandler, HTTPServer
	11	import requests
	12	import requests_cache
	13	import paperdoorknob
	14
	15	TIMEOUT = 8
	16
	17
	18	class FakeGlowficHTTPRequestHandler(BaseHTTPRequestHandler):
	19
	20	def _notify_test(self) -> None:
	21	raise NotImplementedError()
	22
	23	def _response_code(self) -> int:
	24	if self.path == "/not_found":
	25	return 404
	26	if self.path == "/server_error":
	27	return 500
	28	return 200
	29
	30	def do_GET(self) -> None:
	31	body = b'''<html>
	32	<body>
	33	<div class="post-container post-post">
	34	<div class="post-edit-box">We don't want edit boxes</div>
	35	This is glowfic
	36	<div class="post-footer">We don't want footers</div>
	37	</div>
	38	<div class="flat-post-replies">
	39	<div class="post-container post-reply">
	40	<div class="post-edit-box">We don't want edit boxes</div>
	41	You sure?
	42	<div class="post-footer">We don't want footers</div>
	43	</div>
	44	<div class="post-container post-reply">
	45	Pretty sure.
	46	</div>
	47	</div>
	48	</body>
	49	</html>'''
	50	self.send_response(self._response_code())
	51	self.send_header("Content-type", "text/html")
	52	self.send_header("Content-Length", str(len(body)))
	53	self.end_headers()
	54	self.wfile.write(body)
	55	self._notify_test()
	56
	57
	58	class TestFetch(unittest.TestCase):
	59	def _port(self) -> int:
	60	port = self._web_server.socket.getsockname()[1]
	61	assert isinstance(port, int)
	62	return port
	63
	64	def _count_request(self) -> None:
	65	self._request_counter += 1
	66
	67	def setUp(self) -> None:
	68	self._request_counter = 0
	69	handler = type("Handler", (FakeGlowficHTTPRequestHandler,), {
	70	'_notify_test': lambda _: self._count_request()})
	71	self._web_server = HTTPServer(('', 0), handler)
	72	self._thread = threading.Thread(target=self._web_server.serve_forever)
	73	self._thread.start()
	74
	75	def tearDown(self) -> None:
	76	self._web_server.shutdown()
	77	self._thread.join()
	78	self._web_server.server_close()
	79
	80	def testFetch(self) -> None:
	81	with requests.session() as s:
	82	paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT)
	83	self.assertEqual(self._request_counter, 1)
	84	paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT)
	85	self.assertEqual(self._request_counter, 2)
	86
	87	def testFetchCaching(self) -> None:
	88	with requests_cache.CachedSession() as s:
	89	paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT)
	90	self.assertEqual(self._request_counter, 1)
	91	paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT)
	92	self.assertEqual(self._request_counter, 1)
	93
	94	def testFetchPersistentCaching(self) -> None:
	95	with requests_cache.CachedSession() as s:
	96	paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT)
	97	self.assertEqual(self._request_counter, 1)
	98	with requests_cache.CachedSession() as s:
	99	paperdoorknob.fetch(f"http://localhost:{self._port()}", s, TIMEOUT)
	100	self.assertEqual(self._request_counter, 1)
	101
	102	def testProcess(self) -> None:
	103	with requests.session() as s:
	104	replies = paperdoorknob.process(
	105	f"http://localhost:{self._port()}", s, TIMEOUT)
	106	self.assertEqual([r.text.strip() for r in replies],
	107	["This is glowfic", "You sure?", "Pretty sure."])
	108
	109	def testFetchErrors(self) -> None:
	110	with requests.session() as s:
	111	with self.assertRaises(requests.HTTPError):
	112	paperdoorknob.fetch(
	113	f"http://localhost:{self._port()}/not_found", s, TIMEOUT)
	114	with self.assertRaises(requests.HTTPError):
	115	paperdoorknob.fetch(
	116	f"http://localhost:{self._port()}/server_error", s, TIMEOUT)
	117
	118
	119	if __name__ == '__main__':
	120	unittest.main()