Testen von Elasticsearch mit Python-tcptest

Gestern wurde die Version von python-tcptest aktualisiert und kann in der Python3-Serie verwendet werden. (Obwohl ich selbst Pururi-Kun geworfen habe)

Zum Gedenken werde ich Ihnen zeigen, wie Sie mit tcptest Ihren Python-Code mit Elasticsearch testen können.

Was ist tcptest?

1. Suchen Sie einen freien Port
2. Starten Sie die für diesen Port angegebene Serversoftware
3. Löschen Sie die Serversoftware, wenn der Test beendet ist

Das ist. Ich denke, die folgenden drei Punkte werden geschätzt.

• Da die eigentliche Serversoftware verwendet wird, müssen Sie sich keine Gedanken über den Unterschied zwischen dem Schein und dem tatsächlichen machen.
• Beseitigt Fehler wie das versehentliche Löschen von Daten in der Datenbank, die nicht zum Testen dienen. ――Sie müssen sich nicht die Mühe machen, die Serversoftware zum Testen einzurichten oder zu löschen (keine menschliche Wärme).

So fügen Sie Python-tcptest ein

$ pip install tcptest

Python-Tcptest-Fluss

Am Anfang

1. TestServer .__ init__: Legen Sie eine Zeitüberschreitung oder etwas anderes fest (Wenn der Start lange dauert, ist es besser, ihn länger zu machen).
2. TestServer._before_start: Schreiben Sie den Prozess (temporäre Verzeichniserstellung usw.), der vor dem Starten des Servers ausgeführt werden soll, hier (standardmäßig nichts tun).
3. TestServer.start: Suchen Sie einen freien Port und klicken Sie auf den Befehl TestServer.build_command.
4. TestServer._after_start: Schreiben Sie den Prozess (Grundeinstellung usw.), der unmittelbar nach dem Starten des Servers ausgeführt werden soll, hier (standardmäßig nichts tun).

Wenn Sie fertig sind

1. TestServer._before_stop: Schreiben Sie hier, was zu tun ist, bevor der Server heruntergefahren wird (standardmäßig nichts)
2. TestServer.stop: Stoppen Sie den Server.
3. TestServer._after_start: Schreiben Sie hier, was nach dem Herunterfahren des Servers zu tun ist (löschen Sie das temporäre Verzeichnis) (standardmäßig nichts tun)

Schreiben wir für Elasticsearch

python-tcptest enthält von Anfang an Redis, Fluentd und memcached Testklassen. Dieses Mal möchte ich die elastische Suche testen, damit ich meine eigene Klasse erstellen kann.

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)

Erben Sie einfach tcptest.TestServer und schreiben Sie 3 Methoden neu!

Bestätigungscode

Lassen Sie uns anhand von elasticsearch-py example überprüfen.

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))

Wenn Sie den obigen Code ausführen, sieht es so aus ↓

[{'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

Ich habe Elasticsearch an einem freien Port gestartet, den ich nicht verstehe, ein Datenverzeichnis und ein Protokollverzeichnis an einem ordentlichen Ort erstellt und es hat einwandfrei funktioniert. Als es fertig war, stellte ich fest, dass es sauber war.

das ist alles.