pool/python-Flask-SocketIO
SHA256
2
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- version 5.4.1

Browse Source 
* source tarball renamed
  * Removed a debugging statement 
  * Support catch-all events with socketio.on('*') #2095 
  * Prevent crash when client sends empty event 
  * Handle events with catch-all namespace #2061
  * Add unit test for background tasks
  * Drop Python 3.7, add Python 3.12, upgrade to pypy 3.10

OBS-URL: https://build.opensuse.org/package/show/devel:languages:python:flask/python-Flask-SocketIO?expand=0&rev=15
This commit is contained in:
Markéta Machová 2024-11-11 11:40:41 +00:00 committed by Git OBS Bridge
commit bc0ff6a985
7 changed files with 953 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
.gitattributes vendored Normal file
View File

@ -0,0 +1,23 @@
## Default LFS
*.7z filter=lfs diff=lfs merge=lfs -text
*.bsp filter=lfs diff=lfs merge=lfs -text
*.bz2 filter=lfs diff=lfs merge=lfs -text
*.gem filter=lfs diff=lfs merge=lfs -text
*.gz filter=lfs diff=lfs merge=lfs -text
*.jar filter=lfs diff=lfs merge=lfs -text
*.lz filter=lfs diff=lfs merge=lfs -text
*.lzma filter=lfs diff=lfs merge=lfs -text
*.obscpio filter=lfs diff=lfs merge=lfs -text
*.oxt filter=lfs diff=lfs merge=lfs -text
*.pdf filter=lfs diff=lfs merge=lfs -text
*.png filter=lfs diff=lfs merge=lfs -text
*.rpm filter=lfs diff=lfs merge=lfs -text
*.tbz filter=lfs diff=lfs merge=lfs -text
*.tbz2 filter=lfs diff=lfs merge=lfs -text
*.tgz filter=lfs diff=lfs merge=lfs -text
*.ttf filter=lfs diff=lfs merge=lfs -text
*.txz filter=lfs diff=lfs merge=lfs -text
*.whl filter=lfs diff=lfs merge=lfs -text
*.xz filter=lfs diff=lfs merge=lfs -text
*.zip filter=lfs diff=lfs merge=lfs -text
*.zst filter=lfs diff=lfs merge=lfs -text

1
.gitignore vendored Normal file
View File

@ -0,0 +1 @@
.osc

3
Flask-SocketIO-5.3.6.tar.gz Normal file
View File

@ -0,0 +1,3 @@
version https://git-lfs.github.com/spec/v1
oid sha256:bb8f9f9123ef47632f5ce57a33514b0c0023ec3696b2384457f0fcaa5b70501c
size 17167

3
flask_socketio-5.4.1.tar.gz Normal file
View File

@ -0,0 +1,3 @@
version https://git-lfs.github.com/spec/v1
oid sha256:2e9b8864a5be37ca54f6c76a4d06b1ac5e0df61fde12d03afc81ab4057e1eb86
size 37261

95
python-Flask-SocketIO.changes Normal file
View File

