<s> __author__ = '<STR_LIT>' <EOL> from nose . plugins . attrib import attr <EOL> from mock import Mock , MagicMock <EOL> import unittest <EOL> from pyon . core . governance . policy . policy_decision import PolicyDecisionPointManager <EOL> from pyon . core . exception import NotFound <EOL> from pyon . util . unit_test import PyonTestCase <EOL> @ attr ( '<STR_LIT>' ) <EOL> class PolicyDecisionUnitTest ( PyonTestCase ) : <EOL> permit_ION_MANAGER_rule = '''<STR_LIT>''' <EOL> deny_ION_MANAGER_rule = '''<STR_LIT>''' <EOL> deny_message_parameter_rule = '''<STR_LIT>''' <EOL> deny_message_parameter_function_rule = '''<STR_LIT>''' <EOL> def test_resource_policies ( self ) : <EOL> gc = Mock ( ) <EOL> resource_id = '<STR_LIT>' <EOL> pdpm = PolicyDecisionPointManager ( gc ) <EOL> self . assertEqual ( pdpm . get_resource_pdp ( resource_id ) , pdpm . empty_pdp ) <EOL> pdpm . load_resource_policy_rules ( resource_id , self . permit_ION_MANAGER_rule ) <EOL> self . assertNotEqual ( pdpm . get_resource_pdp ( resource_id ) , pdpm . empty_pdp ) <EOL> invocation = MagicMock ( ) <EOL> invocation . message_annotations = { } <EOL> with self . assertRaises ( NotFound ) as chk_res : <EOL> pdpm . check_resource_request_policies ( invocation , None ) <EOL> self . assertIn ( chk_res . exception . message , '<STR_LIT>' ) <EOL> mock_header = Mock ( ) <EOL> def get_header_value ( key , default ) : <EOL> return invocation . headers . get ( key , default ) <EOL> mock_header . side_effect = get_header_value <EOL> invocation . get_header_value = mock_header <EOL> mock_args = Mock ( ) <EOL> class MockProcess ( Mock ) : <EOL> def check_test_value ( self , process , message , headers ) : <EOL> if message [ '<STR_LIT>' ] > <NUM_LIT:3> : <EOL> return False , '<STR_LIT>' <EOL> return True , '<STR_LIT>' <EOL> mock_process = MockProcess ( ) <EOL> mock_process . org_governance_name = '<STR_LIT>' <EOL> invocation . args = { '<STR_LIT>' : mock_process } <EOL> def get_arg_value ( key , default ) : <EOL> return invocation . args . get ( key , default ) <EOL> mock_args . side_effect = get_arg_value <EOL> invocation . get_arg_value = mock_args <EOL> invocation . message_annotations = { } <EOL> invocation . message = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : mock_process , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . get_message_sender . return_value = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> gc . system_root_org_name = '<STR_LIT>' <EOL> response = pdpm . check_resource_request_policies ( invocation , resource_id ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : mock_process , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . message_annotations = { } <EOL> response = pdpm . check_resource_request_policies ( invocation , resource_id ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> pdpm . load_resource_policy_rules ( resource_id , self . deny_message_parameter_rule ) <EOL> invocation . message = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : mock_process , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . message_annotations = { } <EOL> response = pdpm . check_resource_request_policies ( invocation , resource_id ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> self . assertEqual ( invocation . message_annotations . has_key ( '<STR_LIT>' ) , True ) <EOL> self . assertEqual ( invocation . message_annotations [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> invocation . message = { '<STR_LIT>' : <NUM_LIT:5> } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : mock_process , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . message_annotations = { } <EOL> response = pdpm . check_resource_request_policies ( invocation , resource_id ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> self . assertEqual ( invocation . message_annotations . has_key ( '<STR_LIT>' ) , False ) <EOL> pdpm . load_resource_policy_rules ( resource_id , self . deny_message_parameter_function_rule ) <EOL> invocation . message = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : mock_process , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . message_annotations = { } <EOL> response = pdpm . check_resource_request_policies ( invocation , resource_id ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> self . assertEqual ( invocation . message_annotations . has_key ( '<STR_LIT>' ) , False ) <EOL> invocation . message = { '<STR_LIT>' : <NUM_LIT:5> } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : mock_process , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . message_annotations = { } <EOL> response = pdpm . check_resource_request_policies ( invocation , resource_id ) <EOL> self . assertEqual ( invocation . message_annotations . has_key ( '<STR_LIT>' ) , True ) <EOL> self . assertEqual ( invocation . message_annotations [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_service_policies ( self ) : <EOL> gc = Mock ( ) <EOL> service_key = '<STR_LIT>' <EOL> pdpm = PolicyDecisionPointManager ( gc ) <EOL> self . assertEqual ( pdpm . get_service_pdp ( service_key ) , pdpm . load_common_service_pdp ) <EOL> pdpm . load_service_policy_rules ( service_key , self . permit_ION_MANAGER_rule ) <EOL> self . assertNotEqual ( pdpm . get_service_pdp ( service_key ) , pdpm . load_common_service_pdp ) <EOL> invocation = Mock ( ) <EOL> invocation . message_annotations = { } <EOL> invocation . get_message_receiver . return_value = None <EOL> with self . assertRaises ( NotFound ) as chk_res : <EOL> pdpm . check_service_request_policies ( invocation ) <EOL> self . assertIn ( chk_res . exception . message , '<STR_LIT>' ) <EOL> mock_header = Mock ( ) <EOL> invocation . message_annotations = { } <EOL> invocation . message = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . get_message_receiver . return_value = '<STR_LIT>' <EOL> invocation . get_service_name . return_value = '<STR_LIT>' <EOL> invocation . get_message_sender . return_value = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> def get_header_value ( key , default ) : <EOL> return invocation . headers . get ( key , default ) <EOL> mock_header . side_effect = get_header_value <EOL> invocation . get_header_value = mock_header <EOL> mock_args = Mock ( ) <EOL> process = Mock ( ) <EOL> process . org_governance_name = '<STR_LIT>' <EOL> invocation . args = { '<STR_LIT>' : process } <EOL> def get_arg_value ( key , default ) : <EOL> return invocation . args . get ( key , default ) <EOL> mock_args . side_effect = get_arg_value <EOL> invocation . get_arg_value = mock_args <EOL> gc . system_root_org_name = '<STR_LIT>' <EOL> response = pdpm . check_service_request_policies ( invocation ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> invocation . message_annotations = { } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> response = pdpm . check_service_request_policies ( invocation ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> def test_agent_policies ( self ) : <EOL> gc = Mock ( ) <EOL> service_key = '<STR_LIT>' <EOL> resource_id = '<STR_LIT>' <EOL> pdpm = PolicyDecisionPointManager ( gc ) <EOL> invocation = Mock ( ) <EOL> mock_header = Mock ( ) <EOL> invocation . message_annotations = { } <EOL> invocation . message = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> invocation . headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> invocation . get_message_receiver . return_value = '<STR_LIT>' <EOL> invocation . get_service_name . return_value = '<STR_LIT>' <EOL> invocation . get_message_sender . return_value = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> def get_header_value ( key , default ) : <EOL> return invocation . headers . get ( key , default ) <EOL> mock_header . side_effect = get_header_value <EOL> invocation . get_header_value = mock_header <EOL> mock_args = Mock ( ) <EOL> process = Mock ( ) <EOL> process . org_governance_name = '<STR_LIT>' <EOL> process . resource_id = '<STR_LIT>' <EOL> invocation . args = { '<STR_LIT>' : process } <EOL> def get_arg_value ( key , default = '<STR_LIT>' ) : <EOL> return invocation . args . get ( key , default ) <EOL> mock_args . side_effect = get_arg_value <EOL> invocation . get_arg_value = mock_args <EOL> gc . system_root_org_name = '<STR_LIT>' <EOL> pdpm . load_service_policy_rules ( service_key , self . deny_ION_MANAGER_rule ) <EOL> pdpm . load_resource_policy_rules ( resource_id , self . permit_ION_MANAGER_rule ) <EOL> response = pdpm . check_agent_request_policies ( invocation ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> pdpm . load_service_policy_rules ( service_key , self . permit_ION_MANAGER_rule ) <EOL> pdpm . load_resource_policy_rules ( resource_id , self . deny_ION_MANAGER_rule ) <EOL> response = pdpm . check_agent_request_policies ( invocation ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) <EOL> pdpm . load_service_policy_rules ( service_key , self . permit_ION_MANAGER_rule ) <EOL> pdpm . load_resource_policy_rules ( resource_id , self . permit_ION_MANAGER_rule ) <EOL> response = pdpm . check_agent_request_policies ( invocation ) <EOL> self . assertEqual ( response . value , "<STR_LIT>" ) </s>
