<s> import json <EOL> import mock <EOL> from oslo_service import loopingcall <EOL> from oslotest import base as test_base <EOL> from ironic_python_agent import errors <EOL> from ironic_python_agent import hardware <EOL> from ironic_python_agent import ironic_api_client <EOL> API_URL = '<STR_LIT>' <EOL> DRIVER = '<STR_LIT>' <EOL> class FakeResponse ( object ) : <EOL> def __init__ ( self , content = None , status_code = <NUM_LIT:200> , headers = None ) : <EOL> content = content or { } <EOL> self . content = json . dumps ( content ) <EOL> self . status_code = status_code <EOL> self . headers = headers or { } <EOL> class TestBaseIronicPythonAgent ( test_base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestBaseIronicPythonAgent , self ) . setUp ( ) <EOL> self . api_client = ironic_api_client . APIClient ( API_URL , DRIVER ) <EOL> self . hardware_info = { <EOL> '<STR_LIT>' : [ <EOL> hardware . NetworkInterface ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> hardware . NetworkInterface ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] , <EOL> '<STR_LIT>' : hardware . CPU ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : [ <EOL> hardware . BlockDevice ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , False ) , <EOL> hardware . BlockDevice ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , False ) , <EOL> ] , <EOL> '<STR_LIT>' : hardware . Memory ( total = '<STR_LIT>' , <EOL> physical_mb = '<STR_LIT>' ) , <EOL> } <EOL> def test_successful_heartbeat ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT> ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> self . api_client . heartbeat ( <EOL> uuid = '<STR_LIT>' , <EOL> advertise_address = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ) <EOL> heartbeat_path = '<STR_LIT>' '<STR_LIT>' <EOL> request_args = self . api_client . session . request . call_args [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT:POST>' , request_args [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( API_URL + heartbeat_path , request_args [ <NUM_LIT:1> ] ) <EOL> def test_heartbeat_requests_exception ( self ) : <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . side_effect = Exception ( '<STR_LIT>' ) <EOL> self . assertRaises ( errors . HeartbeatError , <EOL> self . api_client . heartbeat , <EOL> uuid = '<STR_LIT>' , <EOL> advertise_address = ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_heartbeat_invalid_status_code ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT> ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> self . assertRaises ( errors . HeartbeatError , <EOL> self . api_client . heartbeat , <EOL> uuid = '<STR_LIT>' , <EOL> advertise_address = ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_heartbeat_409_status_code ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT> ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> self . assertRaises ( errors . HeartbeatConflictError , <EOL> self . api_client . heartbeat , <EOL> uuid = '<STR_LIT>' , <EOL> advertise_address = ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_lookup_node ( self , lookup_mock , sleep_mock ) : <EOL> content = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> '<STR_LIT>' : <NUM_LIT> <EOL> } <EOL> lookup_mock . side_effect = loopingcall . LoopingCallDone ( <EOL> retvalue = content ) <EOL> returned_content = self . api_client . lookup_node ( <EOL> hardware_info = self . hardware_info , <EOL> timeout = <NUM_LIT> , <EOL> starting_interval = <NUM_LIT:1> ) <EOL> self . assertEqual ( content , returned_content ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_lookup_timeout ( self , lookup_mock , sleep_mock ) : <EOL> lookup_mock . side_effect = loopingcall . LoopingCallTimeOut ( ) <EOL> self . assertRaises ( errors . LookupNodeError , <EOL> self . api_client . lookup_node , <EOL> hardware_info = self . hardware_info , <EOL> timeout = <NUM_LIT> , <EOL> starting_interval = <NUM_LIT:1> ) <EOL> def test_do_lookup ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT:200> , content = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> '<STR_LIT>' : <NUM_LIT> <EOL> } ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> self . assertRaises ( loopingcall . LoopingCallDone , <EOL> self . api_client . _do_lookup , <EOL> hardware_info = self . hardware_info , <EOL> node_uuid = None ) <EOL> url = '<STR_LIT>' . format ( <EOL> api_url = API_URL , driver = DRIVER ) <EOL> request_args = self . api_client . session . request . call_args [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT:POST>' , request_args [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( url , request_args [ <NUM_LIT:1> ] ) <EOL> data = self . api_client . session . request . call_args [ <NUM_LIT:1> ] [ '<STR_LIT:data>' ] <EOL> content = json . loads ( data ) <EOL> self . assertNotIn ( '<STR_LIT>' , content ) <EOL> self . assertEqual ( self . api_client . payload_version , content [ '<STR_LIT:version>' ] ) <EOL> self . assertEqual ( { <EOL> u'<STR_LIT>' : [ <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : True , <EOL> } , <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : True , <EOL> } <EOL> ] , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:count>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } , <EOL> u'<STR_LIT>' : [ <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT:size>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> } , <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT:size>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> } <EOL> ] , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } , <EOL> } , content [ '<STR_LIT>' ] ) <EOL> def test_do_lookup_bad_response_code ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT> , content = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> error = self . api_client . _do_lookup ( self . hardware_info , <EOL> node_uuid = None ) <EOL> self . assertFalse ( error ) <EOL> def test_do_lookup_bad_response_data ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT:200> , content = { <EOL> '<STR_LIT>' : <NUM_LIT> <EOL> } ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> error = self . api_client . _do_lookup ( self . hardware_info , <EOL> node_uuid = None ) <EOL> self . assertFalse ( error ) <EOL> def test_do_lookup_no_heartbeat_timeout ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT:200> , content = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> error = self . api_client . _do_lookup ( self . hardware_info , <EOL> node_uuid = None ) <EOL> self . assertFalse ( error ) <EOL> def test_do_lookup_bad_response_body ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT:200> , content = { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> error = self . api_client . _do_lookup ( self . hardware_info , <EOL> node_uuid = None ) <EOL> self . assertFalse ( error ) <EOL> def test_do_lookup_with_node_uuid ( self ) : <EOL> response = FakeResponse ( status_code = <NUM_LIT:200> , content = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> '<STR_LIT>' : <NUM_LIT> <EOL> } ) <EOL> self . api_client . session . request = mock . Mock ( ) <EOL> self . api_client . session . request . return_value = response <EOL> self . assertRaises ( loopingcall . LoopingCallDone , <EOL> self . api_client . _do_lookup , <EOL> hardware_info = self . hardware_info , <EOL> node_uuid = '<STR_LIT>' ) <EOL> url = '<STR_LIT>' . format ( <EOL> api_url = API_URL , driver = DRIVER ) <EOL> request_args = self . api_client . session . request . call_args [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT:POST>' , request_args [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( url , request_args [ <NUM_LIT:1> ] ) <EOL> data = self . api_client . session . request . call_args [ <NUM_LIT:1> ] [ '<STR_LIT:data>' ] <EOL> content = json . loads ( data ) <EOL> self . assertEqual ( '<STR_LIT>' , content [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . api_client . payload_version , content [ '<STR_LIT:version>' ] ) <EOL> self . assertEqual ( { <EOL> u'<STR_LIT>' : [ <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : True , <EOL> } , <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : True , <EOL> } <EOL> ] , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:count>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } , <EOL> u'<STR_LIT>' : [ <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT:size>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> } , <EOL> { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT:size>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : None , <EOL> } <EOL> ] , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } , <EOL> } , content [ '<STR_LIT>' ] ) </s>
