third_party/twisted_8_1/twisted/conch/test/test_telnet.py - Issue 12261012: Remove third_party/twisted_8_1

Unified Diff: third_party/twisted_8_1/twisted/conch/test/test_telnet.py

Issue 12261012: Remove third_party/twisted_8_1 (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/tools/build

Patch Set: Created 7 years, 10 months ago

Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.

Jump to:

View side-by-side diff with in-line comments

« no previous file with comments | « third_party/twisted_8_1/twisted/conch/test/test_ssh.py ('k') | third_party/twisted_8_1/twisted/conch/test/test_text.py » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Hide Comments ('s')

Index: third_party/twisted_8_1/twisted/conch/test/test_telnet.py

diff --git a/third_party/twisted_8_1/twisted/conch/test/test_telnet.py b/third_party/twisted_8_1/twisted/conch/test/test_telnet.py

deleted file mode 100644

index a2fc4e77e84686dca010ccd101228b85aab0ebc2..0000000000000000000000000000000000000000

--- a/third_party/twisted_8_1/twisted/conch/test/test_telnet.py

+++ /dev/null

@@ -1,672 +0,0 @@

-# -*- test-case-name: twisted.conch.test.test_telnet -*-

-# See LICENSE for details.

-"""

-Tests for L{twisted.conch.telnet}.

-"""

-from zope.interface import implements

-from twisted.internet import defer

-from twisted.conch import telnet

-from twisted.trial import unittest

-from twisted.test import proto_helpers

-class TestProtocol:

- implements(telnet.ITelnetProtocol)

- localEnableable = ()

- remoteEnableable = ()

- def __init__(self):

- self.bytes = ''

- self.subcmd = ''

- self.calls = []

- self.enabledLocal = []

- self.enabledRemote = []

- self.disabledLocal = []

- self.disabledRemote = []

- def makeConnection(self, transport):

- d = transport.negotiationMap = {}

- d['\x12'] = self.neg_TEST_COMMAND

- d = transport.commandMap = transport.commandMap.copy()

- for cmd in ('NOP', 'DM', 'BRK', 'IP', 'AO', 'AYT', 'EC', 'EL', 'GA'):

- d[getattr(telnet, cmd)] = lambda arg, cmd=cmd: self.calls.append(cmd)

- def dataReceived(self, bytes):

- self.bytes += bytes

- def connectionLost(self, reason):

- pass

- def neg_TEST_COMMAND(self, payload):

- self.subcmd = payload

- def enableLocal(self, option):

- if option in self.localEnableable:

- self.enabledLocal.append(option)

- return True

- return False

- def disableLocal(self, option):

- self.disabledLocal.append(option)

- def enableRemote(self, option):

- if option in self.remoteEnableable:

- self.enabledRemote.append(option)

- return True

- return False

- def disableRemote(self, option):

- self.disabledRemote.append(option)

-class TelnetTransportTestCase(unittest.TestCase):

- """

- Tests for L{telnet.TelnetTransport}.

- """

- def setUp(self):

- self.p = telnet.TelnetTransport(TestProtocol)

- self.t = proto_helpers.StringTransport()

- self.p.makeConnection(self.t)

- def testRegularBytes(self):

- # Just send a bunch of bytes. None of these do anything

- # with telnet. They should pass right through to the

- # application layer.

- h = self.p.protocol

- L = ["here are some bytes la la la",

- "some more arrive here",

- "lots of bytes to play with",

- "la la la",

- "ta de da",

- "dum"]

- for b in L:

- self.p.dataReceived(b)

- self.assertEquals(h.bytes, ''.join(L))

- def testNewlineHandling(self):

- # Send various kinds of newlines and make sure they get translated

- # into \n.

- h = self.p.protocol

- L = ["here is the first line\r\n",

- "here is the second line\r\0",

- "here is the third line\r\n",

- "here is the last line\r\0"]

- for b in L:

- self.p.dataReceived(b)

- self.assertEquals(h.bytes, L[0][:-2] + '\n' +

- L[1][:-2] + '\r' +

- L[2][:-2] + '\n' +

- L[3][:-2] + '\r')

- def testIACEscape(self):

- # Send a bunch of bytes and a couple quoted \xFFs. Unquoted,

- # \xFF is a telnet command. Quoted, one of them from each pair

- # should be passed through to the application layer.

- h = self.p.protocol

- L = ["here are some bytes\xff\xff with an embedded IAC",

- "and here is a test of a border escape\xff",

- "\xff did you get that IAC?"]

- for b in L:

- self.p.dataReceived(b)

- self.assertEquals(h.bytes, ''.join(L).replace('\xff\xff', '\xff'))

- def _simpleCommandTest(self, cmdName):

- # Send a single simple telnet command and make sure

- # it gets noticed and the appropriate method gets

- # called.

- h = self.p.protocol

- cmd = telnet.IAC + getattr(telnet, cmdName)

- L = ["Here's some bytes, tra la la",

- "But ono!" + cmd + " an interrupt"]

- for b in L:

- self.p.dataReceived(b)

- self.assertEquals(h.calls, [cmdName])

- self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))

