| Index: third_party/twisted_8_1/twisted/conch/test/test_manhole.py
|
| diff --git a/third_party/twisted_8_1/twisted/conch/test/test_manhole.py b/third_party/twisted_8_1/twisted/conch/test/test_manhole.py
|
| deleted file mode 100644
|
| index 9d70d2d4ef1e00c9580dd7998d27da65802ba537..0000000000000000000000000000000000000000
|
| --- a/third_party/twisted_8_1/twisted/conch/test/test_manhole.py
|
| +++ /dev/null
|
| @@ -1,345 +0,0 @@
|
| -# -*- test-case-name: twisted.conch.test.test_manhole -*-
|
| -# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
|
| -# See LICENSE for details.
|
| -
|
| -"""
|
| -Tests for L{twisted.conch.manhole}.
|
| -"""
|
| -
|
| -import traceback
|
| -
|
| -from twisted.trial import unittest
|
| -from twisted.internet import error, defer
|
| -from twisted.test.proto_helpers import StringTransport
|
| -from twisted.conch.test.test_recvline import _TelnetMixin, _SSHMixin, _StdioMixin, stdio, ssh
|
| -from twisted.conch import manhole
|
| -from twisted.conch.insults import insults
|
| -
|
| -
|
| -def determineDefaultFunctionName():
|
| - """
|
| - Return the string used by Python as the name for code objects which are
|
| - compiled from interactive input or at the top-level of modules.
|
| - """
|
| - try:
|
| - 1 / 0
|
| - except:
|
| - return traceback.extract_stack()[0][2]
|
| -defaultFunctionName = determineDefaultFunctionName()
|
| -
|
| -
|
| -
|
| -class ManholeInterpreterTests(unittest.TestCase):
|
| - """
|
| - Tests for L{manhole.ManholeInterpreter}.
|
| - """
|
| - def test_resetBuffer(self):
|
| - """
|
| - L{ManholeInterpreter.resetBuffer} should empty the input buffer.
|
| - """
|
| - interpreter = manhole.ManholeInterpreter(None)
|
| - interpreter.buffer.extend(["1", "2"])
|
| - interpreter.resetBuffer()
|
| - self.assertFalse(interpreter.buffer)
|
| -
|
| -
|
| -
|
| -class ManholeProtocolTests(unittest.TestCase):
|
| - """
|
| - Tests for L{manhole.Manhole}.
|
| - """
|
| - def test_interruptResetsInterpreterBuffer(self):
|
| - """
|
| - L{manhole.Manhole.handle_INT} should cause the interpreter input buffer
|
| - to be reset.
|
| - """
|
| - transport = StringTransport()
|
| - terminal = insults.ServerProtocol(manhole.Manhole)
|
| - terminal.makeConnection(transport)
|
| - protocol = terminal.terminalProtocol
|
| - interpreter = protocol.interpreter
|
| - interpreter.buffer.extend(["1", "2"])
|
| - protocol.handle_INT()
|
| - self.assertFalse(interpreter.buffer)
|
| -
|
| -
|
| -
|
| -class WriterTestCase(unittest.TestCase):
|
| - def testInteger(self):
|
| - manhole.lastColorizedLine("1")
|
| -
|
| -
|
| - def testDoubleQuoteString(self):
|
| - manhole.lastColorizedLine('"1"')
|
| -
|
| -
|
| - def testSingleQuoteString(self):
|
| - manhole.lastColorizedLine("'1'")
|
| -
|
| -
|
| - def testTripleSingleQuotedString(self):
|
| - manhole.lastColorizedLine("'''1'''")
|
| -
|
| -
|
| - def testTripleDoubleQuotedString(self):
|
| - manhole.lastColorizedLine('"""1"""')
|
| -
|
| -
|
| - def testFunctionDefinition(self):
|
| - manhole.lastColorizedLine("def foo():")
|
| -
|
| -
|
| - def testClassDefinition(self):
|
| - manhole.lastColorizedLine("class foo:")
|
| -
|
| -
|
| -class ManholeLoopbackMixin:
|
| - serverProtocol = manhole.ColoredManhole
|
| -
|
| - def wfd(self, d):
|
| - return defer.waitForDeferred(d)
|
| -
|
| - def testSimpleExpression(self):
|
| - done = self.recvlineClient.expect("done")
|
| -
|
| - self._testwrite(
|
| - "1 + 1\n"
|
| - "done")
|
| -
|
| - def finished(ign):
|
| - self._assertBuffer(
|
| - [">>> 1 + 1",
|
| - "2",
|
| - ">>> done"])
|
| -
|
| - return done.addCallback(finished)
|
| -
|
| - def testTripleQuoteLineContinuation(self):
|
| - done = self.recvlineClient.expect("done")
|
| -
|
| - self._testwrite(
|
| - "'''\n'''\n"
|
| - "done")
|
| -
|
| - def finished(ign):
|
| - self._assertBuffer(
|
| - [">>> '''",
|
| - "... '''",
|
| - "'\\n'",
|
| - ">>> done"])
|
| -
|
| - return done.addCallback(finished)
|
| -
|
| - def testFunctionDefinition(self):
|
| - done = self.recvlineClient.expect("done")
|
| -
|
| - self._testwrite(
|
| - "def foo(bar):\n"
|
| - "\tprint bar\n\n"
|
| - "foo(42)\n"
|
| - "done")
|
| -
|
| - def finished(ign):
|
| - self._assertBuffer(
|
| - [">>> def foo(bar):",
|
| - "... print bar",
|
| - "... ",
|
| - ">>> foo(42)",
|
| - "42",
|
| - ">>> done"])
|
| -
|
| - return done.addCallback(finished)
|
| -
|
| - def testClassDefinition(self):
|
| - done = self.recvlineClient.expect("done")
|
| -
|
| - self._testwrite(
|
| - "class Foo:\n"
|
| - "\tdef bar(self):\n"
|
| - "\t\tprint 'Hello, world!'\n\n"
|
| - "Foo().bar()\n"
|
| - "done")
|
| -
|
| - def finished(ign):
|
| - self._assertBuffer(
|
| - [">>> class Foo:",
|
| - "... def bar(self):",
|
| - "... print 'Hello, world!'",
|
| - "... ",
|
| - ">>> Foo().bar()",
|
| - "Hello, world!",
|
| - ">>> done"])
|
| -
|
| - return done.addCallback(finished)
|
| -
|
| - def testException(self):
|
| - done = self.recvlineClient.expect("done")
|
| -
|
| - self._testwrite(
|
| - "1 / 0\n"
|
| - "done")
|
| -
|
| - def finished(ign):
|
| - self._assertBuffer(
|
| - [">>> 1 / 0",
|
| - "Traceback (most recent call last):",
|
| - ' File "<console>", line 1, in ' + defaultFunctionName,
|
| - "ZeroDivisionError: integer division or modulo by zero",
|
| - ">>> done"])
|
| -
|
| - return done.addCallback(finished)
|
| -
|
| - def testControlC(self):
|
| - done = self.recvlineClient.expect("done")
|
| -
|
| - self._testwrite(
|
| - "cancelled line" + manhole.CTRL_C +
|
| - "done")
|
| -
|
| - def finished(ign):
|
| - self._assertBuffer(
|
| - [">>> cancelled line",
|
| - "KeyboardInterrupt",
|
| - ">>> done"])
|
| -
|
| - return done.addCallback(finished)
|
| -
|
| -
|
| - def test_interruptDuringContinuation(self):
|
| - """
|
| - Sending ^C to Manhole while in a state where more input is required to
|
| - complete a statement should discard the entire ongoing statement and
|
| - reset the input prompt to the non-continuation prompt.
|
| - """
|
| - continuing = self.recvlineClient.expect("things")
|
| -
|
| - self._testwrite("(\nthings")
|
| -
|
| - def gotContinuation(ignored):
|
| - self._assertBuffer(
|
| - [">>> (",
|
| - "... things"])
|
| - interrupted = self.recvlineClient.expect(">>> ")
|
| - self._testwrite(manhole.CTRL_C)
|
| - return interrupted
|
| - continuing.addCallback(gotContinuation)
|
| -
|
| - def gotInterruption(ignored):
|
| - self._assertBuffer(
|
| - [">>> (",
|
| - "... things",
|
| - "KeyboardInterrupt",
|
| - ">>> "])
|
| - continuing.addCallback(gotInterruption)
|
| - return continuing
|
| -
|
| -
|
| - def testControlBackslash(self):
|
| - self._testwrite("cancelled line")
|
| - partialLine = self.recvlineClient.expect("cancelled line")
|
| -
|
| - def gotPartialLine(ign):
|
| - self._assertBuffer(
|
| - [">>> cancelled line"])
|
| - self._testwrite(manhole.CTRL_BACKSLASH)
|
| -
|
| - d = self.recvlineClient.onDisconnection
|
| - return self.assertFailure(d, error.ConnectionDone)
|
| -
|
| - def gotClearedLine(ign):
|
| - self._assertBuffer(
|
| - [""])
|
| -
|
| - return partialLine.addCallback(gotPartialLine).addCallback(gotClearedLine)
|
| -
|
| - def testControlD(self):
|
| - self._testwrite("1 + 1")
|
| - helloWorld = self.wfd(self.recvlineClient.expect(r"\+ 1"))
|
| - yield helloWorld
|
| - helloWorld.getResult()
|
| - self._assertBuffer([">>> 1 + 1"])
|
| -
|
| - self._testwrite(manhole.CTRL_D + " + 1")
|
| - cleared = self.wfd(self.recvlineClient.expect(r"\+ 1"))
|
| - yield cleared
|
| - cleared.getResult()
|
| - self._assertBuffer([">>> 1 + 1 + 1"])
|
| -
|
| - self._testwrite("\n")
|
| - printed = self.wfd(self.recvlineClient.expect("3\n>>> "))
|
| - yield printed
|
| - printed.getResult()
|
| -
|
| - self._testwrite(manhole.CTRL_D)
|
| - d = self.recvlineClient.onDisconnection
|
| - disconnected = self.wfd(self.assertFailure(d, error.ConnectionDone))
|
| - yield disconnected
|
| - disconnected.getResult()
|
| - testControlD = defer.deferredGenerator(testControlD)
|
| -
|
| -
|
| - def testControlL(self):
|
| - """
|
| - CTRL+L is generally used as a redraw-screen command in terminal
|
| - applications. Manhole doesn't currently respect this usage of it,
|
| - but it should at least do something reasonable in response to this
|
| - event (rather than, say, eating your face).
|
| - """
|
| - # Start off with a newline so that when we clear the display we can
|
| - # tell by looking for the missing first empty prompt line.
|
| - self._testwrite("\n1 + 1")
|
| - helloWorld = self.wfd(self.recvlineClient.expect(r"\+ 1"))
|
| - yield helloWorld
|
| - helloWorld.getResult()
|
| - self._assertBuffer([">>> ", ">>> 1 + 1"])
|
| -
|
| - self._testwrite(manhole.CTRL_L + " + 1")
|
| - redrew = self.wfd(self.recvlineClient.expect(r"1 \+ 1 \+ 1"))
|
| - yield redrew
|
| - redrew.getResult()
|
| - self._assertBuffer([">>> 1 + 1 + 1"])
|
| - testControlL = defer.deferredGenerator(testControlL)
|
| -
|
| -
|
| - def testDeferred(self):
|
| - self._testwrite(
|
| - "from twisted.internet import defer, reactor\n"
|
| - "d = defer.Deferred()\n"
|
| - "d\n")
|
| -
|
| - deferred = self.wfd(self.recvlineClient.expect("<Deferred #0>"))
|
| - yield deferred
|
| - deferred.getResult()
|
| -
|
| - self._testwrite(
|
| - "c = reactor.callLater(0.1, d.callback, 'Hi!')\n")
|
| - delayed = self.wfd(self.recvlineClient.expect(">>> "))
|
| - yield delayed
|
| - delayed.getResult()
|
| -
|
| - called = self.wfd(self.recvlineClient.expect("Deferred #0 called back: 'Hi!'\n>>> "))
|
| - yield called
|
| - called.getResult()
|
| - self._assertBuffer(
|
| - [">>> from twisted.internet import defer, reactor",
|
| - ">>> d = defer.Deferred()",
|
| - ">>> d",
|
| - "<Deferred #0>",
|
| - ">>> c = reactor.callLater(0.1, d.callback, 'Hi!')",
|
| - "Deferred #0 called back: 'Hi!'",
|
| - ">>> "])
|
| -
|
| - testDeferred = defer.deferredGenerator(testDeferred)
|
| -
|
| -class ManholeLoopbackTelnet(_TelnetMixin, unittest.TestCase, ManholeLoopbackMixin):
|
| - pass
|
| -
|
| -class ManholeLoopbackSSH(_SSHMixin, unittest.TestCase, ManholeLoopbackMixin):
|
| - if ssh is None:
|
| - skip = "Crypto requirements missing, can't run manhole tests over ssh"
|
| -
|
| -class ManholeLoopbackStdio(_StdioMixin, unittest.TestCase, ManholeLoopbackMixin):
|
| - if stdio is None:
|
| - skip = "Terminal requirements missing, can't run manhole tests over stdio"
|
| - else:
|
| - serverProtocol = stdio.ConsoleManhole
|
|
|