Skip to content

Commit

Permalink
more tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@miguelgrinberg
miguelgrinberg committed Feb 26, 2014
1 parent 22d21ef commit fa78162
Show file tree
Hide file tree
Showing 3 changed files with 145 additions and 23 deletions.
41 changes: 20 additions & 21 deletions flask_socketio/test_client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,37 +67,36 @@ def send(self, message, json=False, callback=None):
class SocketIOTestClient(object):
server = TestServer()

def __init__(self, app, socketio, ns_name=None):
def __init__(self, app, socketio, namespace=None):
self.socketio = socketio
self.socket = self.server.new_socket()
self.connect(app, ns_name)
self.connect(app, namespace)

def __del__(self):
self.server.remove_socket(self.socket)

def connect(self, app, ns_name=None):
if self.socket.namespace.get(ns_name):
self.disconnect(ns_name)
key_ns_name = ns_name
if ns_name is None or ns_name == '/':
ns_name = ''
self.socket.namespace[ns_name] = \
def connect(self, app, namespace=None):
if self.socket.namespace.get(namespace):
self.disconnect(namespace)
if namespace is None or namespace == '/':
namespace = ''
self.socket.namespace[namespace] = \
self.socketio.get_namespaces(
TestBaseNamespace)[ns_name](ns_name, self.socket, app)
self.socket[ns_name].recv_connect()
TestBaseNamespace)[namespace](namespace, self.socket, app)
self.socket[namespace].recv_connect()

def disconnect(self, ns_name=None):
if ns_name is None or ns_name == '/':
ns_name = ''
if self.socket[ns_name]:
self.socket[ns_name].recv_disconnect()
del self.socket.namespace[ns_name]
def disconnect(self, namespace=None):
if namespace is None or namespace == '/':
namespace = ''
if self.socket[namespace]:
self.socket[namespace].recv_disconnect()
del self.socket.namespace[namespace]

def emit(self, event, *args, **kwargs):
ns_name = kwargs.pop('ns_name', None)
if ns_name is None or ns_name == '/':
ns_name = ''
return self.socket[ns_name].process_event({'name': event, 'args': args})
namespace = kwargs.pop('namespace', None)
if namespace is None or namespace == '/':
namespace = ''
return self.socket[namespace].process_event({'name': event, 'args': args})