@ -0,0 +1,95 @@
-------------------------------------------------------------------
Fri Oct 4 13:47:23 UTC 2024 - Axel Braun <axel.braun@gmx.de>
- version 5.4.1
* source tarball renamed
* Removed a debugging statement
* Support catch-all events with socketio.on('*') #2095
* Prevent crash when client sends empty event
* Handle events with catch-all namespace #2061
* Add unit test for background tasks
* Drop Python 3.7, add Python 3.12, upgrade to pypy 3.10
-------------------------------------------------------------------
Mon Feb 12 05:54:19 UTC 2024 - Steve Kowalik <steven.kowalik@suse.com>
- Add missing BuildRequires on setuptools.
-------------------------------------------------------------------
Mon Dec 18 12:19:34 UTC 2023 - Dirk Müller <dmueller@suse.com>
- update to 5.3.6:
* Fixes in the test client to support recent changes in Socket.IO
dependencies
-------------------------------------------------------------------
Fri Dec 1 16:29:35 UTC 2023 - Antonio Larrosa <alarrosa@suse.com>
- Add %{?sle15_python_module_pythons}
-------------------------------------------------------------------
Thu Aug 10 22:27:27 UTC 2023 - Arun Persaud <arun@gmx.de>
- update to version 5.3.5:
* Prevent allow_unsafe_werkzeug option from being passed to web
servers #2001 (commit)
-------------------------------------------------------------------
Wed Jun 21 21:36:05 UTC 2023 - Matej Cepl <mcepl@suse.com>
- Clean up SPEC file, make rpmlint happy.
-------------------------------------------------------------------
Wed Jun 21 19:56:26 UTC 2023 - Axel Braun <axel.braun@gmx.de>
- update to version 5.3.4
* For a full changelog see https://github.com/miguelgrinberg/Flask-SocketIO/blob/main/CHANGES.md
-------------------------------------------------------------------
Sat Oct 23 15:18:45 UTC 2021 - Axel Braun <axel.braun@gmx.de>
- Update to version 5.1.1
* Only use SSL socket if at least one SSL kwarg is not None #1639 (commit) (thanks JT Raber!)
* Remove unused SSL arguments from eventlet server options #1639 (commit)
* Remove executable permissions from files that lack shebang lines #1621 (commit) (thanks Ben Beasley!)
* Improved project structure (commit)
Release 5.1.0 - 2021-05-28
* Add reloader_options argument to socketio.run#1556 (commit)
* Pass auth data from client in connect event handler #1555 (commit)
* Do not show simple-websocket install prompt if it is already installed (commit)
* Fix namespace bug in example #1543 (commit)
* Added index to documentation #724 (commit)
* Fixed typo in documentation #1551 (commit) (thanks Mayank Anuragi!)
-------------------------------------------------------------------
Tue May 25 17:24:48 UTC 2021 - Ben Greiner <code@bnavigator.de>
- Update to 5.0.3
* Document the use of simple-websocket with the development web server (commit)
* Show transport in example apps (commit)
* Added Open Collective funding option (commit)
* Silence deprecation warning from Werkzeug 2.x #1549 (commit)
* Updated server options in documentation (commit)
* Updated socketio javascript client versions in documentation (commit)
* Fix typo in documentation #1524 (commit) (thanks lennertcl!)
* Change room documentation from room= to to= #1519 (commit) (thanks David McInnis!)
* Fixed a type in the Kafka URL in the documentation #1476 (commit) (thanks VVakko!)
* Fixed typos in documentation #1447 (commit)
* Add test_socketio.py from Github repo for testing, because it
is not included into the PyPI sdist anymore.
-------------------------------------------------------------------
Wed May 5 08:19:02 UTC 2021 - Axel Braun <axel.braun@gmx.de>
- conversion to singlespec
update to version 5.0.1
-------------------------------------------------------------------
Thu Feb 1 15:46:27 UTC 2018 - gsmith@suse.com
- Update references to python-socketio
-------------------------------------------------------------------
Wed Jan 31 21:01:27 UTC 2018 - gsmith@suse.com
- Initial version

61
python-Flask-SocketIO.spec Normal file
View File

@ -0,0 +1,61 @@
#
# spec file for package python-Flask-SocketIO
#
# Copyright (c) 2024 SUSE LLC
#
# All modifications and additions to the file contributed by third parties
# remain the property of their copyright owners, unless otherwise agreed
# upon. The license for this file, and modifications and additions to the
# file, is the same license as for the pristine package itself (unless the
# license for the pristine package is not an Open Source License, in which
# case the license is the MIT License). An "Open Source License" is a
# license that conforms to the Open Source Definition (Version 1.9)
# published by the Open Source Initiative.
# Please submit bugfixes or comments via https://bugs.opensuse.org/
#
%{?sle15_python_module_pythons}
Name: python-Flask-SocketIO
Version: 5.4.1
Release: 0
Summary: SocketIO integration for Flask applications
License: MIT
URL: https://github.com/miguelgrinberg/Flask-SocketIO/
Source: https://files.pythonhosted.org/packages/source/f/flask-socketio/flask_socketio-%{version}.tar.gz
BuildRequires: %{python_module Flask >= 0.9}
BuildRequires: %{python_module pip}
BuildRequires: %{python_module python-socketio >= 5.0.2}
BuildRequires: %{python_module redis}
BuildRequires: %{python_module setuptools}
BuildRequires: %{python_module wheel}
BuildRequires: fdupes
Requires: python-Flask >= 0.9
Requires: python-python-socketio >= 5.0.2
BuildArch: noarch
%python_subpackages
%description
Socket.IO integration for Flask applications.
%prep
%autosetup -n flask_socketio-%{version}
%build
%pyproject_wheel
%install
%pyproject_install
%python_expand %fdupes %{buildroot}%{$python_sitelib}
%check
%pyunittest -v test_socketio.py
%files %{python_files}
%doc README.md
%license LICENSE
%{python_sitelib}/flask_socketio
%{python_sitelib}/Flask_SocketIO-%{version}.dist-info
%changelog

767
test_socketio.py Normal file
View File

@ -0,0 +1,767 @@
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()