Andreas/evennia
1
0
Fork 0
mirror of https://github.com/evennia/evennia.git synced 2026-03-28 10:37:16 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Websocket unit tests

Browse source
This commit is contained in:
trhr 2020-05-09 22:22:08 -05:00
parent 38c7aa40fe
commit 813d5f2b09
1 changed files with 59 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

59
evennia/server/portal/tests.py
View file

@ -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'], {}]))