<s> import sys <EOL> import time <EOL> import unittest <EOL> from StringIO import StringIO <EOL> from supervisor . process import ProcessStates <EOL> from superlance . tests . dummy import * <EOL> _NOW = time . time ( ) <EOL> _FAIL = [ { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : <NUM_LIT:11> , <EOL> '<STR_LIT:state>' : ProcessStates . RUNNING , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:start>' : _NOW - <NUM_LIT:100> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _NOW , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : <NUM_LIT:11> , <EOL> '<STR_LIT:state>' : ProcessStates . RUNNING , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:start>' : _NOW - <NUM_LIT:100> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _NOW , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> } , ] <EOL> def make_connection ( response , exc = None ) : <EOL> class TestConnection : <EOL> def __init__ ( self , hostport ) : <EOL> self . hostport = hostport <EOL> def request ( self , method , path ) : <EOL> if exc : <EOL> raise ValueError ( '<STR_LIT:foo>' ) <EOL> self . method = method <EOL> self . path = path <EOL> def getresponse ( self ) : <EOL> return response <EOL> return TestConnection <EOL> class HTTPOkTests ( unittest . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from superlance . httpok import HTTPOk <EOL> return HTTPOk <EOL> def _makeOne ( self , * opts ) : <EOL> return self . _getTargetClass ( ) ( * opts ) <EOL> def _makeOnePopulated ( self , programs , any , response = None , exc = None , <EOL> gcore = None , coredir = None , eager = True ) : <EOL> if response is None : <EOL> response = DummyResponse ( ) <EOL> rpc = DummyRPCServer ( ) <EOL> sendmail = '<STR_LIT>' <EOL> email = '<STR_LIT>' <EOL> url = '<STR_LIT>' <EOL> timeout = <NUM_LIT:10> <EOL> status = '<STR_LIT>' <EOL> inbody = None <EOL> gcore = gcore <EOL> coredir = coredir <EOL> prog = self . _makeOne ( rpc , programs , any , url , timeout , status , <EOL> inbody , email , sendmail , coredir , gcore , eager ) <EOL> prog . stdin = StringIO ( ) <EOL> prog . stdout = StringIO ( ) <EOL> prog . stderr = StringIO ( ) <EOL> prog . connclass = make_connection ( response , exc = exc ) <EOL> return prog <EOL> def test_listProcesses_no_programs ( self ) : <EOL> programs = [ ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any ) <EOL> specs = list ( prog . listProcesses ( ) ) <EOL> self . assertEqual ( len ( specs ) , <NUM_LIT:0> ) <EOL> def test_listProcesses_w_RUNNING_programs_default_state ( self ) : <EOL> programs = [ '<STR_LIT:foo>' ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any ) <EOL> specs = list ( prog . listProcesses ( ) ) <EOL> self . assertEqual ( len ( specs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( specs [ <NUM_LIT:0> ] , <EOL> DummySupervisorRPCNamespace . all_process_info [ <NUM_LIT:0> ] ) <EOL> def test_listProcesses_w_nonRUNNING_programs_default_state ( self ) : <EOL> programs = [ '<STR_LIT:bar>' ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any ) <EOL> specs = list ( prog . listProcesses ( ) ) <EOL> self . assertEqual ( len ( specs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( specs [ <NUM_LIT:0> ] , <EOL> DummySupervisorRPCNamespace . all_process_info [ <NUM_LIT:1> ] ) <EOL> def test_listProcesses_w_nonRUNNING_programs_RUNNING_state ( self ) : <EOL> programs = [ '<STR_LIT:bar>' ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any ) <EOL> specs = list ( prog . listProcesses ( ProcessStates . RUNNING ) ) <EOL> self . assertEqual ( len ( specs ) , <NUM_LIT:0> , ( prog . programs , specs ) ) <EOL> def test_runforever_eager_notatick ( self ) : <EOL> programs = { '<STR_LIT:foo>' : <NUM_LIT:0> , '<STR_LIT:bar>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:0> } <EOL> groups = { } <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any ) <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> self . assertEqual ( prog . stderr . getvalue ( ) , '<STR_LIT>' ) <EOL> def test_runforever_eager_error_on_request_some ( self ) : <EOL> programs = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any , exc = True ) <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> lines = prog . stderr . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:0> ] , <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> ) <EOL> self . assertEqual ( lines [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:4> ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:5> ] , <EOL> "<STR_LIT>" ) <EOL> mailed = prog . mailed . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( len ( mailed ) , <NUM_LIT:12> ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> def test_runforever_eager_error_on_request_any ( self ) : <EOL> programs = [ ] <EOL> any = True <EOL> prog = self . _makeOnePopulated ( programs , any , exc = True ) <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> lines = prog . stderr . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:4> ] , <EOL> '<STR_LIT>' ) <EOL> mailed = prog . mailed . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( len ( mailed ) , <NUM_LIT:11> ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> def test_runforever_eager_error_on_process_stop ( self ) : <EOL> programs = [ '<STR_LIT>' ] <EOL> any = False <EOL> prog = self . _makeOnePopulated ( programs , any , exc = True ) <EOL> prog . rpc . supervisor . all_process_info = _FAIL <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> lines = prog . stderr . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertEqual ( lines [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:2> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( lines [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> mailed = prog . mailed . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( len ( mailed ) , <NUM_LIT:10> ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> def test_runforever_eager_error_on_process_start ( self ) : <EOL> programs = [ '<STR_LIT>' ] <EOL> any = False <EOL> prog = self . _makeOnePopulated ( programs , any , exc = True ) <EOL> prog . rpc . supervisor . all_process_info = _FAIL <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> lines = prog . stderr . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( lines [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:2> ] , <EOL> "<STR_LIT>" ) <EOL> mailed = prog . mailed . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( len ( mailed ) , <NUM_LIT:9> ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> def test_runforever_eager_gcore ( self ) : <EOL> programs = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any , exc = True , gcore = "<STR_LIT:true>" , <EOL> coredir = "<STR_LIT>" ) <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> lines = prog . stderr . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:0> ] , <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> ) <EOL> self . assertEqual ( lines [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:3> ] , '<STR_LIT:U+0020>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:4> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:5> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:6> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:7> ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:8> ] , <EOL> "<STR_LIT>" ) <EOL> mailed = prog . mailed . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( len ( mailed ) , <NUM_LIT:15> ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> def test_runforever_not_eager_none_running ( self ) : <EOL> programs = [ '<STR_LIT:bar>' , '<STR_LIT>' ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any , exc = True , gcore = "<STR_LIT:true>" , <EOL> coredir = "<STR_LIT>" , eager = False ) <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> lines = filter ( None , prog . stderr . getvalue ( ) . split ( '<STR_LIT:\n>' ) ) <EOL> self . assertEqual ( len ( lines ) , <NUM_LIT:0> , lines ) <EOL> self . failIf ( '<STR_LIT>' in prog . __dict__ ) <EOL> def test_runforever_not_eager_running ( self ) : <EOL> programs = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] <EOL> any = None <EOL> prog = self . _makeOnePopulated ( programs , any , exc = True , eager = False ) <EOL> prog . stdin . write ( '<STR_LIT>' ) <EOL> prog . stdin . seek ( <NUM_LIT:0> ) <EOL> prog . runforever ( test = True ) <EOL> lines = filter ( None , prog . stderr . getvalue ( ) . split ( '<STR_LIT:\n>' ) ) <EOL> self . assertEqual ( lines [ <NUM_LIT:0> ] , <EOL> ( "<STR_LIT>" ) <EOL> ) <EOL> self . assertEqual ( lines [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( lines [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> mailed = prog . mailed . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( len ( mailed ) , <NUM_LIT:10> ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( mailed [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
