<s> import unittest <EOL> import socket <EOL> import time <EOL> import statsd <EOL> class mock_udp_socket ( object ) : <EOL> def __init__ ( self , family , socktype ) : <EOL> assert family == socket . AF_INET <EOL> assert socktype == socket . SOCK_DGRAM <EOL> def sendto ( self , data , addr ) : <EOL> self . data = data <EOL> class TestStatsd ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> statsd . socket = mock_udp_socket <EOL> statsd . init_statsd ( ) <EOL> def tearDown ( self ) : <EOL> statsd . STATSD_HOST = '<STR_LIT:localhost>' <EOL> statsd . STATSD_PORT = <NUM_LIT> <EOL> statsd . STATSD_SAMPLE_RATE = None <EOL> statsd . STATSD_BUCKET_PREFIX = None <EOL> def test_init_statsd ( self ) : <EOL> settings = { '<STR_LIT>' : '<STR_LIT:127.0.0.1>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> statsd . init_statsd ( settings ) <EOL> self . assertEqual ( statsd . STATSD_HOST , '<STR_LIT:127.0.0.1>' ) <EOL> self . assertEqual ( statsd . STATSD_PORT , <NUM_LIT> ) <EOL> self . assertEqual ( statsd . STATSD_SAMPLE_RATE , <NUM_LIT> ) <EOL> self . assertEqual ( statsd . STATSD_BUCKET_PREFIX , '<STR_LIT>' ) <EOL> def test_exception_in_send ( self ) : <EOL> def mock_sendto_raise_error ( data , addr ) : <EOL> mock_sendto_raise_error . exception_raised = True <EOL> raise socket . gaierror <EOL> statsd . _statsd . _socket . sendto = mock_sendto_raise_error <EOL> statsd . decrement ( '<STR_LIT>' ) <EOL> self . assertTrue ( mock_sendto_raise_error . exception_raised ) <EOL> def test_decrement ( self ) : <EOL> statsd . decrement ( '<STR_LIT>' ) <EOL> self . assertEqual ( statsd . _statsd . _socket . data , b'<STR_LIT>' ) <EOL> statsd . decrement ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertEqual ( statsd . _statsd . _socket . data , b'<STR_LIT>' ) <EOL> statsd . decrement ( '<STR_LIT>' , <NUM_LIT:5> , <NUM_LIT> ) <EOL> self . assertTrue ( statsd . _statsd . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> if statsd . _statsd . _socket . data != b'<STR_LIT>' : <EOL> self . assertTrue ( statsd . _statsd . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_increment ( self ) : <EOL> statsd . increment ( '<STR_LIT>' ) <EOL> self . assertEqual ( statsd . _statsd . _socket . data , b'<STR_LIT>' ) <EOL> statsd . increment ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertEqual ( statsd . _statsd . _socket . data , b'<STR_LIT>' ) <EOL> statsd . increment ( '<STR_LIT>' , <NUM_LIT:5> , <NUM_LIT> ) <EOL> self . assertTrue ( statsd . _statsd . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> if statsd . _statsd . _socket . data != b'<STR_LIT>' : <EOL> self . assertTrue ( statsd . _statsd . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_gauge ( self ) : <EOL> statsd . gauge ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> self . assertEqual ( statsd . _statsd . _socket . data , b'<STR_LIT>' ) <EOL> statsd . gauge ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertEqual ( statsd . _statsd . _socket . data , b'<STR_LIT>' ) <EOL> statsd . gauge ( '<STR_LIT>' , - <NUM_LIT:5> , <NUM_LIT> ) <EOL> self . assertTrue ( statsd . _statsd . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> if statsd . _statsd . _socket . data != b'<STR_LIT>' : <EOL> self . assertTrue ( statsd . _statsd . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_timing ( self ) : <EOL> statsd . timing ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . assertEqual ( statsd . _statsd . _socket . data , b'<STR_LIT>' ) <EOL> statsd . timing ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> ) <EOL> self . assertTrue ( statsd . _statsd . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> if statsd . _statsd . _socket . data != b'<STR_LIT>' : <EOL> self . assertTrue ( statsd . _statsd . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> class TestStatsdClient ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> statsd . socket = mock_udp_socket <EOL> def test_prefix ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> client . _send ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> client . _send ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> client . _send ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> def test_decr ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> client . decr ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> def test_decr_sample_rate ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = <NUM_LIT> ) <EOL> client . decr ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> if client . _socket . data != '<STR_LIT>' : <EOL> self . assertTrue ( client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_incr ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> client . incr ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> def test_incr_sample_rate ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = <NUM_LIT> ) <EOL> client . incr ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> if client . _socket . data != '<STR_LIT>' : <EOL> self . assertTrue ( client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_send ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> client . _send ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> def test_send_sample_rate ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = <NUM_LIT> ) <EOL> client . _send ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> if client . _socket . data != '<STR_LIT>' : <EOL> self . assertTrue ( client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_timing ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> client . timing ( '<STR_LIT>' , <NUM_LIT:100> ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> def test_timing_sample_rate ( self ) : <EOL> client = statsd . StatsdClient ( '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = <NUM_LIT> ) <EOL> client . timing ( '<STR_LIT>' , <NUM_LIT:100> ) <EOL> self . assertEqual ( client . _socket . data , b'<STR_LIT>' ) <EOL> if client . _socket . data != '<STR_LIT>' : <EOL> self . assertTrue ( client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> class TestStatsdCounter ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> statsd . socket = mock_udp_socket <EOL> def test_add ( self ) : <EOL> counter = statsd . StatsdCounter ( '<STR_LIT>' , '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> counter += <NUM_LIT:1> <EOL> self . assertEqual ( counter . _client . _socket . data , b'<STR_LIT>' ) <EOL> counter += <NUM_LIT:5> <EOL> self . assertEqual ( counter . _client . _socket . data , b'<STR_LIT>' ) <EOL> def test_sub ( self ) : <EOL> counter = statsd . StatsdCounter ( '<STR_LIT>' , '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> counter -= <NUM_LIT:1> <EOL> self . assertEqual ( counter . _client . _socket . data , b'<STR_LIT>' ) <EOL> counter -= <NUM_LIT:5> <EOL> self . assertEqual ( counter . _client . _socket . data , b'<STR_LIT>' ) <EOL> class TestStatsdTimer ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> statsd . socket = mock_udp_socket <EOL> def test_startstop ( self ) : <EOL> timer = statsd . StatsdTimer ( '<STR_LIT>' , '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> timer . start ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> timer . stop ( ) <EOL> self . assertTrue ( timer . _client . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> self . assertTrue ( timer . _client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_split ( self ) : <EOL> timer = statsd . StatsdTimer ( '<STR_LIT>' , '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> timer . start ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> timer . split ( '<STR_LIT>' ) <EOL> self . assertTrue ( timer . _client . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> self . assertTrue ( timer . _client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> timer . stop ( ) <EOL> self . assertTrue ( timer . _client . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> self . assertTrue ( timer . _client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> def test_wrap ( self ) : <EOL> class TC ( object ) : <EOL> @ statsd . StatsdTimer . wrap ( '<STR_LIT>' ) <EOL> def do ( self ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> return <NUM_LIT:1> <EOL> tc = TC ( ) <EOL> result = tc . do ( ) <EOL> self . assertEqual ( result , <NUM_LIT:1> ) <EOL> def test_with ( self ) : <EOL> timer = statsd . StatsdTimer ( '<STR_LIT>' , '<STR_LIT:localhost>' , <NUM_LIT> , prefix = '<STR_LIT>' , sample_rate = None ) <EOL> with timer : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertTrue ( timer . _client . _socket . data . startswith ( b'<STR_LIT>' ) ) <EOL> self . assertTrue ( timer . _client . _socket . data . endswith ( b'<STR_LIT>' ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
