快速检测或模拟WSAECONNREFUSED

当涉及WSAECONNREFUSED（意味着积压已满或端口不可用，请参阅https://stackoverflow.com/a/10308338/851737）时，Windows套接字有一些奇怪的行为。如果Windows检测到这些情况之一，则会以0.5秒的间隔重试（最多）两次。这意味着在套接字连接尝试（http://support.microsoft.com/kb/175523/en-us）上检测WSAECONNREFUSED至少需要1秒。快速检测或模拟WSAECONNREFUSED

有什么办法可以加速这种检测而不会搞乱注册表值？我需要在单元测试中模拟拒绝套接字连接。类似于使用原始套接字模拟拒绝连接的解决方法也是可以接受的。

下面是一个简单的Python脚本演示该问题：

import errno 
import socket 
import time 

PORT = 5


def main(): 
    s = socket.socket() 
    s.bind(('127.0.0.1', PORT)) 
    s.listen(0) 
    client = socket.socket() 
    client.connect(('127.0.0.1', PORT)) 

    client2 = socket.socket() 
    start = time.time() 

    try: 
     client2.connect(('127.0.0.1', PORT)) 
    except socket.error as e: 
     assert e.errno == errno.WSAECONNREFUSED 
     print 'connection attempt took', time.time() - start 
    finally: 
     client2.close() 
     client.close() 
     s.close() 


if __name__ == '__main__': 
    main()

来源

2014-01-16 schlamar

这不是你问什么。但是如果你只需要单元测试，mock库会很有用。

import errno 
import socket 
import time 
import mock 

PORT = 5


def connect_mock(*agrs): 
    raise socket.error(errno.WSAECONNREFUSED, "Testing") 


def main(): 
    s = socket.socket() 
    s.bind(('127.0.0.1', PORT)) 
    s.listen(0) 
    client = socket.socket() 
    client.connect(('127.0.0.1', PORT)) 

    client2 = socket.socket() 
    start = time.time() 

    with mock.patch('socket.socket.connect', connect_mock): 
     try: 
      client2.connect(('127.0.0.1', PORT)) 
      print "done" 
     except socket.error as e: 
      assert e.errno == errno.WSAECONNREFUSED 
      print 'connection attempt took', time.time() - start 
     finally: 
      client2.close() 
      client.close() 
      s.close() 


if __name__ == '__main__': 
    main()

来源

2014-01-21 20:33:54

其实，我测试的重新连接程序，所以第二个呼叫连接有成功（因为我发现第一次连接尝试失败后启动服务器）。但是这应该可以通过更智能的'connect_mock'版本来实现。我会测试这个并报告我的进度。感谢迄今的灵感。 – schlamar

嗯，这对我来说完美无缺:)你肯定赢得了奖金，但我会保持这个开放几天，因为可能有更通用的解决方案。 – schlamar

这里是基于梅德-vakhrushev的answer这是打补丁的连接方法更智能的我的解决方案：

if sys.platform == 'win32': 
    n_calls = [0] 
    org_connect = socket.socket.connect 

    def refusing_connect(*args): 
     if n_calls[0] < 2: 
      n_calls[0] += 1 
      raise socket.error(errno.WSAECONNREFUSED, "Testing") 
     return org_connect(*args) 

    # patch socket.connect to speed up WSAECONNREFUSED detection 
    patcher = mock.patch('socket.socket.connect', refusing_connect) 
    patcher.start() 
    self.addCleanup(patcher.stop)

来源

2014-01-22 14:39:35 schlamar

快速检测或模拟WSAECONNREFUSED

回答

相关问题