<s> from __future__ import with_statement <EOL> import random <EOL> import re <EOL> import socket <EOL> import mock <EOL> from nose . tools import eq_ <EOL> from statsd import StatsClient <EOL> from statsd import TCPStatsClient <EOL> ADDR = ( socket . gethostbyname ( '<STR_LIT:localhost>' ) , <NUM_LIT> ) <EOL> send_method = { <EOL> '<STR_LIT>' : lambda x : x . sendto , <EOL> '<STR_LIT>' : lambda x : x . sendall , <EOL> } <EOL> make_val = { <EOL> '<STR_LIT>' : lambda x , addr : mock . call ( str . encode ( x ) , addr ) , <EOL> '<STR_LIT>' : lambda x , addr : mock . call ( str . encode ( x + '<STR_LIT:\n>' ) ) , <EOL> } <EOL> def _udp_client ( prefix = None , addr = None , port = None , ipv6 = False ) : <EOL> if not addr : <EOL> addr = ADDR [ <NUM_LIT:0> ] <EOL> if not port : <EOL> port = ADDR [ <NUM_LIT:1> ] <EOL> sc = StatsClient ( host = addr , port = port , prefix = prefix , ipv6 = ipv6 ) <EOL> sc . _sock = mock . Mock ( ) <EOL> return sc <EOL> def _tcp_client ( prefix = None , addr = None , port = None , timeout = None , ipv6 = False ) : <EOL> if not addr : <EOL> addr = ADDR [ <NUM_LIT:0> ] <EOL> if not port : <EOL> port = ADDR [ <NUM_LIT:1> ] <EOL> sc = TCPStatsClient ( host = addr , port = port , prefix = prefix , timeout = timeout , <EOL> ipv6 = ipv6 ) <EOL> sc . _sock = mock . Mock ( ) <EOL> return sc <EOL> def _timer_check ( sock , count , proto , start , end ) : <EOL> send = send_method [ proto ] ( sock ) <EOL> eq_ ( send . call_count , count ) <EOL> value = send . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . decode ( '<STR_LIT:ascii>' ) <EOL> exp = re . compile ( '<STR_LIT>' % ( start , end ) ) <EOL> assert exp . match ( value ) <EOL> def _sock_check ( sock , count , proto , val = None , addr = None ) : <EOL> send = send_method [ proto ] ( sock ) <EOL> eq_ ( send . call_count , count ) <EOL> if not addr : <EOL> addr = ADDR <EOL> if val is not None : <EOL> eq_ ( <EOL> send . call_args , <EOL> make_val [ proto ] ( val , addr ) , <EOL> ) <EOL> class assert_raises ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * exc_cls ) : <EOL> self . exc_cls = exc_cls <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , typ , value , tb ) : <EOL> assert typ , '<STR_LIT>' <EOL> assert typ in self . exc_cls , '<STR_LIT>' % ( <EOL> typ . __name__ , [ e . __name__ for e in self . exc_cls ] ) <EOL> self . exc_type = typ <EOL> self . exception = value <EOL> self . traceback = tb <EOL> return True <EOL> def _test_incr ( cl , proto ) : <EOL> cl . incr ( '<STR_LIT:foo>' ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , val = '<STR_LIT>' ) <EOL> cl . incr ( '<STR_LIT:foo>' , <NUM_LIT:10> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:2> , proto , val = '<STR_LIT>' ) <EOL> cl . incr ( '<STR_LIT:foo>' , <NUM_LIT> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:3> , proto , val = '<STR_LIT>' ) <EOL> cl . incr ( '<STR_LIT:foo>' , <NUM_LIT:10> , rate = <NUM_LIT:0.5> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:4> , proto , val = '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_incr_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_incr ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_incr_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_incr ( cl , '<STR_LIT>' ) <EOL> def _test_decr ( cl , proto ) : <EOL> cl . decr ( '<STR_LIT:foo>' ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> cl . decr ( '<STR_LIT:foo>' , <NUM_LIT:10> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:2> , proto , '<STR_LIT>' ) <EOL> cl . decr ( '<STR_LIT:foo>' , <NUM_LIT> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:3> , proto , '<STR_LIT>' ) <EOL> cl . decr ( '<STR_LIT:foo>' , <NUM_LIT:1> , rate = <NUM_LIT:0.5> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:4> , proto , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_decr_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_decr ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_decr_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_decr ( cl , '<STR_LIT>' ) <EOL> def _test_gauge ( cl , proto ) : <EOL> cl . gauge ( '<STR_LIT:foo>' , <NUM_LIT:30> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> cl . gauge ( '<STR_LIT:foo>' , <NUM_LIT> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:2> , proto , '<STR_LIT>' ) <EOL> cl . gauge ( '<STR_LIT:foo>' , <NUM_LIT> , rate = <NUM_LIT:0.5> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:3> , proto , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_gauge_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_gauge ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_gauge_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_gauge ( cl , '<STR_LIT>' ) <EOL> def _test_ipv6 ( cl , proto , addr ) : <EOL> cl . gauge ( '<STR_LIT:foo>' , <NUM_LIT:30> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' , addr = addr ) <EOL> def test_ipv6_udp ( ) : <EOL> """<STR_LIT>""" <EOL> addr = ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> cl = _udp_client ( addr = addr [ <NUM_LIT:0> ] , ipv6 = True ) <EOL> _test_ipv6 ( cl , '<STR_LIT>' , addr ) <EOL> def test_ipv6_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> addr = ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> cl = _tcp_client ( addr = addr [ <NUM_LIT:0> ] , ipv6 = True ) <EOL> _test_ipv6 ( cl , '<STR_LIT>' , addr ) <EOL> def _test_resolution ( cl , proto , addr ) : <EOL> cl . incr ( '<STR_LIT:foo>' ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' , addr = addr ) <EOL> def test_ipv6_resolution_udp ( ) : <EOL> cl = _udp_client ( addr = '<STR_LIT:localhost>' , ipv6 = True ) <EOL> _test_resolution ( cl , '<STR_LIT>' , ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> def test_ipv6_resolution_tcp ( ) : <EOL> cl = _tcp_client ( addr = '<STR_LIT:localhost>' , ipv6 = True ) <EOL> _test_resolution ( cl , '<STR_LIT>' , ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> def test_ipv4_resolution_udp ( ) : <EOL> cl = _udp_client ( addr = '<STR_LIT:localhost>' ) <EOL> _test_resolution ( cl , '<STR_LIT>' , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) <EOL> def test_ipv4_resolution_tcp ( ) : <EOL> cl = _tcp_client ( addr = '<STR_LIT:localhost>' ) <EOL> _test_resolution ( cl , '<STR_LIT>' , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) <EOL> def _test_gauge_delta ( cl , proto ) : <EOL> tests = ( <EOL> ( <NUM_LIT:12> , '<STR_LIT>' ) , <EOL> ( - <NUM_LIT> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , '<STR_LIT>' ) , <EOL> ( - <NUM_LIT> , '<STR_LIT>' ) , <EOL> ) <EOL> def _check ( num , result ) : <EOL> cl . _sock . reset_mock ( ) <EOL> cl . gauge ( '<STR_LIT:foo>' , num , delta = True ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' % result ) <EOL> for num , result in tests : <EOL> _check ( num , result ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_gauge_delta_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_gauge_delta ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_gauge_delta_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_gauge_delta ( cl , '<STR_LIT>' ) <EOL> def _test_gauge_absolute_negative ( cl , proto ) : <EOL> cl . gauge ( '<STR_LIT:foo>' , - <NUM_LIT:5> , delta = False ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_gauge_absolute_negative_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_gauge_delta ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_gauge_absolute_negative_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_gauge_delta ( cl , '<STR_LIT>' ) <EOL> def _test_gauge_absolute_negative_rate ( cl , proto , mock_random ) : <EOL> mock_random . return_value = - <NUM_LIT:1> <EOL> cl . gauge ( '<STR_LIT:foo>' , - <NUM_LIT:1> , rate = <NUM_LIT:0.5> , delta = False ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> mock_random . return_value = <NUM_LIT:2> <EOL> cl . gauge ( '<STR_LIT:foo>' , - <NUM_LIT:2> , rate = <NUM_LIT:0.5> , delta = False ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' ) <EOL> def test_gauge_absolute_negative_rate_udp ( mock_random ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_gauge_absolute_negative_rate ( cl , '<STR_LIT>' , mock_random ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' ) <EOL> def test_gauge_absolute_negative_rate_tcp ( mock_random ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_gauge_absolute_negative_rate ( cl , '<STR_LIT>' , mock_random ) <EOL> def _test_set ( cl , proto ) : <EOL> cl . set ( '<STR_LIT:foo>' , <NUM_LIT:10> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> cl . set ( '<STR_LIT:foo>' , <NUM_LIT> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:2> , proto , '<STR_LIT>' ) <EOL> cl . set ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:3> , proto , '<STR_LIT>' ) <EOL> cl . set ( '<STR_LIT:foo>' , <NUM_LIT> , <NUM_LIT:0.5> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:4> , proto , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_set_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_set ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_set_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_set ( cl , '<STR_LIT>' ) <EOL> def _test_timing ( cl , proto ) : <EOL> cl . timing ( '<STR_LIT:foo>' , <NUM_LIT:100> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> cl . timing ( '<STR_LIT:foo>' , <NUM_LIT> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:2> , proto , '<STR_LIT>' ) <EOL> cl . timing ( '<STR_LIT:foo>' , <NUM_LIT:100> , rate = <NUM_LIT:0.5> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:3> , proto , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timing_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timing ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timing_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timing ( cl , '<STR_LIT>' ) <EOL> def _test_prepare ( cl , proto ) : <EOL> tests = ( <EOL> ( '<STR_LIT>' , ( '<STR_LIT:foo>' , '<STR_LIT>' , <NUM_LIT:1> ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT:bar>' , '<STR_LIT>' , <NUM_LIT:0.5> ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> ) ) , <EOL> ) <EOL> def _check ( o , s , v , r ) : <EOL> with mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) : <EOL> eq_ ( o , cl . _prepare ( s , v , r ) ) <EOL> for o , ( s , v , r ) in tests : <EOL> _check ( o , s , v , r ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_prepare_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_prepare ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_prepare_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_prepare ( cl , '<STR_LIT>' ) <EOL> def _test_prefix ( cl , proto ) : <EOL> cl . incr ( '<STR_LIT:bar>' ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_prefix_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( prefix = '<STR_LIT:foo>' ) <EOL> _test_prefix ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_prefix_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( prefix = '<STR_LIT:foo>' ) <EOL> _test_prefix ( cl , '<STR_LIT>' ) <EOL> def _test_timer_manager ( cl , proto ) : <EOL> with cl . timer ( '<STR_LIT:foo>' ) : <EOL> pass <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_timer_manager_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_manager ( cl , '<STR_LIT>' ) <EOL> def test_timer_manager_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_manager ( cl , '<STR_LIT>' ) <EOL> def _test_timer_decorator ( cl , proto ) : <EOL> @ cl . timer ( '<STR_LIT:foo>' ) <EOL> def foo ( a , b ) : <EOL> return [ a , b ] <EOL> @ cl . timer ( '<STR_LIT:bar>' ) <EOL> def bar ( a , b ) : <EOL> return [ b , a ] <EOL> eq_ ( [ <NUM_LIT:4> , <NUM_LIT:2> ] , foo ( <NUM_LIT:4> , <NUM_LIT:2> ) ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> eq_ ( [ <NUM_LIT:2> , <NUM_LIT:4> ] , bar ( <NUM_LIT:4> , <NUM_LIT:2> ) ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:2> , proto , '<STR_LIT:bar>' , '<STR_LIT>' ) <EOL> eq_ ( [ <NUM_LIT:6> , <NUM_LIT:5> ] , bar ( <NUM_LIT:5> , <NUM_LIT:6> ) ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:3> , proto , '<STR_LIT:bar>' , '<STR_LIT>' ) <EOL> def test_timer_decorator_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_decorator ( cl , '<STR_LIT>' ) <EOL> def test_timer_decorator_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_decorator ( cl , '<STR_LIT>' ) <EOL> def _test_timer_capture ( cl , proto ) : <EOL> with cl . timer ( '<STR_LIT>' ) as result : <EOL> eq_ ( result . ms , None ) <EOL> assert isinstance ( result . ms , float ) <EOL> def test_timer_capture_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_capture ( cl , '<STR_LIT>' ) <EOL> def test_timer_capture_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_capture ( cl , '<STR_LIT>' ) <EOL> def _test_timer_context_rate ( cl , proto ) : <EOL> with cl . timer ( '<STR_LIT:foo>' , rate = <NUM_LIT:0.5> ) : <EOL> pass <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timer_context_rate_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_context_rate ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timer_context_rate_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_context_rate ( cl , '<STR_LIT>' ) <EOL> def _test_timer_decorator_rate ( cl , proto ) : <EOL> @ cl . timer ( '<STR_LIT:foo>' , rate = <NUM_LIT:0.1> ) <EOL> def foo ( a , b ) : <EOL> return [ b , a ] <EOL> @ cl . timer ( '<STR_LIT:bar>' , rate = <NUM_LIT> ) <EOL> def bar ( a , b = <NUM_LIT:2> , c = <NUM_LIT:3> ) : <EOL> return [ c , b , a ] <EOL> eq_ ( [ <NUM_LIT:2> , <NUM_LIT:4> ] , foo ( <NUM_LIT:4> , <NUM_LIT:2> ) ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> eq_ ( [ <NUM_LIT:3> , <NUM_LIT:2> , <NUM_LIT:5> ] , bar ( <NUM_LIT:5> ) ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:2> , proto , '<STR_LIT:bar>' , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timer_decorator_rate_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_decorator_rate ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timer_decorator_rate_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_decorator_rate ( cl , '<STR_LIT>' ) <EOL> def _test_timer_context_exceptions ( cl , proto ) : <EOL> with assert_raises ( socket . timeout ) : <EOL> with cl . timer ( '<STR_LIT:foo>' ) : <EOL> raise socket . timeout ( ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_timer_context_exceptions_udp ( ) : <EOL> cl = _udp_client ( ) <EOL> _test_timer_context_exceptions ( cl , '<STR_LIT>' ) <EOL> def test_timer_context_exceptions_tcp ( ) : <EOL> cl = _tcp_client ( ) <EOL> _test_timer_context_exceptions ( cl , '<STR_LIT>' ) <EOL> def _test_timer_decorator_exceptions ( cl , proto ) : <EOL> @ cl . timer ( '<STR_LIT:foo>' ) <EOL> def foo ( ) : <EOL> raise ValueError ( ) <EOL> with assert_raises ( ValueError ) : <EOL> foo ( ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_timer_decorator_exceptions_udp ( ) : <EOL> cl = _udp_client ( ) <EOL> _test_timer_decorator_exceptions ( cl , '<STR_LIT>' ) <EOL> def test_timer_decorator_exceptions_tcp ( ) : <EOL> cl = _tcp_client ( ) <EOL> _test_timer_decorator_exceptions ( cl , '<STR_LIT>' ) <EOL> def _test_timer_object ( cl , proto ) : <EOL> t = cl . timer ( '<STR_LIT:foo>' ) . start ( ) <EOL> t . stop ( ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_timer_object_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_object ( cl , '<STR_LIT>' ) <EOL> def test_timer_object_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_object ( cl , '<STR_LIT>' ) <EOL> def _test_timer_object_no_send ( cl , proto ) : <EOL> t = cl . timer ( '<STR_LIT:foo>' ) . start ( ) <EOL> t . stop ( send = False ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:0> , proto ) <EOL> t . send ( ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_timer_object_no_send_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_object_no_send ( cl , '<STR_LIT>' ) <EOL> def test_timer_object_no_send_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_object_no_send ( cl , '<STR_LIT>' ) <EOL> def _test_timer_object_rate ( cl , proto ) : <EOL> t = cl . timer ( '<STR_LIT:foo>' , rate = <NUM_LIT:0.5> ) <EOL> t . start ( ) <EOL> t . stop ( ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timer_object_rate_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_object_rate ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : - <NUM_LIT:1> ) <EOL> def test_timer_object_rate_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_object_rate ( cl , '<STR_LIT>' ) <EOL> def _test_timer_object_no_send_twice ( cl ) : <EOL> t = cl . timer ( '<STR_LIT:foo>' ) . start ( ) <EOL> t . stop ( ) <EOL> with assert_raises ( RuntimeError ) : <EOL> t . send ( ) <EOL> def test_timer_object_no_send_twice_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_object_no_send_twice ( cl ) <EOL> def test_timer_object_no_send_twice_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_object_no_send_twice ( cl ) <EOL> def _test_timer_send_without_stop ( cl ) : <EOL> with cl . timer ( '<STR_LIT:foo>' ) as t : <EOL> assert t . ms is None <EOL> with assert_raises ( RuntimeError ) : <EOL> t . send ( ) <EOL> t = cl . timer ( '<STR_LIT:bar>' ) . start ( ) <EOL> assert t . ms is None <EOL> with assert_raises ( RuntimeError ) : <EOL> t . send ( ) <EOL> def test_timer_send_without_stop_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_send_without_stop ( cl ) <EOL> def test_timer_send_without_stop_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_send_without_stop ( cl ) <EOL> def _test_timer_object_stop_without_start ( cl ) : <EOL> with assert_raises ( RuntimeError ) : <EOL> cl . timer ( '<STR_LIT:foo>' ) . stop ( ) <EOL> def test_timer_object_stop_without_start_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_timer_object_stop_without_start ( cl ) <EOL> def test_timer_object_stop_without_start_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_timer_object_stop_without_start ( cl ) <EOL> def _test_pipeline ( cl , proto ) : <EOL> pipe = cl . pipeline ( ) <EOL> pipe . incr ( '<STR_LIT:foo>' ) <EOL> pipe . decr ( '<STR_LIT:bar>' ) <EOL> pipe . timing ( '<STR_LIT>' , <NUM_LIT> ) <EOL> pipe . send ( ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> def test_pipeline_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline ( cl , '<STR_LIT>' ) <EOL> def test_pipeline_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline ( cl , '<STR_LIT>' ) <EOL> def _test_pipeline_null ( cl , proto ) : <EOL> pipe = cl . pipeline ( ) <EOL> pipe . send ( ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:0> , proto ) <EOL> def test_pipeline_null_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline_null ( cl , '<STR_LIT>' ) <EOL> def test_pipeline_null_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline_null ( cl , '<STR_LIT>' ) <EOL> def _test_pipeline_manager ( cl , proto ) : <EOL> with cl . pipeline ( ) as pipe : <EOL> pipe . incr ( '<STR_LIT:foo>' ) <EOL> pipe . decr ( '<STR_LIT:bar>' ) <EOL> pipe . gauge ( '<STR_LIT>' , <NUM_LIT:15> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> def test_pipeline_manager_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline_manager ( cl , '<STR_LIT>' ) <EOL> def test_pipeline_manager_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline_manager ( cl , '<STR_LIT>' ) <EOL> def _test_pipeline_timer_manager ( cl , proto ) : <EOL> with cl . pipeline ( ) as pipe : <EOL> with pipe . timer ( '<STR_LIT:foo>' ) : <EOL> pass <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_pipeline_timer_manager_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline_timer_manager ( cl , '<STR_LIT>' ) <EOL> def test_pipeline_timer_manager_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline_timer_manager ( cl , '<STR_LIT>' ) <EOL> def _test_pipeline_timer_decorator ( cl , proto ) : <EOL> with cl . pipeline ( ) as pipe : <EOL> @ pipe . timer ( '<STR_LIT:foo>' ) <EOL> def foo ( ) : <EOL> pass <EOL> foo ( ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_pipeline_timer_decorator_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline_timer_decorator ( cl , '<STR_LIT>' ) <EOL> def test_pipeline_timer_decorator_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline_timer_decorator ( cl , '<STR_LIT>' ) <EOL> def _test_pipeline_timer_object ( cl , proto ) : <EOL> with cl . pipeline ( ) as pipe : <EOL> t = pipe . timer ( '<STR_LIT:foo>' ) . start ( ) <EOL> t . stop ( ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:0> , proto ) <EOL> _timer_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_pipeline_timer_object_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline_timer_object ( cl , '<STR_LIT>' ) <EOL> def test_pipeline_timer_object_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline_timer_object ( cl , '<STR_LIT>' ) <EOL> def _test_pipeline_empty ( cl ) : <EOL> with cl . pipeline ( ) as pipe : <EOL> pipe . incr ( '<STR_LIT:foo>' ) <EOL> eq_ ( <NUM_LIT:1> , len ( pipe . _stats ) ) <EOL> eq_ ( <NUM_LIT:0> , len ( pipe . _stats ) ) <EOL> def test_pipeline_empty_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline_empty ( cl ) <EOL> def test_pipeline_empty_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline_empty ( cl ) <EOL> def _test_pipeline_negative_absolute_gauge ( cl , proto ) : <EOL> with cl . pipeline ( ) as pipe : <EOL> pipe . gauge ( '<STR_LIT:foo>' , - <NUM_LIT:10> , delta = False ) <EOL> pipe . incr ( '<STR_LIT:bar>' ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , '<STR_LIT>' ) <EOL> def test_pipeline_negative_absolute_gauge_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_pipeline_negative_absolute_gauge ( cl , '<STR_LIT>' ) <EOL> def test_pipeline_negative_absolute_gauge_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_pipeline_negative_absolute_gauge ( cl , '<STR_LIT>' ) <EOL> def _test_big_numbers ( cl , proto ) : <EOL> num = <NUM_LIT> <EOL> tests = ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ) <EOL> def _check ( method , result ) : <EOL> cl . _sock . reset_mock ( ) <EOL> getattr ( cl , method ) ( '<STR_LIT:foo>' , num ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , proto , result ) <EOL> for method , result in tests : <EOL> _check ( method , result ) <EOL> def test_big_numbers_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_big_numbers ( cl , '<STR_LIT>' ) <EOL> def test_big_numbers_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_big_numbers ( cl , '<STR_LIT>' ) <EOL> def _test_rate_no_send ( cl , proto ) : <EOL> cl . incr ( '<STR_LIT:foo>' , rate = <NUM_LIT:0.5> ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:0> , proto ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : <NUM_LIT:2> ) <EOL> def test_rate_no_send_udp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> _test_rate_no_send ( cl , '<STR_LIT>' ) <EOL> @ mock . patch . object ( random , '<STR_LIT>' , lambda : <NUM_LIT:2> ) <EOL> def test_rate_no_send_tcp ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _tcp_client ( ) <EOL> _test_rate_no_send ( cl , '<STR_LIT>' ) <EOL> def test_socket_error ( ) : <EOL> """<STR_LIT>""" <EOL> cl = _udp_client ( ) <EOL> cl . _sock . sendto . side_effect = socket . timeout ( ) <EOL> cl . incr ( '<STR_LIT:foo>' ) <EOL> _sock_check ( cl . _sock , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_pipeline_packet_size ( ) : <EOL> """<STR_LIT>""" <EOL> sc = _udp_client ( ) <EOL> pipe = sc . pipeline ( ) <EOL> for x in range ( <NUM_LIT:32> ) : <EOL> pipe . incr ( '<STR_LIT>' ) <EOL> pipe . send ( ) <EOL> eq_ ( <NUM_LIT:2> , sc . _sock . sendto . call_count ) <EOL> assert len ( sc . _sock . sendto . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <= <NUM_LIT> <EOL> assert len ( sc . _sock . sendto . call_args_list [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <= <NUM_LIT> <EOL> @ mock . patch . object ( socket , '<STR_LIT>' ) <EOL> def test_tcp_raises_exception_to_user ( mock_socket ) : <EOL> """<STR_LIT>""" <EOL> addr = ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) <EOL> cl = _tcp_client ( addr = addr [ <NUM_LIT:0> ] , port = addr [ <NUM_LIT:1> ] ) <EOL> cl . incr ( '<STR_LIT:foo>' ) <EOL> eq_ ( <NUM_LIT:1> , cl . _sock . sendall . call_count ) <EOL> cl . _sock . sendall . side_effect = socket . error <EOL> with assert_raises ( socket . error ) : <EOL> cl . incr ( '<STR_LIT:foo>' ) <EOL> @ mock . patch . object ( socket , '<STR_LIT>' ) <EOL> def test_tcp_timeout ( mock_socket ) : <EOL> """<STR_LIT>""" <EOL> test_timeout = <NUM_LIT> <EOL> cl = TCPStatsClient ( timeout = test_timeout ) <EOL> cl . incr ( '<STR_LIT:foo>' ) <EOL> cl . _sock . settimeout . assert_called_once_with ( test_timeout ) </s>
