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.
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)
$ pip install tcptest
TestServer.__init__
: Set a timeout or something (longer ones that take a long time to start)TestServer._before_start
: Write the process (temporary directory creation, etc.) to be done before starting the server here (do nothing by default)TestServer.start
: Find a free port and hit the TestServer.build_command
command.TestServer._after_start
: Write the process (initial settings, etc.) to be done immediately after starting the server here (does nothing by default)TestServer._before_stop
: Write here what to do before the server shuts down (does nothing by default)TestServer.stop
: Stop the server.TestServer._after_start
: Write here what to do after the server is shut down (delete the temporary directory) (do nothing by default)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!
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.
Recommended Posts