momota.txt

hello, hello, hello, how low?

Python Unittest で Flask (チャットボット) の単体テスト

最近 Slack クローンの mattermost のチャットボットを Python で作った。 そのコードをテストしたかったときに unittest を覚えたのでそのメモ。

1
2

$ python -V
Python 3.6.4 :: Anaconda, Inc.

ディレクトリ構成は以下。

1
2
3
4
5
6

$ tree
.
|____log
| |____mattermost_bot.log
|____mattermostbot.py
|____test_mattermostbot.py
  • mattermostbot.py: ボット本体
  • test_mattermostbot.py: テストコード
  • log/ : ログ保存用のディレクトリ

チャットボット

mattermost の統合機能から「外向きのウェブフック」を設定する。

webhook

チャットボットは bot_name COMMAND ARGUMENT のように呼び出す。 ここではコマンドは以下の5種類をテストすることにする。

  • echo: ARGUMENT をそのまま返す
  • hoge: “hoge” と返す
  • ping: “pong :ping_pong:” と返す
  • sushi: “（っ’–‘)╮ =͟͟͞͞ :sushi: ﾌﾞｫﾝ” を返す
  • tenki: 今日と明日の天気を返す (Livedoor の天気 API を利用)

チャットボットのコード mattermostbot.py は以下。Slack のボットとしても機能するはず。

Flask を使って HTTP POST で送られてくる JSON データを mattermost から受け取って、処理している。 bot() メソッドがボットメイン処理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

import json
import logging
import requests
from flask import Flask, request
from logging.handlers import RotatingFileHandler

app = Flask(__name__)

@app.route('/bot', methods=['POST'])
def bot():
    # mattermost -> bot へ送信される JSON データの取得
    post_dict = request.form
    app.logger.info(post_dict)

    # JSONから token と text (ユーザが入力したメッセージ) を取得
    token = post_dict['token']
    income_text = post_dict['text']

    # income_text は `bot_name COMMAND ARGUMENT` のような形式なので
    # 半角スペースで分割し、それぞれの要素を変数に格納する
    text_array = income_text.split(' ')
    bot_name = text_array[0]
    command = text_array[1]
    arg = " ".join(text_array[2:])

    payload_text = ""

    # command によって処理を分岐する
    if command == "echo":
        payload_text = echo(arg)
    elif command == "hoge":
        payload_text = hoge()
    elif command == "ping":
        payload_text = pong()
    elif command == "sushi":
        payload_text = sushi()
    elif command == "tenki":
        payload_text = tenki()

    app.logger.info(payload_text)

    # レスポンス用の JSON を組み立てる
    payload = {
        'username': bot_name,
        'icon_url': 'http://your-server/images/bot_icon.png',
        'text': payload_text,
        'MATTERMOST_TOKEN': token
    }
    json_payload = json.dumps(payload)

    return json_payload

# ログ出力用メソッド
def log(app):
    handler = RotatingFileHandler('log/mattermost_bot.log', maxBytes=10000, backupCount=2)
    handler.setLevel(logging.INFO)
    formatter = logging.Formatter('%(asctime)s\t%(lineno)d\t%(levelname)s\t%(name)s\t%(message)s')
    handler.setFormatter(formatter)
    app.logger.addHandler(handler)

# -------------------------------------------------------
# echo command
# -------------------------------------------------------
def echo(text):
    return text

# -------------------------------------------------------
# hoge command
# -------------------------------------------------------
def hoge():
    return "hoge"

# -------------------------------------------------------
# ping command
# -------------------------------------------------------
def pong():
    pong_msg = "pong :ping_pong:"
    return pong_msg

# -------------------------------------------------------
# sushi command
# -------------------------------------------------------
def sushi():
    return "（っ'-')╮ =͟͟͞͞ :sushi: ﾌﾞｫﾝ"

