在Python中控制TCP服务器

我使用unittests来检查TCP客户端通信的正确性，所以我需要一些TCP服务器，我可以控制它并发送给客户端（以及何时）。
一个简单的测试应该是这样的：在Python中控制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的，但我没有访问既没有数据，也没有控制它..

来源

2013-03-02 DominiCane

我不知道是否有人需要这个，但无论如何：我用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)

来源

2013-03-04 12:21:03 DominiCane

在Python中控制TCP服务器

回答

相关问题