Tester Elasticsearch avec python-tcptest

Hier, la version de python-tcptest a été mise à jour et peut être utilisée dans la série Python3. (Bien que j'aie jeté Pururi-kun moi-même)

Pour commémorer cela, je vais vous montrer comment utiliser tcptest pour tester votre code Python à l'aide d'Elasticsearch.

Qu'est-ce que tcptest?

1. Trouvez un port gratuit
2. Lancez le logiciel serveur spécifié pour ce port
3. Supprimez le logiciel serveur une fois le test terminé

C'est. Je pense que les trois points suivants sont appréciés.

• Étant donné que le logiciel serveur réel est utilisé, vous n'avez pas à vous soucier de la différence entre le simulacre et le réel.
• Élimine les erreurs telles que l'effacement accidentel de données dans la base de données qui ne sont pas à tester. ――Vous n'avez pas à vous soucier de configurer ou de supprimer le logiciel serveur pour le test (pas de chaleur humaine)

Comment insérer python-tcptest

$ pip install tcptest

flux python-tcptest

Au début

1. TestServer .__ init__: Définissez un délai d'expiration ou quelque chose du genre (s'il faut beaucoup de temps pour démarrer, il est préférable de l'allonger)
2. TestServer._before_start: Écrivez le processus (création de répertoire temporaire, etc.) à faire avant de démarrer le serveur ici (ne rien faire par défaut)
3. TestServer.start: Trouvez un port libre et appuyez sur la commande TestServer.build_command.
4. TestServer._after_start: Écrivez ici le processus (réglage initial, etc.) à effectuer immédiatement après le démarrage du serveur (ne rien faire par défaut)

Une fois terminé

1. TestServer._before_stop: écrivez ici ce qu'il faut faire avant que le serveur ne s'arrête (ne fait rien par défaut)
2. TestServer.stop: Arrêtez le serveur.
3. TestServer._after_start: écrivez ici ce qu'il faut faire après l'arrêt du serveur (supprimer le répertoire temporaire) (ne rien faire par défaut)

Écrivons pour Elasticsearch

python-tcptest contient les classes de test Redis, Fluentd et memcached depuis le début. Cette fois, je veux tester la recherche élastique, je vais donc créer ma propre classe.

import os
import shutil
import tempfile
import tcptest


class ESTestServer(tcptest.TestServer):

    def build_command(self):
        return ('elasticsearch',
                '-Des.network.bind_host=127.0.0.1',
                '-Des.network.host=127.0.0.1',
                '-Des.http.port=%s' % self.port,
                "-Des.node.master=true",
                "-Des.node.local=true",
                '-Des.path.data=%s' % self.data_dir,
                '-Des.path.logs=%s' % self.logs_dir
               )

    def _before_start(self):
        self.data_dir = tempfile.mkdtemp(prefix='esdata')
        self.logs_dir = tempfile.mkdtemp(prefix='eslogs')

    def _after_stop(self):
        for path in filter(os.path.exists, (self.data_dir, self.logs_dir)):
            shutil.rmtree(path)

Héritez simplement tcptest.TestServer et réécrivez 3 méthodes!

Code de confirmation

Vérifions sur la base de exemple elasticsearch-py.

from datetime import datetime
import os
from elasticsearch import Elasticsearch


with ESTestServer(timeout=30) as server:
    es = Elasticsearch(['localhost:%s' % server.port])
    print(es.transport.hosts)

    # from https://elasticsearch-py.readthedocs.org/en/master/#example-usage
    doc = {
        'author': 'kimchy',
        'text': 'Elasticsearch: cool. bonsai cool.',
        'timestamp': datetime.now(),
    }
    res = es.index(index="test-index", doc_type='tweet', id=1, body=doc)
    print(res['created'])

    res = es.get(index="test-index", doc_type='tweet', id=1)
    print(res['_source'])

    es.indices.refresh(index="test-index")

    res = es.search(index="test-index", body={"query": {"match_all": {}}})
    print("Got %d Hits:" % res['hits']['total'])
    for hit in res['hits']['hits']:
        print("%(timestamp)s %(author)s: %(text)s" % hit["_source"])

    print('data_dir contains:', os.listdir(server.data_dir))
    print('logs_dir contains:', os.listdir(server.logs_dir))
    data_dir = server.data_dir
    logs_dir = server.logs_dir

print('data_dir exists:', os.path.exists(data_dir))
print('logs_dir exists:', os.path.exists(logs_dir))

Lorsque vous exécutez le code ci-dessus, cela ressemble à ceci ↓

[{'port': 55003, 'host': 'localhost'}]
True
{'timestamp': '2015-10-30T14:10:16.703556', 'author': 'kimchy', 'text': 'Elasticsearch: cool. bonsai cool.'}
Got 1 Hits:
2015-10-30T14:10:16.703556 kimchy: Elasticsearch: cool. bonsai cool.
data_dir contains: ['elasticsearch_brew']
logs_dir contains: ['elasticsearch_brew.log', 'elasticsearch_brew_index_indexing_slowlog.log', 'elasticsearch_brew_index_search_slowlog.log']
data_dir exists: False
logs_dir exists: False

J'ai commencé Elasticsearch sur un port libre que je ne comprends pas, j'ai créé un répertoire de données et un répertoire de journaux dans un endroit soigné, et cela a bien fonctionné, et une fois terminé, j'ai trouvé que c'était propre.

c'est tout.