<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> from twisted . conch . insults import insults <EOL> from twisted . conch import recvline <EOL> from twisted . python import reflect , components , filepath <EOL> from twisted . internet import defer , error <EOL> from twisted . trial import unittest <EOL> from twisted . cred import portal <EOL> from twisted . test . proto_helpers import StringTransport <EOL> class ArrowsTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . underlyingTransport = StringTransport ( ) <EOL> self . pt = insults . ServerProtocol ( ) <EOL> self . p = recvline . HistoricRecvLine ( ) <EOL> self . pt . protocolFactory = lambda : self . p <EOL> self . pt . factory = self <EOL> self . pt . makeConnection ( self . underlyingTransport ) <EOL> def test_printableCharacters ( self ) : <EOL> """<STR_LIT>""" <EOL> self . p . keystrokeReceived ( '<STR_LIT:x>' , None ) <EOL> self . p . keystrokeReceived ( '<STR_LIT:y>' , None ) <EOL> self . p . keystrokeReceived ( '<STR_LIT:z>' , None ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_horizontalArrows ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . RIGHT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT:z>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT:x>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . RIGHT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT:x>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . RIGHT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT:z>' ) ) <EOL> kR ( self . pt . RIGHT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . RIGHT_ARROW ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_newline ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' ) , ( ) ) ) <EOL> kR ( '<STR_LIT:c>' ) <EOL> kR ( '<STR_LIT:b>' ) <EOL> kR ( '<STR_LIT:a>' ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' ) , ( ) ) ) <EOL> kR ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' , '<STR_LIT>' ) , ( ) ) ) <EOL> def test_verticalArrows ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' ) , ( ) ) ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . UP_ARROW ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT:abc>' ) , ( '<STR_LIT>' , ) ) ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . UP_ARROW ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( '<STR_LIT>' , ) , ( '<STR_LIT:abc>' , '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT:abc>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . UP_ARROW ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( ) , ( '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . UP_ARROW ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( ) , ( '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> for i in range ( <NUM_LIT:4> ) : <EOL> kR ( self . pt . DOWN_ARROW ) <EOL> self . assertEqual ( self . p . currentHistoryBuffer ( ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' ) , ( ) ) ) <EOL> def test_home ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . HOME ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_end ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . HOME ) <EOL> kR ( self . pt . END ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_backspace ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . BACKSPACE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( self . pt . BACKSPACE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT:y>' ) ) <EOL> kR ( self . pt . BACKSPACE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT:y>' ) ) <EOL> def test_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . DELETE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( self . pt . DELETE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( self . pt . DELETE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT:x>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( self . pt . DELETE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . DELETE ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_insert ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( '<STR_LIT:A>' ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT:z>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( '<STR_LIT:B>' ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_typeover ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> for ch in '<STR_LIT>' : <EOL> kR ( ch ) <EOL> kR ( self . pt . INSERT ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( '<STR_LIT:A>' ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> kR ( self . pt . LEFT_ARROW ) <EOL> kR ( '<STR_LIT:B>' ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_unprintableCharacters ( self ) : <EOL> """<STR_LIT>""" <EOL> kR = lambda ch : self . p . keystrokeReceived ( ch , None ) <EOL> pt = self . pt <EOL> for ch in ( pt . F1 , pt . F2 , pt . F3 , pt . F4 , pt . F5 , pt . F6 , pt . F7 , pt . F8 , <EOL> pt . F9 , pt . F10 , pt . F11 , pt . F12 , pt . PGUP , pt . PGDN ) : <EOL> kR ( ch ) <EOL> self . assertEqual ( self . p . currentLineBuffer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> from twisted . conch import telnet <EOL> from twisted . conch . insults import helper <EOL> from twisted . protocols import loopback <EOL> class EchoServer ( recvline . HistoricRecvLine ) : <EOL> def lineReceived ( self , line ) : <EOL> self . terminal . write ( line + '<STR_LIT:\n>' + self . ps [ self . pn ] ) <EOL> left = "<STR_LIT>" <EOL> right = "<STR_LIT>" <EOL> up = "<STR_LIT>" <EOL> down = "<STR_LIT>" <EOL> insert = "<STR_LIT>" <EOL> home = "<STR_LIT>" <EOL> delete = "<STR_LIT>" <EOL> end = "<STR_LIT>" <EOL> backspace = "<STR_LIT>" <EOL> from twisted . cred import checkers <EOL> try : <EOL> from twisted . conch . ssh import ( userauth , transport , channel , connection , <EOL> session , keys ) <EOL> from twisted . conch . manhole_ssh import TerminalUser , TerminalSession , TerminalRealm , TerminalSessionTransport , ConchFactory <EOL> except ImportError : <EOL> ssh = False <EOL> else : <EOL> ssh = True <EOL> class SessionChannel ( channel . SSHChannel ) : <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , protocolFactory , protocolArgs , protocolKwArgs , width , height , * a , ** kw ) : <EOL> channel . SSHChannel . __init__ ( self , * a , ** kw ) <EOL> self . protocolFactory = protocolFactory <EOL> self . protocolArgs = protocolArgs <EOL> self . protocolKwArgs = protocolKwArgs <EOL> self . width = width <EOL> self . height = height <EOL> def channelOpen ( self , data ) : <EOL> term = session . packRequest_pty_req ( "<STR_LIT>" , ( self . height , self . width , <NUM_LIT:0> , <NUM_LIT:0> ) , '<STR_LIT>' ) <EOL> self . conn . sendRequest ( self , '<STR_LIT>' , term ) <EOL> self . conn . sendRequest ( self , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _protocolInstance = self . protocolFactory ( * self . protocolArgs , ** self . protocolKwArgs ) <EOL> self . _protocolInstance . factory = self <EOL> self . _protocolInstance . makeConnection ( self ) <EOL> def closed ( self ) : <EOL> self . _protocolInstance . connectionLost ( error . ConnectionDone ( ) ) <EOL> def dataReceived ( self , data ) : <EOL> self . _protocolInstance . dataReceived ( data ) <EOL> class TestConnection ( connection . SSHConnection ) : <EOL> def __init__ ( self , protocolFactory , protocolArgs , protocolKwArgs , width , height , * a , ** kw ) : <EOL> connection . SSHConnection . __init__ ( self , * a , ** kw ) <EOL> self . protocolFactory = protocolFactory <EOL> self . protocolArgs = protocolArgs <EOL> self . protocolKwArgs = protocolKwArgs <EOL> self . width = width <EOL> self . height = height <EOL> def serviceStarted ( self ) : <EOL> self . __channel = SessionChannel ( self . protocolFactory , self . protocolArgs , self . protocolKwArgs , self . width , self . height ) <EOL> self . openChannel ( self . __channel ) <EOL> def write ( self , bytes ) : <EOL> return self . __channel . write ( bytes ) <EOL> class TestAuth ( userauth . SSHUserAuthClient ) : <EOL> def __init__ ( self , username , password , * a , ** kw ) : <EOL> userauth . SSHUserAuthClient . __init__ ( self , username , * a , ** kw ) <EOL> self . password = password <EOL> def getPassword ( self ) : <EOL> return defer . succeed ( self . password ) <EOL> class TestTransport ( transport . SSHClientTransport ) : <EOL> def __init__ ( self , protocolFactory , protocolArgs , protocolKwArgs , username , password , width , height , * a , ** kw ) : <EOL> self . protocolFactory = protocolFactory <EOL> self . protocolArgs = protocolArgs <EOL> self . protocolKwArgs = protocolKwArgs <EOL> self . username = username <EOL> self . password = password <EOL> self . width = width <EOL> self . height = height <EOL> def verifyHostKey ( self , hostKey , fingerprint ) : <EOL> return defer . succeed ( True ) <EOL> def connectionSecure ( self ) : <EOL> self . __connection = TestConnection ( self . protocolFactory , self . protocolArgs , self . protocolKwArgs , self . width , self . height ) <EOL> self . requestService ( <EOL> TestAuth ( self . username , self . password , self . __connection ) ) <EOL> def write ( self , bytes ) : <EOL> return self . __connection . write ( bytes ) <EOL> class TestSessionTransport ( TerminalSessionTransport ) : <EOL> def protocolFactory ( self ) : <EOL> return self . avatar . conn . transport . factory . serverProtocol ( ) <EOL> class TestSession ( TerminalSession ) : <EOL> transportFactory = TestSessionTransport <EOL> class TestUser ( TerminalUser ) : <EOL> pass <EOL> components . registerAdapter ( TestSession , TestUser , session . ISession ) <EOL> class LoopbackRelay ( loopback . LoopbackRelay ) : <EOL> clearCall = None <EOL> def logPrefix ( self ) : <EOL> return "<STR_LIT>" % ( self . target . __class__ . __name__ , ) <EOL> def write ( self , bytes ) : <EOL> loopback . LoopbackRelay . write ( self , bytes ) <EOL> if self . clearCall is not None : <EOL> self . clearCall . cancel ( ) <EOL> from twisted . internet import reactor <EOL> self . clearCall = reactor . callLater ( <NUM_LIT:0> , self . _clearBuffer ) <EOL> def _clearBuffer ( self ) : <EOL> self . clearCall = None <EOL> loopback . LoopbackRelay . clearBuffer ( self ) <EOL> class NotifyingExpectableBuffer ( helper . ExpectableBuffer ) : <EOL> def __init__ ( self ) : <EOL> self . onConnection = defer . Deferred ( ) <EOL> self . onDisconnection = defer . Deferred ( ) <EOL> def connectionMade ( self ) : <EOL> helper . ExpectableBuffer . connectionMade ( self ) <EOL> self . onConnection . callback ( self ) <EOL> def connectionLost ( self , reason ) : <EOL> self . onDisconnection . errback ( reason ) <EOL> class _BaseMixin : <EOL> WIDTH = <NUM_LIT> <EOL> HEIGHT = <NUM_LIT> <EOL> def _assertBuffer ( self , lines ) : <EOL> receivedLines = str ( self . recvlineClient ) . splitlines ( ) <EOL> expectedLines = lines + ( [ '<STR_LIT>' ] * ( self . HEIGHT - len ( lines ) - <NUM_LIT:1> ) ) <EOL> self . assertEqual ( len ( receivedLines ) , len ( expectedLines ) ) <EOL> for i in range ( len ( receivedLines ) ) : <EOL> self . assertEqual ( <EOL> receivedLines [ i ] , expectedLines [ i ] , <EOL> str ( receivedLines [ max ( <NUM_LIT:0> , i - <NUM_LIT:1> ) : i + <NUM_LIT:1> ] ) + <EOL> "<STR_LIT>" + <EOL> str ( expectedLines [ max ( <NUM_LIT:0> , i - <NUM_LIT:1> ) : i + <NUM_LIT:1> ] ) ) <EOL> def _trivialTest ( self , input , output ) : <EOL> done = self . recvlineClient . expect ( "<STR_LIT>" ) <EOL> self . _testwrite ( input ) <EOL> def finished ( ign ) : <EOL> self . _assertBuffer ( output ) <EOL> return done . addCallback ( finished ) <EOL> class _SSHMixin ( _BaseMixin ) : <EOL> def setUp ( self ) : <EOL> if not ssh : <EOL> raise unittest . SkipTest ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> u , p = '<STR_LIT>' , '<STR_LIT>' <EOL> rlm = TerminalRealm ( ) <EOL> rlm . userFactory = TestUser <EOL> rlm . chainedProtocolFactory = lambda : insultsServer <EOL> ptl = portal . Portal ( <EOL> rlm , <EOL> [ checkers . InMemoryUsernamePasswordDatabaseDontUse ( ** { u : p } ) ] ) <EOL> sshFactory = ConchFactory ( ptl ) <EOL> sshKey = keys . _getPersistentRSAKey ( filepath . FilePath ( self . mktemp ( ) ) , <EOL> keySize = <NUM_LIT> ) <EOL> sshFactory . publicKeys [ "<STR_LIT>" ] = sshKey <EOL> sshFactory . privateKeys [ "<STR_LIT>" ] = sshKey <EOL> sshFactory . serverProtocol = self . serverProtocol <EOL> sshFactory . startFactory ( ) <EOL> recvlineServer = self . serverProtocol ( ) <EOL> insultsServer = insults . ServerProtocol ( lambda : recvlineServer ) <EOL> sshServer = sshFactory . buildProtocol ( None ) <EOL> clientTransport = LoopbackRelay ( sshServer ) <EOL> recvlineClient = NotifyingExpectableBuffer ( ) <EOL> insultsClient = insults . ClientProtocol ( lambda : recvlineClient ) <EOL> sshClient = TestTransport ( lambda : insultsClient , ( ) , { } , u , p , self . WIDTH , self . HEIGHT ) <EOL> serverTransport = LoopbackRelay ( sshClient ) <EOL> sshClient . makeConnection ( clientTransport ) <EOL> sshServer . makeConnection ( serverTransport ) <EOL> self . recvlineClient = recvlineClient <EOL> self . sshClient = sshClient <EOL> self . sshServer = sshServer <EOL> self . clientTransport = clientTransport <EOL> self . serverTransport = serverTransport <EOL> return recvlineClient . onConnection <EOL> def _testwrite ( self , bytes ) : <EOL> self . sshClient . write ( bytes ) <EOL> from twisted . conch . test import test_telnet <EOL> class TestInsultsClientProtocol ( insults . ClientProtocol , <EOL> test_telnet . TestProtocol ) : <EOL> pass <EOL> class TestInsultsServerProtocol ( insults . ServerProtocol , <EOL> test_telnet . TestProtocol ) : <EOL> pass <EOL> class _TelnetMixin ( _BaseMixin ) : <EOL> def setUp ( self ) : <EOL> recvlineServer = self . serverProtocol ( ) <EOL> insultsServer = TestInsultsServerProtocol ( lambda : recvlineServer ) <EOL> telnetServer = telnet . TelnetTransport ( lambda : insultsServer ) <EOL> clientTransport = LoopbackRelay ( telnetServer ) <EOL> recvlineClient = NotifyingExpectableBuffer ( ) <EOL> insultsClient = TestInsultsClientProtocol ( lambda : recvlineClient ) <EOL> telnetClient = telnet . TelnetTransport ( lambda : insultsClient ) <EOL> serverTransport = LoopbackRelay ( telnetClient ) <EOL> telnetClient . makeConnection ( clientTransport ) <EOL> telnetServer . makeConnection ( serverTransport ) <EOL> serverTransport . clearBuffer ( ) <EOL> clientTransport . clearBuffer ( ) <EOL> self . recvlineClient = recvlineClient <EOL> self . telnetClient = telnetClient <EOL> self . clientTransport = clientTransport <EOL> self . serverTransport = serverTransport <EOL> return recvlineClient . onConnection <EOL> def _testwrite ( self , bytes ) : <EOL> self . telnetClient . write ( bytes ) <EOL> try : <EOL> from twisted . conch import stdio <EOL> except ImportError : <EOL> stdio = None <EOL> class _StdioMixin ( _BaseMixin ) : <EOL> def setUp ( self ) : <EOL> testTerminal = NotifyingExpectableBuffer ( ) <EOL> insultsClient = insults . ClientProtocol ( lambda : testTerminal ) <EOL> processClient = stdio . TerminalProcessProtocol ( insultsClient ) <EOL> exe = sys . executable <EOL> module = stdio . __file__ <EOL> if module . endswith ( '<STR_LIT>' ) or module . endswith ( '<STR_LIT>' ) : <EOL> module = module [ : - <NUM_LIT:1> ] <EOL> args = [ exe , module , reflect . qual ( self . serverProtocol ) ] <EOL> env = os . environ . copy ( ) <EOL> env [ "<STR_LIT>" ] = os . pathsep . join ( sys . path ) <EOL> from twisted . internet import reactor <EOL> clientTransport = reactor . spawnProcess ( processClient , exe , args , <EOL> env = env , usePTY = True ) <EOL> self . recvlineClient = self . testTerminal = testTerminal <EOL> self . processClient = processClient <EOL> self . clientTransport = clientTransport <EOL> return defer . gatherResults ( filter ( None , [ <EOL> processClient . onConnection , <EOL> testTerminal . expect ( "<STR_LIT>" ) ] ) ) <EOL> def tearDown ( self ) : <EOL> try : <EOL> self . clientTransport . signalProcess ( "<STR_LIT>" ) <EOL> except ( error . ProcessExitedAlready , OSError ) : <EOL> pass <EOL> def trap ( failure ) : <EOL> failure . trap ( error . ProcessTerminated ) <EOL> self . assertEqual ( failure . value . exitCode , None ) <EOL> self . assertEqual ( failure . value . status , <NUM_LIT:9> ) <EOL> return self . testTerminal . onDisconnection . addErrback ( trap ) <EOL> def _testwrite ( self , bytes ) : <EOL> self . clientTransport . write ( bytes ) <EOL> class RecvlineLoopbackMixin : <EOL> serverProtocol = EchoServer <EOL> def testSimple ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testLeftArrow ( self ) : <EOL> return self . _trivialTest ( <EOL> insert + '<STR_LIT>' + left * <NUM_LIT:4> + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testRightArrow ( self ) : <EOL> return self . _trivialTest ( <EOL> insert + '<STR_LIT>' + left * <NUM_LIT:4> + right * <NUM_LIT:2> + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testBackspace ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" + backspace * <NUM_LIT:4> + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testDelete ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" + left * <NUM_LIT:4> + delete * <NUM_LIT:4> + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testInsert ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" + left * <NUM_LIT:3> + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testTypeover ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" + left * <NUM_LIT:4> + insert + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testHome ( self ) : <EOL> return self . _trivialTest ( <EOL> insert + "<STR_LIT>" + home + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testEnd ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" + left * <NUM_LIT:4> + end + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> class RecvlineLoopbackTelnetTests ( _TelnetMixin , unittest . TestCase , RecvlineLoopbackMixin ) : <EOL> pass <EOL> class RecvlineLoopbackSSHTests ( _SSHMixin , unittest . TestCase , RecvlineLoopbackMixin ) : <EOL> pass <EOL> class RecvlineLoopbackStdioTests ( _StdioMixin , unittest . TestCase , RecvlineLoopbackMixin ) : <EOL> if stdio is None : <EOL> skip = "<STR_LIT>" <EOL> class HistoricRecvlineLoopbackMixin : <EOL> serverProtocol = EchoServer <EOL> def testUpArrow ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" + up + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def testDownArrow ( self ) : <EOL> return self . _trivialTest ( <EOL> "<STR_LIT>" + up * <NUM_LIT:2> + down + "<STR_LIT>" , <EOL> [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> class HistoricRecvlineLoopbackTelnetTests ( _TelnetMixin , unittest . TestCase , HistoricRecvlineLoopbackMixin ) : <EOL> pass <EOL> class HistoricRecvlineLoopbackSSHTests ( _SSHMixin , unittest . TestCase , HistoricRecvlineLoopbackMixin ) : <EOL> pass <EOL> class HistoricRecvlineLoopbackStdioTests ( _StdioMixin , unittest . TestCase , HistoricRecvlineLoopbackMixin ) : <EOL> if stdio is None : <EOL> skip = "<STR_LIT>" </s>