- def testInterrupt(self):

- self._simpleCommandTest("IP")

- def testNoOperation(self):

- self._simpleCommandTest("NOP")

- def testDataMark(self):

- self._simpleCommandTest("DM")

- def testBreak(self):

- self._simpleCommandTest("BRK")

- def testAbortOutput(self):

- self._simpleCommandTest("AO")

- def testAreYouThere(self):

- self._simpleCommandTest("AYT")

- def testEraseCharacter(self):

- self._simpleCommandTest("EC")

- def testEraseLine(self):

- self._simpleCommandTest("EL")

- def testGoAhead(self):

- self._simpleCommandTest("GA")

- def testSubnegotiation(self):

- # Send a subnegotiation command and make sure it gets

- # parsed and that the correct method is called.

- h = self.p.protocol

- cmd = telnet.IAC + telnet.SB + '\x12hello world' + telnet.IAC + telnet.SE

- L = ["These are some bytes but soon" + cmd,

- "there will be some more"]

- for b in L:

- self.p.dataReceived(b)

- self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))

- self.assertEquals(h.subcmd, list("hello world"))

- def testSubnegotiationWithEmbeddedSE(self):

- # Send a subnegotiation command with an embedded SE. Make sure

- # that SE gets passed to the correct method.

- h = self.p.protocol

- cmd = (telnet.IAC + telnet.SB +

- '\x12' + telnet.SE +

- telnet.IAC + telnet.SE)

- L = ["Some bytes are here" + cmd + "and here",

- "and here"]

- for b in L:

- self.p.dataReceived(b)

- self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))

- self.assertEquals(h.subcmd, [telnet.SE])

- def testBoundarySubnegotiation(self):

- # Send a subnegotiation command. Split it at every possible byte boundary

- # and make sure it always gets parsed and that it is passed to the correct

- # method.

- cmd = (telnet.IAC + telnet.SB +

- '\x12' + telnet.SE + 'hello' +

- telnet.IAC + telnet.SE)

- for i in range(len(cmd)):

- h = self.p.protocol = TestProtocol()

- h.makeConnection(self.p)

- a, b = cmd[:i], cmd[i:]

- L = ["first part" + a,

- b + "last part"]

- for bytes in L:

- self.p.dataReceived(bytes)

- self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))

- self.assertEquals(h.subcmd, [telnet.SE] + list('hello'))

- def _enabledHelper(self, o, eL=[], eR=[], dL=[], dR=[]):

- self.assertEquals(o.enabledLocal, eL)

- self.assertEquals(o.enabledRemote, eR)

- self.assertEquals(o.disabledLocal, dL)

- self.assertEquals(o.disabledRemote, dR)

- def testRefuseWill(self):

- # Try to enable an option. The server should refuse to enable it.

- cmd = telnet.IAC + telnet.WILL + '\x12'

- bytes = "surrounding bytes" + cmd + "to spice things up"

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x12')

- self._enabledHelper(self.p.protocol)

- def testRefuseDo(self):

- # Try to enable an option. The server should refuse to enable it.

- cmd = telnet.IAC + telnet.DO + '\x12'

- bytes = "surrounding bytes" + cmd + "to spice things up"

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), telnet.IAC + telnet.WONT + '\x12')

- self._enabledHelper(self.p.protocol)

- def testAcceptDo(self):

- # Try to enable an option. The option is in our allowEnable

- # list, so we will allow it to be enabled.

- cmd = telnet.IAC + telnet.DO + '\x19'

- bytes = 'padding' + cmd + 'trailer'

- h = self.p.protocol

- h.localEnableable = ('\x19',)

