我正在编写一个聊天服务器,在对其进行单元测试时遇到了以下问题。在我的一个单元测试中,我将许多测试客户端连接到我的服务器。当连接的用户数达到 511 时,服务器停止响应而没有任何错误消息。在这个阶段,一切都在 PC 上本地运行。
我准备了一个简单的服务器、测试客户端和单元测试代码粘贴到论坛中。
知道为什么服务器挂断了吗?任何帮助深表感谢
这段代码基本上来自于twisted simple chat tutorial。简单服务器:
from twisted.internet.protocol import Factory
from twisted.protocols.basic import LineReceiver
from twisted.internet import reactor
class Chat(LineReceiver):
def __init__(self, users, userNum):
self.users = users
self.userNum = userNum
def connectionMade(self):
print "Connected to user %d" % (self.userNum)
self.users[self.userNum] = self
def connectionLost(self, reason):
print "Connection to user %d lost" % (self.userNum)
if self.users.has_key(self.userNum):
del self.users[self.userNum]
def lineReceived(self, line):
for user in self.users:
if user == self.userNum:
continue
self.users[user].sendLine("%d - %s" % (self.userNum, line))
class ChatFactory(Factory):
def __init__(self):
self.users = {} # maps user names to Chat instances
self.nUsers = 0
def buildProtocol(self, addr):
self.nUsers += 1
return Chat(self.users, self.nUsers)
def clientConnectionFailed(self, connector, reason):
print 'connection failed:', reason.getErrorMessage()
reactor.stop()
def clientConnectionLost(self, connector, reason):
print 'connection lost:', reason.getErrorMessage()
reactor.stop()
reactor.listenTCP(8123, ChatFactory())
reactor.run()
这是我的测试客户。这个客户端被单元测试多次实例化。
import socket
HOST = "localhost"
PORT = 8123
class TestClient:
def __init__(self):
self.connected = False
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
except socket.error, msg:
print("Socket error %s" % msg)
def connect(self):
try:
self.socket.settimeout(10)
self.socket.connect((HOST, PORT))
self.connected = True
except socket.error, msg:
print("Socket error %s" % msg)
self.connected = False
def disconnect(self):
self.socket.close()
def connected(self):
return self.connected
最后是单元测试代码文件:
import unittest
from TestClient import TestClient
class TestSequenceFunctions(unittest.TestCase):
def test_manyUsers(self):
users = []
number_of_users = 1000
for u in range(number_of_users):
# create client
users.append(TestClient())
# connect client to server
users[-1].connect()
# check connected state
self.assertTrue(users[-1].connected, "User %d is not connected" % (u))
# close connection of all users
for user in users:
user.disconnect()
if __name__ == '__main__':
unittest.main()