File test_socketio.py of Package python-Flask-SocketIO

Overview Repositories Revisions Requests Users Attributes Meta

File test_socketio.py of Package python-Flask-SocketIO

import json
import unittest

from flask import Flask, session, request, json as flask_json
from flask_socketio import SocketIO, send, emit, join_room, leave_room, \
    Namespace, disconnect, ConnectionRefusedError

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret'
socketio = SocketIO(app)
disconnected = None

@socketio.on('connect')
def on_connect(auth):
    if auth != {'foo': 'bar'}:  # pragma: no cover
        return False
    if request.args.get('fail'):
        raise ConnectionRefusedError('failed!')
    send('connected')
    send(json.dumps(request.args.to_dict(flat=False)))
    send(json.dumps({h: request.headers[h] for h in request.headers.keys()
                     if h not in ['Host', 'Content-Type', 'Content-Length']}))
    emit('dummy', to='nobody')

@socketio.on('disconnect')
def on_disconnect():
    global disconnected
    disconnected = '/'

@socketio.event(namespace='/test')
def connect():
    send('connected-test')
    send(json.dumps(request.args.to_dict(flat=False)))
    send(json.dumps({h: request.headers[h] for h in request.headers.keys()
                     if h not in ['Host', 'Content-Type', 'Content-Length']}))

@socketio.on('disconnect', namespace='/test')
def on_disconnect_test():
    global disconnected
    disconnected = '/test'

@socketio.event
def message(message):
    send(message)
    if message == 'test session':
        if not socketio.manage_session and 'a' in session:
            raise RuntimeError('session is being stored')
        if 'a' not in session:
            session['a'] = 'b'
        else:
            session['a'] = 'c'
    if message not in "test noackargs":
        return message

@socketio.on('json')
def on_json(data):
    send(data, json=True, broadcast=True)
    if not data.get('noackargs'):
        return data

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

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

@socketio.on('my custom event')
def on_custom_event(data):
    emit('my custom response', data)
    if not data.get('noackargs'):
        return data

@socketio.on('other custom event')
@socketio.on('and another custom event')
def get_request_event(data):
    global request_event_data
    request_event_data = request.event
    emit('my custom response', data)

def get_request_event2(data):
    global request_event_data
    request_event_data = request.event
    emit('my custom response', data)

socketio.on_event('yet another custom event', get_request_event2)

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

def on_custom_event_test2(data):
    emit('my custom namespace response', data, namespace='/test')

socketio.on_event('yet another custom namespace event', on_custom_event_test2,
                  namespace='/test')

@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)

@socketio.on('join room')
def on_join_room(data):
    join_room(data['room'])

@socketio.on('leave room')
def on_leave_room(data):
    leave_room(data['room'])

@socketio.on('join room', namespace='/test')
def on_join_room_namespace(data):
    join_room(data['room'])

@socketio.on('leave room', namespace='/test')
def on_leave_room_namespace(data):
    leave_room(data['room'])

@socketio.on('my room event')
def on_room_event(data):
    room = data.pop('room')
    emit('my room response', data, room=room)

@socketio.on('my room namespace event', namespace='/test')
def on_room_namespace_event(data):
    room = data.pop('room')
    send('room message', room=room)

@socketio.on('bad response')
def on_bad_response():
    emit('my custom response', {'foo': socketio})

@socketio.on('bad callback')
def on_bad_callback():
    return {'foo': socketio}

@socketio.on('changing response')
def on_changing_response():
    data = {'foo': 'bar'}
    emit('my custom response', data)
    data['foo'] = 'baz'
    return data

@socketio.on_error()
def error_handler(value):
    if isinstance(value, AssertionError):
        global error_testing
        error_testing = True
    else:
        raise value
    return 'error'

@socketio.on('error testing')
def raise_error(data):
    raise AssertionError()

@socketio.on_error('/test')
def error_handler_namespace(value):
    if isinstance(value, AssertionError):
        global error_testing_namespace
        error_testing_namespace = True
    else:
        raise value
    return 'error/test'

@socketio.on("error testing", namespace='/test')
def raise_error_namespace(data):
    raise AssertionError()

@socketio.on_error_default
def error_handler_default(value):
    if isinstance(value, AssertionError):
        global error_testing_default
        error_testing_default = True
    else:
        raise value
    return 'error/default'

@socketio.on("error testing", namespace='/unused_namespace')
def raise_error_default(data):
    raise AssertionError()

