0

我使用单元测试来检查 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，但我既不能访问数据，也不能控制它。

4

1 回答 1

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

然后 unittest 看起来像这样：

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)
于 2013-03-04T12:21:03.113 回答