4
연결 해제 후 서버에 다시 연결하는 코드를 테스트하려고합니다. 이것은 테스트 밖에서는 완벽하게 작동하지만 테스트를 실행할 때 소켓이 단절되었음을 알리는 데 실패합니다.유닛 테스트에서 소켓 재사용 테스트를 죽이는 방법
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
을 그리고 내 테스트는 다음과 같습니다 :
내가 진짜 듣기 서버를 조롱하는 Gevent 스트림 서버를 사용하고 그것은 assert not client.socket_connected에 실패
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
.
나는 recv 동안 "not data"를 감지한다. 그것이 None이라면, 다른 코드가 재 연결 여부를 결정할 수 있도록 일부 변수를 설정합니다 (user_disconnect 인 경우 다시 연결하지 마십시오). 이 동작은 작동하며 과거에는 항상 저에게 효과가있었습니다. 지금까지는 테스트를 해본 적이 없었습니다. 뭔가 소켓 연결 및 로컬 함수 범위 또는 뭔가 이상한 있습니까? 마치 서버를 멈춘 후에도 연결이 여전히 존재하는 것과 같습니다.
내가 테스트하는 데 노력하고있어 코드가 열려 : 당신이 테스트를 실행하면 https://github.com/kyleterry/tenyks.git
, 당신은 내가 실패 해결하기 위해 노력하고있어 하나를 볼 수 있습니다.
포장은 고름입니다. 파이썬에는 [unittest.mock.patch'] (http://docs.python.org/dev/library/unittest.mock#id4)와 같은 외부 객체를 대체 할 수있는 좀 더 세련된 솔루션이 있습니다. –