<s> from twisted . internet import protocol <EOL> from twisted . trial import unittest <EOL> from twimp . amf0 import encode as encode_amf <EOL> from twimp import chunks <EOL> from twimp import const <EOL> from twimp . proto import DispatchProtocol <EOL> from twimp . helpers import vb <EOL> from test . helpers import StringTransport , unvb <EOL> class TestDemuxerMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def inject ( self , cs_id , time , type_ , ms_id , body ) : <EOL> h = chunks . Header ( cs_id , time , len ( body ) , type_ , ms_id ) <EOL> if cs_id == <NUM_LIT:2> and <NUM_LIT:0> < type_ < <NUM_LIT:8> and ms_id == <NUM_LIT:0> : <EOL> self . controlMessageReceived ( h , body ) <EOL> else : <EOL> self . protocol . messageReceived ( h , body ) <EOL> def sinject ( self , cs_id , time , type_ , ms_id , data ) : <EOL> body = vb ( data ) <EOL> self . inject ( cs_id , time , type_ , ms_id , body ) <EOL> def mixin ( base_class , mixin_class , name = None ) : <EOL> if not name : <EOL> name = '<STR_LIT>' + base_class . __name__ <EOL> return type ( name , ( mixin_class , base_class ) , { } ) <EOL> class TestMuxer ( object ) : <EOL> def __init__ ( self , transport ) : <EOL> self . transport = transport <EOL> self . chunk_size = chunks . DEFAULT_CHUNK_SIZE <EOL> self . messages = [ ] <EOL> def set_chunk_size ( self , new_chunk_size ) : <EOL> self . chunk_size = new_chunk_size <EOL> def sendMessage ( self , time , type_ , ms_id , body , absolute = False ) : <EOL> self . messages . append ( ( time , type_ , ms_id , body . read ( len ( body ) ) , <EOL> absolute ) ) <EOL> class TestHandshaker ( object ) : <EOL> def __init__ ( self , protocol , epoch_base , is_client = False ) : <EOL> self . protocol = protocol <EOL> def gen_handler ( self ) : <EOL> s = yield <NUM_LIT:1> <EOL> s . read ( <NUM_LIT:1> ) <EOL> self . protocol . handshakeSucceeded ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> class TestFactory ( protocol . Factory ) : <EOL> init_time = <NUM_LIT:0> <EOL> class _ProtocolTestBase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . messages = [ ] <EOL> def clear ( self ) : <EOL> self . messages [ : ] = [ ] <EOL> def make_protocol_class ( self ) : <EOL> raise NotImplementedError ( ) <EOL> def make_factory ( self ) : <EOL> return TestFactory ( ) <EOL> def make_protocol ( self ) : <EOL> f = self . make_factory ( ) <EOL> f . protocol = self . make_protocol_class ( ) <EOL> p = f . buildProtocol ( None ) <EOL> return p <EOL> def connect_protocol ( self , protocol ) : <EOL> t = StringTransport ( ) <EOL> protocol . makeConnection ( t ) <EOL> protocol . dataReceived ( '<STR_LIT:.>' ) <EOL> return t <EOL> def build_proto ( self ) : <EOL> p = self . make_protocol ( ) <EOL> t = self . connect_protocol ( p ) <EOL> return p , t , p . _demuxer , p . muxer <EOL> class TestDispatchProtocol ( _ProtocolTestBase ) : <EOL> def make_protocol_class ( self ) : <EOL> tself = self <EOL> class TDispatchProtocol ( DispatchProtocol ) : <EOL> demuxer_class = mixin ( DispatchProtocol . demuxer_class , <EOL> TestDemuxerMixin ) <EOL> muxer_class = TestMuxer <EOL> handshaker_class = TestHandshaker <EOL> def doCommand ( self , ts , ms_id , args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , args ) ) <EOL> def doMeta ( self , ts , ms_id , args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , args ) ) <EOL> def doData ( self , type_ , ts , ms_id , body ) : <EOL> tself . messages . append ( ( '<STR_LIT:data>' , ts , ms_id , type_ , unvb ( body ) ) ) <EOL> def unknownMessageType ( self , header , body ) : <EOL> header_args = ( header . cs_id , header . abs_time , header . size , <EOL> header . type , header . ms_id ) <EOL> tself . messages . append ( ( '<STR_LIT>' , header_args , unvb ( body ) ) ) <EOL> return TDispatchProtocol <EOL> def test_dispatch_data ( self ) : <EOL> p , t , d , m = self . build_proto ( ) <EOL> d . sinject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_AUDIO , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> d . sinject ( <NUM_LIT:4> , <NUM_LIT:0> , const . RTMP_VIDEO , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEquals ( self . messages , [ ( '<STR_LIT:data>' , <NUM_LIT:0> , <NUM_LIT:1> , const . RTMP_AUDIO , <EOL> '<STR_LIT>' ) , <EOL> ( '<STR_LIT:data>' , <NUM_LIT:0> , <NUM_LIT:1> , const . RTMP_VIDEO , <EOL> '<STR_LIT>' ) ] ) <EOL> def test_dispatch_meta ( self ) : <EOL> p , t , d , m = self . build_proto ( ) <EOL> d . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_DATA , <NUM_LIT:1> , encode_amf ( '<STR_LIT>' , None ) ) <EOL> self . assertEquals ( self . messages , [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , [ '<STR_LIT>' , None ] ) ] ) <EOL> def test_dispatch_command ( self ) : <EOL> p , t , d , m = self . build_proto ( ) <EOL> d . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None ) ) <EOL> self . assertEquals ( self . messages , [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , <EOL> [ '<STR_LIT>' , <NUM_LIT:0.0> , None ] ) ] ) <EOL> def test_dispatch_unknown ( self ) : <EOL> p , t , d , m = self . build_proto ( ) <EOL> d . sinject ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:4> , <NUM_LIT> , <NUM_LIT:1> ) , '<STR_LIT>' ) ] ) </s>
