<s> try : <EOL> import ConfigParser <EOL> from StringIO import StringIO <EOL> except ImportError : <EOL> import configparser as ConfigParser <EOL> from io import StringIO <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . plugins . sli_plugin import SLIMetricFactory <EOL> from txstatsd . metrics . slimetric import ( <EOL> SLIMetricReporter , BetweenCondition , AboveCondition , BelowCondition ) <EOL> from txstatsd import service <EOL> from txstatsd . tests . test_processor import TestMessageProcessor <EOL> class TestConditions ( TestCase ) : <EOL> def test_below ( self ) : <EOL> c = BelowCondition ( <NUM_LIT:5> ) <EOL> self . assertEquals ( c ( <NUM_LIT:2> ) , True ) <EOL> self . assertEquals ( c ( <NUM_LIT:6> ) , False ) <EOL> def test_above ( self ) : <EOL> c = AboveCondition ( <NUM_LIT:5> ) <EOL> self . assertEquals ( c ( <NUM_LIT:2> ) , False ) <EOL> self . assertEquals ( c ( <NUM_LIT:6> ) , True ) <EOL> def test_between ( self ) : <EOL> c = BetweenCondition ( <NUM_LIT> , <NUM_LIT:5> ) <EOL> self . assertEquals ( c ( <NUM_LIT:2> ) , False ) <EOL> self . assertEquals ( c ( <NUM_LIT:6> ) , False ) <EOL> self . assertEquals ( c ( <NUM_LIT> ) , True ) <EOL> def test_below_linear ( self ) : <EOL> c = BelowCondition ( <NUM_LIT:5> , <NUM_LIT:1> ) <EOL> self . assertEquals ( c ( <NUM_LIT> , <NUM_LIT:1> ) , True ) <EOL> self . assertEquals ( c ( <NUM_LIT> , <NUM_LIT:2> ) , True ) <EOL> self . assertEquals ( c ( <NUM_LIT> , <NUM_LIT:3> ) , False ) <EOL> def test_above_linear ( self ) : <EOL> c = AboveCondition ( <NUM_LIT:4> , <NUM_LIT:1> ) <EOL> self . assertEquals ( c ( <NUM_LIT> , <NUM_LIT:1> ) , True ) <EOL> self . assertEquals ( c ( <NUM_LIT> , <NUM_LIT:2> ) , True ) <EOL> self . assertEquals ( c ( <NUM_LIT:7> , <NUM_LIT:3> ) , False ) <EOL> class TestParsing ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . processor = TestMessageProcessor ( ) <EOL> def test_parse ( self ) : <EOL> self . processor . process ( '<STR_LIT>' ) <EOL> self . processor . process ( '<STR_LIT>' ) <EOL> self . processor . process ( '<STR_LIT>' ) <EOL> class TestMetric ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . sli = SLIMetricReporter ( '<STR_LIT:test>' , { <EOL> "<STR_LIT>" : BelowCondition ( <NUM_LIT:5> ) , <EOL> "<STR_LIT>" : BelowCondition ( <NUM_LIT:3> ) } ) <EOL> def test_count_all ( self ) : <EOL> self . sli . update ( <NUM_LIT:1> ) <EOL> self . sli . update ( <NUM_LIT:1> ) <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:2> ) <EOL> def test_count_error ( self ) : <EOL> self . sli . update ( <NUM_LIT:1> ) <EOL> self . sli . update ( "<STR_LIT:error>" ) <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . sli . error , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> def test_count_threshold ( self ) : <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT:7> ) : <EOL> self . sli . update ( i ) <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:6> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:4> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> def test_reports ( self ) : <EOL> self . test_count_threshold ( ) <EOL> rows = sorted ( self . sli . flush ( <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> self . assertEquals ( <EOL> [ ( "<STR_LIT>" , <NUM_LIT:6> , <NUM_LIT:0> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:4> , <NUM_LIT:0> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:2> , <NUM_LIT:0> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> ) ] , <EOL> rows ) <EOL> def test_clear ( self ) : <EOL> self . sli . update ( <NUM_LIT:1> ) <EOL> self . sli . update ( <NUM_LIT:1> ) <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:2> ) <EOL> self . sli . flush ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> self . sli . update ( <NUM_LIT:1> ) <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:1> ) <EOL> class TestMetricLinear ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . sli = SLIMetricReporter ( '<STR_LIT:test>' , { <EOL> "<STR_LIT>" : BelowCondition ( <NUM_LIT:5> , <NUM_LIT:1> ) , <EOL> "<STR_LIT>" : BelowCondition ( <NUM_LIT:3> , <NUM_LIT:1> ) } ) <EOL> def test_count_threshold ( self ) : <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT:7> ) : <EOL> self . sli . update ( <NUM_LIT:7> , i ) <EOL> self . assertEquals ( self . sli . count , <NUM_LIT:6> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:4> ) <EOL> self . assertEquals ( self . sli . counts [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> class TestFactory ( TestCase ) : <EOL> def test_configure ( self ) : <EOL> class TestOptions ( service . OptionsGlue ) : <EOL> optParameters = [ [ "<STR_LIT:test>" , "<STR_LIT:t>" , "<STR_LIT:default>" , "<STR_LIT>" ] ] <EOL> config_section = "<STR_LIT>" <EOL> o = TestOptions ( ) <EOL> config_file = ConfigParser . RawConfigParser ( ) <EOL> config_file . readfp ( StringIO ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> o . configure ( config_file ) <EOL> smf = SLIMetricFactory ( ) <EOL> smf . configure ( o ) <EOL> smr = smf . build_metric ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> rc = smr . conditions [ "<STR_LIT>" ] <EOL> self . assertTrue ( isinstance ( rc , BelowCondition ) ) <EOL> self . assertEquals ( rc . value , <NUM_LIT:5> ) <EOL> gc = smr . conditions [ "<STR_LIT>" ] <EOL> self . assertTrue ( isinstance ( gc , BetweenCondition ) ) <EOL> self . assertEquals ( gc . hi , <NUM_LIT:3> ) <EOL> self . assertEquals ( gc . low , <NUM_LIT:0.1> ) <EOL> smr = smf . build_metric ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> rc = smr . conditions [ "<STR_LIT>" ] <EOL> self . assertTrue ( isinstance ( rc , AboveCondition ) ) <EOL> self . assertEquals ( rc . value , <NUM_LIT:4> ) <EOL> def test_configure_linear ( self ) : <EOL> class TestOptions ( service . OptionsGlue ) : <EOL> optParameters = [ [ "<STR_LIT:test>" , "<STR_LIT:t>" , "<STR_LIT:default>" , "<STR_LIT>" ] ] <EOL> config_section = "<STR_LIT>" <EOL> o = TestOptions ( ) <EOL> config_file = ConfigParser . RawConfigParser ( ) <EOL> config_file . readfp ( StringIO ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> o . configure ( config_file ) <EOL> smf = SLIMetricFactory ( ) <EOL> smf . configure ( o ) <EOL> smr = smf . build_metric ( "<STR_LIT>" , "<STR_LIT:test>" ) <EOL> rc = smr . conditions [ "<STR_LIT>" ] <EOL> self . assertTrue ( isinstance ( rc , BelowCondition ) ) <EOL> self . assertEquals ( rc . value , <NUM_LIT:5> ) <EOL> self . assertEquals ( rc . slope , <NUM_LIT:1> ) <EOL> rc = smr . conditions [ "<STR_LIT>" ] <EOL> self . assertTrue ( isinstance ( rc , AboveCondition ) ) <EOL> self . assertEquals ( rc . value , <NUM_LIT:3> ) <EOL> self . assertEquals ( rc . slope , <NUM_LIT:1> ) </s>
