tests: fix test_server.py

master
Roman Zeyde 7 years ago
parent fde50f04ab
commit b1f302151b
No known key found for this signature in database
GPG Key ID: 87CAE5FA46917CBB

@ -1,4 +1,3 @@
import functools
import io import io
import os import os
import socket import socket
@ -74,26 +73,27 @@ def test_handle():
def test_server_thread(): def test_server_thread():
connections = [FakeSocket()] sock = FakeSocket()
connections = [sock]
quit_event = threading.Event() quit_event = threading.Event()
class FakeServer(object): class FakeServer(object):
def accept(self): # pylint: disable=no-self-use def accept(self): # pylint: disable=no-self-use
if connections: if connections:
return connections.pop(), 'address' return connections.pop(), 'address'
quit_event.set()
raise socket.timeout() raise socket.timeout()
def getsockname(self): # pylint: disable=no-self-use def getsockname(self): # pylint: disable=no-self-use
return 'fake_server' return 'fake_server'
handler = protocol.Handler(conn=empty_device()), def handle_conn(conn):
handle_conn = functools.partial(server.handle_connection, assert conn is sock
handler=handler, quit_event.set()
mutex=None)
server.server_thread(sock=FakeServer(), server.server_thread(sock=FakeServer(),
handle_conn=handle_conn, handle_conn=handle_conn,
quit_event=quit_event) quit_event=quit_event)
quit_event.wait()
def test_spawn(): def test_spawn():

Loading…
Cancel
Save