Websocket unit tests

evennia/server/portal/tests.py

@@ -1,3 +1,4 @@
+
 try:
     from django.utils.unittest import TestCase
 except ImportError:
@@ -12,8 +13,12 @@ import sys
 import string
 import mock
 import pickle
+import json
+
 from mock import Mock, MagicMock
 from evennia.server.portal import irc
+from evennia.utils.test_resources import mockdelay, EvenniaTest, patch
+from twisted import internet
 
 from twisted.conch.telnet import IAC, WILL, DONT, SB, SE, NAWS, DO
 from twisted.test import proto_helpers
@@ -32,6 +37,9 @@ from .telnet_oob import MSDP, MSDP_VAL, MSDP_VAR
 from .amp import AMPMultiConnectionProtocol, MsgServer2Portal, MsgPortal2Server, AMP_MAXLEN
 from .amp_server import AMPServerFactory
 
+from autobahn.twisted.websocket import WebSocketServerFactory, WebSocketServerProtocol
+from .webclient import WebSocketClient
+
 
 class TestAMPServer(TwistedTestCase):
     """
@@ -269,3 +277,54 @@ class TestTelnet(TwistedTestCase):
         self.proto.nop_keep_alive.stop()
         self.proto._handshake_delay.cancel()
         return d
+
+
+class WebSocket(EvenniaTest):
+    def setUp(self):
+        super().setUp()
+        # from autobahn.websocket.protocol import WebSocketProtocol
+        # WebSocketServerProtocol.log = Mock()
+        # internet.base.DelayedCall.debug = True
+        # self.proto = WebSocketClient()
+        # self.proto.factory = WebSocketServerFactory()
+        # self.proto.factory.sessionhandler = PORTAL_SESSIONS
+        # self.proto.sessionhandler = PORTAL_SESSIONS
+        # self.proto.sessionhandler.portal = Mock()
+        # self.proto.data_in = MagicMock()
+        # self.proto.data_out = MagicMock()
+        # self.transport = WebSocketProtocol()
+        # self.transport.client = ["localhost"]
+        # self.transport.setTcpKeepAlive = Mock()
+        # self.proto.state = WebSocketProtocol.STATE_OPEN
+        self.proto = WebSocketClient()
+        self.proto.factory = WebSocketServerFactory()
+        self.proto.factory.sessionhandler = PORTAL_SESSIONS
+        self.proto.sessionhandler = PORTAL_SESSIONS
+        self.proto.sessionhandler.portal = Mock()
+        self.proto.transport = proto_helpers.StringTransport()
+        #self.proto.transport = proto_helpers.FakeDatagramTransport()
+        self.proto.transport.client = ["localhost"]
+        self.proto.transport.setTcpKeepAlive = Mock()
+        self.proto.state = MagicMock()
+        self.addCleanup(self.proto.factory.sessionhandler.disconnect_all)
+
+    def tearDown(self):
+        super().tearDown()
+
+    def test_data_in(self):
+        self.proto.sessionhandler.data_in = MagicMock()
+        self.proto.onOpen()
+        msg = json.dumps(["logged_in", (), {}]).encode()
+        self.proto.onMessage(msg, isBinary=False)
+        self.proto.sessionhandler.data_in.assert_called_with(self.proto, logged_in=[[], {}])
+        sendStr = "You can get anything you want at Alice's Restaurant."
+        msg = json.dumps(["text", (sendStr,), {}]).encode()
+        self.proto.onMessage(msg, isBinary=False)
+        self.proto.sessionhandler.data_in.assert_called_with(self.proto, text=[[sendStr], {}])
+
+    def test_data_out(self):
+        self.proto.onOpen()
+        self.proto.sendLine = MagicMock()
+        msg = json.dumps(["logged_in", (), {}])
+        self.proto.sessionhandler.data_out(self.proto, text=[["Excepting Alice"], {}])
+        self.proto.sendLine.assert_called_with(json.dumps(['text', ['Excepting Alice'], {}]))