class MyNamespace(Namespace):
    def on_connect(self):
        send('connected-ns')
        send(json.dumps(request.args.to_dict(flat=False)))
        send(json.dumps(
            {h: request.headers[h] for h in request.headers.keys()
             if h not in ['Host', 'Content-Type', 'Content-Length']}))

def on_disconnect(self):
        global disconnected
        disconnected = '/ns'

def on_message(self, message):
        send(message)
        if message == 'test session':
            session['a'] = 'b'
        if message not in "test noackargs":
            return message

def on_json(self, data):
        send(data, json=True, broadcast=True)
        if not data.get('noackargs'):
            return data

def on_exit(self, data):
        disconnect()

def on_my_custom_event(self, data):
        emit('my custom response', data)
        if not data.get('noackargs'):
            return data

def on_other_custom_event(self, data):
        global request_event_data
        request_event_data = request.event
        emit('my custom response', data)

socketio.on_namespace(MyNamespace('/ns'))

@app.route('/session')
def session_route():
    session['foo'] = 'bar'
    return ''

class TestSocketIO(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        pass

@classmethod
    def tearDownClass(cls):
        pass

def setUp(self):
        pass

def tearDown(self):
        pass

def test_connect(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client2 = socketio.test_client(app, auth={'foo': 'bar'})
        self.assertTrue(client.is_connected())
        self.assertTrue(client2.is_connected())
        self.assertNotEqual(client.eio_sid, client2.eio_sid)
        received = client.get_received()
        self.assertEqual(len(received), 3)
        self.assertEqual(received[0]['args'], 'connected')
        self.assertEqual(received[1]['args'], '{}')
        self.assertEqual(received[2]['args'], '{}')
        client.disconnect()
        self.assertFalse(client.is_connected())
        self.assertTrue(client2.is_connected())
        client2.disconnect()
        self.assertFalse(client2.is_connected())

def test_connect_query_string_and_headers(self):
        client = socketio.test_client(
            app, query_string='?foo=bar&foo=baz',
            headers={'Authorization': 'Bearer foobar'},
            auth={'foo': 'bar'})
        received = client.get_received()
        self.assertEqual(len(received), 3)
        self.assertEqual(received[0]['args'], 'connected')
        self.assertEqual(received[1]['args'], '{"foo": ["bar", "baz"]}')
        self.assertEqual(received[2]['args'],
                         '{"Authorization": "Bearer foobar"}')
        client.disconnect()

def test_connect_namespace(self):
        client = socketio.test_client(app, namespace='/test')
        self.assertTrue(client.is_connected('/test'))
        received = client.get_received('/test')
        self.assertEqual(len(received), 3)
        self.assertEqual(received[0]['args'], 'connected-test')
        self.assertEqual(received[1]['args'], '{}')
        self.assertEqual(received[2]['args'], '{}')
        client.disconnect(namespace='/test')
        self.assertFalse(client.is_connected('/test'))

def test_connect_namespace_query_string_and_headers(self):
        client = socketio.test_client(
            app, namespace='/test', query_string='foo=bar',
            headers={'My-Custom-Header': 'Value'})
        received = client.get_received('/test')
        self.assertEqual(len(received), 3)
        self.assertEqual(received[0]['args'], 'connected-test')
        self.assertEqual(received[1]['args'], '{"foo": ["bar"]}')
        self.assertEqual(received[2]['args'], '{"My-Custom-Header": "Value"}')
        client.disconnect(namespace='/test')

def test_connect_rejected(self):
        client = socketio.test_client(app, query_string='fail=1',
                                      auth={'foo': 'bar'})
        self.assertFalse(client.is_connected())

def test_disconnect(self):
        global disconnected
        disconnected = None
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client.disconnect()
        self.assertEqual(disconnected, '/')

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

def test_message_queue_options(self):
        app = Flask(__name__)
        socketio = SocketIO(app, message_queue='redis://')
        self.assertFalse(socketio.server_options['client_manager'].write_only)

app = Flask(__name__)
        socketio = SocketIO(app)
        socketio.init_app(app, message_queue='redis://')
        self.assertFalse(socketio.server_options['client_manager'].write_only)

app = Flask(__name__)
        socketio = SocketIO(message_queue='redis://')
        self.assertTrue(socketio.server_options['client_manager'].write_only)

app = Flask(__name__)
        socketio = SocketIO()
        socketio.init_app(None, message_queue='redis://')
        self.assertTrue(socketio.server_options['client_manager'].write_only)

def test_send(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client.get_received()
        client.send('echo this message back')
        received = client.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args'], 'echo this message back')

def test_send_json(self):
        client1 = socketio.test_client(app, auth={'foo': 'bar'})
        client2 = socketio.test_client(app, auth={'foo': 'bar'})
        client1.get_received()
        client2.get_received()
        client1.send({'a': 'b'}, json=True)
        received = client1.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args']['a'], 'b')
        received = client2.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args']['a'], 'b')

def test_send_namespace(self):
        client = socketio.test_client(app, namespace='/test')
        client.get_received('/test')
        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')
        client.get_received('/test')
        client.send({'a': 'b'}, json=True, namespace='/test')
        received = client.get_received('/test')
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args']['a'], 'b')

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

def test_emit_binary(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client.get_received()
        client.emit('my custom event', {u'a': b'\x01\x02\x03'})
        received = client.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(len(received[0]['args']), 1)
        self.assertEqual(received[0]['name'], 'my custom response')
        self.assertEqual(received[0]['args'][0]['a'], b'\x01\x02\x03')

def test_request_event_data(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client.get_received()
        global request_event_data
        request_event_data = None
        client.emit('other custom event', 'foo')
        expected_data = {'message': 'other custom event', 'args': ('foo',)}
        self.assertEqual(request_event_data, expected_data)
        client.emit('and another custom event', 'bar')
        expected_data = {'message': 'and another custom event',
                         'args': ('bar',)}
        self.assertEqual(request_event_data, expected_data)

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

def test_broadcast(self):
        client1 = socketio.test_client(app, auth={'foo': 'bar'})
        client2 = socketio.test_client(app, auth={'foo': 'bar'})
        client3 = socketio.test_client(app, namespace='/test')
        client2.get_received()
        client3.get_received('/test')
        client1.emit('my custom broadcast event', {'a': 'b'}, broadcast=True)
        received = client2.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(len(received[0]['args']), 1)
        self.assertEqual(received[0]['name'], 'my custom response')
        self.assertEqual(received[0]['args'][0]['a'], 'b')
        self.assertEqual(len(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, auth={'foo': 'bar'})
        client2.get_received('/test')
        client3.get_received()
        client1.emit('my custom broadcast namespace event', {'a': 'b'},
                     namespace='/test')
        received = client2.get_received('/test')
        self.assertEqual(len(received), 1)
        self.assertEqual(len(received[0]['args']), 1)
        self.assertEqual(received[0]['name'], 'my custom namespace response')
        self.assertEqual(received[0]['args'][0]['a'], 'b')
        self.assertEqual(len(client3.get_received()), 0)

def test_managed_session(self):
        flask_client = app.test_client()
        flask_client.get('/session')
        client = socketio.test_client(app, flask_test_client=flask_client,
                                      auth={'foo': 'bar'})
        client.get_received()
        client.send('echo this message back')
        self.assertEqual(
            socketio.server.environ[client.eio_sid]['saved_session'],
            {'foo': 'bar'})
        client.send('test session')
        self.assertEqual(
            socketio.server.environ[client.eio_sid]['saved_session'],
            {'a': 'b', 'foo': 'bar'})
        client.send('test session')
        self.assertEqual(
            socketio.server.environ[client.eio_sid]['saved_session'],
            {'a': 'c', 'foo': 'bar'})

def test_unmanaged_session(self):
        socketio.manage_session = False
        flask_client = app.test_client()
        flask_client.get('/session')
        client = socketio.test_client(app, flask_test_client=flask_client,
                                      auth={'foo': 'bar'})
        client.get_received()
        client.send('test session')
        client.send('test session')
        socketio.manage_session = True

def test_room(self):
        client1 = socketio.test_client(app, auth={'foo': 'bar'})
        client2 = socketio.test_client(app, auth={'foo': 'bar'})
        client3 = socketio.test_client(app, namespace='/test')
        client1.get_received()
        client2.get_received()
        client3.get_received('/test')
        client1.emit('join room', {'room': 'one'})
        client2.emit('join room', {'room': 'one'})
        client3.emit('join room', {'room': 'one'}, namespace='/test')
        client1.emit('my room event', {'a': 'b', 'room': 'one'})
        received = client1.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(len(received[0]['args']), 1)
        self.assertEqual(received[0]['name'], 'my room response')
        self.assertEqual(received[0]['args'][0]['a'], 'b')
        self.assertEqual(received, client2.get_received())
        received = client3.get_received('/test')
        self.assertEqual(len(received), 0)
        client1.emit('leave room', {'room': 'one'})
        client1.emit('my room event', {'a': 'b', 'room': 'one'})
        received = client1.get_received()
        self.assertEqual(len(received), 0)
        received = client2.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(len(received[0]['args']), 1)
        self.assertEqual(received[0]['name'], 'my room response')
        self.assertEqual(received[0]['args'][0]['a'], 'b')
        client2.disconnect()
        socketio.emit('my room event', {'a': 'b'}, room='one')
        received = client1.get_received()
        self.assertEqual(len(received), 0)
        received = client3.get_received('/test')
        self.assertEqual(len(received), 0)
        client3.emit('my room namespace event', {'room': 'one'},
                     namespace='/test')
        received = client3.get_received('/test')
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['name'], 'message')
        self.assertEqual(received[0]['args'], 'room message')
        socketio.close_room('one', namespace='/test')
        client3.emit('my room namespace event', {'room': 'one'},
                     namespace='/test')
        received = client3.get_received('/test')
        self.assertEqual(len(received), 0)

def test_error_handling(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client.get_received()
        global error_testing
        error_testing = False
        client.emit("error testing", "")
        self.assertTrue(error_testing)

def test_error_handling_namespace(self):
        client = socketio.test_client(app, namespace='/test')
        client.get_received('/test')
        global error_testing_namespace
        error_testing_namespace = False
        client.emit("error testing", "", namespace='/test')
        self.assertTrue(error_testing_namespace)

def test_error_handling_default(self):
        client = socketio.test_client(app, namespace='/unused_namespace')
        client.get_received('/unused_namespace')
        global error_testing_default
        error_testing_default = False
        client.emit("error testing", "", namespace='/unused_namespace')
        self.assertTrue(error_testing_default)

def test_ack(self):
        client1 = socketio.test_client(app, auth={'foo': 'bar'})
        client2 = socketio.test_client(app, auth={'foo': 'bar'})
        client3 = socketio.test_client(app, auth={'foo': 'bar'})
        ack = client1.send('echo this message back', callback=True)
        self.assertEqual(ack, 'echo this message back')
        ack = client1.send('test noackargs', callback=True)
        # python-socketio releases before 1.5 did not correctly implement
        # callbacks with no arguments. Here we check for [] (the correct, 1.5
        # and up response) and None (the wrong pre-1.5 response).
        self.assertTrue(ack == [] or ack is None)
        ack2 = client2.send({'a': 'b'}, json=True, callback=True)
        self.assertEqual(ack2, {'a': 'b'})
        ack3 = client3.emit('my custom event', {'a': 'b'}, callback=True)
        self.assertEqual(ack3, {'a': 'b'})

def test_noack(self):
        client1 = socketio.test_client(app, auth={'foo': 'bar'})
        client2 = socketio.test_client(app, auth={'foo': 'bar'})
        client3 = socketio.test_client(app, auth={'foo': 'bar'})
        no_ack_dict = {'noackargs': True}
        noack = client1.send("test noackargs", callback=False)
        self.assertIsNone(noack)
        noack2 = client2.send(no_ack_dict, json=True, callback=False)
        self.assertIsNone(noack2)
        noack3 = client3.emit('my custom event', no_ack_dict)
        self.assertIsNone(noack3)

def test_error_handling_ack(self):
        client1 = socketio.test_client(app, auth={'foo': 'bar'})
        client2 = socketio.test_client(app, namespace='/test')
        client3 = socketio.test_client(app, namespace='/unused_namespace')
        errorack = client1.emit("error testing", "", callback=True)
        self.assertEqual(errorack, 'error')
        errorack_namespace = client2.emit("error testing", "",
                                          namespace='/test', callback=True)
        self.assertEqual(errorack_namespace, 'error/test')
        errorack_default = client3.emit("error testing", "",
                                        namespace='/unused_namespace',
                                        callback=True)
        self.assertEqual(errorack_default, 'error/default')

def test_on_event(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client.get_received()
        global request_event_data
        request_event_data = None
        client.emit('yet another custom event', 'foo')
        expected_data = {'message': 'yet another custom event',
                         'args': ('foo',)}
        self.assertEqual(request_event_data, expected_data)

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

def test_connect_class_based(self):
        client = socketio.test_client(app, namespace='/ns')
        received = client.get_received('/ns')
        self.assertEqual(len(received), 3)
        self.assertEqual(received[0]['args'], 'connected-ns')
        self.assertEqual(received[1]['args'], '{}')
        self.assertEqual(received[2]['args'], '{}')
        client.disconnect('/ns')

def test_connect_class_based_query_string_and_headers(self):
        client = socketio.test_client(
            app, namespace='/ns', query_string='foo=bar',
            headers={'Authorization': 'Basic foobar'})
        received = client.get_received('/ns')
        self.assertEqual(len(received), 3)
        self.assertEqual(received[0]['args'], 'connected-ns')
        self.assertEqual(received[1]['args'], '{"foo": ["bar"]}')
        self.assertEqual(received[2]['args'],
                         '{"Authorization": "Basic foobar"}')
        client.disconnect('/ns')

def test_disconnect_class_based(self):
        global disconnected
        disconnected = None
        client = socketio.test_client(app, namespace='/ns')
        client.disconnect('/ns')
        self.assertEqual(disconnected, '/ns')

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

def test_send_json_class_based(self):
        client = socketio.test_client(app, namespace='/ns')
        client.get_received('/ns')
        client.send({'a': 'b'}, json=True, namespace='/ns')
        received = client.get_received('/ns')
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args']['a'], 'b')

def test_server_disconnected(self):
        client = socketio.test_client(app, namespace='/ns')
        client2 = socketio.test_client(app, namespace='/ns')
        client.get_received('/ns')
        client2.get_received('/ns')
        client.emit('exit', {}, namespace='/ns')
        self.assertFalse(client.is_connected('/ns'))
        self.assertTrue(client2.is_connected('/ns'))
        with self.assertRaises(RuntimeError):
            client.emit('hello', {}, namespace='/ns')
        client2.emit('exit', {}, namespace='/ns')
        self.assertFalse(client2.is_connected('/ns'))
        with self.assertRaises(RuntimeError):
            client2.emit('hello', {}, namespace='/ns')

def test_emit_class_based(self):
        client = socketio.test_client(app, namespace='/ns')
        client.get_received('/ns')
        client.emit('my_custom_event', {'a': 'b'}, namespace='/ns')
        received = client.get_received('/ns')
        self.assertEqual(len(received), 1)
        self.assertEqual(len(received[0]['args']), 1)
        self.assertEqual(received[0]['name'], 'my custom response')
        self.assertEqual(received[0]['args'][0]['a'], 'b')

def test_request_event_data_class_based(self):
        client = socketio.test_client(app, namespace='/ns')
        client.get_received('/ns')
        global request_event_data
        request_event_data = None
        client.emit('other_custom_event', 'foo', namespace='/ns')
        expected_data = {'message': 'other_custom_event', 'args': ('foo',)}
        self.assertEqual(request_event_data, expected_data)

def test_delayed_init(self):
        app = Flask(__name__)
        socketio = SocketIO(allow_upgrades=False, json=flask_json)

@socketio.on('connect')
        def on_connect():
            send({'connected': 'foo'}, json=True)

socketio.init_app(app, cookie='foo')
        self.assertFalse(socketio.server.eio.allow_upgrades)
        self.assertEqual(socketio.server.eio.cookie, 'foo')

client = socketio.test_client(app, auth={'foo': 'bar'})
        received = client.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args'], {'connected': 'foo'})

def test_encode_decode(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        client.get_received()
        data = {'foo': 'bar', 'invalid': socketio}
        self.assertRaises(TypeError, client.emit, 'my custom event', data,
                          callback=True)
        data = {'foo': 'bar'}
        ack = client.emit('my custom event', data, callback=True)
        data['foo'] = 'baz'
        received = client.get_received()
        self.assertEqual(ack, {'foo': 'bar'})
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args'][0], {'foo': 'bar'})

def test_encode_decode_2(self):
        client = socketio.test_client(app, auth={'foo': 'bar'})
        self.assertRaises(TypeError, client.emit, 'bad response')
        self.assertRaises(TypeError, client.emit, 'bad callback',
                          callback=True)
        client.get_received()
        ack = client.emit('changing response', callback=True)
        received = client.get_received()
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0]['args'][0], {'foo': 'bar'})
        self.assertEqual(ack, {'foo': 'baz'})

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

Places

File test_socketio.py of Package python-Flask-SocketIO

Places