Pytestを使ってAPNsのテストを実装した時のメモ


APNsのレスポンスをチェックしたくて色々と調べた結果、以下の方法、ちょっと力技なやり方だと思うが、これで実装してみた。

pyapnsがtestを意識したインターフェースになっていればよいがそうでもないので、global変数を使って、 コールバックの引数をMainThreadから参照できるようにしてみた。

pyapnsを拡張してごにょごにょしても良さげだが、このやり方が一番手っ取り早いと思うので、このやり方にしてみることに。

もっと良いやり方があればどなたか教えて欲しい。

## サンプルコード

```python:test_apns.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

from apns import APNs, Payload
import logging
import time

logging.basicConfig(level=logging.DEBUG)
mylogger = logging.getLogger('mylogger')

# test_production の内で enhanced=True にして threading を使って処理
# response listener を使って受けた引数をグローバルの変数で持つことにしてみた
# Main thread へ値を渡すのはこれが一番楽？他に良い方法があれば知りたい
_response_listener_response = {}

# apns.gateway_server.register_response_listener に登録する関数
def response_listener(error_response):
    mylogger.debug('register_response_listener...')
    global _response_listener_response
    mylogger.debug(error_response)

    # 2,3,4,5,6,7送信前にチェックされてる？
    error_messages = {
        1: "Processing error",
        2: "Missing device token",
        3: "Missing topic",
        4: "Missing payload",
        5: "Invalid token size",
        6: "Invalid topic size",
        7: "Invalid payload size",
        8: "Invalid token",
        10: "Shutdown",
        255: "unknown"
    }
    _response_listener_response = error_response
    _response_listener_response.update(
        {'message': error_messages.get(error_response['status'])}
    )


# class なくても良いが、とりあえず class を宣言
class TestClass:

    def test_production(self):
        mylogger.debug('test_production...')

        # APNs を宣言する
        # enhanced=True をして threading を有効にする
        apns = APNs(use_sandbox=False,
                    cert_file='cert.pem',
                    key_file='key.pem',
                    enhanced=True
                    )

        # payload の準備
        payload = Payload(
            alert="Hello World!",
            custom={'url': 'snapdish://dish?id=xxxxxxxxxxxxxxxxxx'},
            sound="default",
            badge=1)

        try:
            # 通知を別threadで送る
            token_hex = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
            apns.gateway_server.send_notification(token_hex, payload)

        except Exception, e:
            mylogger.debug(str(e))
            # send_notificationでexceptionがある時
            # 強制的に connection を張ったことにして
            # close_read_thread を行う
            # でないと、thread targetのループから抜けない
            apns.gateway_server.connection_alive = True
            apns.gateway_server.force_close()
            assert False

        # response_listerコールバックを指定する
        # レスポンスは別threadで受けるので、コールバック内でglobal変数をセットする
        # mainthreadへの通知は改造しないとできない（違っていたら教えて欲しい）
        # それやるのは、結構厳しい
        apns.gateway_server.register_response_listener(response_listener)

        # 通知の処理の完了まで約最大1秒待って、threadを止める
        # この場合、1秒越えると正常系として評価されてしまうので、この辺りが課題
        time.sleep(1)
        apns.gateway_server.force_close()

        # global _response_lister_responseをチェックしてassert処理を適応する
        global _response_listener_response
        mylogger.debug(_response_listener_response)

        if 'status' in _response_listener_response:
            assert _response_listener_response['status'] == 0

        # 無事終了
        # 通知は無事実機に飛んでるかな？最終的に目視も必要
        assert True
```

# testの実行

```bash
$ PYTHONPATH=`pwd` py.test test_apns.py
```

# threadingの時のテストって？

pyapnsのthreadingを読んでみて、testを考えた時threadingの処理をどう実装すべきか考えてみた。

今回は global変数を使ってcallbackの引数をmainthreadからsleepを使って強引に参照できるようにしてみることはできる。
ただ、これだと、完全ではない。

やはり、threadをsubclassingして、そのsubclassを利用してthreadの処理を制御したほうが良いのかなと思った。

pyapnsはよくできている方だが、この点は若干改良の余地がありそう。

あと、余談だが、実際app serverからpushを送る場合は、job_queueに`enhanced=Ture`でjobを登録して処理をすると通知の結果を無理なく処理することができる。多分。まだやっていなが。次にそれをやってみよう。