Remarques sur la mise en œuvre des tests APN à l'aide de Pytest

À la suite de diverses enquêtes pour vérifier la réponse des APN, je pense que la méthode suivante est un peu délicate, mais je l'ai implémentée avec cela.

Ce serait bien si pyapns avait une interface prenant en charge les tests, mais ce n'est pas le cas, j'ai donc utilisé une variable globale pour rendre les arguments de rappel visibles à MainThread.

Il n'y a pas de mal à étendre les pyapns, mais je pense que c'est le moyen le plus rapide de le faire, alors j'ai décidé de l'essayer.

S'il vous plaît laissez-moi savoir s'il existe une meilleure façon.

Exemple de 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_Amélioré dans la production=Définir sur True et traiter à l'aide du threading
#J'ai décidé de recevoir l'argument en utilisant l'écouteur de réponse comme variable globale
#Est-ce le moyen le plus simple de transmettre une valeur au thread principal? Je veux savoir s'il y a un autre bon moyen
_response_listener_response = {}

# apns.gateway_server.register_response_Fonction pour s'inscrire dans l'auditeur
def response_listener(error_response):
    mylogger.debug('register_response_listener...')
    global _response_listener_response
    mylogger.debug(error_response)

    # 2,3,4,5,6,7 Est-il vérifié avant l'envoi?
    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'])}
    )


#Vous n'êtes pas obligé d'avoir une classe, mais déclarez une classe pour le moment
class TestClass:

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

        #Déclarer les APN
        # enhanced=True pour activer le thread
        apns = APNs(use_sandbox=False,
                    cert_file='cert.pem',
                    key_file='key.pem',
                    enhanced=True
                    )

        #Préparation de la charge utile
        payload = Payload(
            alert="Hello World!",
            custom={'url': 'snapdish://dish?id=xxxxxxxxxxxxxxxxxx'},
            sound="default",
            badge=1)

        try:
            #Envoyer une notification dans un autre fil
            token_hex = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
            apns.gateway_server.send_notification(token_hex, payload)

        except Exception, e:
            mylogger.debug(str(e))
            # send_Lorsqu'il y a une exception dans la notification
            #J'ai décidé d'établir une connexion de force
            # close_read_faire du fil
            #Sinon, vous ne pourrez pas sortir de la boucle cible du thread
            apns.gateway_server.connection_alive = True
            apns.gateway_server.force_close()
            assert False

        # response_spécifier le rappel de l'auditeur
        #Puisque la réponse est reçue dans un autre thread, définissez la variable globale dans le rappel
        #La notification à mainthread ne peut pas être effectuée sans modification (veuillez me le faire savoir si elle est différente)
        #C'est assez difficile à faire
        apns.gateway_server.register_response_listener(response_listener)

        #Attendez jusqu'à 1 seconde pour que le traitement des notifications se termine et arrêtez le thread
        #Dans ce cas, s'il dépasse 1 seconde, il sera évalué comme un système normal, donc cette zone est un problème
        time.sleep(1)
        apns.gateway_server.force_close()

        # global _response_lister_Vérifier la réponse et appliquer le traitement des assertions
        global _response_listener_response
        mylogger.debug(_response_listener_response)

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

        #Réussite
        #La notification vole-t-elle à l'avion innocent? Enfin une inspection visuelle est également nécessaire
        assert True

Lancer le test

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

Quel est le test au moment du filetage?

J'ai lu le threading dans pyapns et j'ai réfléchi à la manière d'implémenter le traitement du threading en pensant au test.

Cette fois, vous pouvez utiliser la variable globale pour faire référence de force à l'argument de rappel du thread principal en utilisant sleep. Cependant, ce n'est pas parfait.

Après tout, j'ai pensé qu'il serait préférable de sous-classer le thread et d'utiliser cette sous-classe pour contrôler le traitement du thread.

pyapns est une bonne personne, mais il semble qu'il y ait place à amélioration à cet égard.

De plus, lors de l'envoi de push depuis le serveur d'application, si vous enregistrez un travail dans job_queue avec ʻenhanced = True` et le traitez, vous pouvez traiter le résultat de la notification de manière raisonnable. Peut-être. Je ne l'ai pas encore fait. Faisons-le ensuite.