From fe969111ce274aab16d6baf2edd0adebc359e8be Mon Sep 17 00:00:00 2001 From: Tehom Date: Tue, 16 Oct 2018 19:51:43 -0400 Subject: [PATCH 1/4] Add stub for testing Telnet --- evennia/server/portal/tests.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/evennia/server/portal/tests.py b/evennia/server/portal/tests.py index be400144c6..b09f71ae03 100644 --- a/evennia/server/portal/tests.py +++ b/evennia/server/portal/tests.py @@ -11,6 +11,11 @@ except ImportError: import string from evennia.server.portal import irc +from twisted.test import proto_helpers +from twisted.trial.unittest import TestCase as TwistedTestCase + +from .telnet import TelnetServerFactory + class TestIRC(TestCase): @@ -73,3 +78,15 @@ class TestIRC(TestCase): s = r'|wthis|Xis|gis|Ma|C|complex|*string' self.assertEqual(irc.parse_irc_to_ansi(irc.parse_ansi_to_irc(s)), s) + + +class TestTelnet(TwistedTestCase): + def setUp(self): + super(TestTelnet, self).setUp() + factory = TelnetServerFactory() + self.proto = factory.buildProtocol(("localhost", 0)) + self.transport = proto_helpers.StringTransport() + + def test_connect(self): + self.proto.makeConnection(self.transport) + # TODO: Add rest of stuff for testing connection From 40a37e501f1ffb94f448e886a13220abf402caa0 Mon Sep 17 00:00:00 2001 From: Tehom Date: Tue, 16 Oct 2018 20:33:12 -0400 Subject: [PATCH 2/4] Add some cleanup steps to prevent unclean reactor --- evennia/server/portal/telnet.py | 2 +- evennia/server/portal/tests.py | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/evennia/server/portal/telnet.py b/evennia/server/portal/telnet.py index 955ea5e918..83e2fa03a2 100644 --- a/evennia/server/portal/telnet.py +++ b/evennia/server/portal/telnet.py @@ -84,7 +84,7 @@ class TelnetProtocol(Telnet, StatefulTelnetProtocol, Session): from evennia.utils.utils import delay # timeout the handshakes in case the client doesn't reply at all - delay(2, callback=self.handshake_done, timeout=True) + self._handshake_delay = delay(2, callback=self.handshake_done, timeout=True) # TCP/IP keepalive watches for dead links self.transport.setTcpKeepAlive(1) diff --git a/evennia/server/portal/tests.py b/evennia/server/portal/tests.py index b09f71ae03..53a732f121 100644 --- a/evennia/server/portal/tests.py +++ b/evennia/server/portal/tests.py @@ -8,13 +8,15 @@ try: except ImportError: import unittest +from mock import Mock import string from evennia.server.portal import irc from twisted.test import proto_helpers from twisted.trial.unittest import TestCase as TwistedTestCase -from .telnet import TelnetServerFactory +from .telnet import TelnetServerFactory, TelnetProtocol +from .portal import PORTAL_SESSIONS class TestIRC(TestCase): @@ -84,9 +86,19 @@ class TestTelnet(TwistedTestCase): def setUp(self): super(TestTelnet, self).setUp() factory = TelnetServerFactory() + factory.protocol = TelnetProtocol + factory.sessionhandler = PORTAL_SESSIONS + factory.sessionhandler.portal = Mock() self.proto = factory.buildProtocol(("localhost", 0)) self.transport = proto_helpers.StringTransport() + self.addCleanup(factory.sessionhandler.disconnect_all) def test_connect(self): - self.proto.makeConnection(self.transport) + self.transport.client = ["localhost"] + self.transport.setTcpKeepAlive = Mock() + d = self.proto.makeConnection(self.transport) # TODO: Add rest of stuff for testing connection + # clean up to prevent Unclean reactor + self.proto.nop_keep_alive.stop() + self.proto._handshake_delay.cancel() + return d From b4383592016e532ebedb315616353712988bd257 Mon Sep 17 00:00:00 2001 From: Tehom Date: Wed, 17 Oct 2018 01:50:57 -0400 Subject: [PATCH 3/4] Add test of NOGOAHEAD --- evennia/server/portal/tests.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/evennia/server/portal/tests.py b/evennia/server/portal/tests.py index 53a732f121..4142ca297e 100644 --- a/evennia/server/portal/tests.py +++ b/evennia/server/portal/tests.py @@ -12,11 +12,13 @@ from mock import Mock import string from evennia.server.portal import irc +from twisted.conch.telnet import IAC, NOP, LINEMODE, GA, WILL, WONT, ECHO, NULL, DONT from twisted.test import proto_helpers from twisted.trial.unittest import TestCase as TwistedTestCase from .telnet import TelnetServerFactory, TelnetProtocol from .portal import PORTAL_SESSIONS +from .suppress_ga import SUPPRESS_GA class TestIRC(TestCase): @@ -98,6 +100,10 @@ class TestTelnet(TwistedTestCase): self.transport.setTcpKeepAlive = Mock() d = self.proto.makeConnection(self.transport) # TODO: Add rest of stuff for testing connection + self.assertTrue(self.proto.protocol_flags["NOGOAHEAD"]) + self.proto.dataReceived(IAC + DONT+ SUPPRESS_GA) + self.assertFalse(self.proto.protocol_flags["NOGOAHEAD"]) + self.assertEqual(self.proto.handshakes, 7) # clean up to prevent Unclean reactor self.proto.nop_keep_alive.stop() self.proto._handshake_delay.cancel() From ef0e0e0b4c6e2827f8832512632c580280c1af28 Mon Sep 17 00:00:00 2001 From: Tehom Date: Wed, 17 Oct 2018 12:49:48 -0400 Subject: [PATCH 4/4] Add various simple tests for different handshakes --- evennia/server/portal/tests.py | 49 +++++++++++++++++++++++++++++++--- 1 file changed, 45 insertions(+), 4 deletions(-) diff --git a/evennia/server/portal/tests.py b/evennia/server/portal/tests.py index 4142ca297e..791e5172a4 100644 --- a/evennia/server/portal/tests.py +++ b/evennia/server/portal/tests.py @@ -12,13 +12,19 @@ from mock import Mock import string from evennia.server.portal import irc -from twisted.conch.telnet import IAC, NOP, LINEMODE, GA, WILL, WONT, ECHO, NULL, DONT +from twisted.conch.telnet import IAC, WILL, DONT, SB, SE, NAWS, DO from twisted.test import proto_helpers from twisted.trial.unittest import TestCase as TwistedTestCase from .telnet import TelnetServerFactory, TelnetProtocol from .portal import PORTAL_SESSIONS from .suppress_ga import SUPPRESS_GA +from .naws import DEFAULT_HEIGHT, DEFAULT_WIDTH +from .ttype import TTYPE, IS +from .mccp import MCCP +from .mssp import MSSP +from .mxp import MXP +from .telnet_oob import MSDP, MSDP_VAL, MSDP_VAR class TestIRC(TestCase): @@ -95,15 +101,50 @@ class TestTelnet(TwistedTestCase): self.transport = proto_helpers.StringTransport() self.addCleanup(factory.sessionhandler.disconnect_all) - def test_connect(self): + def test_mudlet_ttype(self): self.transport.client = ["localhost"] self.transport.setTcpKeepAlive = Mock() d = self.proto.makeConnection(self.transport) - # TODO: Add rest of stuff for testing connection + # test suppress_ga self.assertTrue(self.proto.protocol_flags["NOGOAHEAD"]) - self.proto.dataReceived(IAC + DONT+ SUPPRESS_GA) + self.proto.dataReceived(IAC + DONT + SUPPRESS_GA) self.assertFalse(self.proto.protocol_flags["NOGOAHEAD"]) self.assertEqual(self.proto.handshakes, 7) + # test naws + self.assertEqual(self.proto.protocol_flags['SCREENWIDTH'], {0: DEFAULT_WIDTH}) + self.assertEqual(self.proto.protocol_flags['SCREENHEIGHT'], {0: DEFAULT_HEIGHT}) + self.proto.dataReceived(IAC + WILL + NAWS) + self.proto.dataReceived([IAC, SB, NAWS, '', 'x', '', 'd', IAC, SE]) + self.assertEqual(self.proto.protocol_flags['SCREENWIDTH'][0], 120) + self.assertEqual(self.proto.protocol_flags['SCREENHEIGHT'][0], 100) + self.assertEqual(self.proto.handshakes, 6) + # test ttype + self.assertTrue(self.proto.protocol_flags["FORCEDENDLINE"]) + self.assertFalse(self.proto.protocol_flags["TTYPE"]) + self.assertTrue(self.proto.protocol_flags["ANSI"]) + self.proto.dataReceived(IAC + WILL + TTYPE) + self.proto.dataReceived([IAC, SB, TTYPE, IS, "MUDLET", IAC, SE]) + self.assertTrue(self.proto.protocol_flags["XTERM256"]) + self.assertEqual(self.proto.protocol_flags["CLIENTNAME"], "MUDLET") + self.proto.dataReceived([IAC, SB, TTYPE, IS, "XTERM", IAC, SE]) + self.proto.dataReceived([IAC, SB, TTYPE, IS, "MTTS 137", IAC, SE]) + self.assertEqual(self.proto.handshakes, 5) + # test mccp + self.proto.dataReceived(IAC + DONT + MCCP) + self.assertFalse(self.proto.protocol_flags['MCCP']) + self.assertEqual(self.proto.handshakes, 4) + # test mssp + self.proto.dataReceived(IAC + DONT + MSSP) + self.assertEqual(self.proto.handshakes, 3) + # test oob + self.proto.dataReceived(IAC + DO + MSDP) + self.proto.dataReceived([IAC, SB, MSDP, MSDP_VAR, "LIST", MSDP_VAL, "COMMANDS", IAC, SE]) + self.assertTrue(self.proto.protocol_flags['OOB']) + self.assertEqual(self.proto.handshakes, 2) + # test mxp + self.proto.dataReceived(IAC + DONT + MXP) + self.assertFalse(self.proto.protocol_flags['MXP']) + self.assertEqual(self.proto.handshakes, 1) # clean up to prevent Unclean reactor self.proto.nop_keep_alive.stop() self.proto._handshake_delay.cancel()