<s> from test import CollectorTestCase <EOL> from test import get_collector_config <EOL> from test import unittest <EOL> from mock import Mock <EOL> from mock import patch <EOL> from diamond . collector import Collector <EOL> from kafka_jolokia import KafkaJolokiaCollector <EOL> def find_metric ( metric_list , metric_name ) : <EOL> return filter ( lambda metric : metric [ "<STR_LIT:name>" ] . find ( metric_name ) > - <NUM_LIT:1> , metric_list ) <EOL> def find_by_dimension ( metric_list , key , val ) : <EOL> return filter ( lambda metric : metric [ "<STR_LIT>" ] [ key ] == val , metric_list ) [ <NUM_LIT:0> ] <EOL> def list_request ( ) : <EOL> return { '<STR_LIT:value>' : { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT:status>' : <NUM_LIT:200> } <EOL> class TestKafkaJolokiaCollector ( CollectorTestCase ) : <EOL> def setUp ( self ) : <EOL> config = get_collector_config ( '<STR_LIT>' , { } ) <EOL> self . collector = KafkaJolokiaCollector ( config , None ) <EOL> self . collector . list_request = list_request <EOL> def test_import ( self ) : <EOL> self . assertTrue ( KafkaJolokiaCollector ) <EOL> @ patch . object ( Collector , '<STR_LIT>' ) <EOL> def test_should_create_type ( self , publish_mock ) : <EOL> def se ( url ) : <EOL> return self . getFixture ( "<STR_LIT>" ) <EOL> patch_urlopen = patch ( '<STR_LIT>' , Mock ( side_effect = se ) ) <EOL> with patch_urlopen : <EOL> self . collector . collect ( ) <EOL> self . assertEquals ( len ( self . collector . payload ) , <NUM_LIT> ) <EOL> metrics = find_metric ( self . collector . payload , "<STR_LIT>" ) <EOL> self . assertNotEqual ( len ( metrics ) , <NUM_LIT:0> ) <EOL> metric = find_by_dimension ( metrics , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( metric [ "<STR_LIT:type>" ] , "<STR_LIT>" ) <EOL> metrics_dots = find_metric ( self . collector . payload , <EOL> "<STR_LIT>" ) <EOL> self . assertNotEqual ( len ( metrics_dots ) , <NUM_LIT:0> ) <EOL> @ patch . object ( Collector , '<STR_LIT>' ) <EOL> def test_blacklisting ( self , publish_mock ) : <EOL> def se ( url ) : <EOL> return self . getFixture ( "<STR_LIT>" ) <EOL> patch_urlopen = patch ( '<STR_LIT>' , Mock ( side_effect = se ) ) <EOL> with patch_urlopen : <EOL> self . collector . collect ( ) <EOL> self . assertEquals ( len ( self . collector . payload ) , <NUM_LIT> ) <EOL> metrics = find_metric ( self . collector . payload , "<STR_LIT>" ) <EOL> self . assertEquals ( len ( metrics ) , <NUM_LIT:0> ) <EOL> @ patch . object ( Collector , '<STR_LIT>' ) <EOL> def test_total_topic ( self , publish_mock ) : <EOL> def se ( url ) : <EOL> return self . getFixture ( "<STR_LIT>" ) <EOL> patch_urlopen = patch ( '<STR_LIT>' , Mock ( side_effect = se ) ) <EOL> with patch_urlopen : <EOL> self . collector . collect ( ) <EOL> self . assertEquals ( len ( self . collector . payload ) , <NUM_LIT> ) <EOL> metrics = find_metric ( self . collector . payload , "<STR_LIT>" ) <EOL> self . assertNotEqual ( len ( metrics ) , <NUM_LIT:0> ) <EOL> metric = find_by_dimension ( metrics , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( metric [ "<STR_LIT:type>" ] , "<STR_LIT>" ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
