#NEWLINE# @ patch ( 'pyon.net.endpoint.RPCResponseEndpointUnit._message_received' ) #NEWLINE# def test_message_received( self , mockmr ) : #NEWLINE# #INDENT# procmock = Mock ( ) #NEWLINE# procmock . push_context = MagicMock ( ) #NEWLINE# ep = ProcessRPCResponseEndpointUnit ( process = procmock , interceptors = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# ep . _routing_obj = cm #NEWLINE# msg_dict = { #NEWLINE# #INDENT# 'iam' : 'adict' , #NEWLINE# #UNINDENT# } #NEWLINE# header_dict = { #NEWLINE# #INDENT# 'op' : 'anyop' , #NEWLINE# #UNINDENT# } #NEWLINE# ep . message_received ( msg_dict , header_dict ) #NEWLINE# ep . _routing_obj . anyop . assert_called_once_with ( iam = 'adict' ) #NEWLINE# def deny_anyop( self , operation , id = None ) : #NEWLINE# #INDENT# raise Unauthorized ( 'The anyop operation has been denied' ) #NEWLINE# #UNINDENT# msg_dict2 = { #NEWLINE# #INDENT# 'iam' : 'adict2' , #NEWLINE# #UNINDENT# } #NEWLINE# ep . _routing_obj . _service_op_preconditions = { #NEWLINE# #INDENT# 'anyop' : 'deny_anyop' , #NEWLINE# #UNINDENT# } #NEWLINE# ep . _routing_obj . container . governance_controller . check_process_operation_preconditions = deny_anyop #NEWLINE# with self . assertRaises ( Unauthorized ) as cm : #NEWLINE# #INDENT# ep . message_received ( msg_dict2 , header_dict ) #NEWLINE# #UNINDENT# self . assertIn ( 'The anyop operation has been denied' , cm . exception . message ) #NEWLINE# ep . _routing_obj . anyop . assert_called_once_with ( iam = 'adict' )