# -------------------------------------------------------
# tenki command
# -------------------------------------------------------
def tenki():
    api_url = "http://weather.livedoor.com/forecast/webservice/json/v1?city="
    # 横浜の city id
    # その他のIDはここから探して: http://weather.livedoor.com/forecast/rss/primary_area.xml
    city_id = "140010"
    api_res = requests.get(api_url + city_id)
    json_res = json.loads(api_res.text)

    today = json_res['forecasts'][0]
    tomorrow = json_res['forecasts'][1]

    tenki_info = ""
    tenki_info += format_tenki_json(today)
    tenki_info += format_tenki_json(tomorrow)
    return tenki_info

def format_tenki_json(j):
    tenki_info = "# "
    tenki_info += j['dateLabel'] + " (" + j['date'] + ") の横浜の天気は" + j['telop']
    tenki_info += " ![](" + j['image']['url'] + ")\n"

    t_min = j['temperature']['min']
    t_max = j['temperature']['max']
    t_min_str = " -- " if t_min is None else str(t_min['celsius'])
    t_max_str = " -- " if t_max is None else str(t_max['celsius'])
    tenki_info += "- 最低気温は " + t_min_str +"℃\n"
    tenki_info += "- 最高気温は " + t_max_str  +"℃\n\n"
    return tenki_info


if __name__ == '__main__':
    log(app)
    app.debug = True
    app.run(host='0.0.0.0')

テスト

以下を参考にした。

Python の Flask で作ったアプリケーションをテストする | CUBE SUGAR STORAGE

各チャットボットコマンドに対応するテストケースを以下のように書ける。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

import datetime
import json
import mattermostbot
import unittest

class TestMattermostBot(unittest.TestCase):
    # mattermost -> bot へのリクエスト JSON のダミーデータ雛形
    data = dict(
            channel_id = 'xxxxxxxxxxxxxxxxxxxxxxxxxx',
            channel_name = 'some_channel',
            file_ids = '',
            post_id = 'xxxxxxxxxxxxxxxxxxxxxxxxxx',
            team_domain = 'some_team',
            team_id = 'xxxxxxxxxxxxxxxxxxxxxxxxxx',
            text = 'your_bot_name ',
            timestamp = '9999999999',
            token = 'xxxxxxxxxxxxxxxxxxxxxxxxxx',
            trigger_word = 'your_bot_name',
            user_id = 'xxxxxxxxxxxxxxxxxxxxxxxxxx',
            user_name = 'your_name')

    def setUp(self):
        self.app = mattermostbot.app.test_client()

    def test_post_echo(self):
        response = TestMattermostBot.req(self, 'echo fuga')
        response_data = json.loads(response.data)
        assert response.status_code == 200
        assert response_data['text'] == 'fuga'

    def test_post_hoge(self):
        response = TestMattermostBot.req(self, 'hoge')
        response_data = json.loads(response.data)
        assert response.status_code == 200
        assert response_data['text'] == 'hoge'

    def test_post_ping(self):
        response = TestMattermostBot.req(self, 'ping')
        response_data = json.loads(response.data)
        assert response.status_code == 200
        assert response_data['text'] == 'pong :ping_pong:'

    def test_post_sushi(self):
        response = TestMattermostBot.req(self, 'sushi')
        response_data = json.loads(response.data)
        assert response.status_code == 200
        assert response_data['text'] == "（っ'-')╮ =͟͟͞͞ :sushi: ﾌﾞｫﾝ"

    def test_post_tenki(self):
        response = TestMattermostBot.req(self, 'tenki')
        response_data = json.loads(response.data)
        assert response.status_code == 200
        assert datetime.date.today().strftime('%Y-%m-%d') in response_data['text']

    def req(self, command):
        data = TestMattermostBot.data.copy()
        data['text'] = data['text'] + command
        return self.app.post('/bot', data=data)

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

テストを実行するとこんな感じになる。

1
2
3
4
5
6

$ python test_mattermostbot.py
.....
----------------------------------------------------------------------
Ran 5 tests in 0.115s

OK

Comments