- self.p.dataReceived(bytes)

- self.assertEquals(self.t.value(), telnet.IAC + telnet.WILL + '\x19')

- self._enabledHelper(h, eL=['\x19'])

- def testAcceptWill(self):

- # Same as testAcceptDo, but reversed.

- cmd = telnet.IAC + telnet.WILL + '\x91'

- bytes = 'header' + cmd + 'padding'

- h = self.p.protocol

- h.remoteEnableable = ('\x91',)

- self.p.dataReceived(bytes)

- self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x91')

- self._enabledHelper(h, eR=['\x91'])

- def testAcceptWont(self):

- # Try to disable an option. The server must allow any option to

- # be disabled at any time. Make sure it disables it and sends

- # back an acknowledgement of this.

- cmd = telnet.IAC + telnet.WONT + '\x29'

- # Jimmy it - after these two lines, the server will be in a state

- # such that it believes the option to have been previously enabled

- # via normal negotiation.

- s = self.p.getOptionState('\x29')

- s.him.state = 'yes'

- bytes = "fiddle dee" + cmd

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x29')

- self.assertEquals(s.him.state, 'no')

- self._enabledHelper(self.p.protocol, dR=['\x29'])

- def testAcceptDont(self):

- # Try to disable an option. The server must allow any option to

- # be disabled at any time. Make sure it disables it and sends

- # back an acknowledgement of this.

- cmd = telnet.IAC + telnet.DONT + '\x29'

- # Jimmy it - after these two lines, the server will be in a state

- # such that it believes the option to have beenp previously enabled

- # via normal negotiation.

- s = self.p.getOptionState('\x29')

- s.us.state = 'yes'

- bytes = "fiddle dum " + cmd

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), telnet.IAC + telnet.WONT + '\x29')

- self.assertEquals(s.us.state, 'no')

- self._enabledHelper(self.p.protocol, dL=['\x29'])

- def testIgnoreWont(self):

- # Try to disable an option. The option is already disabled. The

- # server should send nothing in response to this.

- cmd = telnet.IAC + telnet.WONT + '\x47'

- bytes = "dum de dum" + cmd + "tra la la"

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), '')

- self._enabledHelper(self.p.protocol)

- def testIgnoreDont(self):

- # Try to disable an option. The option is already disabled. The

- # server should send nothing in response to this. Doing so could

- # lead to a negotiation loop.

- cmd = telnet.IAC + telnet.DONT + '\x47'

- bytes = "dum de dum" + cmd + "tra la la"

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), '')

- self._enabledHelper(self.p.protocol)

- def testIgnoreWill(self):

- # Try to enable an option. The option is already enabled. The

- # server should send nothing in response to this. Doing so could

- # lead to a negotiation loop.

- cmd = telnet.IAC + telnet.WILL + '\x56'

- # Jimmy it - after these two lines, the server will be in a state

- # such that it believes the option to have been previously enabled

- # via normal negotiation.

- s = self.p.getOptionState('\x56')

- s.him.state = 'yes'

- bytes = "tra la la" + cmd + "dum de dum"

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), '')

- self._enabledHelper(self.p.protocol)

- def testIgnoreDo(self):

- # Try to enable an option. The option is already enabled. The

- # server should send nothing in response to this. Doing so could

- # lead to a negotiation loop.

- cmd = telnet.IAC + telnet.DO + '\x56'

- # Jimmy it - after these two lines, the server will be in a state

- # such that it believes the option to have been previously enabled

- # via normal negotiation.

- s = self.p.getOptionState('\x56')

- s.us.state = 'yes'

- bytes = "tra la la" + cmd + "dum de dum"

- self.p.dataReceived(bytes)

- self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))

- self.assertEquals(self.t.value(), '')

- self._enabledHelper(self.p.protocol)

- def testAcceptedEnableRequest(self):

- # Try to enable an option through the user-level API. This

- # returns a Deferred that fires when negotiation about the option

- # finishes. Make sure it fires, make sure state gets updated

- # properly, make sure the result indicates the option was enabled.

- d = self.p.do('\x42')

- h = self.p.protocol

- h.remoteEnableable = ('\x42',)

- self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x42')

- self.p.dataReceived(telnet.IAC + telnet.WILL + '\x42')

- d.addCallback(self.assertEquals, True)

- d.addCallback(lambda _: self._enabledHelper(h, eR=['\x42']))

