[PYTHON] Hinweise zur Implementierung von APNs-Tests mit Pytest

Aufgrund verschiedener Untersuchungen zur Überprüfung der Reaktion von APNs halte ich die folgende Methode für etwas schwierig, habe sie jedoch damit implementiert.

Es wäre schön, wenn Pyapns eine testbewusste Schnittstelle hätten, aber das ist nicht der Fall. Deshalb habe ich eine globale Variable verwendet, um die Rückrufargumente für MainThread sichtbar zu machen.

Es ist in Ordnung, Pyapns zu erweitern, aber ich denke, dies ist der schnellste Weg, also habe ich beschlossen, es zu versuchen.

Bitte lassen Sie mich wissen, ob es einen besseren Weg gibt.

Beispielcode

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_In der Produktion verbessert=Auf True setzen und mit Threading verarbeiten
#Ich habe beschlossen, das Argument mithilfe des Antwort-Listeners als globale Variable empfangen zu lassen
#Ist dies der einfachste Weg, einen Wert an den Haupt-Thread zu übergeben? Ich möchte wissen, ob es einen anderen guten Weg gibt
_response_listener_response = {}

# apns.gateway_server.register_response_Funktion zur Registrierung im 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 Wird es vor dem Senden überprüft?
    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'])}
    )


#Sie müssen keine Klasse haben, aber vorerst eine Klasse deklarieren
class TestClass:

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

        #APNs deklarieren
        # enhanced=True, um das Threading zu aktivieren
        apns = APNs(use_sandbox=False,
                    cert_file='cert.pem',
                    key_file='key.pem',
                    enhanced=True
                    )

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

        try:
            #Benachrichtigung in einem anderen Thread senden
            token_hex = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
            apns.gateway_server.send_notification(token_hex, payload)

        except Exception, e:
            mylogger.debug(str(e))
            # send_Wenn die Benachrichtigung eine Ausnahme enthält
            #Ich beschloss, gewaltsam eine Verbindung herzustellen
            # close_read_Faden machen
            #Andernfalls können Sie die Thread-Zielschleife nicht verlassen
            apns.gateway_server.connection_alive = True
            apns.gateway_server.force_close()
            assert False

        # response_Geben Sie den Lister-Rückruf an
        #Da die Antwort in einem anderen Thread empfangen wird, legen Sie die globale Variable im Rückruf fest
        #Die Benachrichtigung des Wartungskopfs kann nicht ohne Änderung erfolgen (bitte lassen Sie mich wissen, wenn es anders ist)
        #Es ist ziemlich schwer zu tun
        apns.gateway_server.register_response_listener(response_listener)

        #Warten Sie bis zu 1 Sekunde, bis die Benachrichtigungsverarbeitung abgeschlossen ist, und stoppen Sie den Thread
        #In diesem Fall wird es, wenn es länger als 1 Sekunde ist, als normales System ausgewertet, sodass dieser Bereich ein Problem darstellt
        time.sleep(1)
        apns.gateway_server.force_close()

        # global _response_lister_Überprüfen Sie die Antwort und wenden Sie die Assert-Verarbeitung an
        global _response_listener_response
        mylogger.debug(_response_listener_response)

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

        #Erfolgreiche Fertigstellung
        #Fliegt die Benachrichtigung zum unschuldigen Flugzeug? Schließlich ist auch eine Sichtprüfung erforderlich
        assert True

Lauftest

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

Was ist der Test zum Zeitpunkt des Einfädelns?

Ich las Threading in Pyapns und dachte darüber nach, wie man Threading-Verarbeitung implementiert, wenn ich über Test nachdenke.

Dieses Mal können Sie die globale Variable verwenden, um mithilfe des Ruhezustands zwangsweise auf das Rückrufargument aus dem Hauptthread zu verweisen. Dies ist jedoch nicht perfekt.

Schließlich dachte ich, es wäre besser, den Thread zu unterordnen und diese Unterklasse zu verwenden, um die Thread-Verarbeitung zu steuern.

pyapns ist ein guter Mensch, aber in dieser Hinsicht scheint es Raum für Verbesserungen zu geben.

Abgesehen davon können Sie beim Senden eines Pushs vom App-Server, wenn Sie einen Job in job_queue mit "advanced = True" registrieren und verarbeiten, das Benachrichtigungsergebnis angemessen verarbeiten. Vielleicht. Ich habe es noch nicht getan. Lass es uns als nächstes tun.