Annikka/evennia
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add some cleanup steps to prevent unclean reactor

Browse source
This commit is contained in:
Tehom 2018-10-16 20:33:12 -04:00
parent fe969111ce
commit 40a37e501f
2 changed files with 15 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

2
evennia/server/portal/telnet.py
View file

@ -84,7 +84,7 @@ class TelnetProtocol(Telnet, StatefulTelnetProtocol, Session):
from evennia.utils.utils import delay from evennia.utils.utils import delay
# timeout the handshakes in case the client doesn't reply at all # 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 # TCP/IP keepalive watches for dead links
self.transport.setTcpKeepAlive(1) self.transport.setTcpKeepAlive(1)

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

@ -8,13 +8,15 @@ try:
except ImportError: except ImportError:
import unittest import unittest
from mock import Mock
import string import string
from evennia.server.portal import irc from evennia.server.portal import irc
from twisted.test import proto_helpers from twisted.test import proto_helpers
from twisted.trial.unittest import TestCase as TwistedTestCase 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): class TestIRC(TestCase):
@ -84,9 +86,19 @@ class TestTelnet(TwistedTestCase):
def setUp(self): def setUp(self):
super(TestTelnet, self).setUp() super(TestTelnet, self).setUp()
factory = TelnetServerFactory() factory = TelnetServerFactory()
factory.protocol = TelnetProtocol
factory.sessionhandler = PORTAL_SESSIONS
factory.sessionhandler.portal = Mock()
self.proto = factory.buildProtocol(("localhost", 0)) self.proto = factory.buildProtocol(("localhost", 0))
self.transport = proto_helpers.StringTransport() self.transport = proto_helpers.StringTransport()
self.addCleanup(factory.sessionhandler.disconnect_all)
def test_connect(self): 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 # 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