<s> import signal <EOL> import time <EOL> import pyuv <EOL> from gaffer . error import ProcessError , CommandError , CommandNotFound <EOL> from gaffer . controller import Controller , Command <EOL> from gaffer . manager import Manager <EOL> from gaffer . process import ProcessConfig <EOL> from test_manager import dummy_cmd <EOL> class TestCommand ( Command ) : <EOL> def __init__ ( self , name , args = None , kwargs = None ) : <EOL> super ( TestCommand , self ) . __init__ ( name , args = args , kwargs = kwargs ) <EOL> self . result = None <EOL> self . error = None <EOL> def reply ( self , result ) : <EOL> self . result = result <EOL> def reply_error ( self , error ) : <EOL> self . error = error <EOL> def init ( ) : <EOL> m = Manager ( ) <EOL> m . start ( ) <EOL> ctl = Controller ( m ) <EOL> testfile , cmd , args , wdir = dummy_cmd ( ) <EOL> config = ProcessConfig ( "<STR_LIT>" , cmd , args = args , cwd = wdir ) <EOL> return m , ctl , config <EOL> def init1 ( ) : <EOL> m = Manager ( ) <EOL> m . start ( ) <EOL> ctl = Controller ( m ) <EOL> testfile , cmd , args , wdir = dummy_cmd ( ) <EOL> config = ProcessConfig ( "<STR_LIT>" , cmd , args = args , cwd = wdir ) <EOL> return m , ctl , testfile , config <EOL> def test_basic ( ) : <EOL> m , ctl , conf = init ( ) <EOL> config = conf . to_dict ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ config ] , { "<STR_LIT:start>" : False } ) <EOL> ctl . process_command ( cmd ) <EOL> state = m . _get_locked_state ( "<STR_LIT>" ) <EOL> assert state . numprocesses == <NUM_LIT:1> <EOL> assert state . name == "<STR_LIT>" <EOL> assert state . cmd == config [ '<STR_LIT>' ] <EOL> assert state . config [ '<STR_LIT:args>' ] == config [ '<STR_LIT:args>' ] <EOL> assert state . config [ '<STR_LIT>' ] == config [ '<STR_LIT>' ] <EOL> cmd1 = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd1 ) <EOL> assert m . jobs ( ) == [ ] <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert cmd . error == None <EOL> assert cmd . result == { "<STR_LIT>" : True } <EOL> assert cmd1 . error == None <EOL> assert cmd1 . result == { "<STR_LIT>" : True } <EOL> def test_basic_error ( ) : <EOL> m , ctl , config = init ( ) <EOL> cmd = TestCommand ( "<STR_LIT:info>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert cmd . result == None <EOL> assert cmd . error == { "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : "<STR_LIT>" } <EOL> def test_command_not_found ( ) : <EOL> m , ctl , config = init ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" ) <EOL> ctl . process_command ( cmd ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert cmd . result == None <EOL> assert cmd . error == { "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : "<STR_LIT>" } <EOL> def test_bad_command ( ) : <EOL> m , ctl , config = init ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" ) <EOL> ctl . process_command ( cmd ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert cmd . result == None <EOL> assert cmd . error [ '<STR_LIT>' ] == <NUM_LIT> <EOL> def test_jobs ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> cmd = TestCommand ( "<STR_LIT>" ) <EOL> ctl . process_command ( cmd ) <EOL> cmd1 = TestCommand ( "<STR_LIT>" , [ "<STR_LIT:default>" ] ) <EOL> ctl . process_command ( cmd1 ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert "<STR_LIT>" in cmd . result <EOL> assert cmd . result [ '<STR_LIT>' ] == [ "<STR_LIT>" ] <EOL> assert isinstance ( cmd1 . result , dict ) <EOL> assert "<STR_LIT>" in cmd1 . result <EOL> assert cmd1 . result [ '<STR_LIT>' ] == [ "<STR_LIT>" ] <EOL> def test_sessions ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> cmd = TestCommand ( "<STR_LIT>" ) <EOL> ctl . process_command ( cmd ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert "<STR_LIT>" in cmd . result <EOL> assert cmd . result [ '<STR_LIT>' ] == [ "<STR_LIT:default>" ] <EOL> def test_pids ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> cmd = TestCommand ( "<STR_LIT>" ) <EOL> ctl . process_command ( cmd ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert "<STR_LIT>" in cmd . result <EOL> assert cmd . result [ '<STR_LIT>' ] == [ <NUM_LIT:1> ] <EOL> def test_reload ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> pids = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> pids1 = m . pids ( ) <EOL> jobs = m . jobs ( ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert cmd . result == { "<STR_LIT>" : True } <EOL> assert pids != pids1 <EOL> assert "<STR_LIT>" in jobs <EOL> def test_update ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> pids = m . pids ( ) <EOL> config [ '<STR_LIT>' ] = <NUM_LIT:2> <EOL> cmd = TestCommand ( "<STR_LIT>" , [ config . to_dict ( ) ] ) <EOL> ctl . process_command ( cmd ) <EOL> m . manage ( "<STR_LIT>" ) <EOL> pids1 = m . pids ( ) <EOL> jobs = m . jobs ( ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert cmd . result == { "<STR_LIT>" : True } <EOL> assert pids != pids1 <EOL> assert "<STR_LIT>" in jobs <EOL> assert len ( pids ) == <NUM_LIT:1> <EOL> assert len ( pids1 ) == <NUM_LIT:2> <EOL> def test_start_job ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config , start = False ) <EOL> pids = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> pids1 = m . pids ( ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert cmd . result == { "<STR_LIT>" : True } <EOL> assert pids != pids1 <EOL> assert len ( pids ) == <NUM_LIT:0> <EOL> assert len ( pids1 ) == <NUM_LIT:1> <EOL> def test_stop_job ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> pids = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> pids1 = m . pids ( ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert cmd . result == { "<STR_LIT>" : True } <EOL> assert pids != pids1 <EOL> assert len ( pids ) == <NUM_LIT:1> <EOL> assert len ( pids1 ) == <NUM_LIT:0> <EOL> def test_scale ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> pids = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" , <NUM_LIT:1> ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> pids1 = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" , - <NUM_LIT:1> ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> pids2 = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> pids3 = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> pids4 = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> pids5 = m . pids ( ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert len ( pids ) == <NUM_LIT:1> <EOL> assert len ( pids1 ) == <NUM_LIT:2> <EOL> assert len ( pids2 ) == <NUM_LIT:1> <EOL> assert len ( pids3 ) == <NUM_LIT:5> <EOL> assert len ( pids4 ) == <NUM_LIT:4> <EOL> assert len ( pids5 ) == <NUM_LIT:1> <EOL> def test_info ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> cmd = TestCommand ( "<STR_LIT:info>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> m . scale ( "<STR_LIT>" , <NUM_LIT:2> ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> cmd1 = TestCommand ( "<STR_LIT:info>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd1 ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert "<STR_LIT:info>" in cmd . result <EOL> info = cmd . result [ '<STR_LIT:info>' ] <EOL> assert info [ '<STR_LIT:name>' ] == "<STR_LIT>" <EOL> assert info [ '<STR_LIT>' ] == True <EOL> assert info [ '<STR_LIT>' ] == <NUM_LIT:1> <EOL> assert info [ '<STR_LIT>' ] == <NUM_LIT:1> <EOL> assert "<STR_LIT>" in info <EOL> assert info [ '<STR_LIT>' ] == config . to_dict ( ) <EOL> assert isinstance ( cmd1 . result , dict ) <EOL> assert "<STR_LIT:info>" in cmd1 . result <EOL> info1 = cmd1 . result [ '<STR_LIT:info>' ] <EOL> assert info1 [ '<STR_LIT:name>' ] == "<STR_LIT>" <EOL> assert info1 [ '<STR_LIT>' ] == <NUM_LIT:3> <EOL> assert info1 [ '<STR_LIT>' ] == config . to_dict ( ) <EOL> def test_stats ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> m . scale ( "<STR_LIT>" , <NUM_LIT:2> ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> cmd1 = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd1 ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert "<STR_LIT>" in cmd . result <EOL> stats = cmd . result [ '<STR_LIT>' ] <EOL> assert isinstance ( stats , dict ) <EOL> assert stats [ '<STR_LIT:name>' ] == "<STR_LIT>" <EOL> assert "<STR_LIT>" in stats <EOL> assert len ( stats [ '<STR_LIT>' ] ) == <NUM_LIT:1> <EOL> assert stats [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == <NUM_LIT:1> <EOL> assert isinstance ( cmd1 . result , dict ) <EOL> stats = cmd1 . result [ '<STR_LIT>' ] <EOL> assert isinstance ( stats , dict ) <EOL> assert stats [ '<STR_LIT:name>' ] == "<STR_LIT>" <EOL> assert "<STR_LIT>" in stats <EOL> assert len ( stats [ '<STR_LIT>' ] ) == <NUM_LIT:3> <EOL> def test_stopall ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> pids = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> jobs = m . jobs ( ) <EOL> pids1 = m . pids ( ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert len ( pids ) == <NUM_LIT:1> <EOL> assert jobs == [ '<STR_LIT>' ] <EOL> assert len ( pids1 ) == <NUM_LIT:0> <EOL> def test_killall ( ) : <EOL> m , ctl , testfile , config = init1 ( ) <EOL> m . load ( config ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" , signal . SIGHUP ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> m . stop_job ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert cmd . result == { "<STR_LIT>" : True } <EOL> with open ( testfile , '<STR_LIT:r>' ) as f : <EOL> res = f . read ( ) <EOL> assert res == '<STR_LIT>' <EOL> def test_process_info ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ <NUM_LIT:1> ] ) <EOL> ctl . process_command ( cmd ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert "<STR_LIT:info>" in cmd . result <EOL> info = cmd . result [ '<STR_LIT:info>' ] <EOL> assert info [ '<STR_LIT:name>' ] == "<STR_LIT>" <EOL> assert info [ '<STR_LIT>' ] == True <EOL> assert info [ '<STR_LIT>' ] == <NUM_LIT:1> <EOL> assert info [ '<STR_LIT>' ] == config [ '<STR_LIT>' ] <EOL> assert info [ '<STR_LIT:args>' ] == config [ '<STR_LIT:args>' ] <EOL> def test_process_stats ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ <NUM_LIT:1> ] ) <EOL> ctl . process_command ( cmd ) <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert isinstance ( cmd . result , dict ) <EOL> assert "<STR_LIT>" in cmd . result <EOL> stats = cmd . result [ '<STR_LIT>' ] <EOL> assert "<STR_LIT>" in stats <EOL> def test_stop_process ( ) : <EOL> m , ctl , config = init ( ) <EOL> m . load ( config ) <EOL> pids = m . pids ( ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ <NUM_LIT:1> ] ) <EOL> ctl . process_command ( cmd ) <EOL> result = [ ] <EOL> def wait ( h ) : <EOL> result . append ( ( m . jobs ( ) , m . pids ( ) ) ) <EOL> m . stop ( ) <EOL> t = pyuv . Timer ( m . loop ) <EOL> t . start ( wait , <NUM_LIT> , <NUM_LIT:0.0> ) <EOL> m . run ( ) <EOL> assert len ( result ) == <NUM_LIT:1> <EOL> jobs , pids1 = result [ <NUM_LIT:0> ] <EOL> assert len ( pids ) == <NUM_LIT:1> <EOL> assert jobs == [ '<STR_LIT>' ] <EOL> assert pids != pids1 <EOL> assert len ( pids1 ) == <NUM_LIT:1> <EOL> def test_kill_process ( ) : <EOL> m , ctl , testfile , config = init1 ( ) <EOL> m . load ( config ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ <NUM_LIT:1> , signal . SIGHUP ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> m . stop_job ( "<STR_LIT>" ) <EOL> t = pyuv . Timer ( m . loop ) <EOL> t . start ( lambda h : m . stop ( ) , <NUM_LIT> , <NUM_LIT:0.0> ) <EOL> m . run ( ) <EOL> assert cmd . result == { "<STR_LIT>" : True } <EOL> with open ( testfile , '<STR_LIT:r>' ) as f : <EOL> res = f . read ( ) <EOL> assert res == '<STR_LIT>' <EOL> def test_process_commit ( ) : <EOL> m , ctl , config = init ( ) <EOL> config [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> m . load ( config , start = False ) <EOL> cmd = TestCommand ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctl . process_command ( cmd ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> state = m . _get_locked_state ( "<STR_LIT>" ) <EOL> assert len ( state . running ) == <NUM_LIT:0> <EOL> assert state . numprocesses == <NUM_LIT:0> <EOL> assert len ( state . running_out ) == <NUM_LIT:1> <EOL> assert m . pids ( ) == [ <NUM_LIT:1> ] <EOL> m . stop ( ) <EOL> m . run ( ) <EOL> assert cmd . result [ "<STR_LIT>" ] == <NUM_LIT:1> </s>
