#NEWLINE# def test_redis_replication_service_check( self ) : #NEWLINE# #INDENT# check_name = 'redis.replication.master_link_status' #NEWLINE# r = load_check ( 'redisdb' , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# def extract_check( instance ) : #NEWLINE# #INDENT# r . check ( instance ) #NEWLINE# checks = [ c for c in r . get_service_checks ( ) if ( c [ 'check' ] == check_name ) ] #NEWLINE# return ( ( checks and checks [ 0 ] ) or None ) #NEWLINE# #UNINDENT# time . sleep ( 5 ) #NEWLINE# check = extract_check ( { #NEWLINE# #INDENT# 'host' : 'localhost' , #NEWLINE# 'port' : SLAVE_HEALTHY_PORT , #NEWLINE# #UNINDENT# } ) #NEWLINE# assert check , ( '%s service check not returned' % check_name ) #NEWLINE# self . assertEqual ( check [ 'status' ] , AgentCheck . OK , ( 'Value of %s service check should be OK' % check_name ) ) #NEWLINE# check = extract_check ( { #NEWLINE# #INDENT# 'host' : 'localhost' , #NEWLINE# 'port' : SLAVE_UNHEALTHY_PORT , #NEWLINE# #UNINDENT# } ) #NEWLINE# self . assertEqual ( check [ 'status' ] , AgentCheck . CRITICAL , ( 'Value of %s service check should be CRITICAL' % check_name ) )