<s> import subprocess <EOL> import shlex <EOL> from multiprocessing import Process , Queue <EOL> from tornado . testing import gen_test <EOL> from tornado . gen import coroutine , Return <EOL> from circus . circusctl import USAGE , VERSION , CircusCtl <EOL> from circus . tests . support import ( TestCircus , async_poll_for , EasyTestSuite , <EOL> skipIf , DEBUG , PYTHON , SLEEP ) <EOL> from circus . util import tornado_sleep , DEFAULT_ENDPOINT_DEALER <EOL> from circus . py3compat import b , s <EOL> def run_ctl ( args , queue = None , stdin = '<STR_LIT>' , endpoint = DEFAULT_ENDPOINT_DEALER ) : <EOL> cmd = '<STR_LIT>' % PYTHON <EOL> if '<STR_LIT>' not in args : <EOL> args = '<STR_LIT>' % endpoint + args <EOL> proc = subprocess . Popen ( cmd . split ( ) + shlex . split ( args ) , <EOL> stdin = subprocess . PIPE if stdin else None , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> stdout , stderr = proc . communicate ( b ( stdin ) if stdin else None ) <EOL> stdout = s ( stdout ) <EOL> stderr = s ( stderr ) <EOL> if queue : <EOL> queue . put ( stderr ) <EOL> queue . put ( stdout ) <EOL> try : <EOL> import gevent <EOL> if hasattr ( gevent , '<STR_LIT>' ) : <EOL> gevent . shutdown ( ) <EOL> except ImportError : <EOL> pass <EOL> return stdout , stderr <EOL> @ coroutine <EOL> def async_run_ctl ( args , stdin = '<STR_LIT>' , endpoint = DEFAULT_ENDPOINT_DEALER ) : <EOL> """<STR_LIT>""" <EOL> queue = Queue ( ) <EOL> circusctl_process = Process ( target = run_ctl , <EOL> args = ( args , queue , stdin , <EOL> endpoint ) ) <EOL> circusctl_process . start ( ) <EOL> while queue . empty ( ) : <EOL> yield tornado_sleep ( <NUM_LIT> ) <EOL> stderr = queue . get ( ) <EOL> stdout = queue . get ( ) <EOL> raise Return ( ( stdout , stderr ) ) <EOL> class CommandlineTest ( TestCircus ) : <EOL> @ skipIf ( DEBUG , '<STR_LIT>' ) <EOL> @ gen_test <EOL> def test_help_switch_no_command ( self ) : <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT>' ) <EOL> if stderr : <EOL> self . assertIn ( '<STR_LIT>' , stderr ) <EOL> output = stdout . splitlines ( ) <EOL> self . assertEqual ( output [ <NUM_LIT:0> ] , '<STR_LIT>' + USAGE ) <EOL> self . assertEqual ( output [ <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( output [ <NUM_LIT:4> ] , '<STR_LIT>' ) <EOL> @ gen_test <EOL> def test_help_invalid_command ( self ) : <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT:foo>' ) <EOL> self . assertEqual ( stdout , '<STR_LIT>' ) <EOL> err = stderr . splitlines ( ) <EOL> while err and '<STR_LIT>' in err [ <NUM_LIT:0> ] : <EOL> del err [ <NUM_LIT:0> ] <EOL> self . assertEqual ( err [ <NUM_LIT:0> ] , '<STR_LIT>' + USAGE ) <EOL> self . assertEqual ( err [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> @ skipIf ( DEBUG , '<STR_LIT>' ) <EOL> @ gen_test <EOL> def test_help_for_add_command ( self ) : <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT>' ) <EOL> if stderr : <EOL> self . assertIn ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( stdout . splitlines ( ) [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> @ skipIf ( DEBUG , '<STR_LIT>' ) <EOL> @ gen_test <EOL> def test_add ( self ) : <EOL> yield self . start_arbiter ( ) <EOL> async_poll_for ( self . test_file , '<STR_LIT>' ) <EOL> ep = self . arbiter . endpoint <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT>' % SLEEP % <NUM_LIT:1> , <EOL> endpoint = ep ) <EOL> if stderr : <EOL> self . assertIn ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( stdout . strip ( ) , '<STR_LIT>' ) <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT>' , <EOL> endpoint = ep ) <EOL> if stderr : <EOL> self . assertIn ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( stdout . strip ( ) , '<STR_LIT>' ) <EOL> yield self . stop_arbiter ( ) <EOL> @ skipIf ( DEBUG , '<STR_LIT>' ) <EOL> @ gen_test <EOL> def test_add_start ( self ) : <EOL> yield self . start_arbiter ( ) <EOL> async_poll_for ( self . test_file , '<STR_LIT>' ) <EOL> ep = self . arbiter . endpoint <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT>' <EOL> % SLEEP % <NUM_LIT:1> , <EOL> endpoint = ep ) <EOL> if stderr : <EOL> self . assertIn ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( stdout . strip ( ) , '<STR_LIT>' ) <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT>' , <EOL> endpoint = ep ) <EOL> if stderr : <EOL> self . assertIn ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( stdout . strip ( ) , '<STR_LIT>' ) <EOL> yield self . stop_arbiter ( ) <EOL> class CLITest ( TestCircus ) : <EOL> @ coroutine <EOL> def run_ctl ( self , command = '<STR_LIT>' , endpoint = DEFAULT_ENDPOINT_DEALER ) : <EOL> """<STR_LIT>""" <EOL> if command : <EOL> command += '<STR_LIT:\n>' <EOL> stdout , stderr = yield async_run_ctl ( '<STR_LIT>' , command + '<STR_LIT>' , <EOL> endpoint = endpoint ) <EOL> raise Return ( ( stdout , stderr ) ) <EOL> @ skipIf ( DEBUG , '<STR_LIT>' ) <EOL> @ gen_test <EOL> def test_launch_cli ( self ) : <EOL> yield self . start_arbiter ( ) <EOL> async_poll_for ( self . test_file , '<STR_LIT>' ) <EOL> stdout , stderr = yield self . run_ctl ( endpoint = self . arbiter . endpoint ) <EOL> if stderr : <EOL> self . assertIn ( '<STR_LIT>' , stderr ) <EOL> output = stdout . splitlines ( ) <EOL> self . assertEqual ( output [ <NUM_LIT:0> ] , VERSION ) <EOL> if not output [ <NUM_LIT:2> ] . startswith ( CircusCtl . prompt ) : <EOL> prompt = output [ <NUM_LIT:2> ] [ - len ( CircusCtl . prompt ) : ] <EOL> self . assertEqual ( prompt , CircusCtl . prompt ) <EOL> yield self . stop_arbiter ( ) <EOL> @ gen_test <EOL> def test_cli_help ( self ) : <EOL> yield self . start_arbiter ( ) <EOL> stdout , stderr = yield self . run_ctl ( '<STR_LIT>' , <EOL> endpoint = self . arbiter . endpoint ) <EOL> self . assertEqual ( stderr , '<STR_LIT>' ) <EOL> prompt = stdout . splitlines ( ) <EOL> self . assertEqual ( prompt [ <NUM_LIT:3> ] , "<STR_LIT>" ) <EOL> yield self . stop_arbiter ( ) <EOL> test_suite = EasyTestSuite ( __name__ ) </s>
