快速檢測或模擬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

回答

相關問題