<s> from unittest import TestCase <EOL> from mock import Mock , PropertyMock , patch <EOL> from carbon . cache import _MetricCache , DrainStrategy , MaxStrategy , RandomStrategy , SortedStrategy <EOL> class MetricCacheTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> settings = { <EOL> '<STR_LIT>' : float ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : float ( '<STR_LIT>' ) <EOL> } <EOL> self . _settings_patch = patch . dict ( '<STR_LIT>' , settings ) <EOL> self . _settings_patch . start ( ) <EOL> self . strategy_mock = Mock ( spec = DrainStrategy ) <EOL> self . metric_cache = _MetricCache ( self . strategy_mock ) <EOL> def tearDown ( self ) : <EOL> self . _settings_patch . stop ( ) <EOL> def test_cache_is_a_dict ( self ) : <EOL> self . assertTrue ( issubclass ( _MetricCache , dict ) ) <EOL> def test_initial_size ( self ) : <EOL> self . assertEqual ( <NUM_LIT:0> , self . metric_cache . size ) <EOL> def test_store_new_metric ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . metric_cache . size ) <EOL> self . assertEqual ( [ ( <NUM_LIT> , <NUM_LIT:1.0> ) ] , self . metric_cache [ '<STR_LIT:foo>' ] . items ( ) ) <EOL> def test_store_multiple_datapoints ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , self . metric_cache . size ) <EOL> result = self . metric_cache [ '<STR_LIT:foo>' ] . items ( ) <EOL> self . assertTrue ( ( <NUM_LIT> , <NUM_LIT:1.0> ) in result ) <EOL> self . assertTrue ( ( <NUM_LIT> , <NUM_LIT> ) in result ) <EOL> def test_store_duplicate_timestamp ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . metric_cache . size ) <EOL> self . assertEqual ( [ ( <NUM_LIT> , <NUM_LIT> ) ] , self . metric_cache [ '<STR_LIT:foo>' ] . items ( ) ) <EOL> def test_store_checks_fullness ( self ) : <EOL> is_full_mock = PropertyMock ( ) <EOL> with patch . object ( _MetricCache , '<STR_LIT>' , is_full_mock ) : <EOL> with patch ( '<STR_LIT>' ) : <EOL> metric_cache = _MetricCache ( ) <EOL> metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , is_full_mock . call_count ) <EOL> def test_store_on_full_triggers_events ( self ) : <EOL> is_full_mock = PropertyMock ( return_value = True ) <EOL> with patch . object ( _MetricCache , '<STR_LIT>' , is_full_mock ) : <EOL> with patch ( '<STR_LIT>' ) as events_mock : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> events_mock . cacheFull . assert_called_with ( ) <EOL> def test_pop_multiple_datapoints ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> result = self . metric_cache . pop ( '<STR_LIT:foo>' ) <EOL> self . assertTrue ( ( <NUM_LIT> , <NUM_LIT:1.0> ) in result ) <EOL> self . assertTrue ( ( <NUM_LIT> , <NUM_LIT> ) in result ) <EOL> def test_pop_reduces_size ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . pop ( '<STR_LIT:foo>' ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . metric_cache . size ) <EOL> def test_pop_triggers_space_check ( self ) : <EOL> with patch . object ( self . metric_cache , '<STR_LIT>' ) as check_space_mock : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . pop ( '<STR_LIT:foo>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , check_space_mock . call_count ) <EOL> def test_pop_triggers_space_event ( self ) : <EOL> with patch ( '<STR_LIT>' , new = Mock ( return_value = True ) ) : <EOL> with patch ( '<STR_LIT>' ) as events_mock : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . pop ( '<STR_LIT:foo>' ) <EOL> events_mock . cacheSpaceAvailable . assert_called_with ( ) <EOL> def test_pop_returns_sorted_timestamps ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> result = self . metric_cache . pop ( '<STR_LIT:foo>' ) <EOL> expected = [ ( <NUM_LIT> , <NUM_LIT:1.0> ) , ( <NUM_LIT> , <NUM_LIT> ) , ( <NUM_LIT> , <NUM_LIT> ) ] <EOL> self . assertEqual ( expected , result ) <EOL> def test_pop_raises_on_missing ( self ) : <EOL> self . assertRaises ( KeyError , self . metric_cache . pop , '<STR_LIT:foo>' ) <EOL> def test_get_datapoints ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertEqual ( [ ( <NUM_LIT> , <NUM_LIT:1.0> ) ] , self . metric_cache . get_datapoints ( '<STR_LIT:foo>' ) ) <EOL> def test_get_datapoints_doesnt_pop ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertEqual ( [ ( <NUM_LIT> , <NUM_LIT:1.0> ) ] , self . metric_cache . get_datapoints ( '<STR_LIT:foo>' ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . metric_cache . size ) <EOL> self . assertEqual ( [ ( <NUM_LIT> , <NUM_LIT:1.0> ) ] , self . metric_cache . get_datapoints ( '<STR_LIT:foo>' ) ) <EOL> def test_get_datapoints_returns_empty_on_missing ( self ) : <EOL> self . assertEqual ( [ ] , self . metric_cache . get_datapoints ( '<STR_LIT:foo>' ) ) <EOL> def test_get_datapoints_returns_sorted_timestamps ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> result = self . metric_cache . get_datapoints ( '<STR_LIT:foo>' ) <EOL> expected = [ ( <NUM_LIT> , <NUM_LIT:1.0> ) , ( <NUM_LIT> , <NUM_LIT> ) , ( <NUM_LIT> , <NUM_LIT> ) ] <EOL> self . assertEqual ( expected , result ) <EOL> def test_drain_metric_respects_strategy ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . strategy_mock . return_value . choose_item . side_effect = [ '<STR_LIT:bar>' , '<STR_LIT>' , '<STR_LIT:foo>' ] <EOL> self . assertEqual ( '<STR_LIT:bar>' , self . metric_cache . drain_metric ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , self . metric_cache . drain_metric ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , self . metric_cache . drain_metric ( ) [ <NUM_LIT:0> ] ) <EOL> def test_drain_metric_works_without_strategy ( self ) : <EOL> metric_cache = _MetricCache ( ) <EOL> metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , metric_cache . drain_metric ( ) [ <NUM_LIT:0> ] ) <EOL> def test_is_full_short_circuits_on_inf ( self ) : <EOL> with patch . object ( self . metric_cache , '<STR_LIT:size>' ) as size_mock : <EOL> self . metric_cache . is_full <EOL> size_mock . assert_not_called ( ) <EOL> def test_is_full ( self ) : <EOL> self . _settings_patch . values [ '<STR_LIT>' ] = <NUM_LIT> <EOL> self . _settings_patch . start ( ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> self . assertFalse ( self . metric_cache . is_full ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertFalse ( self . metric_cache . is_full ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertTrue ( self . metric_cache . is_full ) <EOL> def test_counts_one_datapoint ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . assertEqual ( [ ( '<STR_LIT:foo>' , <NUM_LIT:1> ) ] , self . metric_cache . counts ) <EOL> def test_counts_two_datapoints ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assertEqual ( [ ( '<STR_LIT:foo>' , <NUM_LIT:2> ) ] , self . metric_cache . counts ) <EOL> def test_counts_multiple_datapoints ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assertTrue ( ( '<STR_LIT:foo>' , <NUM_LIT:2> ) in self . metric_cache . counts ) <EOL> self . assertTrue ( ( '<STR_LIT:bar>' , <NUM_LIT:1> ) in self . metric_cache . counts ) <EOL> class DrainStrategyTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . metric_cache = _MetricCache ( ) <EOL> def test_max_strategy ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> max_strategy = MaxStrategy ( self . metric_cache ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , max_strategy . choose_item ( ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assertEqual ( '<STR_LIT:bar>' , max_strategy . choose_item ( ) ) <EOL> self . metric_cache . pop ( '<STR_LIT:foo>' ) <EOL> self . metric_cache . pop ( '<STR_LIT:bar>' ) <EOL> self . assertEqual ( '<STR_LIT>' , max_strategy . choose_item ( ) ) <EOL> def test_sorted_strategy_static_cache ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> sorted_strategy = SortedStrategy ( self . metric_cache ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , sorted_strategy . choose_item ( ) ) <EOL> self . assertEqual ( '<STR_LIT:bar>' , sorted_strategy . choose_item ( ) ) <EOL> self . assertEqual ( '<STR_LIT>' , sorted_strategy . choose_item ( ) ) <EOL> def test_sorted_strategy_changing_sizes ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> sorted_strategy = SortedStrategy ( self . metric_cache ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , sorted_strategy . choose_item ( ) ) <EOL> self . metric_cache . store ( '<STR_LIT>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assertEqual ( '<STR_LIT:bar>' , sorted_strategy . choose_item ( ) ) <EOL> self . assertEqual ( '<STR_LIT>' , sorted_strategy . choose_item ( ) ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , sorted_strategy . choose_item ( ) ) <EOL> self . assertEqual ( '<STR_LIT:bar>' , sorted_strategy . choose_item ( ) ) <EOL> self . assertEqual ( '<STR_LIT>' , sorted_strategy . choose_item ( ) ) <EOL> class RandomStrategyTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . metric_cache = _MetricCache ( ) <EOL> def test_random_strategy ( self ) : <EOL> self . metric_cache . store ( '<STR_LIT:foo>' , ( <NUM_LIT> , <NUM_LIT:1.0> ) ) <EOL> self . metric_cache . store ( '<STR_LIT:bar>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . metric_cache . store ( '<STR_LIT>' , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> strategy = RandomStrategy ( self . metric_cache ) <EOL> for _i in range ( <NUM_LIT:3> ) : <EOL> item = strategy . choose_item ( ) <EOL> self . assertTrue ( item in self . metric_cache ) <EOL> self . metric_cache . pop ( item ) </s>
