#NEWLINE# def test_status( self ) : #NEWLINE# #INDENT# conf_files = ( 'test-server/1.conf' , 'test-server/2.conf' , 'test-server/3.conf' , 'test-server/4.conf' ) #NEWLINE# pid_files = ( ( 'test-server/1.pid' , 1 ) , ( 'test-server/2.pid' , 2 ) , ( 'test-server/3.pid' , 3 ) , ( 'test-server/4.pid' , 4 ) ) #NEWLINE# with temptree ( conf_files ) as swift_dir : #NEWLINE# #INDENT# manager . SWIFT_DIR = swift_dir #NEWLINE# ( files , pids ) = zip ( * pid_files ) #NEWLINE# with temptree ( files , pids ) as t : #NEWLINE# #INDENT# manager . RUN_DIR = t #NEWLINE# server = manager . Server ( 'test' , run_dir = t ) #NEWLINE# old_stdout = sys . stdout #NEWLINE# try : #NEWLINE# #INDENT# with open ( os . path . join ( t , 'output' ) , 'w+' ) as f : #NEWLINE# #INDENT# sys . stdout = f #NEWLINE# manager . os = MockOs ( pids ) #NEWLINE# proc_files = ( ( '1/cmdline' , 'swift-test-server' ) , ( '2/cmdline' , 'swift-test-server' ) , ( '3/cmdline' , 'swift-test-server' ) , ( '4/cmdline' , 'swift-test-server' ) ) #NEWLINE# ( files , contents ) = zip ( * proc_files ) #NEWLINE# with temptree ( files , contents ) as t : #NEWLINE# #INDENT# manager . PROC_DIR = t #NEWLINE# self . assertEqual ( server . status ( ) , 0 ) #NEWLINE# output = pop_stream ( f ) . strip ( ) . splitlines ( ) #NEWLINE# self . assertEqual ( len ( output ) , 4 ) #NEWLINE# for line in output : #NEWLINE# #INDENT# self . assertTrue ( ( 'test-server running' in line ) ) #NEWLINE# #UNINDENT# #UNINDENT# with temptree ( [ ] , [ ] ) as t : #NEWLINE# #INDENT# manager . PROC_DIR = t #NEWLINE# self . assertEqual ( server . status ( number = 4 ) , 0 ) #NEWLINE# output = pop_stream ( f ) . strip ( ) . splitlines ( ) #NEWLINE# self . assertEqual ( len ( output ) , 1 ) #NEWLINE# line = output [ 0 ] #NEWLINE# self . assertTrue ( ( 'test-server running' in line ) ) #NEWLINE# conf_four = self . join_swift_dir ( conf_files [ 3 ] ) #NEWLINE# self . assertTrue ( ( ( '4 - %s' % conf_four ) in line ) ) #NEWLINE# #UNINDENT# manager . os = MockOs ( [ 1 , 2 , 3 ] ) #NEWLINE# proc_files = ( ( '1/cmdline' , 'swift-test-server' ) , ( '2/cmdline' , 'swift-test-server' ) , ( '3/cmdline' , 'swift-test-server' ) ) #NEWLINE# ( files , contents ) = zip ( * proc_files ) #NEWLINE# with temptree ( files , contents ) as t : #NEWLINE# #INDENT# manager . PROC_DIR = t #NEWLINE# self . assertEqual ( server . status ( ) , 0 ) #NEWLINE# output = pop_stream ( f ) . strip ( ) . splitlines ( ) #NEWLINE# self . assertEqual ( len ( output ) , 3 ) #NEWLINE# for line in output : #NEWLINE# #INDENT# self . assertTrue ( ( 'test-server running' in line ) ) #NEWLINE# #UNINDENT# #UNINDENT# manager . os = MockOs ( [ 1 , 2 ] ) #NEWLINE# proc_files = ( ( '1/cmdline' , 'swift-test-server' ) , ( '2/cmdline' , 'swift-test-server' ) ) #NEWLINE# ( files , contents ) = zip ( * proc_files ) #NEWLINE# with temptree ( files , contents ) as t : #NEWLINE# #INDENT# manager . PROC_DIR = t #NEWLINE# self . assertEqual ( server . status ( number = 3 ) , 1 ) #NEWLINE# output = pop_stream ( f ) . strip ( ) . splitlines ( ) #NEWLINE# self . assertEqual ( len ( output ) , 1 ) #NEWLINE# line = output [ 0 ] #NEWLINE# self . assertTrue ( ( 'not running' in line ) ) #NEWLINE# conf_three = self . join_swift_dir ( conf_files [ 2 ] ) #NEWLINE# self . assertTrue ( ( conf_three in line ) ) #NEWLINE# #UNINDENT# manager . os = MockOs ( [ ] ) #NEWLINE# with temptree ( [ ] , [ ] ) as t : #NEWLINE# #INDENT# manager . PROC_DIR = t #NEWLINE# self . assertEqual ( server . status ( ) , 1 ) #NEWLINE# output = pop_stream ( f ) . lower ( ) #NEWLINE# self . assertTrue ( ( 'no test-server running' in output ) ) #NEWLINE# #UNINDENT# pids = { #NEWLINE# #INDENT# 1 : '1.pid' , #NEWLINE# 2 : '2.pid' , #NEWLINE# #UNINDENT# } #NEWLINE# called = [ ] #NEWLINE# def mock( * args , **kwargs ) : #NEWLINE# #INDENT# called . append ( True ) #NEWLINE# #UNINDENT# server . get_running_pids = mock #NEWLINE# status = server . status ( pids = pids ) #NEWLINE# self . assertEqual ( status , 0 ) #NEWLINE# self . assertFalse ( called ) #NEWLINE# output = pop_stream ( f ) . strip ( ) . splitlines ( ) #NEWLINE# self . assertEqual ( len ( output ) , 2 ) #NEWLINE# for line in output : #NEWLINE# #INDENT# self . assertTrue ( ( 'test-server running' in line ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# finally : #NEWLINE# #INDENT# sys . stdout = old_stdout