<s> import copy <EOL> import datetime as dt <EOL> from dateutil import parser <EOL> import mock <EOL> from rally import exceptions <EOL> from rally . plugins . openstack . scenarios . ceilometer import utils <EOL> from tests . unit import test <EOL> CEILOMETER_UTILS = "<STR_LIT>" <EOL> class CeilometerScenarioTestCase ( test . ScenarioTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( CeilometerScenarioTestCase , self ) . setUp ( ) <EOL> self . scenario = utils . CeilometerScenario ( self . context ) <EOL> def test__make_samples_no_batch_size ( self ) : <EOL> self . scenario . generate_random_name = mock . Mock ( <EOL> return_value = "<STR_LIT>" ) <EOL> test_timestamp = dt . datetime ( <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT:20> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> result = list ( self . scenario . _make_samples ( count = <NUM_LIT:2> , interval = <NUM_LIT> , <EOL> timestamp = test_timestamp ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( result ) ) <EOL> expected = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:%>" , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : test_timestamp . isoformat ( ) } <EOL> self . assertEqual ( expected , result [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> samples_int = ( parser . parse ( result [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ) - <EOL> parser . parse ( result [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ "<STR_LIT>" ] ) ) . seconds <EOL> self . assertEqual ( <NUM_LIT> , samples_int ) <EOL> def test__make_samples_batch_size ( self ) : <EOL> self . scenario . generate_random_name = mock . Mock ( <EOL> return_value = "<STR_LIT>" ) <EOL> test_timestamp = dt . datetime ( <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT:20> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> result = list ( self . scenario . _make_samples ( count = <NUM_LIT:4> , interval = <NUM_LIT> , <EOL> batch_size = <NUM_LIT:2> , <EOL> timestamp = test_timestamp ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( result ) ) <EOL> expected = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:%>" , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : test_timestamp . isoformat ( ) } <EOL> self . assertEqual ( expected , result [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> samples_int = ( parser . parse ( result [ <NUM_LIT:0> ] [ - <NUM_LIT:1> ] [ "<STR_LIT>" ] ) - <EOL> parser . parse ( result [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ) ) . seconds <EOL> self . assertEqual ( <NUM_LIT> , samples_int ) <EOL> def test__make_timestamp_query ( self ) : <EOL> start_time = "<STR_LIT>" <EOL> end_time = "<STR_LIT>" <EOL> expected_start = [ <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] <EOL> expected_end = [ <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> ] <EOL> actual = self . scenario . _make_timestamp_query ( start_time , end_time ) <EOL> self . assertEqual ( expected_start + expected_end , actual ) <EOL> self . assertRaises ( exceptions . InvalidArgumentsException , <EOL> self . scenario . _make_timestamp_query , <EOL> end_time , start_time ) <EOL> self . assertEqual ( <EOL> expected_start , <EOL> self . scenario . _make_timestamp_query ( start_time = start_time ) ) <EOL> self . assertEqual ( <EOL> expected_end , <EOL> self . scenario . _make_timestamp_query ( end_time = end_time ) ) <EOL> def test__list_alarms_by_id ( self ) : <EOL> self . assertEqual ( self . clients ( "<STR_LIT>" ) . alarms . get . return_value , <EOL> self . scenario . _list_alarms ( "<STR_LIT>" ) ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . get . assert_called_once_with ( <EOL> "<STR_LIT>" ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_alarms ( self ) : <EOL> self . assertEqual ( self . clients ( "<STR_LIT>" ) . alarms . list . return_value , <EOL> self . scenario . _list_alarms ( ) ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . list . assert_called_once_with ( ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__create_alarm ( self ) : <EOL> alarm_dict = { "<STR_LIT>" : "<STR_LIT>" } <EOL> orig_alarm_dict = copy . copy ( alarm_dict ) <EOL> self . scenario . generate_random_name = mock . Mock ( ) <EOL> self . assertEqual ( self . scenario . _create_alarm ( "<STR_LIT>" , <NUM_LIT:100> , <EOL> alarm_dict ) , <EOL> self . clients ( "<STR_LIT>" ) . alarms . create . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . create . assert_called_once_with ( <EOL> meter_name = "<STR_LIT>" , <EOL> threshold = <NUM_LIT:100> , <EOL> description = "<STR_LIT>" , <EOL> alarm_id = "<STR_LIT>" , <EOL> name = self . scenario . generate_random_name . return_value ) <EOL> self . assertDictEqual ( alarm_dict , orig_alarm_dict ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__delete_alarms ( self ) : <EOL> self . scenario . _delete_alarm ( "<STR_LIT>" ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . delete . assert_called_once_with ( <EOL> "<STR_LIT>" ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__update_alarm ( self ) : <EOL> alarm_diff = { "<STR_LIT:description>" : "<STR_LIT>" } <EOL> orig_alarm_diff = copy . copy ( alarm_diff ) <EOL> self . scenario . _update_alarm ( "<STR_LIT>" , alarm_diff ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . update . assert_called_once_with ( <EOL> "<STR_LIT>" , ** alarm_diff ) <EOL> self . assertDictEqual ( alarm_diff , orig_alarm_diff ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__get_alarm_history ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _get_alarm_history ( "<STR_LIT>" ) , <EOL> self . clients ( "<STR_LIT>" ) . alarms . get_history . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . get_history . assert_called_once_with ( <EOL> "<STR_LIT>" ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__get_alarm_state ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _get_alarm_state ( "<STR_LIT>" ) , <EOL> self . clients ( "<STR_LIT>" ) . alarms . get_state . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . get_state . assert_called_once_with ( <EOL> "<STR_LIT>" ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__set_alarm_state ( self ) : <EOL> alarm = mock . Mock ( ) <EOL> self . clients ( "<STR_LIT>" ) . alarms . create . return_value = alarm <EOL> return_alarm = self . scenario . _set_alarm_state ( alarm , "<STR_LIT>" , <NUM_LIT:100> ) <EOL> self . mock_wait_for . mock . assert_called_once_with ( <EOL> alarm , <EOL> ready_statuses = [ "<STR_LIT>" ] , <EOL> update_resource = self . mock_get_from_manager . mock . return_value , <EOL> timeout = <NUM_LIT:100> , check_interval = <NUM_LIT:1> ) <EOL> self . mock_get_from_manager . mock . assert_called_once_with ( ) <EOL> self . assertEqual ( self . mock_wait_for . mock . return_value , return_alarm ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_events ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _list_events ( ) , <EOL> self . admin_clients ( "<STR_LIT>" ) . events . list . return_value <EOL> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__get_events ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _get_event ( event_id = "<STR_LIT>" ) , <EOL> self . admin_clients ( "<STR_LIT>" ) . events . get . return_value <EOL> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_event_types ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _list_event_types ( ) , <EOL> self . admin_clients ( "<STR_LIT>" ) . event_types . list . return_value <EOL> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_event_traits ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _list_event_traits ( <EOL> event_type = "<STR_LIT>" , trait_name = "<STR_LIT>" ) , <EOL> self . admin_clients ( "<STR_LIT>" ) . traits . list . return_value <EOL> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_event_trait_descriptions ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _list_event_trait_descriptions ( <EOL> event_type = "<STR_LIT>" <EOL> ) , <EOL> self . admin_clients ( "<STR_LIT>" ) . trait_descriptions . list . <EOL> return_value <EOL> ) <EOL> self . _test_atomic_action_timer ( <EOL> self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_meters ( self ) : <EOL> self . assertEqual ( self . scenario . _list_meters ( ) , <EOL> self . clients ( "<STR_LIT>" ) . meters . list . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . meters . list . assert_called_once_with ( <EOL> q = None , limit = None ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_resources ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _list_resources ( ) , <EOL> self . clients ( "<STR_LIT>" ) . resources . list . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . resources . list . assert_called_once_with ( <EOL> q = None , limit = None ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_samples ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _list_samples ( ) , <EOL> self . clients ( "<STR_LIT>" ) . samples . list . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . samples . list . assert_called_once_with ( <EOL> q = None , limit = None ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__list_samples_with_query ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _list_samples ( query = [ { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] , <EOL> limit = <NUM_LIT:10> ) , <EOL> self . clients ( "<STR_LIT>" ) . samples . list . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . samples . list . assert_called_once_with ( <EOL> q = [ { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ] , limit = <NUM_LIT:10> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__get_resource ( self ) : <EOL> self . assertEqual ( self . scenario . _get_resource ( "<STR_LIT>" ) , <EOL> self . clients ( "<STR_LIT>" ) . resources . get . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . resources . get . assert_called_once_with ( <EOL> "<STR_LIT>" ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__get_stats ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _get_stats ( "<STR_LIT>" ) , <EOL> self . clients ( "<STR_LIT>" ) . statistics . list . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . statistics . list . assert_called_once_with ( <EOL> "<STR_LIT>" , q = None , period = None , groupby = None , aggregates = None ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__create_meter ( self ) : <EOL> self . scenario . generate_random_name = mock . Mock ( ) <EOL> self . assertEqual ( <EOL> self . scenario . _create_meter ( fakearg = "<STR_LIT>" ) , <EOL> self . clients ( "<STR_LIT>" ) . samples . create . return_value [ <NUM_LIT:0> ] ) <EOL> self . clients ( "<STR_LIT>" ) . samples . create . assert_called_once_with ( <EOL> counter_name = self . scenario . generate_random_name . return_value , <EOL> fakearg = "<STR_LIT>" ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__query_alarms ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _query_alarms ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:10> ) , <EOL> self . clients ( "<STR_LIT>" ) . query_alarms . query . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . query_alarms . query . assert_called_once_with ( <EOL> "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:10> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__query_alarm_history ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _query_alarm_history ( <EOL> "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:10> ) , <EOL> self . clients ( "<STR_LIT>" ) . query_alarm_history . query . return_value ) <EOL> self . clients ( <EOL> "<STR_LIT>" ) . query_alarm_history . query . assert_called_once_with ( <EOL> "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:10> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__query_samples ( self ) : <EOL> self . assertEqual ( <EOL> self . scenario . _query_samples ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:10> ) , <EOL> self . clients ( "<STR_LIT>" ) . query_samples . query . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . query_samples . query . assert_called_once_with ( <EOL> "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:10> ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__create_sample_no_resource_id ( self ) : <EOL> self . scenario . generate_random_name = mock . Mock ( ) <EOL> created_sample = self . scenario . _create_sample ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> created_sample , <EOL> self . clients ( "<STR_LIT>" ) . samples . create . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . samples . create . assert_called_once_with ( <EOL> counter_name = "<STR_LIT>" , <EOL> counter_type = "<STR_LIT>" , <EOL> counter_unit = "<STR_LIT>" , <EOL> counter_volume = "<STR_LIT>" , <EOL> resource_id = self . scenario . generate_random_name . return_value ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__create_sample ( self ) : <EOL> created_sample = self . scenario . _create_sample ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> created_sample , <EOL> self . clients ( "<STR_LIT>" ) . samples . create . return_value ) <EOL> self . clients ( "<STR_LIT>" ) . samples . create . assert_called_once_with ( <EOL> counter_name = "<STR_LIT>" , <EOL> counter_type = "<STR_LIT>" , <EOL> counter_unit = "<STR_LIT>" , <EOL> counter_volume = "<STR_LIT>" , <EOL> resource_id = "<STR_LIT>" ) <EOL> self . _test_atomic_action_timer ( self . scenario . atomic_actions ( ) , <EOL> "<STR_LIT>" ) <EOL> def test__make_general_query ( self ) : <EOL> self . scenario . context = { <EOL> "<STR_LIT:user>" : { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" , "<STR_LIT>" : [ "<STR_LIT>" ] } } <EOL> metadata = { "<STR_LIT>" : "<STR_LIT>" } <EOL> expected = [ <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> ] <EOL> actual = self . scenario . _make_general_query ( True , True , True , metadata ) <EOL> self . assertEqual ( expected , actual ) <EOL> def test__make_query_item ( self ) : <EOL> expected = { "<STR_LIT>" : "<STR_LIT:foo>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT:bar>" } <EOL> self . assertEqual ( expected , <EOL> self . scenario . _make_query_item ( "<STR_LIT:foo>" , value = "<STR_LIT:bar>" ) ) <EOL> def test__make_profiler_key ( self ) : <EOL> query = [ <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT:bar>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:value>" : None } <EOL> ] <EOL> limit = <NUM_LIT:100> <EOL> method = "<STR_LIT>" <EOL> actual = self . scenario . _make_profiler_key ( method , query , limit ) <EOL> self . assertEqual ( "<STR_LIT>" , actual ) <EOL> actual = self . scenario . _make_profiler_key ( method , query , None ) <EOL> self . assertEqual ( "<STR_LIT>" , actual ) <EOL> self . assertEqual ( method , <EOL> self . scenario . _make_profiler_key ( method , None , None ) ) </s>
