Miguel Grinberg
1 year ago
Failed to extract signature
3 changed files with
11 additions and
8 deletions
-
src/socketio/base_server.py
-
tests/async/test_server.py
-
tests/common/test_server.py
|
|
@ -187,15 +187,18 @@ class BaseServer: |
|
|
|
namespace = namespace or '/' |
|
|
|
return self.manager.get_rooms(sid, namespace) |
|
|
|
|
|
|
|
def transport(self, sid): |
|
|
|
def transport(self, sid, namespace=None): |
|
|
|
"""Return the name of the transport used by the client. |
|
|
|
|
|
|
|
The two possible values returned by this function are ``'polling'`` |
|
|
|
and ``'websocket'``. |
|
|
|
|
|
|
|
:param sid: The session of the client. |
|
|
|
:param namespace: The Socket.IO namespace. If this argument is omitted |
|
|
|
the default namespace is used. |
|
|
|
""" |
|
|
|
return self.eio.transport(sid) |
|
|
|
eio_sid = self.manager.eio_sid_from_sid(sid, namespace or '/') |
|
|
|
return self.eio.transport(eio_sid) |
|
|
|
|
|
|
|
def get_environ(self, sid, namespace=None): |
|
|
|
"""Return the WSGI environ dictionary for a client. |
|
|
|
|
|
@ -307,9 +307,9 @@ class TestAsyncServer(unittest.TestCase): |
|
|
|
eio.return_value.send = AsyncMock() |
|
|
|
s = async_server.AsyncServer() |
|
|
|
s.eio.transport = mock.MagicMock(return_value='polling') |
|
|
|
_run(s._handle_eio_connect('foo', 'environ')) |
|
|
|
assert s.transport('foo') == 'polling' |
|
|
|
s.eio.transport.assert_called_once_with('foo') |
|
|
|
sid_foo = _run(s.manager.connect('123', '/foo')) |
|
|
|
assert s.transport(sid_foo, '/foo') == 'polling' |
|
|
|
s.eio.transport.assert_called_once_with('123') |
|
|
|
|
|
|
|
def test_handle_connect(self, eio): |
|
|
|
eio.return_value.send = AsyncMock() |
|
|
|
|
|
@ -292,9 +292,9 @@ class TestServer(unittest.TestCase): |
|
|
|
def test_transport(self, eio): |
|
|
|
s = server.Server() |
|
|
|
s.eio.transport = mock.MagicMock(return_value='polling') |
|
|
|
s._handle_eio_connect('foo', 'environ') |
|
|
|
assert s.transport('foo') == 'polling' |
|
|
|
s.eio.transport.assert_called_once_with('foo') |
|
|
|
sid_foo = s.manager.connect('123', '/foo') |
|
|
|
assert s.transport(sid_foo, '/foo') == 'polling' |
|
|
|
s.eio.transport.assert_called_once_with('123') |
|
|
|
|
|
|
|
def test_handle_connect(self, eio): |
|
|
|
s = server.Server() |
|
|
|