<s> import sys <EOL> import os <EOL> import tempfile <EOL> import six <EOL> from copy import copy <EOL> from circus . circusd import get_maxfd , daemonize , main <EOL> from circus import circusd <EOL> from circus . arbiter import Arbiter <EOL> from circus . util import REDIRECT_TO <EOL> from circus import util <EOL> from circus . tests . support import ( has_gevent , TestCase , skipIf , EasyTestSuite , <EOL> IS_WINDOWS ) <EOL> CIRCUS_INI = os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> class TestCircusd ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . saved = dict ( sys . modules ) <EOL> self . argv = copy ( sys . argv ) <EOL> self . starter = Arbiter . start <EOL> Arbiter . start = lambda x : None <EOL> self . exit = sys . exit <EOL> sys . exit = lambda x : None <EOL> self . _files = [ ] <EOL> if not IS_WINDOWS : <EOL> self . fork = os . fork <EOL> os . fork = self . _forking <EOL> self . setsid = os . setsid <EOL> os . setsid = lambda : None <EOL> self . dup2 = os . dup2 <EOL> os . dup2 = lambda x , y : None <EOL> self . forked = <NUM_LIT:0> <EOL> self . closerange = circusd . closerange <EOL> circusd . closerange = lambda x , y : None <EOL> self . open = os . open <EOL> os . open = self . _open <EOL> self . stop = Arbiter . stop <EOL> Arbiter . stop = lambda x : None <EOL> self . config = util . configure_logger <EOL> circusd . configure_logger = util . configure_logger = self . _logger <EOL> def _logger ( self , * args , ** kw ) : <EOL> pass <EOL> def _open ( self , path , * args , ** kw ) : <EOL> if path == REDIRECT_TO : <EOL> return <EOL> return self . open ( path , * args , ** kw ) <EOL> def tearDown ( self ) : <EOL> circusd . configure_logger = util . configure_logger = self . config <EOL> Arbiter . stop = self . stop <EOL> sys . argv = self . argv <EOL> os . open = self . open <EOL> circusd . closerange = self . closerange <EOL> sys . modules = self . saved <EOL> Arbiter . start = self . starter <EOL> sys . exit = self . exit <EOL> if not IS_WINDOWS : <EOL> os . fork = self . fork <EOL> os . dup2 = self . dup2 <EOL> os . setsid = self . setsid <EOL> for file in self . _files : <EOL> if os . path . exists ( file ) : <EOL> os . remove ( file ) <EOL> self . forked = <NUM_LIT:0> <EOL> def _forking ( self ) : <EOL> self . forked += <NUM_LIT:1> <EOL> return <NUM_LIT:0> <EOL> @ skipIf ( '<STR_LIT>' in os . environ , '<STR_LIT>' ) <EOL> @ skipIf ( not has_gevent ( ) , "<STR_LIT>" ) <EOL> def test_daemon ( self ) : <EOL> self . assertRaises ( ValueError , daemonize ) <EOL> for module in sys . modules . keys ( ) : <EOL> if module . startswith ( '<STR_LIT>' ) : <EOL> del sys . modules [ module ] <EOL> import gevent <EOL> sys . modules [ '<STR_LIT>' ] = gevent <EOL> self . assertRaises ( ValueError , daemonize ) <EOL> def test_maxfd ( self ) : <EOL> max = get_maxfd ( ) <EOL> self . assertTrue ( isinstance ( max , six . integer_types ) ) <EOL> @ skipIf ( has_gevent ( ) , "<STR_LIT>" ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> def test_daemonize ( self ) : <EOL> daemonize ( ) <EOL> self . assertEqual ( self . forked , <NUM_LIT:2> ) <EOL> def _get_file ( self ) : <EOL> fd , path = tempfile . mkstemp ( ) <EOL> os . close ( fd ) <EOL> self . _files . append ( path ) <EOL> return path <EOL> def test_main ( self ) : <EOL> def _check_pid ( cls ) : <EOL> self . assertTrue ( os . path . exists ( pid_file ) ) <EOL> Arbiter . start = _check_pid <EOL> pid_file = self . _get_file ( ) <EOL> sys . argv = [ '<STR_LIT>' , CIRCUS_INI , '<STR_LIT>' , pid_file ] <EOL> main ( ) <EOL> self . assertFalse ( os . path . exists ( pid_file ) ) <EOL> test_suite = EasyTestSuite ( __name__ ) </s>
