4

我正在尝试测试一些在断开连接后重新连接到服务器的代码。这在测试之外工作得很好，但是在运行测试时它无法确认套接字已断开连接。

我正在使用 Gevent Stream Server 来模拟一个真正的监听服务器：

import gevent.server
from gevent import queue


class TestServer(gevent.server.StreamServer):

    def __init__(self, *args, **kwargs):
        super(TestServer, self).__init__(*args, **kwargs)
        self.sockets = {}

    def handle(self, socket, address):
        self.sockets[address] = (socket, queue.Queue())
        socket.sendall('testing the connection\r\n')
        gevent.spawn(self.recv, address)

    def recv(self, address):
        socket = self.sockets[address][0]
        queue = self.sockets[address][1]
        print 'Connection accepted %s:%d' % address
        try:
            for data in socket.recv(1024):
                queue.put(data)
        except:
            pass

    def murder(self):
        self.stop()
        for sock in self.sockets.iteritems():
            print sock
            sock[1][0].shutdown(socket.SHUT_RDWR)
            sock[1][0].close()
        self.sockets = {}


def run_server():
    test_server = TestServer(('127.0.0.1', 10666))
    test_server.start()
    return test_server

我的测试看起来像这样：

def test_can_reconnect(self):
    test_server = run_server()
    client_config = {'host': '127.0.0.1', 'port': 10666}
    client = Connection('test client', client_config, get_config())
    client.connect()
    assert client.socket_connected
    test_server.murder()
    #time.sleep(4) #tried sleeping. no dice.
    assert not client.socket_connected
    assert client.server_disconnect
    test_server = run_server()
    client.reconnect()
    assert client.socket_connected

它在assert not client.socket_connected.

我在接收期间检测到“非数据”。如果它是None，那么我设置一些变量，以便其他代码可以决定是否重新连接（如果是user_disconnect，则不要重新连接等等）。这种行为有效并且过去一直对我有效，直到现在我才尝试对其进行测试。套接字连接和本地函数范围或其他东西有什么奇怪的吗？就像停止服务器后连接仍然存在一样。

我要测试的代码是开放的：https ://github.com/kyleterry/tenyks.git

如果你运行测试，你会看到我试图修复的那个失败。

4

4 回答 4

2

尝试使用真正的套接字运行单元测试是一件很难的事情。这将很棘手，因为一次只能运行一组测试，因为将使用服务器端口，并且随着套接字的建立和拆除，它会很慢。最重要的是，如果这真的是一个单元测试，你不想测试套接字，只是使用套接字的代码。

如果您模拟套接字调用，您可以从模拟代码中随意抛出异常，并确保使用套接字的代码执行正确的操作。您不需要真正的套接字来确保被测类执行正确的操作，如果您可以将套接字调用包装在对象中，则可以伪造它。在构造类时传入对套接字对象的引用，然后就可以开始了。

我的建议是将套接字调用包装在一个支持 sendall、recv 以及您在套接字上调用的所有方法的类中。然后，您可以用 TestReconnectSocket （或其他）替换实际的 Socket 类并运行您的测试。

看看mox，一个 python 模拟框架。

于 2012-10-25T18:13:05.340 回答
0

像这样测试套接字时要记住的一件事是，操作系统不喜欢在套接字使用后立即重新打开它。您可以设置一个套接字选项来告诉它继续并重用它。创建套接字后立即设置套接字的选项：

mysocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

希望这可以解决您的问题。您可能必须在服务器端和客户端都执行此操作，具体取决于哪个给您带来问题。

于 2012-10-25T18:00:39.087 回答
0
  • 您正在打电话shutdown(socket.SHUT_RDWR)，所以这似乎不是recv 阻塞的问题。
  • 但是，您正在使用gevent.socket.socket.recv，因此请检查您的gevent版本，如果底层文件描述符关闭（版本 < ） ，则存在recv()导致其阻塞的问题v0.13.0
  • 您可能仍然需要gevent.sleep()进行合作收益并给客户一个退出recv()通话的机会。
于 2012-10-25T21:33:48.250 回答
0

模糊的响应，但我的直接反应是您的 recv() 调用正在阻塞并保持套接字处于活动状态 - 您是否尝试过使套接字变为非阻塞，并在关闭时捕获错误？

于 2012-10-25T09:43:12.127 回答