<s> import copy <EOL> import mock <EOL> from oslo_serialization import jsonutils <EOL> from nova . cells import utils as cells_utils <EOL> from nova import compute <EOL> from nova import context <EOL> from nova import exception <EOL> from nova import objects <EOL> from nova import test <EOL> from nova . tests . unit import fake_notifier <EOL> from nova . tests . unit . objects import test_objects <EOL> from nova . tests . unit . objects import test_service <EOL> class ComputeHostAPITestCase ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ComputeHostAPITestCase , self ) . setUp ( ) <EOL> self . host_api = compute . HostAPI ( ) <EOL> self . ctxt = context . get_admin_context ( ) <EOL> fake_notifier . stub_notifier ( self . stubs ) <EOL> self . addCleanup ( fake_notifier . reset ) <EOL> def _compare_obj ( self , obj , db_obj ) : <EOL> test_objects . compare_obj ( self , obj , db_obj , <EOL> allow_missing = test_service . OPTIONAL ) <EOL> def _compare_objs ( self , obj_list , db_obj_list ) : <EOL> for index , obj in enumerate ( obj_list ) : <EOL> self . _compare_obj ( obj , db_obj_list [ index ] ) <EOL> def test_set_host_enabled ( self ) : <EOL> fake_notifier . NOTIFICATIONS = [ ] <EOL> @ mock . patch . object ( self . host_api . rpcapi , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_assert_host_exists , mock_set_host_enabled ) : <EOL> result = self . host_api . set_host_enabled ( self . ctxt , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( fake_notifier . NOTIFICATIONS ) ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , msg . event_type ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . publisher_id ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . priority ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , msg . event_type ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . publisher_id ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . priority ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> _do_test ( ) <EOL> def test_host_name_from_assert_hosts_exists ( self ) : <EOL> @ mock . patch . object ( self . host_api . rpcapi , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_assert_host_exists , mock_set_host_enabled ) : <EOL> result = self . host_api . set_host_enabled ( self . ctxt , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> _do_test ( ) <EOL> def test_get_host_uptime ( self ) : <EOL> @ mock . patch . object ( self . host_api . rpcapi , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_assert_host_exists , mock_get_host_uptime ) : <EOL> result = self . host_api . get_host_uptime ( self . ctxt , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> _do_test ( ) <EOL> def test_get_host_uptime_service_down ( self ) : <EOL> @ mock . patch . object ( self . host_api . db , '<STR_LIT>' , <EOL> return_value = dict ( test_service . fake_service , id = <NUM_LIT:1> ) ) <EOL> @ mock . patch . object ( self . host_api . servicegroup_api , '<STR_LIT>' , <EOL> return_value = False ) <EOL> def _do_test ( mock_service_is_up , mock_service_get_by_compute_host ) : <EOL> self . assertRaises ( exception . ComputeServiceUnavailable , <EOL> self . host_api . get_host_uptime , self . ctxt , <EOL> '<STR_LIT>' ) <EOL> _do_test ( ) <EOL> def test_host_power_action ( self ) : <EOL> fake_notifier . NOTIFICATIONS = [ ] <EOL> @ mock . patch . object ( self . host_api . rpcapi , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_assert_host_exists , mock_host_power_action ) : <EOL> result = self . host_api . host_power_action ( self . ctxt , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( fake_notifier . NOTIFICATIONS ) ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , msg . event_type ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . publisher_id ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . priority ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT:action>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , msg . event_type ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . publisher_id ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . priority ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT:action>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> _do_test ( ) <EOL> def test_set_host_maintenance ( self ) : <EOL> fake_notifier . NOTIFICATIONS = [ ] <EOL> @ mock . patch . object ( self . host_api . rpcapi , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_assert_host_exists , mock_host_maintenance_mode ) : <EOL> result = self . host_api . set_host_maintenance ( self . ctxt , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( fake_notifier . NOTIFICATIONS ) ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , msg . event_type ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . publisher_id ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . priority ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> msg = fake_notifier . NOTIFICATIONS [ <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , msg . event_type ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . publisher_id ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . priority ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , msg . payload [ '<STR_LIT>' ] ) <EOL> _do_test ( ) <EOL> def test_service_get_all_no_zones ( self ) : <EOL> services = [ dict ( test_service . fake_service , <EOL> id = <NUM_LIT:1> , topic = '<STR_LIT>' , host = '<STR_LIT>' ) , <EOL> dict ( test_service . fake_service , <EOL> topic = '<STR_LIT>' , host = '<STR_LIT>' ) ] <EOL> @ mock . patch . object ( self . host_api . db , '<STR_LIT>' ) <EOL> def _do_test ( mock_service_get_all ) : <EOL> mock_service_get_all . return_value = services <EOL> result = self . host_api . service_get_all ( self . ctxt ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> disabled = None ) <EOL> self . _compare_objs ( result , services ) <EOL> mock_service_get_all . reset_mock ( ) <EOL> result = self . host_api . service_get_all ( self . ctxt , filters = { } ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> disabled = None ) <EOL> self . _compare_objs ( result , services ) <EOL> mock_service_get_all . reset_mock ( ) <EOL> result = self . host_api . service_get_all ( self . ctxt , <EOL> filters = dict ( host = '<STR_LIT>' ) ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> disabled = None ) <EOL> self . _compare_objs ( result , [ services [ <NUM_LIT:1> ] ] ) <EOL> _do_test ( ) <EOL> def test_service_get_all ( self ) : <EOL> services = [ dict ( test_service . fake_service , <EOL> topic = '<STR_LIT>' , host = '<STR_LIT>' ) , <EOL> dict ( test_service . fake_service , <EOL> topic = '<STR_LIT>' , host = '<STR_LIT>' ) ] <EOL> exp_services = [ ] <EOL> for service in services : <EOL> exp_service = { } <EOL> exp_service . update ( availability_zone = '<STR_LIT>' , ** service ) <EOL> exp_services . append ( exp_service ) <EOL> @ mock . patch . object ( self . host_api . db , '<STR_LIT>' ) <EOL> def _do_test ( mock_service_get_all ) : <EOL> mock_service_get_all . return_value = services <EOL> result = self . host_api . service_get_all ( self . ctxt , set_zones = True ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> disabled = None ) <EOL> self . _compare_objs ( result , exp_services ) <EOL> mock_service_get_all . reset_mock ( ) <EOL> result = self . host_api . service_get_all ( self . ctxt , filters = { } , <EOL> set_zones = True ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> disabled = None ) <EOL> self . _compare_objs ( result , exp_services ) <EOL> mock_service_get_all . reset_mock ( ) <EOL> result = self . host_api . service_get_all ( self . ctxt , <EOL> filters = dict ( host = '<STR_LIT>' ) , <EOL> set_zones = True ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> disabled = None ) <EOL> self . _compare_objs ( result , [ exp_services [ <NUM_LIT:1> ] ] ) <EOL> mock_service_get_all . reset_mock ( ) <EOL> filters = { '<STR_LIT>' : '<STR_LIT>' } <EOL> result = self . host_api . service_get_all ( self . ctxt , <EOL> filters = filters ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> disabled = None ) <EOL> self . _compare_objs ( result , exp_services ) <EOL> _do_test ( ) <EOL> def test_service_get_by_compute_host ( self ) : <EOL> @ mock . patch . object ( self . host_api . db , '<STR_LIT>' , <EOL> return_value = test_service . fake_service ) <EOL> def _do_test ( mock_service_get_by_compute_host ) : <EOL> result = self . host_api . service_get_by_compute_host ( self . ctxt , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( test_service . fake_service [ '<STR_LIT:id>' ] , result . id ) <EOL> _do_test ( ) <EOL> def test_service_update ( self ) : <EOL> host_name = '<STR_LIT>' <EOL> binary = '<STR_LIT>' <EOL> params_to_update = dict ( disabled = True ) <EOL> service_id = <NUM_LIT> <EOL> expected_result = dict ( test_service . fake_service , id = service_id ) <EOL> @ mock . patch . object ( self . host_api . db , '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api . db , '<STR_LIT>' ) <EOL> def _do_test ( mock_service_update , mock_service_get_by_host_and_binary ) : <EOL> mock_service_get_by_host_and_binary . return_value = expected_result <EOL> mock_service_update . return_value = expected_result <EOL> result = self . host_api . service_update ( <EOL> self . ctxt , host_name , binary , params_to_update ) <EOL> self . _compare_obj ( result , expected_result ) <EOL> _do_test ( ) <EOL> @ mock . patch . object ( objects . InstanceList , '<STR_LIT>' , <EOL> return_value = [ '<STR_LIT>' ] ) <EOL> def test_instance_get_all_by_host ( self , mock_get ) : <EOL> result = self . host_api . instance_get_all_by_host ( self . ctxt , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] , result ) <EOL> def test_task_log_get_all ( self ) : <EOL> @ mock . patch . object ( self . host_api . db , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_task_log_get_all ) : <EOL> result = self . host_api . task_log_get_all ( self . ctxt , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> host = '<STR_LIT>' , <EOL> state = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> _do_test ( ) <EOL> def test_service_delete ( self ) : <EOL> with test . nested ( <EOL> mock . patch . object ( objects . Service , '<STR_LIT>' , <EOL> return_value = objects . Service ( ) ) , <EOL> mock . patch . object ( objects . Service , '<STR_LIT>' ) <EOL> ) as ( <EOL> get_by_id , destroy <EOL> ) : <EOL> self . host_api . service_delete ( self . ctxt , <NUM_LIT:1> ) <EOL> get_by_id . assert_called_once_with ( self . ctxt , <NUM_LIT:1> ) <EOL> destroy . assert_called_once_with ( ) <EOL> class ComputeHostAPICellsTestCase ( ComputeHostAPITestCase ) : <EOL> def setUp ( self ) : <EOL> self . flags ( enable = True , group = '<STR_LIT>' ) <EOL> self . flags ( cell_type = '<STR_LIT>' , group = '<STR_LIT>' ) <EOL> super ( ComputeHostAPICellsTestCase , self ) . setUp ( ) <EOL> def test_service_get_all_no_zones ( self ) : <EOL> services = [ <EOL> cells_utils . ServiceProxy ( <EOL> objects . Service ( id = <NUM_LIT:1> , topic = '<STR_LIT>' , host = '<STR_LIT>' ) , <EOL> '<STR_LIT>' ) , <EOL> cells_utils . ServiceProxy ( <EOL> objects . Service ( id = <NUM_LIT:2> , topic = '<STR_LIT>' , host = '<STR_LIT>' ) , <EOL> '<STR_LIT>' ) ] <EOL> fake_filters = { '<STR_LIT:host>' : '<STR_LIT>' } <EOL> @ mock . patch . object ( self . host_api . cells_rpcapi , '<STR_LIT>' ) <EOL> def _do_test ( mock_service_get_all ) : <EOL> mock_service_get_all . return_value = services <EOL> result = self . host_api . service_get_all ( self . ctxt , <EOL> filters = fake_filters ) <EOL> self . assertEqual ( services , result ) <EOL> _do_test ( ) <EOL> def _test_service_get_all ( self , fake_filters , ** kwargs ) : <EOL> service_attrs = dict ( test_service . fake_service ) <EOL> del service_attrs [ '<STR_LIT:version>' ] <EOL> services = [ <EOL> cells_utils . ServiceProxy ( <EOL> objects . Service ( ** dict ( service_attrs , id = <NUM_LIT:1> , <EOL> topic = '<STR_LIT>' , host = '<STR_LIT>' ) ) , <EOL> '<STR_LIT>' ) , <EOL> cells_utils . ServiceProxy ( <EOL> objects . Service ( ** dict ( service_attrs , id = <NUM_LIT:2> , <EOL> topic = '<STR_LIT>' , host = '<STR_LIT>' ) ) , <EOL> '<STR_LIT>' ) ] <EOL> exp_services = [ ] <EOL> for service in services : <EOL> exp_service = copy . copy ( service ) <EOL> exp_service . update ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> exp_services . append ( exp_service ) <EOL> @ mock . patch . object ( self . host_api . cells_rpcapi , '<STR_LIT>' ) <EOL> def _do_test ( mock_service_get_all ) : <EOL> mock_service_get_all . return_value = services <EOL> result = self . host_api . service_get_all ( self . ctxt , <EOL> filters = fake_filters , <EOL> ** kwargs ) <EOL> mock_service_get_all . assert_called_once_with ( self . ctxt , <EOL> filters = fake_filters ) <EOL> self . assertEqual ( jsonutils . to_primitive ( exp_services ) , <EOL> jsonutils . to_primitive ( result ) ) <EOL> _do_test ( ) <EOL> def test_service_get_all ( self ) : <EOL> fake_filters = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _test_service_get_all ( fake_filters ) <EOL> def test_service_get_all_set_zones ( self ) : <EOL> fake_filters = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _test_service_get_all ( fake_filters , set_zones = True ) <EOL> def test_service_get_by_compute_host ( self ) : <EOL> obj = objects . Service ( id = <NUM_LIT:1> , host = '<STR_LIT>' ) <EOL> fake_service = cells_utils . ServiceProxy ( obj , '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api . cells_rpcapi , <EOL> '<STR_LIT>' ) <EOL> def _do_test ( mock_service_get_by_compute_host ) : <EOL> mock_service_get_by_compute_host . return_value = fake_service <EOL> result = self . host_api . service_get_by_compute_host ( self . ctxt , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( fake_service , result ) <EOL> _do_test ( ) <EOL> def test_service_update ( self ) : <EOL> host_name = '<STR_LIT>' <EOL> binary = '<STR_LIT>' <EOL> params_to_update = dict ( disabled = True ) <EOL> obj = objects . Service ( id = <NUM_LIT> , host = '<STR_LIT>' ) <EOL> fake_service = cells_utils . ServiceProxy ( obj , '<STR_LIT>' ) <EOL> @ mock . patch . object ( self . host_api . cells_rpcapi , '<STR_LIT>' ) <EOL> def _do_test ( mock_service_update ) : <EOL> mock_service_update . return_value = fake_service <EOL> result = self . host_api . service_update ( <EOL> self . ctxt , host_name , binary , params_to_update ) <EOL> self . assertEqual ( fake_service , result ) <EOL> _do_test ( ) <EOL> def test_service_delete ( self ) : <EOL> cell_service_id = cells_utils . cell_with_item ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> with mock . patch . object ( self . host_api . cells_rpcapi , <EOL> '<STR_LIT>' ) as service_delete : <EOL> self . host_api . service_delete ( self . ctxt , cell_service_id ) <EOL> service_delete . assert_called_once_with ( <EOL> self . ctxt , cell_service_id ) <EOL> @ mock . patch . object ( objects . InstanceList , '<STR_LIT>' ) <EOL> def test_instance_get_all_by_host ( self , mock_get ) : <EOL> instances = [ dict ( id = <NUM_LIT:1> , cell_name = '<STR_LIT>' , host = '<STR_LIT>' ) , <EOL> dict ( id = <NUM_LIT:2> , cell_name = '<STR_LIT>' , host = '<STR_LIT>' ) , <EOL> dict ( id = <NUM_LIT:3> , cell_name = '<STR_LIT>' , host = '<STR_LIT>' ) ] <EOL> mock_get . return_value = instances <EOL> expected_result = [ instances [ <NUM_LIT:0> ] , instances [ <NUM_LIT:2> ] ] <EOL> cell_and_host = cells_utils . cell_with_item ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> result = self . host_api . instance_get_all_by_host ( self . ctxt , <EOL> cell_and_host ) <EOL> self . assertEqual ( expected_result , result ) <EOL> def test_task_log_get_all ( self ) : <EOL> @ mock . patch . object ( self . host_api . cells_rpcapi , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_task_log_get_all ) : <EOL> result = self . host_api . task_log_get_all ( self . ctxt , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> host = '<STR_LIT>' , <EOL> state = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> _do_test ( ) <EOL> def test_get_host_uptime_service_down ( self ) : <EOL> pass <EOL> def test_get_host_uptime ( self ) : <EOL> @ mock . patch . object ( self . host_api . cells_rpcapi , '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) <EOL> def _do_test ( mock_get_host_uptime ) : <EOL> result = self . host_api . get_host_uptime ( self . ctxt , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> _do_test ( ) </s>