def send(self, message, json=False, namespace=None):
if namespace is None or namespace == '/':
Expand Down
3 changes: 3 additions & 0 deletions setup.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,9 @@
'Flask>=0.9',
'gevent-socketio>=0.3.6'
],
tests_require=[
'coverage'
],
test_suite='test_socketio',
classifiers=[
'Environment :: Web Environment',
Expand Down
124 changes: 122 additions & 2 deletions test_socketio.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,40 @@ def on_connect_test():
@socketio.on('disconnect', namespace='/test')
def on_disconnect_test():
global disconnected
disconnected = 'test'
disconnected = '/test'

@socketio.on('message')
def on_message(message):
send(message)

@socketio.on('json')
def on_json(data):
send(data, json=True)

@socketio.on('message', namespace='/test')
def on_message_test(message):
send(message, json=True)
send(message)

@socketio.on('json', namespace='/test')
def on_json_test(data):
send(data, json=True)

@socketio.on('my custom event')
def on_custom_event(data):
emit('my custom response', data)

@socketio.on('my custom namespace event', namespace='/test')
def on_custom_event_test(data):
emit('my custom namespace response', data)

@socketio.on('my custom broadcast event')
def on_custom_event_broadcast(data):
emit('my custom response', data, broadcast=True)

@socketio.on('my custom broadcast namespace event', namespace='/test')
def on_custom_event_broadcast_test(data):
emit('my custom namespace response', data, namespace='/test',
broadcast=True)


class TestSocketIO(unittest.TestCase):
Expand Down Expand Up @@ -66,7 +91,102 @@ def test_connect_namespace(self):
received = client.get_received('/test')
self.assertTrue(len(received) == 1)
self.assertTrue(received[0]['args'] == 'connected-test')
client.disconnect(namespace='/test')

def test_disconnect(self):
global disconnected
disconnected = None
client = socketio.test_client(app)
client.disconnect()
self.assertTrue(disconnected == '/')

def test_disconnect_namespace(self):
global disconnected
disconnected = None
client = socketio.test_client(app, namespace='/test')
client.disconnect('/test')
self.assertTrue(disconnected == '/test')

def test_send(self):
client = socketio.test_client(app)
received = client.get_received() # clean received
client.send('echo this message back')
received = client.get_received()
self.assertTrue(len(received) == 1)
self.assertTrue(received[0]['args'] == 'echo this message back')

def test_send_json(self):
client = socketio.test_client(app)
received = client.get_received() # clean received
client.send({'a': 'b'}, json=True)
received = client.get_received()
self.assertTrue(len(received) == 1)
self.assertTrue(received[0]['args']['a'] == 'b')

def test_send_namespace(self):
client = socketio.test_client(app, namespace='/test')
received = client.get_received('/test') # clean received
client.send('echo this message back', namespace='/test')
received = client.get_received('/test')
self.assertTrue(len(received) == 1)
self.assertTrue(received[0]['args'] == 'echo this message back')

def test_send_json_namespace(self):
client = socketio.test_client(app, namespace='/test')
received = client.get_received('/test') # clean received
client.send({'a': 'b'}, json=True, namespace='/test')
received = client.get_received('/test')
self.assertTrue(len(received) == 1)
self.assertTrue(received[0]['args']['a'] == 'b')

def test_emit(self):
client = socketio.test_client(app)
received = client.get_received() # clean received
client.emit('my custom event', {'a': 'b'})
received = client.get_received()
self.assertTrue(len(received) == 1)
self.assertTrue(len(received[0]['args']) == 1)
self.assertTrue(received[0]['name'] == 'my custom response')
self.assertTrue(received[0]['args'][0]['a'] == 'b')

def test_emit_namespace(self):
client = socketio.test_client(app, namespace='/test')
received = client.get_received('/test') # clean received
client.emit('my custom namespace event', {'a': 'b'}, namespace='/test')
received = client.get_received('/test')
self.assertTrue(len(received) == 1)
self.assertTrue(len(received[0]['args']) == 1)
self.assertTrue(received[0]['name'] == 'my custom namespace response')
self.assertTrue(received[0]['args'][0]['a'] == 'b')

def test_broadcast(self):
client1 = socketio.test_client(app)
client2 = socketio.test_client(app)
client3 = socketio.test_client(app, namespace='/test')
received = client2.get_received() # clean
received = client3.get_received('/test') # clean
client1.emit('my custom broadcast event', {'a': 'b'}, broadcast=True)
received = client2.get_received()
self.assertTrue(len(received) == 1)
self.assertTrue(len(received[0]['args']) == 1)
self.assertTrue(received[0]['name'] == 'my custom response')
self.assertTrue(received[0]['args'][0]['a'] == 'b')
self.assertTrue(client3.get_received('/test') == 0)

def test_broadcast_namespace(self):
client1 = socketio.test_client(app, namespace='/test')
client2 = socketio.test_client(app, namespace='/test')
client3 = socketio.test_client(app)
received = client2.get_received('/test') # clean
received = client3.get_received() # clean
client1.emit('my custom broadcast namespace event', {'a': 'b'},
namespace='/test', broadcast=True)
received = client2.get_received('/test')
self.assertTrue(len(received) == 1)
self.assertTrue(len(received[0]['args']) == 1)
self.assertTrue(received[0]['name'] == 'my custom response')
self.assertTrue(received[0]['args'][0]['a'] == 'b')
self.assertTrue(client3.get_received() == 0)

if __name__ == '__main__':
unittest.main()

0 comments on commit fa78162

Please sign in to comment.