<s> """<STR_LIT>""" <EOL> from twisted . internet import protocol <EOL> from twisted . python import failure <EOL> from twisted . trial import unittest <EOL> from twisted . words . xish import domish , utility , xmlstream <EOL> class XmlStreamTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . connectionLostMsg = "<STR_LIT>" <EOL> self . outlist = [ ] <EOL> self . xmlstream = xmlstream . XmlStream ( ) <EOL> self . xmlstream . transport = self <EOL> self . xmlstream . transport . write = self . outlist . append <EOL> def loseConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . xmlstream . connectionLost ( failure . Failure ( <EOL> Exception ( self . connectionLostMsg ) ) ) <EOL> def test_send ( self ) : <EOL> """<STR_LIT>""" <EOL> self . xmlstream . connectionMade ( ) <EOL> self . xmlstream . send ( "<STR_LIT>" ) <EOL> self . assertEqual ( self . outlist [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> def test_receiveRoot ( self ) : <EOL> """<STR_LIT>""" <EOL> streamStarted = [ ] <EOL> def streamStartEvent ( rootelem ) : <EOL> streamStarted . append ( None ) <EOL> self . xmlstream . addObserver ( xmlstream . STREAM_START_EVENT , <EOL> streamStartEvent ) <EOL> self . xmlstream . connectionMade ( ) <EOL> self . xmlstream . dataReceived ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( streamStarted ) ) <EOL> def test_receiveBadXML ( self ) : <EOL> """<STR_LIT>""" <EOL> streamError = [ ] <EOL> streamEnd = [ ] <EOL> def streamErrorEvent ( reason ) : <EOL> streamError . append ( reason ) <EOL> def streamEndEvent ( _ ) : <EOL> streamEnd . append ( None ) <EOL> self . xmlstream . addObserver ( xmlstream . STREAM_ERROR_EVENT , <EOL> streamErrorEvent ) <EOL> self . xmlstream . addObserver ( xmlstream . STREAM_END_EVENT , <EOL> streamEndEvent ) <EOL> self . xmlstream . connectionMade ( ) <EOL> self . xmlstream . dataReceived ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( streamError ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( streamEnd ) ) <EOL> self . xmlstream . dataReceived ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( streamError ) ) <EOL> self . assertTrue ( streamError [ <NUM_LIT:0> ] . check ( domish . ParserError ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( streamEnd ) ) <EOL> def test_streamEnd ( self ) : <EOL> """<STR_LIT>""" <EOL> streamEnd = [ ] <EOL> def streamEndEvent ( reason ) : <EOL> streamEnd . append ( reason ) <EOL> self . xmlstream . addObserver ( xmlstream . STREAM_END_EVENT , <EOL> streamEndEvent ) <EOL> self . xmlstream . connectionMade ( ) <EOL> self . loseConnection ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( streamEnd ) ) <EOL> self . assertIsInstance ( streamEnd [ <NUM_LIT:0> ] , failure . Failure ) <EOL> self . assertEqual ( streamEnd [ <NUM_LIT:0> ] . getErrorMessage ( ) , <EOL> self . connectionLostMsg ) <EOL> class DummyProtocol ( protocol . Protocol , utility . EventDispatcher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> self . observers = [ ] <EOL> utility . EventDispatcher . __init__ ( self ) <EOL> class BootstrapMixinTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . factory = xmlstream . BootstrapMixin ( ) <EOL> def test_installBootstraps ( self ) : <EOL> """<STR_LIT>""" <EOL> called = [ ] <EOL> def cb ( data ) : <EOL> called . append ( data ) <EOL> dispatcher = DummyProtocol ( ) <EOL> self . factory . addBootstrap ( '<STR_LIT>' , cb ) <EOL> self . factory . installBootstraps ( dispatcher ) <EOL> dispatcher . dispatch ( None , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( called ) ) <EOL> def test_addAndRemoveBootstrap ( self ) : <EOL> """<STR_LIT>""" <EOL> called = [ ] <EOL> def cb ( data ) : <EOL> called . append ( data ) <EOL> self . factory . addBootstrap ( '<STR_LIT>' , cb ) <EOL> self . factory . removeBootstrap ( '<STR_LIT>' , cb ) <EOL> dispatcher = DummyProtocol ( ) <EOL> self . factory . installBootstraps ( dispatcher ) <EOL> dispatcher . dispatch ( None , '<STR_LIT>' ) <EOL> self . assertFalse ( called ) <EOL> class GenericXmlStreamFactoryTestsMixin ( BootstrapMixinTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . factory = xmlstream . XmlStreamFactory ( ) <EOL> def test_buildProtocolInstallsBootstraps ( self ) : <EOL> """<STR_LIT>""" <EOL> called = [ ] <EOL> def cb ( data ) : <EOL> called . append ( data ) <EOL> self . factory . addBootstrap ( '<STR_LIT>' , cb ) <EOL> xs = self . factory . buildProtocol ( None ) <EOL> xs . dispatch ( None , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( called ) ) <EOL> def test_buildProtocolStoresFactory ( self ) : <EOL> """<STR_LIT>""" <EOL> xs = self . factory . buildProtocol ( None ) <EOL> self . assertIdentical ( self . factory , xs . factory ) <EOL> class XmlStreamFactoryMixinTest ( GenericXmlStreamFactoryTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . factory = xmlstream . XmlStreamFactoryMixin ( None , test = None ) <EOL> self . factory . protocol = DummyProtocol <EOL> def test_buildProtocolFactoryArguments ( self ) : <EOL> """<STR_LIT>""" <EOL> xs = self . factory . buildProtocol ( None ) <EOL> self . assertEqual ( ( None , ) , xs . args ) <EOL> self . assertEqual ( { '<STR_LIT:test>' : None } , xs . kwargs ) </s>
