tcptest

tcptest is following Perl’s “Test::TCP”.

It has the methods like original one, but also has another TestServer implementation.

This package also includes memcached and redis test server support.

SYNOPSIS

import tcptest.memcached
import tcptest.redis
import tcptest.fluentd

# with context
with tcptest.memcached.Server() as server:
    memd = memcache.Client(['127.0.0.1:%d' % server.port])
    ...

with tcptest.redis.Server() as server:
    db = redis.Redis(host='127.0.0.1', port=server.port, db=0)
    ...

with tcptest.fluentd.Server() as server:
    import fluentd.sender
    import fluentd.event
    fluent.sender.setup('app', port=server.port)
    fluent.event.Event('follow', {'foo': 'bar'})
    fluent.event.Event('label', {'hoge': 'fuga'})
print server.logs
# [('app.follow:', {u'foo': u'bar'}), ('app.label:', {u'hoge': u'fuga'})]

# manually handling
server = tcptest.memcached.Server()
server.start()
...
server.stop()

TIPS

custom server

import tcptest

class YourTestServer(tcptest.TestServer):
    def build_command(self):
        return ('your server command', 'arg1', 'arg2', ...)

with YourTestServer() as server:
    # your server works on server.port
    ...

capture server outputs

import tcptest.redis

res = {}
with tcptest.redis.Server(res=res) as server:
    ...
stdout = res['stdout']
stderr = res['stderr']

redis options

import tcptest.redis

with tcptest.redis.Server(settings=dict(databases=4)) as server:
    ...

redis replication

import tcptest.redis

master = tcptest.redis.Server()
master.start()

slave_settings = {'slaveof': 'localhost %d' % master.port}
slave = tcptest.redis.Server(settings=slave_settings)
slave.start()