<s> from twisted . trial import unittest <EOL> from twisted . internet import protocol , reactor , error <EOL> from twisted . python import failure , components <EOL> from twisted . pair import ethernet , raw <EOL> from zope . interface import implements <EOL> class MyProtocol : <EOL> implements ( raw . IRawPacketProtocol ) <EOL> def __init__ ( self , expecting ) : <EOL> self . expecting = list ( expecting ) <EOL> def datagramReceived ( self , data , ** kw ) : <EOL> assert self . expecting , '<STR_LIT>' <EOL> expect = self . expecting . pop ( <NUM_LIT:0> ) <EOL> assert expect == ( data , kw ) , "<STR_LIT>" % ( <EOL> expect , ( data , kw ) , <EOL> ) <EOL> class EthernetTestCase ( unittest . TestCase ) : <EOL> def testPacketParsing ( self ) : <EOL> proto = ethernet . EthernetProtocol ( ) <EOL> p1 = MyProtocol ( [ <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ] ) <EOL> proto . addProto ( <NUM_LIT> , p1 ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:0> ) <EOL> assert not p1 . expecting , '<STR_LIT>' % p1 . expecting <EOL> def testMultiplePackets ( self ) : <EOL> proto = ethernet . EthernetProtocol ( ) <EOL> p1 = MyProtocol ( [ <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ] ) <EOL> proto . addProto ( <NUM_LIT> , p1 ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:0> ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:1> ) <EOL> assert not p1 . expecting , '<STR_LIT>' % p1 . expecting <EOL> def testMultipleSameProtos ( self ) : <EOL> proto = ethernet . EthernetProtocol ( ) <EOL> p1 = MyProtocol ( [ <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ] ) <EOL> p2 = MyProtocol ( [ <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ] ) <EOL> proto . addProto ( <NUM_LIT> , p1 ) <EOL> proto . addProto ( <NUM_LIT> , p2 ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:0> ) <EOL> assert not p1 . expecting , '<STR_LIT>' % p1 . expecting <EOL> assert not p2 . expecting , '<STR_LIT>' % p2 . expecting <EOL> def testWrongProtoNotSeen ( self ) : <EOL> proto = ethernet . EthernetProtocol ( ) <EOL> p1 = MyProtocol ( [ ] ) <EOL> proto . addProto ( <NUM_LIT> , p1 ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:0> ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:1> ) <EOL> def testDemuxing ( self ) : <EOL> proto = ethernet . EthernetProtocol ( ) <EOL> p1 = MyProtocol ( [ <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ] ) <EOL> proto . addProto ( <NUM_LIT> , p1 ) <EOL> p2 = MyProtocol ( [ <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:source>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } ) , <EOL> ] ) <EOL> proto . addProto ( <NUM_LIT> , p2 ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:0> ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:1> ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:0> ) <EOL> proto . datagramReceived ( "<STR_LIT>" , <EOL> partial = <NUM_LIT:1> ) <EOL> assert not p1 . expecting , '<STR_LIT>' % p1 . expecting <EOL> assert not p2 . expecting , '<STR_LIT>' % p2 . expecting <EOL> def testAddingBadProtos_WrongLevel ( self ) : <EOL> """<STR_LIT>""" <EOL> e = ethernet . EthernetProtocol ( ) <EOL> try : <EOL> e . addProto ( <NUM_LIT> , "<STR_LIT>" ) <EOL> except components . CannotAdapt : <EOL> pass <EOL> else : <EOL> raise AssertionError , '<STR_LIT>' <EOL> def testAddingBadProtos_TooSmall ( self ) : <EOL> """<STR_LIT>""" <EOL> e = ethernet . EthernetProtocol ( ) <EOL> try : <EOL> e . addProto ( - <NUM_LIT:1> , MyProtocol ( [ ] ) ) <EOL> except TypeError , e : <EOL> if e . args == ( '<STR_LIT>' , ) : <EOL> pass <EOL> else : <EOL> raise <EOL> else : <EOL> raise AssertionError , '<STR_LIT>' <EOL> def testAddingBadProtos_TooBig ( self ) : <EOL> """<STR_LIT>""" <EOL> e = ethernet . EthernetProtocol ( ) <EOL> try : <EOL> e . addProto ( <NUM_LIT:2> ** <NUM_LIT:16> , MyProtocol ( [ ] ) ) <EOL> except TypeError , e : <EOL> if e . args == ( '<STR_LIT>' , ) : <EOL> pass <EOL> else : <EOL> raise <EOL> else : <EOL> raise AssertionError , '<STR_LIT>' <EOL> def testAddingBadProtos_TooBig2 ( self ) : <EOL> """<STR_LIT>""" <EOL> e = ethernet . EthernetProtocol ( ) <EOL> try : <EOL> e . addProto ( <NUM_LIT:2> ** <NUM_LIT:16> + <NUM_LIT:1> , MyProtocol ( [ ] ) ) <EOL> except TypeError , e : <EOL> if e . args == ( '<STR_LIT>' , ) : <EOL> pass <EOL> else : <EOL> raise <EOL> else : <EOL> raise AssertionError , '<STR_LIT>' </s>