- return d

- def testRefusedEnableRequest(self):

- # Try to enable an option through the user-level API. This

- # returns a Deferred that fires when negotiation about the option

- # finishes. Make sure it fires, make sure state gets updated

- # properly, make sure the result indicates the option was enabled.

- d = self.p.do('\x42')

- self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x42')

- self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')

- d = self.assertFailure(d, telnet.OptionRefused)

- d.addCallback(lambda _: self._enabledHelper(self.p.protocol))

- return d

- def testAcceptedDisableRequest(self):

- # Try to disable an option through the user-level API. This

- # returns a Deferred that fires when negotiation about the option

- # finishes. Make sure it fires, make sure state gets updated

- # properly, make sure the result indicates the option was enabled.

- s = self.p.getOptionState('\x42')

- s.him.state = 'yes'

- d = self.p.dont('\x42')

- self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x42')

- self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')

- d.addCallback(self.assertEquals, True)

- d.addCallback(lambda _: self._enabledHelper(self.p.protocol,

- dR=['\x42']))

- return d

- def testNegotiationBlocksFurtherNegotiation(self):

- # Try to disable an option, then immediately try to enable it, then

- # immediately try to disable it. Ensure that the 2nd and 3rd calls

- # fail quickly with the right exception.

- s = self.p.getOptionState('\x24')

- s.him.state = 'yes'

- d2 = self.p.dont('\x24') # fires after the first line of _final

- def _do(x):

- d = self.p.do('\x24')

- return self.assertFailure(d, telnet.AlreadyNegotiating)

- def _dont(x):

- d = self.p.dont('\x24')

- return self.assertFailure(d, telnet.AlreadyNegotiating)

- def _final(x):

- self.p.dataReceived(telnet.IAC + telnet.WONT + '\x24')

- # an assertion that only passes if d2 has fired

- self._enabledHelper(self.p.protocol, dR=['\x24'])

- # Make sure we allow this

- self.p.protocol.remoteEnableable = ('\x24',)

- d = self.p.do('\x24')

- self.p.dataReceived(telnet.IAC + telnet.WILL + '\x24')

- d.addCallback(self.assertEquals, True)

- d.addCallback(lambda _: self._enabledHelper(self.p.protocol,

- eR=['\x24'],

- dR=['\x24']))

- return d

- d = _do(None)

- d.addCallback(_dont)

- d.addCallback(_final)

- return d

- def testSuperfluousDisableRequestRaises(self):

- # Try to disable a disabled option. Make sure it fails properly.

- d = self.p.dont('\xab')

- return self.assertFailure(d, telnet.AlreadyDisabled)

- def testSuperfluousEnableRequestRaises(self):

- # Try to disable a disabled option. Make sure it fails properly.

- s = self.p.getOptionState('\xab')

- s.him.state = 'yes'

- d = self.p.do('\xab')

- return self.assertFailure(d, telnet.AlreadyEnabled)

- def testLostConnectionFailsDeferreds(self):

- d1 = self.p.do('\x12')

- d2 = self.p.do('\x23')

- d3 = self.p.do('\x34')

- class TestException(Exception):

- pass

- self.p.connectionLost(TestException("Total failure!"))

- d1 = self.assertFailure(d1, TestException)

- d2 = self.assertFailure(d2, TestException)

- d3 = self.assertFailure(d3, TestException)

- return defer.gatherResults([d1, d2, d3])

-class TestTelnet(telnet.Telnet):

- """

- A trivial extension of the telnet protocol class useful to unit tests.

- """

- def __init__(self):

- telnet.Telnet.__init__(self)

- self.events = []

- def applicationDataReceived(self, bytes):

- """

- Record the given data in C{self.events}.

- """

- self.events.append(('bytes', bytes))

- def unhandledCommand(self, command, bytes):

- """

- Record the given command in C{self.events}.

- """

- self.events.append(('command', command, bytes))

- def unhandledSubnegotiation(self, command, bytes):

- """

- Record the given subnegotiation command in C{self.events}.

- """

- self.events.append(('negotiate', command, bytes))

-class TelnetTests(unittest.TestCase):

- """

- Tests for L{telnet.Telnet}.

- L{telnet.Telnet} implements the TELNET protocol (RFC 854), including option

- and suboption negotiation, and option state tracking.

- """

- def setUp(self):

