<s> from twisted . internet import defer , error , reactor <EOL> from twimp . amf0 import encode as encode_amf , decode as decode_amf , Object <EOL> from twimp import chunks <EOL> from twimp import const <EOL> from twimp . dispatch import CommandDispatchProtocol <EOL> from twimp . dispatch import EventDispatchProtocol <EOL> from twimp . dispatch import CallDispatchProtocol <EOL> from twimp . error import UnexpectedStatusError , ProtocolContractError <EOL> from twimp . error import CommandResultError <EOL> from twimp . error import CallAbortedException <EOL> from twimp . helpers import vb , vb_clone <EOL> from test . test_proto import _ProtocolTestBase , mixin <EOL> from test . test_proto import TestDemuxerMixin , TestMuxer , TestHandshaker <EOL> from test . common import ArtificialError , ArtificialRemoteError <EOL> from test . helpers import muxer_messages , unvb <EOL> class TestCommandDispatchProtocol ( _ProtocolTestBase ) : <EOL> timeout = <NUM_LIT:10> <EOL> def make_protocol_class ( self ) : <EOL> tself = self <EOL> CDP = CommandDispatchProtocol <EOL> class TCommandDispatchProtocol ( CDP ) : <EOL> demuxer_class = mixin ( CDP . demuxer_class , TestDemuxerMixin ) <EOL> muxer_class = TestMuxer <EOL> handshaker_class = TestHandshaker <EOL> def unknownMessageType ( self , header , body ) : <EOL> bclone = vb_clone ( body ) <EOL> tself . messages . append ( ( '<STR_LIT>' , header , unvb ( bclone ) ) ) <EOL> CDP . unknownMessageType ( self , header , body ) <EOL> def unknownCommandType ( self , cmd , ts , ms_id , args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , cmd , ts , ms_id , args ) ) <EOL> def unexpectedCallResult ( self , ts , ms_id , trans_id , args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , trans_id , args ) ) <EOL> def unexpectedCallError ( self , ts , ms_id , trans_id , args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , trans_id , args ) ) <EOL> def command_foobar ( self , ts , ms_id , trans_id , arg1 , arg2 ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , trans_id , <EOL> arg1 , arg2 ) ) <EOL> def command_breakdown ( self , ts , ms_id , trans_id , * args ) : <EOL> raise ArtificialError ( '<STR_LIT>' ) <EOL> return TCommandDispatchProtocol <EOL> def test_command_simple ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( lambda _ : self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0.0> , <EOL> '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ] ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_command_simple_after_exception ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( lambda _ : self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0.0> , <EOL> '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ] ) ) <EOL> def check_clean_logs ( result ) : <EOL> self . assertEquals ( len ( self . flushLoggedErrors ( ArtificialError ) ) , <EOL> <NUM_LIT:1> ) <EOL> return result <EOL> d . addBoth ( check_clean_logs ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_command_unknown ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT:a>' , '<STR_LIT>' ) ) <EOL> d = defer . Deferred ( ) <EOL> def verify_unknown ( result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , '<STR_LIT>' , <EOL> <NUM_LIT:0> , <NUM_LIT:1> , [ <NUM_LIT:0.0> , '<STR_LIT:a>' , '<STR_LIT>' ] ) ] ) <EOL> return result <EOL> d . addCallback ( verify_unknown ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_unexpectedCallResult ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:2> , None , '<STR_LIT>' ) ) <EOL> d = defer . Deferred ( ) <EOL> def verify_unexpected ( result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT> , ( None , '<STR_LIT>' ) ) ] ) <EOL> return result <EOL> d . addCallback ( verify_unexpected ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_unexpectedCallError ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:2> , None , '<STR_LIT>' ) ) <EOL> d = defer . Deferred ( ) <EOL> def verify_unexpected ( result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT> , ( None , '<STR_LIT>' ) ) ] ) <EOL> return result <EOL> d . addCallback ( verify_unexpected ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_callRemote_result ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> d = p . callRemote ( <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEquals ( len ( mux . messages ) , <NUM_LIT:1> ) <EOL> msg = mux . messages [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ <NUM_LIT:0> : <NUM_LIT:3> ] + msg [ <NUM_LIT:4> : ] , <EOL> ( <NUM_LIT:0> , chunks . MSG_COMMAND , <NUM_LIT:1> , False ) ) <EOL> decoded = decode_amf ( vb ( msg [ <NUM_LIT:3> ] ) ) <EOL> self . assertEquals ( ( decoded [ <NUM_LIT:0> ] , decoded [ <NUM_LIT:2> ] ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , decoded [ <NUM_LIT:1> ] , None , decoded [ <NUM_LIT:2> ] ) ) <EOL> d . addCallback ( self . assertEquals , ( None , '<STR_LIT>' ) ) <EOL> return d <EOL> def test_callRemote_error ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> d = p . callRemote ( <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEquals ( len ( mux . messages ) , <NUM_LIT:1> ) <EOL> msg = mux . messages [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ <NUM_LIT:0> : <NUM_LIT:3> ] + msg [ <NUM_LIT:4> : ] , <EOL> ( <NUM_LIT:0> , chunks . MSG_COMMAND , <NUM_LIT:1> , False ) ) <EOL> decoded = decode_amf ( vb ( msg [ <NUM_LIT:3> ] ) ) <EOL> self . assertEquals ( ( decoded [ <NUM_LIT:0> ] , decoded [ <NUM_LIT:2> ] ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> err_info = Object ( code = '<STR_LIT>' , level = '<STR_LIT:error>' , desc = '<STR_LIT>' ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , decoded [ <NUM_LIT:1> ] , None , err_info ) ) <EOL> self . assertFailure ( d , CommandResultError ) <EOL> d . addCallback ( lambda r : self . assertEquals ( r . args , ( None , err_info ) ) ) <EOL> return d <EOL> def test_callRemote_disconnect ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> d = p . callRemote ( <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEquals ( len ( mux . messages ) , <NUM_LIT:1> ) <EOL> msg = mux . messages [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ <NUM_LIT:0> : <NUM_LIT:3> ] + msg [ <NUM_LIT:4> : ] , <EOL> ( <NUM_LIT:0> , chunks . MSG_COMMAND , <NUM_LIT:1> , False ) ) <EOL> decoded = decode_amf ( vb ( msg [ <NUM_LIT:3> ] ) ) <EOL> self . assertEquals ( ( decoded [ <NUM_LIT:0> ] , decoded [ <NUM_LIT:2> ] ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> p . connectionLost ( ) <EOL> self . assertFailure ( d , error . ConnectionDone ) <EOL> return d <EOL> info1 = Object ( code = '<STR_LIT>' , level = '<STR_LIT:status>' , desc = '<STR_LIT>' ) <EOL> info2 = Object ( code = '<STR_LIT>' , level = '<STR_LIT:status>' , desc = '<STR_LIT>' ) <EOL> class TestEventDispatchProtocol ( _ProtocolTestBase ) : <EOL> timeout = <NUM_LIT:10> <EOL> def make_protocol_class ( self ) : <EOL> tself = self <EOL> EDP = EventDispatchProtocol <EOL> class TEventDispatchProtocol ( EDP ) : <EOL> demuxer_class = mixin ( EDP . demuxer_class , TestDemuxerMixin ) <EOL> muxer_class = TestMuxer <EOL> handshaker_class = TestHandshaker <EOL> def unhandledOnStatus ( self , ts , ms_id , info ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , info ) ) <EOL> return TEventDispatchProtocol <EOL> def test_waitStatus ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> d = p . waitStatus ( <NUM_LIT:1> , info1 . code ) <EOL> d . addCallback ( self . assertEquals , info1 ) <EOL> d . addCallback ( lambda _ : p . waitStatus ( <NUM_LIT:1> , None ) ) <EOL> d . addCallback ( self . assertEquals , info2 ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info1 ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info2 ) ) <EOL> return d <EOL> def test_waitStatus_mixed ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> d1 = p . waitStatus ( <NUM_LIT:2> , info1 . code ) <EOL> d1 . addCallback ( self . assertEquals , info1 ) <EOL> d2 = p . waitStatus ( <NUM_LIT:1> , None ) <EOL> d2 . addCallback ( self . assertEquals , info2 ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info2 ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:2> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info1 ) ) <EOL> return defer . DeferredList ( [ d1 , d2 ] , fireOnOneErrback = <NUM_LIT:1> ) <EOL> def test_waitStatus_unexpected ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> code_other = '<STR_LIT>' <EOL> d = p . waitStatus ( <NUM_LIT:1> , info1 . code ) <EOL> d . addCallback ( self . assertEquals , info1 ) <EOL> d . addCallback ( lambda _ : p . waitStatus ( <NUM_LIT:1> , code_other ) ) <EOL> self . assertFailure ( d , UnexpectedStatusError ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info1 ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info2 ) ) <EOL> return d <EOL> def test_waitStatus_unhandled ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> d = p . waitStatus ( <NUM_LIT:1> , info2 . code ) <EOL> d . addCallback ( self . assertEquals , info2 ) <EOL> unhandled = ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:2> , info1 ) <EOL> d . addCallback ( lambda _ : self . assertEquals ( self . messages , [ unhandled ] ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:2> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info1 ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , info2 ) ) <EOL> return d <EOL> def test_waitStatus_no_code ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> wrong_info = Object ( level = '<STR_LIT:status>' , desc = '<STR_LIT>' ) <EOL> d = p . waitStatus ( <NUM_LIT:1> , None ) <EOL> self . assertFailure ( d , ProtocolContractError ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , None , wrong_info ) ) <EOL> return d <EOL> def test_waitStatus_disconnected ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> d = p . waitStatus ( <NUM_LIT:1> , None ) <EOL> self . assertFailure ( d , error . ConnectionDone ) <EOL> p . connectionLost ( ) <EOL> return d <EOL> class TestCallDispatchProtocol ( _ProtocolTestBase ) : <EOL> def make_protocol_class ( self ) : <EOL> tself = self <EOL> CDP = CallDispatchProtocol <EOL> class TCallDispatchProtocol ( CDP ) : <EOL> demuxer_class = mixin ( CDP . demuxer_class , TestDemuxerMixin ) <EOL> muxer_class = TestMuxer <EOL> handshaker_class = TestHandshaker <EOL> def session_time ( self ) : <EOL> return <NUM_LIT:0> <EOL> def unknownRemoteCall ( self , cmd , ts , ms_id , args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , cmd , ts , ms_id , args ) ) <EOL> CDP . unknownRemoteCall ( self , cmd , ts , ms_id , args ) <EOL> def command_baz ( self , ts , ms_id , trans_id , arg1 , arg2 ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , trans_id , <EOL> arg1 , arg2 ) ) <EOL> def remote_baz ( self , ts , ms_id , arg1 , arg2 ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , arg1 , arg2 ) ) <EOL> def remote_foo ( self , ts , ms_id , arg1 , arg2 ) : <EOL> tself . messages . append ( ( '<STR_LIT:foo>' , ts , ms_id , arg1 , arg2 ) ) <EOL> return '<STR_LIT:f>' , '<STR_LIT:o>' , '<STR_LIT:o>' <EOL> def remote_bar ( self , ts , ms_id , * args ) : <EOL> tself . messages . append ( ( '<STR_LIT:bar>' , ts , ms_id , args ) ) <EOL> def remote_breakdown1 ( self , ts , ms_id , * args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , args ) ) <EOL> raise ArtificialRemoteError ( '<STR_LIT>' ) <EOL> def remote_breakdown2 ( self , ts , ms_id , * args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , args ) ) <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> def remote_breakdown3 ( self , ts , ms_id , * args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , args ) ) <EOL> raise ArtificialRemoteError ( '<STR_LIT>' , fatal = True ) <EOL> def remote_breakdown4 ( self , ts , ms_id , * args ) : <EOL> tself . messages . append ( ( '<STR_LIT>' , ts , ms_id , args ) ) <EOL> raise CallAbortedException ( "<STR_LIT>" ) <EOL> return TCallDispatchProtocol <EOL> def test_call_unknown ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT:foo>' ) ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:bar>' ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , '<STR_LIT>' , <EOL> <NUM_LIT:0> , <NUM_LIT:1> , [ '<STR_LIT:foo>' ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <EOL> <NUM_LIT:0> , <NUM_LIT:1> , [ '<STR_LIT:bar>' ] ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> msgs = muxer_messages ( mux ) <EOL> self . assertEquals ( msgs , [ ] ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> d . addCallback ( lambda _ : self . assertFalse ( t . disconnecting ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_call_comand ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT:a>' , '<STR_LIT:b>' ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0.0> , '<STR_LIT:a>' , '<STR_LIT:b>' ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> self . assertEquals ( mux . messages , [ ] ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_call_with_return ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT:foo>' , <NUM_LIT:1> , '<STR_LIT:a>' , '<STR_LIT:b>' ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT:foo>' , <NUM_LIT:0> , <NUM_LIT:1> , '<STR_LIT:a>' , '<STR_LIT:b>' ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> msgs = muxer_messages ( mux ) <EOL> self . assertEquals ( msgs , [ ( <NUM_LIT:0> , <NUM_LIT:6> , <NUM_LIT:1> , <EOL> [ '<STR_LIT>' , <NUM_LIT:1.0> , '<STR_LIT:f>' , '<STR_LIT:o>' , '<STR_LIT:o>' ] , <EOL> False ) ] ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_call_without_return ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT:foo>' , <NUM_LIT:0> , '<STR_LIT:a>' , '<STR_LIT:b>' ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT:foo>' , <NUM_LIT:0> , <NUM_LIT:1> , '<STR_LIT:a>' , '<STR_LIT:b>' ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> self . assertEquals ( mux . messages , [ ] ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_call_with_implicit_none_return ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT:bar>' , <NUM_LIT:1> ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT:bar>' , <NUM_LIT:0> , <NUM_LIT:1> , ( ) ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> msgs = muxer_messages ( mux ) <EOL> self . assertEquals ( msgs , [ ( <NUM_LIT:0> , <NUM_LIT:6> , <NUM_LIT:1> , [ '<STR_LIT>' , <NUM_LIT:1.0> , None ] , False ) ] ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def _test_call_raising_error_1 ( self , trans_id ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , trans_id ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , ( ) ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> msgs = muxer_messages ( mux ) <EOL> error_tuple = ( '<STR_LIT>' , '<STR_LIT:error>' ) <EOL> self . assertEquals ( len ( msgs ) , <NUM_LIT:1> ) <EOL> msg = msgs [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ <NUM_LIT:0> : <NUM_LIT:3> ] + msg [ <NUM_LIT:4> : ] , <EOL> ( <NUM_LIT:0> , chunks . MSG_COMMAND , <NUM_LIT:1> , False ) ) <EOL> args = msg [ <NUM_LIT:3> ] <EOL> self . assertEquals ( len ( args ) , <NUM_LIT:4> ) <EOL> self . assertEquals ( args [ : <NUM_LIT:3> ] , [ '<STR_LIT>' , trans_id , None ] ) <EOL> self . assertIsInstance ( args [ <NUM_LIT:3> ] , Object ) <EOL> self . assertEquals ( ( args [ <NUM_LIT:3> ] . code , args [ <NUM_LIT:3> ] . level ) , <EOL> error_tuple ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> d . addCallback ( lambda _ : self . assertFalse ( t . disconnecting ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_call_raising_error_1_unexpected ( self ) : <EOL> return self . _test_call_raising_error_1 ( <NUM_LIT:0> ) <EOL> def test_call_raising_error_1_expected ( self ) : <EOL> return self . _test_call_raising_error_1 ( <NUM_LIT:1> ) <EOL> def test_call_raising_error_2 ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , ( ) ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> msgs = muxer_messages ( mux ) <EOL> error_tuple = ( '<STR_LIT>' , '<STR_LIT:error>' ) <EOL> self . assertEquals ( len ( msgs ) , <NUM_LIT:1> ) <EOL> msg = msgs [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ <NUM_LIT:0> : <NUM_LIT:3> ] + msg [ <NUM_LIT:4> : ] , <EOL> ( <NUM_LIT:0> , chunks . MSG_COMMAND , <NUM_LIT:1> , False ) ) <EOL> args = msg [ <NUM_LIT:3> ] <EOL> self . assertEquals ( len ( args ) , <NUM_LIT:4> ) <EOL> self . assertEquals ( args [ : <NUM_LIT:3> ] , [ '<STR_LIT>' , <NUM_LIT:0.0> , None ] ) <EOL> self . assertIsInstance ( args [ <NUM_LIT:3> ] , Object ) <EOL> self . assertEquals ( ( args [ <NUM_LIT:3> ] . code , args [ <NUM_LIT:3> ] . level ) , <EOL> error_tuple ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> d . addCallback ( lambda _ : self . assertFalse ( t . disconnecting ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_call_raising_error_fatal ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , ( ) ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> msgs = muxer_messages ( mux ) <EOL> error_tuple = ( '<STR_LIT>' , '<STR_LIT:error>' ) <EOL> self . assertEquals ( len ( msgs ) , <NUM_LIT:1> ) <EOL> msg = msgs [ <NUM_LIT:0> ] <EOL> self . assertEquals ( msg [ <NUM_LIT:0> : <NUM_LIT:3> ] + msg [ <NUM_LIT:4> : ] , <EOL> ( <NUM_LIT:0> , chunks . MSG_COMMAND , <NUM_LIT:1> , False ) ) <EOL> args = msg [ <NUM_LIT:3> ] <EOL> self . assertEquals ( len ( args ) , <NUM_LIT:4> ) <EOL> self . assertEquals ( args [ : <NUM_LIT:3> ] , [ '<STR_LIT>' , <NUM_LIT:0.0> , None ] ) <EOL> self . assertIsInstance ( args [ <NUM_LIT:3> ] , Object ) <EOL> self . assertEquals ( ( args [ <NUM_LIT:3> ] . code , args [ <NUM_LIT:3> ] . level ) , <EOL> error_tuple ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> d . addCallback ( lambda _ : self . assertTrue ( t . disconnecting ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d <EOL> def test_call_aborted ( self ) : <EOL> p , t , dmx , mux = self . build_proto ( ) <EOL> dmx . inject ( <NUM_LIT:3> , <NUM_LIT:0> , const . RTMP_COMMAND , <NUM_LIT:1> , <EOL> encode_amf ( '<STR_LIT>' , <NUM_LIT:1> ) ) <EOL> def check_messages ( _result ) : <EOL> self . assertEquals ( self . messages , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:1> , ( ) ) ] ) <EOL> def verify_sent_messages ( _result ) : <EOL> self . assertEquals ( mux . messages , [ ] ) <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( check_messages ) <EOL> d . addCallback ( verify_sent_messages ) <EOL> d . addCallback ( lambda _ : self . assertFalse ( t . disconnecting ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , None ) <EOL> return d </s>
