<s> import os <EOL> from subprocess import PIPE , Popen <EOL> import time <EOL> import memcache <EOL> from nose . plugins . attrib import attr <EOL> from checks import AgentCheck <EOL> from tests . checks . common import AgentCheckTest <EOL> GAUGES = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ] <EOL> RATES = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ] <EOL> SERVICE_CHECK = '<STR_LIT>' <EOL> PORT = <NUM_LIT> <EOL> @ attr ( requires = '<STR_LIT>' ) <EOL> class TestMemCache ( AgentCheckTest ) : <EOL> CHECK_NAME = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> c = memcache . Client ( [ "<STR_LIT>" . format ( PORT ) ] ) <EOL> c . set ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) <EOL> c . get ( "<STR_LIT:foo>" ) <EOL> def testCoverage ( self ) : <EOL> config = { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" } , <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" , '<STR_LIT:port>' : PORT , '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" , '<STR_LIT:port>' : PORT , '<STR_LIT>' : [ '<STR_LIT:foo>' ] } , <EOL> { '<STR_LIT>' : "<STR_LIT>" } <EOL> ] <EOL> } <EOL> self . assertRaises ( Exception , self . run_check , config ) <EOL> tag_set = [ <EOL> [ "<STR_LIT>" ] , <EOL> [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> [ "<STR_LIT>" , "<STR_LIT:foo>" ] <EOL> ] <EOL> for tags in tag_set : <EOL> for m in GAUGES : <EOL> self . assertMetric ( "<STR_LIT>" . format ( m ) , tags = tags , count = <NUM_LIT:1> ) <EOL> good_service_check_tags = [ "<STR_LIT>" , "<STR_LIT>" . format ( PORT ) ] <EOL> bad_service_check_tags = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> self . assertServiceCheck ( <EOL> SERVICE_CHECK , status = AgentCheck . OK , <EOL> tags = good_service_check_tags , count = <NUM_LIT:3> ) <EOL> self . assertServiceCheck ( <EOL> SERVICE_CHECK , status = AgentCheck . CRITICAL , <EOL> tags = bad_service_check_tags , count = <NUM_LIT:1> ) <EOL> self . coverage_report ( ) <EOL> config = { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" } , <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" , '<STR_LIT:port>' : PORT , '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" , '<STR_LIT:port>' : PORT , '<STR_LIT>' : [ '<STR_LIT:foo>' ] } , <EOL> ] <EOL> } <EOL> self . run_check_twice ( config , force_reload = True ) <EOL> for tags in tag_set : <EOL> for m in GAUGES : <EOL> self . assertMetric ( "<STR_LIT>" . format ( m ) , tags = tags , count = <NUM_LIT:1> ) <EOL> for m in RATES : <EOL> self . assertMetric ( <EOL> "<STR_LIT>" . format ( m ) , tags = tags , count = <NUM_LIT:1> ) <EOL> good_service_check_tags = [ "<STR_LIT>" , "<STR_LIT>" . format ( PORT ) ] <EOL> self . assertServiceCheck ( <EOL> SERVICE_CHECK , status = AgentCheck . OK , <EOL> tags = good_service_check_tags , count = <NUM_LIT:3> ) <EOL> self . coverage_report ( ) <EOL> def _countConnections ( self , port ) : <EOL> pid = os . getpid ( ) <EOL> p1 = Popen ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' % pid , '<STR_LIT>' ] , stdout = PIPE ) <EOL> p2 = Popen ( [ "<STR_LIT>" , "<STR_LIT>" % port ] , stdin = p1 . stdout , stdout = PIPE ) <EOL> p3 = Popen ( [ "<STR_LIT>" , "<STR_LIT>" ] , stdin = p2 . stdout , stdout = PIPE ) <EOL> output = p3 . communicate ( ) [ <NUM_LIT:0> ] <EOL> return int ( output . strip ( ) ) <EOL> def testConnectionLeaks ( self ) : <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . assertEquals ( self . _countConnections ( <NUM_LIT> ) , <NUM_LIT:0> ) <EOL> new_conf = { '<STR_LIT>' : { } , '<STR_LIT>' : [ <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" } ] <EOL> } <EOL> self . run_check ( new_conf ) <EOL> self . assertEquals ( self . _countConnections ( <NUM_LIT> ) , <NUM_LIT:0> ) <EOL> def testMemoryLeak ( self ) : <EOL> config = { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" } , <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" , '<STR_LIT:port>' : PORT , '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT:url>' : "<STR_LIT:localhost>" , '<STR_LIT:port>' : PORT , '<STR_LIT>' : [ '<STR_LIT:foo>' ] } , <EOL> ] <EOL> } <EOL> self . run_check ( config ) <EOL> import gc <EOL> if not self . is_travis ( ) : <EOL> gc . set_debug ( gc . DEBUG_LEAK ) <EOL> gc . collect ( ) <EOL> try : <EOL> start = len ( gc . garbage ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> self . run_check ( config ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . check . get_metrics ( ) <EOL> end = len ( gc . garbage ) <EOL> self . assertEquals ( end - start , <NUM_LIT:0> , gc . garbage ) <EOL> finally : <EOL> gc . set_debug ( <NUM_LIT:0> ) </s>
