from unittest import TestCase, main from socketio.namespace import BaseNamespace from socketio.virtsocket import Socket from mock import MagicMock class MockSocketIOServer(object): """Mock a SocketIO server""" def __init__(self, *args, **kwargs): self.sockets = {} def get_socket(self, socket_id=''): return self.sockets.get(socket_id) class MockSocket(Socket): pass class ChatNamespace(BaseNamespace): def __init__(self, *args, **kwargs): self.use_set = args[0] super(ChatNamespace, self).__init__(*args[1:], **kwargs) def get_initial_acl(self): acls = ['on_foo'] if self.use_set == True: return set(acls) else: return acls def on_foo(self): return 'a' def on_bar(self): return 'b' def on_baz(foo, bar, baz): return 'c' class GlobalNamespace(BaseNamespace): def on_woot(self): return '' def on_tobi(self): return '' class TestBaseNamespace(TestCase): def setUp(self): server = MockSocketIOServer() self.environ = {} socket = MockSocket(server, {}) socket.error = MagicMock() self.environ['socketio'] = socket self.ns = GlobalNamespace(self.environ, '/woot') def test_process_packet_disconnect(self): pkt = {'type': 'disconnect', 'endpoint': '/woot' } self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_process_packet_connect(self): """processing a connection packet """ pkt = {'type': 'connect', 'endpoint': '/tobi', 'qs': '' } self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called # processing a connection packet with query string pkt = {'type': 'connect', 'endpoint': '/test', 'qs': '?test=1' } self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_process_packet_heartbeat(self): """processing a heartbeat packet """ pkt = {'type': 'heartbeat', 'endpoint': '' } self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_process_packet_message(self): """processing a message packet """ pkt = {'type': 'message', 'endpoint': '', 'data': 'woot'} data = self.ns.process_packet(pkt) self.assertEqual(data, pkt['data']) assert not self.environ['socketio'].error.called # processing a message packet with id and endpoint pkt = {'type': 'message', 'id': 5, 'ack': True, 'endpoint': '/tobi', 'data': ''} data = self.ns.process_packet(pkt) self.assertEqual(data, pkt['data']) assert not self.environ['socketio'].error.called def test_process_packet_json(self): """processing json packet """ pkt = {'type': 'json', 'endpoint': '', 'data': '2'} data = self.ns.process_packet(pkt) self.assertEqual(data, pkt['data']) assert not self.environ['socketio'].error.called # processing json packet with message id and ack data pkt = {'type': 'json', 'id': 1, 'endpoint': '', 'ack': 'data', 'data': {u'a': u'b'}} data = self.ns.process_packet(pkt) self.assertEqual(data, pkt['data']) assert not self.environ['socketio'].error.called def test_process_packet_event(self): """processing an event packet """ pkt = {'type': 'event', 'name': 'woot', 'endpoint': '', 'args': []} self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called # processing an event packet with message id and ack pkt = {'type': 'event', 'id': 1, 'ack': 'data', 'name': 'tobi', 'endpoint': '', 'args': []} self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_process_packet_ack(self): """processing a ack packet """ pkt = {'type': 'ack', 'ackId': 140, 'endpoint': '', 'args': []} self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_process_packet_error(self): """processing error packet """ pkt = {'type': 'error', 'reason': '', 'advice': '', 'endpoint': ''} self.ns.process_packet(pkt) pkt = {'type': 'error', 'reason': 'transport not supported', 'advice': '', 'endpoint': ''} self.ns.process_packet(pkt) # processing error packet with reason and advice pkt = {'type': 'error', 'reason': 'unauthorized', 'advice': 'reconnect', 'endpoint': ''} self.ns.process_packet(pkt) # processing error packet with endpoint pkt = {'type': 'error', 'reason': '', 'advice': '', 'endpoint': '/woot'} self.ns.process_packet(pkt) def test_process_packet_message_with_new_line(self): """processing a newline in a message""" pkt = {'type': 'message', 'data': '\n', 'endpoint': ''} self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_del_acl_method(self): pkt = {'type': 'event', 'name': 'foo', 'endpoint': '/chat', 'args': []} message = ("Trying to delete an ACL method, but none were" + " defined yet! Or: No ACL restrictions yet, why would you" + " delete one?") try: self.ns.del_acl_method('on_foo') self.ns.process_packet(pkt) except ValueError as e: self.assertEqual( message, e.message, ) else: raise Exception("""We should not be able to delete an acl that doesn't exist""") def test_allowed_event_name_regex(self): pkt = {'type': 'event', 'name': '$foo', 'endpoint': '/chat', 'args': []} self.ns.process_packet(pkt) args = ['unallowed_event_name', 'name must only contains alpha numerical characters', ] kwargs = dict(msg_id=None, endpoint='/woot', quiet=False) self.environ['socketio'].error.assert_called_with(*args, **kwargs) def test_method_not_found(self): """ test calling a method that doesn't exist """ pkt = {'type': 'event', 'name': 'foo', 'endpoint': '/chat', 'args': [] } self.ns.process_packet(pkt) kwargs = dict( msg_id=None, endpoint='/woot', quiet=False ) self.environ['socketio'].error.assert_called_with( 'no_such_method', 'The method "%s" was not found' % 'on_foo', **kwargs ) class TestChatNamespace(TestCase): def setUp(self): server = MockSocketIOServer() self.environ = {} socket = MockSocket(server, {}) socket.error = MagicMock() self.environ['socketio'] = socket self.ns = ChatNamespace( False, self.environ, '/chat' ) def test_allowed_event(self): pkt = {'type': 'event', 'name': 'foo', 'endpoint': '/chat', 'args': []} self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_blocked_event(self): pkt = {'type': 'event', 'name': 'bar', 'endpoint': '/chat', 'args': []} self.ns.process_packet(pkt) args = [ 'method_access_denied', 'You do not have access to method "on_bar"', ] kwargs = dict( msg_id=None, endpoint='/chat', quiet=False ) self.environ['socketio'].error.assert_called_with(*args, **kwargs) def test_add_acl_method(self): pkt = {'type': 'event', 'name': 'bar', 'endpoint': '/chat', 'args': []} self.ns.add_acl_method('on_bar') self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_del_acl_method(self): pkt = {'type': 'event', 'name': 'foo', 'endpoint': '/chat', 'args': []} self.ns.del_acl_method('on_foo') self.ns.process_packet(pkt) args = [ 'method_access_denied', 'You do not have access to method "on_foo"', ] kwargs = dict( msg_id=None, endpoint='/chat', quiet=False ) self.environ['socketio'].error.assert_called_with(*args, **kwargs) def test_lift_acl_restrictions(self): pkt1 = {'type': 'event', 'name': 'foo', 'endpoint': '/chat', 'args': []} self.ns.lift_acl_restrictions() self.ns.process_packet(pkt1) assert not self.environ['socketio'].error.called pkt2 = {'type': 'event', 'name': 'bar', 'endpoint': '/chat', 'args': []} self.ns.process_packet(pkt2) assert not self.environ['socketio'].error.called def test_use_set_on_acl(self): self.ns = ChatNamespace( True, self.environ, '/chat' ) pkt = {'type': 'event', 'name': 'bar', 'endpoint': '/chat', 'args': []} self.ns.add_acl_method('on_bar') self.ns.process_packet(pkt) assert not self.environ['socketio'].error.called def test_call_method_invalid_definition(self): pkt = {'type': 'event', 'name': 'baz', 'endpoint': '/chat', 'args': []} self.ns.add_acl_method('on_baz') self.ns.process_packet(pkt) kwargs = dict(msg_id=None, endpoint='/chat', quiet=False) self.environ['socketio'].error.assert_called_with( "invalid_method_args", "The server-side method is invalid, as it doesn't " "have 'self' as its first argument" , **kwargs) if __name__ == '__main__': main()