在Python中控制TCP服务器

Question

我用单元测试来检查TCP客户端的通信是否正确，所以我需要一个可以控制的TCP服务器，并且它能在我需要的时候向客户端发送数据。
一个简单的测试应该是这样的：

server = Server("localhost", 5555)
server.start()
client.connect()
self.assertTrue("login_message" in server.received_data)
server.send_to_client(reject_messages)
self.assertTrue("login_again" in server.received_data)
time.sleep(10)
self.assertTrue("login_again_and_again" in server.newest_received_data)
server.stop()
self.assertTrue("login failed" in client.logs) 

我需要完全的流程控制，应该用什么来实现服务器呢？
现在我尝试使用多线程的SocketServer，但我既无法访问数据，也无法控制它……

Answer 1

我不知道有没有人需要这个，不过无所谓：
我使用了gevent服务器。

from gevent import sleep, socket
from gevent.server import StreamServer

class Connection:
    def __init__(self, host, port):
        self.server = StreamServer((host, port), self.handle)
        self.data = []
        self.socks = []
        self.pointer = 0

    def handle(self, sock, address):
        self.socks.append(sock)
        while True:
            line = sock.recv(1024)
            if line:
                self.data += [line]
            else:
                break
        sock.close()
        self.socks.remove(sock)

    def send(self, msg):
        if self.socks:
            sock2send = self.socks[-1]
            try:
                sock2send.send(msg)
            except IOError, e:
                print "Can't send message '%s'! Exception:" % msg, e
        else:
            print "No sockets to send the message to"

    def start(self):
        self.server.start()

    def serve_forever(self):
        self.server.serve_forever()

    def close(self):
        self.server.stop()
        for sock in self.socks:
            sock.close()

    def new_data(self):
        newest = self.data[self.pointer:]
        self.pointer = len(self.data)
        return newest

然后单元测试看起来是这样的：

def testTCPClient(self):    
    j = lambda x: "".join(x)

    server = Connection("", 5555)            
    server.start()
    client.run()
    sleep(3)
    data = j(server.new_data())
    self.assertTrue("login" in data)
    sleep(2)
    server.send("login approve")
    sleep(2)
    data = j(server.new_data())
    self.assertTrue("after_login" in data)
    server.send("logout")
    sleep(2)
    data = j(server.new_data())
    self.assertTrue("received_logout" in data)
    server.close()
    self.assertTrue("disconnected" in client.logs)