Pythonのhttp.serverを利用してWebスクレイピングのunittestを書く

- はじめに -

「Webスクレイピングで情報を収集する」という内容は多い。

しかし、Webスクレイピングのコードは肥大化しやすいだけでなく、細かな変更が多くなる。
テストを書いて変更の影響をちゃんと見ておく必要性が高い。

unittestとhttp.serverを使ったテストの実装についてメモしておく。

参考：python - How to stop BaseHTTPServer.serve_forever() in a BaseHTTPRequestHandler subclass? - Stack Overflow

- http.server -

http.serverはPython 2.xではSimpleHTTPServerと呼ばれていたもの。
(http.serverよりSimpleHTTPServerの方がググラビリティ高いかも)

Webサービス等の開発用にローカルサーバとして利用している人も多い。

21.22. http.server — HTTP サーバ — Python 3.6.1 ドキュメント

コマンドでの実行も可能だが、Pythonからだとハンドラを指定したHTTPServerを以下のように記述し、簡易にサーバを立てる事ができる。

from http.server import HTTPServer, SimpleHTTPRequestHandler

httpd = HTTPServer(("localhost", 8888), SimpleHTTPRequestHandler)
httpd.serve_forever()

これによって http://localhost:8888 に、ローカルでのServer環境が整う。

- unittestにかませる -

serve_foreverによって起動されるサーバ(SocketServer.py)は以下のようにWhileで実行されている。

def serve_forever(self):
    """Handle one request at a time until doomsday."""
    while 1:
        self.handle_request()

なのでwrapperとなるclassを書いてやって、server.shutdown及びserver.server_closeを実行できるようによしなに書いておけば良い。

また、HTTPServerが動いているThreadでPythonのコードを動かすのは少し困難なので、別のThreadで動かしてやる。

別ThreadでHTTPServerを実行し、requestを使ってアクセスするunittestは以下のような感じ。

from http.server import HTTPServer, SimpleHTTPRequestHandler
import threading
import unittest

HOST = "localhost"
PORT = 8888

class StoppableHTTPServer(HTTPServer):
    """
    ThreadでSimpleHTTPServerを動かすためのwrapper class.
    Ctrl + Cで終了されるとThreadだけが死んで残る.
    KeyboardInterruptはpassする.
    """
    def run(self):
        try:
            self.serve_forever()
        except KeyboardInterrupt:
            pass
        finally:
            self.server_close()

class TestWebScraping(unittest.TestCase):
    def setUp(self):
        """Use unittest setUp method."""
        self.server = StoppableHTTPServer((HOST, PORT), SimpleHTTPRequestHandler)
        self.url = "http://{}:{}/test_case/index.html".format(HOST, PORT)
        self.thread = threading.Thread(None, self.server.run)
        self.thread.start()

    def test_requests_get(self):
        """
        requestsモジュールでURLを叩くunittest.
        contentが返ってくるか.
        """
        r = requests.get(self.url)
        self.assertEqual(type(resp.content), str)

    def tearDown(self):
        """Use unittest tearDown method."""
        self.server.shutdown()
        self.thread.join()