如何在單元測試中終止套接字以重新連接測試

我試圖測試一些在斷開連接後重新連接到服務器的代碼。這在測試之外很好地工作，但是在運行測試時它無法確認套接字已經斷開連接。如何在單元測試中終止套接字以重新連接測試

我使用的是GEVENT流服務器來模擬一個真實的聆聽服務器：

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。

我在recv期間檢測到「不是數據」。如果它是None，那麼我設置一些變量，以便其他代碼可以決定是否重新連接（如果是user_disconnect等，則不要重新連接）。這種行爲起作用，過去一直對我有用，直到現在我從來沒有試過對它進行測試。套接字連接和本地函數作用域有什麼奇怪的地方？就像停止服務器後連接仍然存在。

我想要測試的代碼是開放的：https://github.com/kyleterry/tenyks.git

如果你運行測試，你會看到一個我試圖修復失敗。

來源

2012-10-17 Kyle

試圖用真正的套接字運行單元測試是一個艱難的行。這將是棘手的，因爲一次只能運行一組測試，因爲服務器端口將被使用，並且隨着套接字被設置並拆除將會變得緩慢。如果這真的是一個單元測試，你不想測試套接字，只需使用套接字的代碼即可。

如果你模擬套接字調用，你可以從模擬代碼中拋出異常，並確保使用套接字的代碼做正確的事情。你不需要一個真正的套接字來確保被測試的類是正確的，如果你可以將套接字調用包裝在一個對象中，你可以僞造它。在構建你的課程時傳入對套接字對象的引用，並且你準備好了。

我的建議是將套接字調用包裝在支持sendall，recv和您在套接字上調用的所有方法的類中。然後你可以用一個TestReconnectSocket（或其他）替換實際的Socket類並運行你的測試。

看看mox，一個python模擬框架。

來源

2012-10-25 18:13:05

包裝是一個kludge。在Python中，有更多優雅的解決方案來替代外部對象，例如['unittest.mock.patch']（http://docs.python.org/dev/library/unittest.mock#id4）。 –

模糊的反應，但我的第一反應會是您的recv（）調用阻止，並保持插座活着 - 你試圖使插座非阻塞，並在接近捕捉錯誤呢？

來源

2012-10-25 09:43:12 KevinL

測試這種套接字時要記住的一件事是，操作系統在使用後不喜歡重新打開套接字。你可以設置一個套接字選項來告訴它繼續並反覆使用它。在創建套接字後立即設置套接字的選項：

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

希望這可以解決您的問題。您可能必須在服務器端和客戶端上執行此操作，具體取決於哪一個可以解決問題。

來源

2012-10-25 18:00:39 dail

實際上，它看起來像你只需要在服務器的套接字上執行此操作，因爲客戶端每次都會使用不同的端口。 – dail

你打電話給shutdown(socket.SHUT_RDWR)所以這看起來不像problem with recv blocking。
但是，您使用的gevent.socket.socket.recv，所以請檢查您的gevent版本，有一個與recv()，導致它block if the underlying file descriptor is closed問題（<版本v0.13.0）
您可能仍然需要gevent.sleep()做合作的產量，並給予客戶有機會退出recv()呼叫。

來源

2012-10-25 21:33:48 dnozay

我在哪裏gevent.sleep（）？ – Kyle

我使用gevent == 0.13.8 – Kyle

，你可以把它放在'TestServer.murder'方法的末尾。 – dnozay

如何在單元測試中終止套接字以重新連接測試

回答

相關問題