[PYTHON] Notes on implementing APNs tests using Pytest

As a result of various investigations to check the response of APNs, I think that the following method is a little brute force method, but I implemented it with this.

It would be nice if pyapns had a test-aware interface, but that's not the case, so I used a global variable to make the callback arguments visible to MainThread.

It's okay to extend pyapns, but I think this is the quickest way to do it, so I decided to try it.

Please let me know if there is a better way.

Sample code

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_Enhanced within production=Set to True and process using threading
#I decided to have the argument received by using the response listener as a global variable
#Is this the easiest way to pass a value to the Main thread? I want to know if there is another good way
_response_listener_response = {}

# apns.gateway_server.register_response_Function to register in 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 Is it checked before sending?
    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'])}
    )


#You don't have to have a class, but declare a class for the time being
class TestClass:

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

        #Declare APNs
        # enhanced=True to enable threading
        apns = APNs(use_sandbox=False,
                    cert_file='cert.pem',
                    key_file='key.pem',
                    enhanced=True
                    )

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

        try:
            #Send notification in another thread
            token_hex = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
            apns.gateway_server.send_notification(token_hex, payload)

        except Exception, e:
            mylogger.debug(str(e))
            # send_When there is an exception in the notification
            #I decided to forcibly establish a connection
            # close_read_do thread
            #Otherwise, you can't get out of the thread target loop
            apns.gateway_server.connection_alive = True
            apns.gateway_server.force_close()
            assert False

        # response_specify a lister callback
        #Since the response is received in another thread, set the global variable in the callback
        #Notification to mainthread cannot be done without modification (please let me know if it is different)
        #It's pretty tough to do
        apns.gateway_server.register_response_listener(response_listener)

        #Wait up to 1 second for notification processing to complete and stop threading
        #In this case, if it exceeds 1 second, it will be evaluated as a normal system, so this area is a problem.
        time.sleep(1)
        apns.gateway_server.force_close()

        # global _response_lister_Check response and apply assert processing
        global _response_listener_response
        mylogger.debug(_response_listener_response)

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

        #Successful completion
        #Is the notification flying to the innocent plane? Finally visual inspection is also necessary
        assert True

Run test

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

What is the test at the time of threading?

I read threading in pyapns and thought about how to implement threading processing when thinking about test.

This time, you can use the global variable to forcibly refer to the callback argument from the main thread using sleep. However, this is not perfect.

After all, I thought it would be better to subclass the thread and control the thread processing using that subclass.

pyapns is a good person, but there seems to be some room for improvement in this regard.

Also, as an aside, when actually sending push from the app server, if you register a job in job_queue with ʻenhanced = True` and process it, you can process the notification result reasonably. Perhaps. I haven't done it yet. Let's do it next.