<s> """<STR_LIT>""" <EOL> from twisted . conch . ssh import channel <EOL> from twisted . trial import unittest <EOL> class MockTransport ( object ) : <EOL> """<STR_LIT>""" <EOL> def getPeer ( self ) : <EOL> return ( '<STR_LIT>' , ) <EOL> def getHost ( self ) : <EOL> return ( '<STR_LIT>' , ) <EOL> class MockConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> transport = MockTransport ( ) <EOL> def __init__ ( self ) : <EOL> self . data = { } <EOL> self . extData = { } <EOL> self . closes = { } <EOL> def logPrefix ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" <EOL> def sendData ( self , channel , data ) : <EOL> """<STR_LIT>""" <EOL> self . data . setdefault ( channel , [ ] ) . append ( data ) <EOL> def sendExtendedData ( self , channel , type , data ) : <EOL> """<STR_LIT>""" <EOL> self . extData . setdefault ( channel , [ ] ) . append ( ( type , data ) ) <EOL> def sendClose ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> self . closes [ channel ] = True <EOL> class ChannelTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . conn = MockConnection ( ) <EOL> self . channel = channel . SSHChannel ( conn = self . conn , <EOL> remoteMaxPacket = <NUM_LIT:10> ) <EOL> self . channel . name = '<STR_LIT>' <EOL> def test_init ( self ) : <EOL> """<STR_LIT>""" <EOL> c = channel . SSHChannel ( conn = self . conn ) <EOL> self . assertEqual ( c . localWindowSize , <NUM_LIT> ) <EOL> self . assertEqual ( c . localWindowLeft , <NUM_LIT> ) <EOL> self . assertEqual ( c . localMaxPacket , <NUM_LIT> ) <EOL> self . assertEqual ( c . remoteWindowLeft , <NUM_LIT:0> ) <EOL> self . assertEqual ( c . remoteMaxPacket , <NUM_LIT:0> ) <EOL> self . assertEqual ( c . conn , self . conn ) <EOL> self . assertEqual ( c . data , None ) <EOL> self . assertEqual ( c . avatar , None ) <EOL> c2 = channel . SSHChannel ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> self . assertEqual ( c2 . localWindowSize , <NUM_LIT:1> ) <EOL> self . assertEqual ( c2 . localWindowLeft , <NUM_LIT:1> ) <EOL> self . assertEqual ( c2 . localMaxPacket , <NUM_LIT:2> ) <EOL> self . assertEqual ( c2 . remoteWindowLeft , <NUM_LIT:3> ) <EOL> self . assertEqual ( c2 . remoteMaxPacket , <NUM_LIT:4> ) <EOL> self . assertEqual ( c2 . conn , <NUM_LIT:5> ) <EOL> self . assertEqual ( c2 . data , <NUM_LIT:6> ) <EOL> self . assertEqual ( c2 . avatar , <NUM_LIT:7> ) <EOL> def test_str ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( str ( self . channel ) , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_logPrefix ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . channel . logPrefix ( ) , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_addWindowBytes ( self ) : <EOL> """<STR_LIT>""" <EOL> cb = [ False ] <EOL> def stubStartWriting ( ) : <EOL> cb [ <NUM_LIT:0> ] = True <EOL> self . channel . startWriting = stubStartWriting <EOL> self . channel . write ( '<STR_LIT:test>' ) <EOL> self . channel . writeExtended ( <NUM_LIT:1> , '<STR_LIT:test>' ) <EOL> self . channel . addWindowBytes ( <NUM_LIT:50> ) <EOL> self . assertEqual ( self . channel . remoteWindowLeft , <NUM_LIT:50> - <NUM_LIT:4> - <NUM_LIT:4> ) <EOL> self . assertTrue ( self . channel . areWriting ) <EOL> self . assertTrue ( cb [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( self . channel . buf , '<STR_LIT>' ) <EOL> self . assertEqual ( self . conn . data [ self . channel ] , [ '<STR_LIT:test>' ] ) <EOL> self . assertEqual ( self . channel . extBuf , [ ] ) <EOL> self . assertEqual ( self . conn . extData [ self . channel ] , [ ( <NUM_LIT:1> , '<STR_LIT:test>' ) ] ) <EOL> cb [ <NUM_LIT:0> ] = False <EOL> self . channel . addWindowBytes ( <NUM_LIT:20> ) <EOL> self . assertFalse ( cb [ <NUM_LIT:0> ] ) <EOL> self . channel . write ( '<STR_LIT:a>' * <NUM_LIT> ) <EOL> self . channel . loseConnection ( ) <EOL> self . channel . addWindowBytes ( <NUM_LIT:20> ) <EOL> self . assertFalse ( cb [ <NUM_LIT:0> ] ) <EOL> def test_requestReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> self . channel . request_test_method = lambda data : data == '<STR_LIT>' <EOL> self . assertTrue ( self . channel . requestReceived ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( self . channel . requestReceived ( '<STR_LIT>' , '<STR_LIT:a>' ) ) <EOL> self . assertFalse ( self . channel . requestReceived ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_closeReceieved ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertFalse ( self . channel . closing ) <EOL> self . channel . closeReceived ( ) <EOL> self . assertTrue ( self . channel . closing ) <EOL> def test_write ( self ) : <EOL> """<STR_LIT>""" <EOL> cb = [ False ] <EOL> def stubStopWriting ( ) : <EOL> cb [ <NUM_LIT:0> ] = True <EOL> self . channel . stopWriting = stubStopWriting <EOL> self . channel . write ( '<STR_LIT:d>' ) <EOL> self . channel . write ( '<STR_LIT:a>' ) <EOL> self . assertFalse ( self . channel . areWriting ) <EOL> self . assertTrue ( cb [ <NUM_LIT:0> ] ) <EOL> self . channel . addWindowBytes ( <NUM_LIT:20> ) <EOL> self . channel . write ( '<STR_LIT>' ) <EOL> data = self . conn . data [ self . channel ] <EOL> self . assertEqual ( data , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . channel . remoteWindowLeft , <NUM_LIT:16> ) <EOL> self . channel . write ( '<STR_LIT>' ) <EOL> self . assertEqual ( data , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:1>' ] ) <EOL> self . assertEqual ( self . channel . remoteWindowLeft , <NUM_LIT:5> ) <EOL> cb [ <NUM_LIT:0> ] = False <EOL> self . channel . write ( '<STR_LIT>' ) <EOL> self . assertFalse ( self . channel . areWriting ) <EOL> self . assertTrue ( cb [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( data , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . channel . buf , '<STR_LIT>' ) <EOL> self . assertEqual ( self . channel . remoteWindowLeft , <NUM_LIT:0> ) <EOL> def test_writeExtended ( self ) : <EOL> """<STR_LIT>""" <EOL> cb = [ False ] <EOL> def stubStopWriting ( ) : <EOL> cb [ <NUM_LIT:0> ] = True <EOL> self . channel . stopWriting = stubStopWriting <EOL> self . channel . writeExtended ( <NUM_LIT:1> , '<STR_LIT:d>' ) <EOL> self . channel . writeExtended ( <NUM_LIT:1> , '<STR_LIT:a>' ) <EOL> self . channel . writeExtended ( <NUM_LIT:2> , '<STR_LIT:t>' ) <EOL> self . assertFalse ( self . channel . areWriting ) <EOL> self . assertTrue ( cb [ <NUM_LIT:0> ] ) <EOL> self . channel . addWindowBytes ( <NUM_LIT:20> ) <EOL> self . channel . writeExtended ( <NUM_LIT:2> , '<STR_LIT:a>' ) <EOL> data = self . conn . extData [ self . channel ] <EOL> self . assertEqual ( data , [ ( <NUM_LIT:1> , '<STR_LIT>' ) , ( <NUM_LIT:2> , '<STR_LIT:t>' ) , ( <NUM_LIT:2> , '<STR_LIT:a>' ) ] ) <EOL> self . assertEqual ( self . channel . remoteWindowLeft , <NUM_LIT:16> ) <EOL> self . channel . writeExtended ( <NUM_LIT:3> , '<STR_LIT>' ) <EOL> self . assertEqual ( data , [ ( <NUM_LIT:1> , '<STR_LIT>' ) , ( <NUM_LIT:2> , '<STR_LIT:t>' ) , ( <NUM_LIT:2> , '<STR_LIT:a>' ) , <EOL> ( <NUM_LIT:3> , '<STR_LIT>' ) , ( <NUM_LIT:3> , '<STR_LIT:1>' ) ] ) <EOL> self . assertEqual ( self . channel . remoteWindowLeft , <NUM_LIT:5> ) <EOL> cb [ <NUM_LIT:0> ] = False <EOL> self . channel . writeExtended ( <NUM_LIT:4> , '<STR_LIT>' ) <EOL> self . assertFalse ( self . channel . areWriting ) <EOL> self . assertTrue ( cb [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( data , [ ( <NUM_LIT:1> , '<STR_LIT>' ) , ( <NUM_LIT:2> , '<STR_LIT:t>' ) , ( <NUM_LIT:2> , '<STR_LIT:a>' ) , <EOL> ( <NUM_LIT:3> , '<STR_LIT>' ) , ( <NUM_LIT:3> , '<STR_LIT:1>' ) , ( <NUM_LIT:4> , '<STR_LIT>' ) ] ) <EOL> self . assertEqual ( self . channel . extBuf , [ [ <NUM_LIT:4> , '<STR_LIT>' ] ] ) <EOL> self . assertEqual ( self . channel . remoteWindowLeft , <NUM_LIT:0> ) <EOL> def test_writeSequence ( self ) : <EOL> """<STR_LIT>""" <EOL> self . channel . addWindowBytes ( <NUM_LIT:20> ) <EOL> self . channel . writeSequence ( map ( str , range ( <NUM_LIT:10> ) ) ) <EOL> self . assertEqual ( self . conn . data [ self . channel ] , [ '<STR_LIT>' ] ) <EOL> def test_loseConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . channel . write ( '<STR_LIT:data>' ) <EOL> self . channel . writeExtended ( <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . channel . loseConnection ( ) <EOL> self . assertEqual ( self . conn . closes . get ( self . channel ) , None ) <EOL> self . channel . addWindowBytes ( <NUM_LIT:4> ) <EOL> self . assertEqual ( self . conn . closes . get ( self . channel ) , None ) <EOL> self . channel . addWindowBytes ( <NUM_LIT:8> ) <EOL> self . assertTrue ( self . conn . closes . get ( self . channel ) ) <EOL> def test_getPeer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . channel . getPeer ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_getHost ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . channel . getHost ( ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) </s>