- """

- Create an unconnected L{telnet.Telnet} to be used by tests.

- """

- self.protocol = TestTelnet()

- def test_enableLocal(self):

- """

- L{telnet.Telnet.enableLocal} should reject all options, since

- L{telnet.Telnet} does not know how to implement any options.

- """

- self.assertFalse(self.protocol.enableLocal('\0'))

- def test_enableRemote(self):

- """

- L{telnet.Telnet.enableRemote} should reject all options, since

- L{telnet.Telnet} does not know how to implement any options.

- """

- self.assertFalse(self.protocol.enableRemote('\0'))

- def test_disableLocal(self):

- """

- It is an error for L{telnet.Telnet.disableLocal} to be called, since

- L{telnet.Telnet.enableLocal} will never allow any options to be enabled

- locally. If a subclass overrides enableLocal, it must also override

- disableLocal.

- """

- self.assertRaises(NotImplementedError, self.protocol.disableLocal, '\0')

- def test_disableRemote(self):

- """

- It is an error for L{telnet.Telnet.disableRemote} to be called, since

- L{telnet.Telnet.enableRemote} will never allow any options to be

- enabled remotely. If a subclass overrides enableRemote, it must also

- override disableRemote.

- """

- self.assertRaises(NotImplementedError, self.protocol.disableRemote, '\0')

- def _deliver(self, bytes, *expected):

- """

- Pass the given bytes to the protocol's C{dataReceived} method and

- assert that the given events occur.

- """

- received = self.protocol.events = []

- self.protocol.dataReceived(bytes)

- self.assertEqual(received, list(expected))

- def test_oneApplicationDataByte(self):

- """

- One application-data byte in the default state gets delivered right

- away.

- """

- self._deliver('a', ('bytes', 'a'))

- def test_twoApplicationDataBytes(self):

- """

- Two application-data bytes in the default state get delivered

- together.

- """

- self._deliver('bc', ('bytes', 'bc'))

- def test_threeApplicationDataBytes(self):

- """

- Three application-data bytes followed by a control byte get

- delivered, but the control byte doesn't.

- """

- self._deliver('def' + telnet.IAC, ('bytes', 'def'))

- def test_escapedControl(self):

- """

- IAC in the escaped state gets delivered and so does another

- application-data byte following it.

- """

- self._deliver(telnet.IAC)

- self._deliver(telnet.IAC + 'g', ('bytes', telnet.IAC + 'g'))

- def test_carriageReturn(self):

- """

- A carriage return only puts the protocol into the newline state. A

- linefeed in the newline state causes just the newline to be

- delivered. A nul in the newline state causes a carriage return to

- be delivered. Anything else causes a carriage return and that thing

- to be delivered.

- """

- self._deliver('\r')

- self._deliver('\n', ('bytes', '\n'))

- self._deliver('\r\n', ('bytes', '\n'))

- self._deliver('\r')

- self._deliver('\0', ('bytes', '\r'))

- self._deliver('\r\0', ('bytes', '\r'))

- self._deliver('\r')

- self._deliver('a', ('bytes', '\ra'))

- self._deliver('\ra', ('bytes', '\ra'))

- def test_applicationDataBeforeSimpleCommand(self):

- """

- Application bytes received before a command are delivered before the

- command is processed.

- """

- self._deliver(

- 'x' + telnet.IAC + telnet.NOP,

- ('bytes', 'x'), ('command', telnet.NOP, None))

- def test_applicationDataBeforeCommand(self):

- """

- Application bytes received before a WILL/WONT/DO/DONT are delivered

- before the command is processed.

- """

- self.protocol.commandMap = {}

- self._deliver(

- 'y' + telnet.IAC + telnet.WILL + '\x00',

- ('bytes', 'y'), ('command', telnet.WILL, '\x00'))

- def test_applicationDataBeforeSubnegotiation(self):

- """

- Application bytes received before a subnegotiation command are

- delivered before the negotiation is processed.

- """

- self._deliver(

- 'z' + telnet.IAC + telnet.SB + 'Qx' + telnet.IAC + telnet.SE,

- ('bytes', 'z'), ('negotiate', 'Q', ['x']))

« no previous file with comments | « third_party/twisted_8_1/twisted/conch/test/test_ssh.py ('k') | third_party/twisted_8_1/twisted/conch/test/test_text.py » ('j') | no next file with comments »