Testing Elasticsearch with python-tcptest

Yesterday, the version of python-tcptest was upgraded and it can be used in Python3 series. (Although I threw Pururi-kun myself)

To commemorate that, I'll show you how to use tcptest to test Python code using Elasticsearch.

What is tcptest?

1. Find a free port
2. Launch the server software specified on that port
3. When the test is over, drop the server software

That is. I think the following three points are appreciated.

--Since the actual server software is used, you do not have to worry about the difference between the mock and the actual one. --Eliminates mistakes such as accidentally erasing data in the DB that is not for testing. ――You don't have to bother to set up or drop the server software for testing (no human warmth)

How to insert python-tcptest

$ pip install tcptest

python-tcptest flow

At the start

1. TestServer.__init__: Set a timeout or something (longer ones that take a long time to start)
2. TestServer._before_start: Write the process (temporary directory creation, etc.) to be done before starting the server here (do nothing by default)
3. TestServer.start: Find a free port and hit the TestServer.build_command command.
4. TestServer._after_start: Write the process (initial settings, etc.) to be done immediately after starting the server here (does nothing by default)

When finished

1. TestServer._before_stop: Write here what to do before the server shuts down (does nothing by default)
2. TestServer.stop: Stop the server.
3. TestServer._after_start: Write here what to do after the server is shut down (delete the temporary directory) (do nothing by default)

Let's write for Elasticsearch

python-tcptest contains Redis, Fluentd and memcached test classes from the beginning. This time I want to test Elasticsearch, so I will create my own class.

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)

Just inherit tcptest.TestServer and rewrite 3 methods!

Confirmation code

Let's check based on elasticsearch-py example.

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

When you execute the above code, it looks like this ↓

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

I started Elasticsearch on a free port that I don't understand, created a data directory and a log directory in a neat place, it worked properly, and when it was finished, I found that it was clean.

that's all.