<s> import os <EOL> import signal <EOL> import time <EOL> import unittest <EOL> import sys <EOL> import errno <EOL> from mock import Mock , patch_object , sentinel <EOL> from supervisor . tests . base import DummyOptions <EOL> from supervisor . tests . base import DummyPConfig <EOL> from supervisor . tests . base import DummyProcess <EOL> from supervisor . tests . base import DummyPGroupConfig <EOL> from supervisor . tests . base import DummyDispatcher <EOL> from supervisor . tests . base import DummyEvent <EOL> from supervisor . tests . base import DummyFCGIGroupConfig <EOL> from supervisor . tests . base import DummySocketConfig <EOL> from supervisor . tests . base import DummyProcessGroup <EOL> from supervisor . tests . base import DummyFCGIProcessGroup <EOL> from supervisor . tests . base import DummySocketManager <EOL> from supervisor . process import Subprocess <EOL> class SubprocessTests ( unittest . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from supervisor . process import Subprocess <EOL> return Subprocess <EOL> def _makeOne ( self , * arg , ** kw ) : <EOL> return self . _getTargetClass ( ) ( * arg , ** kw ) <EOL> def tearDown ( self ) : <EOL> from supervisor . events import clear <EOL> clear ( ) <EOL> def test_getProcessStateDescription ( self ) : <EOL> from supervisor . states import ProcessStates <EOL> from supervisor . process import getProcessStateDescription <EOL> for statename , code in ProcessStates . __dict__ . items ( ) : <EOL> self . assertEqual ( getProcessStateDescription ( code ) , statename ) <EOL> def test_ctor ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' , <EOL> stderr_logfile = '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> self . assertEqual ( instance . config , config ) <EOL> self . assertEqual ( instance . config . options , options ) <EOL> self . assertEqual ( instance . laststart , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . pid , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . laststart , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . laststop , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . delay , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . administrative_stop , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . backoff , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . pipes , { } ) <EOL> self . assertEqual ( instance . dispatchers , { } ) <EOL> self . assertEqual ( instance . spawnerr , None ) <EOL> def test_repr ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> s = repr ( instance ) <EOL> self . assertTrue ( s . startswith ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( s . endswith ( '<STR_LIT>' ) ) <EOL> def test_reopenlogs ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . dispatchers = { <NUM_LIT:0> : DummyDispatcher ( readable = True ) , <EOL> <NUM_LIT:1> : DummyDispatcher ( writable = True ) } <EOL> instance . reopenlogs ( ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:0> ] . logs_reopened , True ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:1> ] . logs_reopened , False ) <EOL> def test_removelogs ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . dispatchers = { <NUM_LIT:0> : DummyDispatcher ( readable = True ) , <EOL> <NUM_LIT:1> : DummyDispatcher ( writable = True ) } <EOL> instance . removelogs ( ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:0> ] . logs_removed , True ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:1> ] . logs_removed , False ) <EOL> def test_drain ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' , <EOL> stderr_logfile = '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . dispatchers = { <NUM_LIT:0> : DummyDispatcher ( readable = True ) , <EOL> <NUM_LIT:1> : DummyDispatcher ( writable = True ) } <EOL> instance . drain ( ) <EOL> self . assertTrue ( instance . dispatchers [ <NUM_LIT:0> ] . read_event_handled ) <EOL> self . assertTrue ( instance . dispatchers [ <NUM_LIT:1> ] . write_event_handled ) <EOL> def test_get_execv_args_abs_missing ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> args = instance . get_execv_args ( ) <EOL> self . assertEqual ( args , ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> def test_get_execv_args_abs_withquotes_missing ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> args = instance . get_execv_args ( ) <EOL> self . assertEqual ( args , ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> def test_get_execv_args_rel_missing ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> args = instance . get_execv_args ( ) <EOL> self . assertEqual ( args , ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> def test_get_execv_args_rel_withquotes_missing ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> args = instance . get_execv_args ( ) <EOL> self . assertEqual ( args , ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> def test_get_execv_args_abs ( self ) : <EOL> executable = '<STR_LIT>' <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , executable ) <EOL> instance = self . _makeOne ( config ) <EOL> args = instance . get_execv_args ( ) <EOL> self . assertEqual ( len ( args ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( args [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( args [ <NUM_LIT:1> ] , [ '<STR_LIT>' , '<STR_LIT:foo>' ] ) <EOL> def test_get_execv_args_rel ( self ) : <EOL> executable = '<STR_LIT>' <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , executable ) <EOL> instance = self . _makeOne ( config ) <EOL> args = instance . get_execv_args ( ) <EOL> self . assertEqual ( len ( args ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( args [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( args [ <NUM_LIT:1> ] , [ '<STR_LIT>' , '<STR_LIT:foo>' ] ) <EOL> def test_record_spawnerr ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . record_spawnerr ( '<STR_LIT:foo>' ) <EOL> self . assertEqual ( instance . spawnerr , '<STR_LIT:foo>' ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> def test_spawn_already_running ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . pid = True <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . RUNNING <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertEqual ( instance . state , ProcessStates . RUNNING ) <EOL> def test_spawn_fail_check_execv_args ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . BACKOFF <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( instance . spawnerr , '<STR_LIT>' ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . failUnless ( instance . delay ) <EOL> self . failUnless ( instance . backoff ) <EOL> from supervisor . states import ProcessStates <EOL> self . assertEqual ( instance . state , ProcessStates . BACKOFF ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> ) <EOL> event1 = L [ <NUM_LIT:0> ] <EOL> event2 = L [ <NUM_LIT:1> ] <EOL> self . assertEqual ( event1 . __class__ , events . ProcessStateStartingEvent ) <EOL> self . assertEqual ( event2 . __class__ , events . ProcessStateBackoffEvent ) <EOL> def test_spawn_fail_make_pipes_emfile ( self ) : <EOL> options = DummyOptions ( ) <EOL> import errno <EOL> options . make_pipes_error = errno . EMFILE <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . BACKOFF <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( instance . spawnerr , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . failUnless ( instance . delay ) <EOL> self . failUnless ( instance . backoff ) <EOL> from supervisor . states import ProcessStates <EOL> self . assertEqual ( instance . state , ProcessStates . BACKOFF ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> ) <EOL> event1 , event2 = L <EOL> self . assertEqual ( event1 . __class__ , events . ProcessStateStartingEvent ) <EOL> self . assertEqual ( event2 . __class__ , events . ProcessStateBackoffEvent ) <EOL> def test_spawn_fail_make_pipes_other ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . make_pipes_error = <NUM_LIT:1> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . BACKOFF <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( instance . spawnerr , '<STR_LIT>' ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . failUnless ( instance . delay ) <EOL> self . failUnless ( instance . backoff ) <EOL> from supervisor . states import ProcessStates <EOL> self . assertEqual ( instance . state , ProcessStates . BACKOFF ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> ) <EOL> event1 , event2 = L <EOL> self . assertEqual ( event1 . __class__ , events . ProcessStateStartingEvent ) <EOL> self . assertEqual ( event2 . __class__ , events . ProcessStateBackoffEvent ) <EOL> def test_spawn_fork_fail_eagain ( self ) : <EOL> options = DummyOptions ( ) <EOL> import errno <EOL> options . fork_error = errno . EAGAIN <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . BACKOFF <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( instance . spawnerr , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( len ( options . parent_pipes_closed ) , <NUM_LIT:6> ) <EOL> self . assertEqual ( len ( options . child_pipes_closed ) , <NUM_LIT:6> ) <EOL> self . failUnless ( instance . delay ) <EOL> self . failUnless ( instance . backoff ) <EOL> from supervisor . states import ProcessStates <EOL> self . assertEqual ( instance . state , ProcessStates . BACKOFF ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> ) <EOL> event1 , event2 = L <EOL> self . assertEqual ( event1 . __class__ , events . ProcessStateStartingEvent ) <EOL> self . assertEqual ( event2 . __class__ , events . ProcessStateBackoffEvent ) <EOL> def test_spawn_fork_fail_other ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . fork_error = <NUM_LIT:1> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . BACKOFF <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( instance . spawnerr , '<STR_LIT>' ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( len ( options . parent_pipes_closed ) , <NUM_LIT:6> ) <EOL> self . assertEqual ( len ( options . child_pipes_closed ) , <NUM_LIT:6> ) <EOL> self . failUnless ( instance . delay ) <EOL> self . failUnless ( instance . backoff ) <EOL> from supervisor . states import ProcessStates <EOL> self . assertEqual ( instance . state , ProcessStates . BACKOFF ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> ) <EOL> event1 , event2 = L <EOL> self . assertEqual ( event1 . __class__ , events . ProcessStateStartingEvent ) <EOL> self . assertEqual ( event2 . __class__ , events . ProcessStateBackoffEvent ) <EOL> def test_spawn_as_child_setuid_ok ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( options . child_pipes_closed , None ) <EOL> self . assertEqual ( options . pgrp_set , True ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> self . assertEqual ( options . written , { } ) <EOL> self . assertEqual ( options . privsdropped , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . execv_args , <EOL> ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> def test_spawn_as_child_setuid_fail ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> options . setuid_msg = '<STR_LIT>' <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( options . child_pipes_closed , None ) <EOL> self . assertEqual ( options . pgrp_set , True ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> self . assertEqual ( options . written , <EOL> { <NUM_LIT:2> : '<STR_LIT>' } ) <EOL> self . assertEqual ( options . privsdropped , None ) <EOL> self . assertEqual ( options . execv_args , <EOL> ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> def test_spawn_as_child_cwd_ok ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> directory = '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( options . child_pipes_closed , None ) <EOL> self . assertEqual ( options . pgrp_set , True ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> self . assertEqual ( options . written , { } ) <EOL> self . assertEqual ( options . execv_args , <EOL> ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> self . assertEqual ( options . changed_directory , True ) <EOL> def test_spawn_as_child_sets_umask ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , umask = <NUM_LIT> <NUM_LIT:2> ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . written , { } ) <EOL> self . assertEqual ( options . execv_args , <EOL> ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> self . assertEqual ( options . umaskset , <NUM_LIT> <NUM_LIT:2> ) <EOL> def test_spawn_as_child_cwd_fail ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> options . chdir_error = <NUM_LIT:2> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> directory = '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( options . child_pipes_closed , None ) <EOL> self . assertEqual ( options . pgrp_set , True ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> self . assertEqual ( options . execv_args , None ) <EOL> self . assertEqual ( options . written , <EOL> { <NUM_LIT:2> : "<STR_LIT>" } ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> self . assertEqual ( options . changed_directory , False ) <EOL> def test_spawn_as_child_execv_fail_oserror ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> options . execv_error = <NUM_LIT:1> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( options . child_pipes_closed , None ) <EOL> self . assertEqual ( options . pgrp_set , True ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> self . assertEqual ( options . written , <EOL> { <NUM_LIT:2> : "<STR_LIT>" } ) <EOL> self . assertEqual ( options . privsdropped , None ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> def test_spawn_as_child_execv_fail_runtime_error ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> options . execv_error = <NUM_LIT:2> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( options . child_pipes_closed , None ) <EOL> self . assertEqual ( options . pgrp_set , True ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> msg = options . written [ <NUM_LIT:2> ] <EOL> self . failUnless ( msg . startswith ( "<STR_LIT>" ) ) <EOL> self . failUnless ( "<STR_LIT>" in msg ) <EOL> self . assertEqual ( options . privsdropped , None ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> def test_spawn_as_child_uses_pconfig_environment ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> environment = { '<STR_LIT>' : '<STR_LIT:1>' } ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . execv_args , ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( options . execv_environment [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> def test_spawn_as_child_environment_supervisor_envvars ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> class Dummy : <EOL> name = '<STR_LIT>' <EOL> instance . group = Dummy ( ) <EOL> instance . group . config = Dummy ( ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . execv_args , ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <EOL> options . execv_environment [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> self . assertEqual ( <EOL> options . execv_environment [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> options . execv_environment [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> options . execv_environment [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_spawn_as_child_stderr_redirected ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> config . redirect_stderr = True <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( options . child_pipes_closed , None ) <EOL> self . assertEqual ( options . pgrp_set , True ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> self . assertEqual ( options . written , { } ) <EOL> self . assertEqual ( options . privsdropped , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . execv_args , <EOL> ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( options . _exitcode , <NUM_LIT> ) <EOL> def test_spawn_as_parent ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:10> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , <NUM_LIT:10> ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:4> ] . __class__ , DummyDispatcher ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:5> ] . __class__ , DummyDispatcher ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:7> ] . __class__ , DummyDispatcher ) <EOL> self . assertEqual ( instance . pipes [ '<STR_LIT>' ] , <NUM_LIT:4> ) <EOL> self . assertEqual ( instance . pipes [ '<STR_LIT>' ] , <NUM_LIT:5> ) <EOL> self . assertEqual ( instance . pipes [ '<STR_LIT>' ] , <NUM_LIT:7> ) <EOL> self . assertEqual ( options . parent_pipes_closed , None ) <EOL> self . assertEqual ( len ( options . child_pipes_closed ) , <NUM_LIT:6> ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertEqual ( instance . spawnerr , None ) <EOL> self . failUnless ( instance . delay ) <EOL> self . assertEqual ( instance . config . options . pidhistory [ <NUM_LIT:10> ] , instance ) <EOL> from supervisor . states import ProcessStates <EOL> self . assertEqual ( instance . state , ProcessStates . STARTING ) <EOL> def test_spawn_redirect_stderr ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:10> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> redirect_stderr = True ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , <NUM_LIT:10> ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:4> ] . __class__ , DummyDispatcher ) <EOL> self . assertEqual ( instance . dispatchers [ <NUM_LIT:5> ] . __class__ , DummyDispatcher ) <EOL> self . assertEqual ( instance . pipes [ '<STR_LIT>' ] , <NUM_LIT:4> ) <EOL> self . assertEqual ( instance . pipes [ '<STR_LIT>' ] , <NUM_LIT:5> ) <EOL> self . assertEqual ( instance . pipes [ '<STR_LIT>' ] , None ) <EOL> def test_write ( self ) : <EOL> executable = '<STR_LIT>' <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , executable ) <EOL> instance = self . _makeOne ( config ) <EOL> sent = '<STR_LIT:a>' * ( <NUM_LIT:1> << <NUM_LIT> ) <EOL> self . assertRaises ( OSError , instance . write , sent ) <EOL> options . forkpid = <NUM_LIT:1> <EOL> result = instance . spawn ( ) <EOL> instance . write ( sent ) <EOL> stdin_fd = instance . pipes [ '<STR_LIT>' ] <EOL> self . assertEqual ( sent , instance . dispatchers [ stdin_fd ] . input_buffer ) <EOL> instance . killing = True <EOL> self . assertRaises ( OSError , instance . write , sent ) <EOL> def test_write_dispatcher_closed ( self ) : <EOL> executable = '<STR_LIT>' <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , executable ) <EOL> instance = self . _makeOne ( config ) <EOL> sent = '<STR_LIT:a>' * ( <NUM_LIT:1> << <NUM_LIT> ) <EOL> self . assertRaises ( OSError , instance . write , sent ) <EOL> options . forkpid = <NUM_LIT:1> <EOL> result = instance . spawn ( ) <EOL> stdin_fd = instance . pipes [ '<STR_LIT>' ] <EOL> instance . dispatchers [ stdin_fd ] . close ( ) <EOL> self . assertRaises ( OSError , instance . write , sent ) <EOL> def test_write_dispatcher_flush_raises_epipe ( self ) : <EOL> executable = '<STR_LIT>' <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , executable ) <EOL> instance = self . _makeOne ( config ) <EOL> sent = '<STR_LIT:a>' * ( <NUM_LIT:1> << <NUM_LIT> ) <EOL> self . assertRaises ( OSError , instance . write , sent ) <EOL> options . forkpid = <NUM_LIT:1> <EOL> result = instance . spawn ( ) <EOL> stdin_fd = instance . pipes [ '<STR_LIT>' ] <EOL> instance . dispatchers [ stdin_fd ] . flush_error = errno . EPIPE <EOL> self . assertRaises ( OSError , instance . write , sent ) <EOL> def dont_test_spawn_and_kill ( self ) : <EOL> from supervisor . tests . base import makeSpew <EOL> try : <EOL> called = <NUM_LIT:0> <EOL> def foo ( * args ) : <EOL> called = <NUM_LIT:1> <EOL> signal . signal ( signal . SIGCHLD , foo ) <EOL> executable = makeSpew ( ) <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , executable ) <EOL> instance = self . _makeOne ( config ) <EOL> result = instance . spawn ( ) <EOL> msg = options . logger . data [ <NUM_LIT:0> ] <EOL> self . failUnless ( msg . startswith ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( len ( instance . pipes ) , <NUM_LIT:6> ) <EOL> self . failUnless ( instance . pid ) <EOL> self . failUnlessEqual ( instance . pid , result ) <EOL> origpid = instance . pid <EOL> import errno <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> data = os . popen ( '<STR_LIT>' ) . read ( ) <EOL> break <EOL> except IOError , why : <EOL> if why [ <NUM_LIT:0> ] != errno . EINTR : <EOL> raise <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . failUnless ( data . find ( ` origpid ` ) != - <NUM_LIT:1> ) <EOL> msg = instance . kill ( signal . SIGTERM ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . assertEqual ( msg , None ) <EOL> pid , sts = os . waitpid ( - <NUM_LIT:1> , os . WNOHANG ) <EOL> data = os . popen ( '<STR_LIT>' ) . read ( ) <EOL> self . assertEqual ( data . find ( ` origpid ` ) , - <NUM_LIT:1> ) <EOL> finally : <EOL> try : <EOL> os . remove ( executable ) <EOL> except : <EOL> pass <EOL> signal . signal ( signal . SIGCHLD , signal . SIG_DFL ) <EOL> def test_stop ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . pid = <NUM_LIT:11> <EOL> dispatcher = DummyDispatcher ( writable = True ) <EOL> instance . dispatchers = { '<STR_LIT:foo>' : dispatcher } <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . RUNNING <EOL> instance . stop ( ) <EOL> self . assertEqual ( instance . administrative_stop , <NUM_LIT:1> ) <EOL> self . failUnless ( instance . delay ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . kills [ <NUM_LIT:11> ] , signal . SIGTERM ) <EOL> self . assertEqual ( dispatcher . write_event_handled , True ) <EOL> def test_give_up ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> L = [ ] <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> instance . state = ProcessStates . BACKOFF <EOL> instance . give_up ( ) <EOL> self . assertEqual ( instance . system_stop , <NUM_LIT:1> ) <EOL> self . assertFalse ( instance . delay ) <EOL> self . assertFalse ( instance . backoff ) <EOL> self . assertEqual ( instance . state , ProcessStates . FATAL ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateFatalEvent ) <EOL> def test_kill_nopid ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . kill ( signal . SIGTERM ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:0> ) <EOL> def test_kill_error ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> options . kill_error = <NUM_LIT:1> <EOL> instance = self . _makeOne ( config ) <EOL> L = [ ] <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> events . subscribe ( events . ProcessStateEvent , <EOL> lambda x : L . append ( x ) ) <EOL> instance . pid = <NUM_LIT:11> <EOL> instance . state = ProcessStates . RUNNING <EOL> instance . kill ( signal . SIGTERM ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . failUnless ( options . logger . data [ <NUM_LIT:1> ] . startswith ( <EOL> '<STR_LIT>' ) ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> ) <EOL> event1 = L [ <NUM_LIT:0> ] <EOL> event2 = L [ <NUM_LIT:1> ] <EOL> self . assertEqual ( event1 . __class__ , events . ProcessStateStoppingEvent ) <EOL> self . assertEqual ( event2 . __class__ , events . ProcessStateUnknownEvent ) <EOL> def test_kill_from_starting ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . pid = <NUM_LIT:11> <EOL> L = [ ] <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> instance . state = ProcessStates . STARTING <EOL> instance . kill ( signal . SIGTERM ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . kills [ <NUM_LIT:11> ] , signal . SIGTERM ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateStoppingEvent ) <EOL> def test_kill_from_running ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . pid = <NUM_LIT:11> <EOL> L = [ ] <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> instance . state = ProcessStates . RUNNING <EOL> instance . kill ( signal . SIGTERM ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . kills [ <NUM_LIT:11> ] , signal . SIGTERM ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateStoppingEvent ) <EOL> def test_kill_from_stopping ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . pid = <NUM_LIT:11> <EOL> L = [ ] <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> events . subscribe ( events . Event , lambda x : L . append ( x ) ) <EOL> instance . state = ProcessStates . STOPPING <EOL> instance . kill ( signal . SIGKILL ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . kills [ <NUM_LIT:11> ] , signal . SIGKILL ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_finish ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . waitstatus = ( <NUM_LIT> , <NUM_LIT:1> ) <EOL> instance . config . options . pidhistory [ <NUM_LIT> ] = instance <EOL> instance . killing = <NUM_LIT:1> <EOL> pipes = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> instance . pipes = pipes <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> instance . state = ProcessStates . STOPPING <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateStoppedEvent , lambda x : L . append ( x ) ) <EOL> instance . pid = <NUM_LIT> <EOL> instance . finish ( <NUM_LIT> , <NUM_LIT:1> ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . pid , <NUM_LIT:0> ) <EOL> self . assertEqual ( options . parent_pipes_closed , pipes ) <EOL> self . assertEqual ( instance . pipes , { } ) <EOL> self . assertEqual ( instance . dispatchers , { } ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . exitstatus , - <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateStoppedEvent ) <EOL> self . assertEqual ( event . extra_values , [ ( '<STR_LIT>' , <NUM_LIT> ) ] ) <EOL> self . assertEqual ( event . from_state , ProcessStates . STOPPING ) <EOL> def test_finish_expected ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . config . options . pidhistory [ <NUM_LIT> ] = instance <EOL> pipes = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> instance . pipes = pipes <EOL> instance . config . exitcodes = [ - <NUM_LIT:1> ] <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> instance . state = ProcessStates . RUNNING <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateExitedEvent , lambda x : L . append ( x ) ) <EOL> instance . pid = <NUM_LIT> <EOL> instance . finish ( <NUM_LIT> , <NUM_LIT:1> ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . pid , <NUM_LIT:0> ) <EOL> self . assertEqual ( options . parent_pipes_closed , pipes ) <EOL> self . assertEqual ( instance . pipes , { } ) <EOL> self . assertEqual ( instance . dispatchers , { } ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . exitstatus , - <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , <EOL> events . ProcessStateExitedEvent ) <EOL> self . assertEqual ( event . expected , True ) <EOL> self . assertEqual ( event . extra_values , [ ( '<STR_LIT>' , True ) , ( '<STR_LIT>' , <NUM_LIT> ) ] ) <EOL> self . assertEqual ( event . from_state , ProcessStates . RUNNING ) <EOL> def test_finish_tooquickly ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' , startsecs = <NUM_LIT:10> ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . config . options . pidhistory [ <NUM_LIT> ] = instance <EOL> pipes = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> instance . pipes = pipes <EOL> instance . config . exitcodes = [ - <NUM_LIT:1> ] <EOL> import time <EOL> instance . laststart = time . time ( ) <EOL> from supervisor . states import ProcessStates <EOL> from supervisor import events <EOL> instance . state = ProcessStates . STARTING <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> instance . pid = <NUM_LIT> <EOL> instance . finish ( <NUM_LIT> , <NUM_LIT:1> ) <EOL> self . assertEqual ( instance . killing , <NUM_LIT:0> ) <EOL> self . assertEqual ( instance . pid , <NUM_LIT:0> ) <EOL> self . assertEqual ( options . parent_pipes_closed , pipes ) <EOL> self . assertEqual ( instance . pipes , { } ) <EOL> self . assertEqual ( instance . dispatchers , { } ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( instance . exitstatus , None ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateBackoffEvent ) <EOL> self . assertEqual ( event . from_state , ProcessStates . STARTING ) <EOL> def test_finish_with_current_event_sends_rejected ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> events . subscribe ( events . EventRejectedEvent , lambda x : L . append ( x ) ) <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' , startsecs = <NUM_LIT:10> ) <EOL> instance = self . _makeOne ( config ) <EOL> from supervisor . states import ProcessStates <EOL> instance . state = ProcessStates . RUNNING <EOL> event = DummyEvent ( ) <EOL> instance . event = event <EOL> instance . finish ( <NUM_LIT> , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> ) <EOL> event1 , event2 = L <EOL> self . assertEqual ( event1 . __class__ , <EOL> events . ProcessStateExitedEvent ) <EOL> self . assertEqual ( event2 . __class__ , events . EventRejectedEvent ) <EOL> self . assertEqual ( event2 . process , instance ) <EOL> self . assertEqual ( event2 . event , event ) <EOL> self . assertEqual ( instance . event , None ) <EOL> def test_set_uid_no_uid ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . set_uid ( ) <EOL> self . assertEqual ( options . privsdropped , None ) <EOL> def test_set_uid ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> msg = instance . set_uid ( ) <EOL> self . assertEqual ( options . privsdropped , <NUM_LIT:1> ) <EOL> self . assertEqual ( msg , None ) <EOL> def test_cmp_bypriority ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' , <EOL> priority = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' , <EOL> priority = <NUM_LIT:2> ) <EOL> instance1 = self . _makeOne ( config ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , <EOL> stdout_logfile = '<STR_LIT>' , <EOL> priority = <NUM_LIT:3> ) <EOL> instance2 = self . _makeOne ( config ) <EOL> L = [ instance2 , instance , instance1 ] <EOL> L . sort ( ) <EOL> self . assertEqual ( L , [ instance , instance1 , instance2 ] ) <EOL> def test_transition_stopped_to_starting_supervisor_stopping ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates , SupervisorStates <EOL> options = DummyOptions ( ) <EOL> options . mood = SupervisorStates . SHUTDOWN <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:0> <EOL> process . state = ProcessStates . STOPPED <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . STOPPED ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_transition_stopped_to_starting_supervisor_running ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates , SupervisorStates <EOL> options = DummyOptions ( ) <EOL> options . mood = SupervisorStates . RUNNING <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:0> <EOL> process . state = ProcessStates . STOPPED <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . STARTING ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateStartingEvent ) <EOL> def test_transition_exited_to_starting_supervisor_stopping ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates , SupervisorStates <EOL> options = DummyOptions ( ) <EOL> options . mood = SupervisorStates . SHUTDOWN <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> from supervisor . datatypes import RestartUnconditionally <EOL> pconfig . autorestart = RestartUnconditionally <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . system_stop = <NUM_LIT:1> <EOL> process . state = ProcessStates . EXITED <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . EXITED ) <EOL> self . assertEqual ( process . system_stop , <NUM_LIT:1> ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_transition_exited_to_starting_uncond_supervisor_running ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> from supervisor . datatypes import RestartUnconditionally <EOL> pconfig . autorestart = RestartUnconditionally <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . state = ProcessStates . EXITED <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . STARTING ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateStartingEvent ) <EOL> def test_transition_exited_to_starting_condit_supervisor_running ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> from supervisor . datatypes import RestartWhenExitUnexpected <EOL> pconfig . autorestart = RestartWhenExitUnexpected <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . state = ProcessStates . EXITED <EOL> process . exitstatus = '<STR_LIT>' <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . STARTING ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateStartingEvent ) <EOL> def test_transition_exited_to_starting_condit_fls_supervisor_running ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> from supervisor . datatypes import RestartWhenExitUnexpected <EOL> pconfig . autorestart = RestartWhenExitUnexpected <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . state = ProcessStates . EXITED <EOL> process . exitstatus = <NUM_LIT:0> <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . EXITED ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_transition_backoff_to_starting_supervisor_stopping ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates , SupervisorStates <EOL> options = DummyOptions ( ) <EOL> options . mood = SupervisorStates . SHUTDOWN <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . delay = <NUM_LIT:0> <EOL> process . backoff = <NUM_LIT:0> <EOL> process . state = ProcessStates . BACKOFF <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . BACKOFF ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_transition_backoff_to_starting_supervisor_running ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates , SupervisorStates <EOL> options = DummyOptions ( ) <EOL> options . mood = SupervisorStates . RUNNING <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . delay = <NUM_LIT:0> <EOL> process . backoff = <NUM_LIT:0> <EOL> process . state = ProcessStates . BACKOFF <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . STARTING ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( L [ <NUM_LIT:0> ] . __class__ , events . ProcessStateStartingEvent ) <EOL> def test_transition_backoff_to_starting_supervisor_running_notyet ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates , SupervisorStates <EOL> options = DummyOptions ( ) <EOL> options . mood = SupervisorStates . RUNNING <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . delay = sys . maxint <EOL> process . backoff = <NUM_LIT:0> <EOL> process . state = ProcessStates . BACKOFF <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . BACKOFF ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_transition_starting_to_running ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . backoff = <NUM_LIT:1> <EOL> process . delay = <NUM_LIT:1> <EOL> process . system_stop = <NUM_LIT:0> <EOL> process . laststart = <NUM_LIT:1> <EOL> process . pid = <NUM_LIT:1> <EOL> process . stdout_buffer = '<STR_LIT:abc>' <EOL> process . stderr_buffer = '<STR_LIT>' <EOL> process . state = ProcessStates . STARTING <EOL> process . transition ( ) <EOL> self . assertEqual ( process . backoff , <NUM_LIT:0> ) <EOL> self . assertEqual ( process . delay , <NUM_LIT:0> ) <EOL> self . assertEqual ( process . system_stop , <NUM_LIT:0> ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateRunningEvent ) <EOL> def test_transition_backoff_to_fatal ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . laststart = <NUM_LIT:1> <EOL> process . backoff = <NUM_LIT> <EOL> process . delay = <NUM_LIT:1> <EOL> process . system_stop = <NUM_LIT:0> <EOL> process . stdout_buffer = '<STR_LIT:abc>' <EOL> process . stderr_buffer = '<STR_LIT>' <EOL> process . state = ProcessStates . BACKOFF <EOL> process . transition ( ) <EOL> self . assertEqual ( process . backoff , <NUM_LIT:0> ) <EOL> self . assertEqual ( process . delay , <NUM_LIT:0> ) <EOL> self . assertEqual ( process . system_stop , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:1> ) <EOL> event = L [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . __class__ , events . ProcessStateFatalEvent ) <EOL> def test_transition_stops_unkillable_notyet ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . delay = sys . maxint <EOL> process . state = ProcessStates . STOPPING <EOL> process . transition ( ) <EOL> self . assertEqual ( process . state , ProcessStates . STOPPING ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_transition_stops_unkillable ( self ) : <EOL> from supervisor import events <EOL> L = [ ] <EOL> events . subscribe ( events . ProcessStateEvent , lambda x : L . append ( x ) ) <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process = self . _makeOne ( pconfig ) <EOL> process . delay = <NUM_LIT:0> <EOL> process . pid = <NUM_LIT:1> <EOL> process . killing = <NUM_LIT:0> <EOL> process . state = ProcessStates . STOPPING <EOL> process . transition ( ) <EOL> self . assertEqual ( process . killing , <NUM_LIT:1> ) <EOL> self . assertNotEqual ( process . delay , <NUM_LIT:0> ) <EOL> self . assertEqual ( process . state , ProcessStates . STOPPING ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> import signal <EOL> self . assertEqual ( options . kills [ <NUM_LIT:1> ] , signal . SIGKILL ) <EOL> self . assertEqual ( L , [ ] ) <EOL> def test_change_state_doesnt_notify_if_no_state_change ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . state = <NUM_LIT:10> <EOL> self . assertEqual ( instance . change_state ( <NUM_LIT:10> ) , False ) <EOL> def test_change_state_sets_backoff_and_delay ( self ) : <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . state = <NUM_LIT:10> <EOL> instance . change_state ( ProcessStates . BACKOFF ) <EOL> self . assertEqual ( instance . backoff , <NUM_LIT:1> ) <EOL> self . failUnless ( instance . delay > <NUM_LIT:0> ) <EOL> class FastCGISubprocessTests ( unittest . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from supervisor . process import FastCGISubprocess <EOL> return FastCGISubprocess <EOL> def _makeOne ( self , * arg , ** kw ) : <EOL> return self . _getTargetClass ( ) ( * arg , ** kw ) <EOL> def tearDown ( self ) : <EOL> from supervisor . events import clear <EOL> clear ( ) <EOL> def test_no_group ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> self . assertRaises ( NotImplementedError , instance . spawn ) <EOL> def test_no_socket_manager ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . group = DummyProcessGroup ( DummyPGroupConfig ( options ) ) <EOL> self . assertRaises ( NotImplementedError , instance . spawn ) <EOL> def test_prepare_child_fds ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> sock_config = DummySocketConfig ( <NUM_LIT:7> ) <EOL> gconfig = DummyFCGIGroupConfig ( options , '<STR_LIT>' , <NUM_LIT> , None , <EOL> sock_config ) <EOL> instance . group = DummyFCGIProcessGroup ( gconfig ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( options . duped [ <NUM_LIT:7> ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( options . duped [ instance . pipes [ '<STR_LIT>' ] ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( options . duped [ instance . pipes [ '<STR_LIT>' ] ] , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> def test_prepare_child_fds_stderr_redirected ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . forkpid = <NUM_LIT:0> <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> config . redirect_stderr = True <EOL> instance = self . _makeOne ( config ) <EOL> sock_config = DummySocketConfig ( <NUM_LIT> ) <EOL> gconfig = DummyFCGIGroupConfig ( options , '<STR_LIT>' , <NUM_LIT> , None , <EOL> sock_config ) <EOL> instance . group = DummyFCGIProcessGroup ( gconfig ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( len ( options . duped ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( options . duped [ <NUM_LIT> ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( options . fds_closed ) , options . minfds - <NUM_LIT:3> ) <EOL> def test_before_spawn_gets_socket_ref ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> sock_config = DummySocketConfig ( <NUM_LIT:7> ) <EOL> gconfig = DummyFCGIGroupConfig ( options , '<STR_LIT>' , <NUM_LIT> , None , <EOL> sock_config ) <EOL> instance . group = DummyFCGIProcessGroup ( gconfig ) <EOL> self . assertTrue ( instance . fcgi_sock is None ) <EOL> instance . before_spawn ( ) <EOL> self . assertFalse ( instance . fcgi_sock is None ) <EOL> def test_after_finish_removes_socket_ref ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . fcgi_sock = '<STR_LIT:hello>' <EOL> instance . after_finish ( ) <EOL> self . assertTrue ( instance . fcgi_sock is None ) <EOL> @ patch_object ( Subprocess , '<STR_LIT>' , Mock ( return_value = sentinel . finish_result ) ) <EOL> def test_finish_override ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . after_finish = Mock ( ) <EOL> result = instance . finish ( sentinel . pid , sentinel . sts ) <EOL> self . assertEqual ( sentinel . finish_result , result , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , instance . after_finish . call_count , <EOL> '<STR_LIT>' ) <EOL> finish_mock = Subprocess . finish <EOL> self . assertEqual ( <NUM_LIT:1> , finish_mock . call_count , <EOL> '<STR_LIT>' ) <EOL> pid_arg = finish_mock . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> sts_arg = finish_mock . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] <EOL> self . assertEqual ( sentinel . pid , pid_arg , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( sentinel . sts , sts_arg , <EOL> '<STR_LIT>' ) <EOL> @ patch_object ( Subprocess , '<STR_LIT>' , Mock ( return_value = sentinel . ppid ) ) <EOL> def test_spawn_override_success ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . before_spawn = Mock ( ) <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( sentinel . ppid , result , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , instance . before_spawn . call_count , <EOL> '<STR_LIT>' ) <EOL> spawn_mock = Subprocess . spawn <EOL> self . assertEqual ( <NUM_LIT:1> , spawn_mock . call_count , <EOL> '<STR_LIT>' ) <EOL> @ patch_object ( Subprocess , '<STR_LIT>' , Mock ( return_value = None ) ) <EOL> def test_spawn_error ( self ) : <EOL> options = DummyOptions ( ) <EOL> config = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , uid = <NUM_LIT:1> ) <EOL> instance = self . _makeOne ( config ) <EOL> instance . before_spawn = Mock ( ) <EOL> instance . fcgi_sock = '<STR_LIT>' <EOL> result = instance . spawn ( ) <EOL> self . assertEqual ( None , result , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , instance . before_spawn . call_count , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( None , instance . fcgi_sock , <EOL> '<STR_LIT>' ) <EOL> class ProcessGroupBaseTests ( unittest . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from supervisor . process import ProcessGroupBase <EOL> return ProcessGroupBase <EOL> def _makeOne ( self , * args , ** kw ) : <EOL> return self . _getTargetClass ( ) ( * args , ** kw ) <EOL> def test_get_unstopped_processes ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STOPPING ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> group = self . _makeOne ( gconfig ) <EOL> group . processes = { '<STR_LIT>' : process1 } <EOL> unstopped = group . get_unstopped_processes ( ) <EOL> self . assertEqual ( unstopped , [ process1 ] ) <EOL> def test_stop_all ( self ) : <EOL> from supervisor . states import ProcessStates <EOL> options = DummyOptions ( ) <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STOPPED ) <EOL> pconfig2 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process2 = DummyProcess ( pconfig2 , state = ProcessStates . RUNNING ) <EOL> pconfig3 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process3 = DummyProcess ( pconfig3 , state = ProcessStates . STARTING ) <EOL> pconfig4 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process4 = DummyProcess ( pconfig4 , state = ProcessStates . BACKOFF ) <EOL> process4 . delay = <NUM_LIT:1000> <EOL> process4 . backoff = <NUM_LIT:10> <EOL> gconfig = DummyPGroupConfig ( <EOL> options , <EOL> pconfigs = [ pconfig1 , pconfig2 , pconfig3 , pconfig4 ] ) <EOL> group = self . _makeOne ( gconfig ) <EOL> group . processes = { '<STR_LIT>' : process1 , '<STR_LIT>' : process2 , <EOL> '<STR_LIT>' : process3 , '<STR_LIT>' : process4 } <EOL> group . stop_all ( ) <EOL> self . assertEqual ( process1 . stop_called , False ) <EOL> self . assertEqual ( process2 . stop_called , True ) <EOL> self . assertEqual ( process3 . stop_called , True ) <EOL> self . assertEqual ( process4 . stop_called , False ) <EOL> self . assertEqual ( process4 . state , ProcessStates . FATAL ) <EOL> def test_get_dispatchers ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STOPPING ) <EOL> process1 . dispatchers = { <NUM_LIT:4> : None } <EOL> pconfig2 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process2 = DummyProcess ( pconfig2 , state = ProcessStates . STOPPING ) <EOL> process2 . dispatchers = { <NUM_LIT:5> : None } <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 , pconfig2 ] ) <EOL> group = self . _makeOne ( gconfig ) <EOL> group . processes = { '<STR_LIT>' : process1 , '<STR_LIT>' : process2 } <EOL> result = group . get_dispatchers ( ) <EOL> self . assertEqual ( result , { <NUM_LIT:4> : None , <NUM_LIT:5> : None } ) <EOL> def test_reopenlogs ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STOPPING ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> group = self . _makeOne ( gconfig ) <EOL> group . processes = { '<STR_LIT>' : process1 } <EOL> group . reopenlogs ( ) <EOL> self . assertEqual ( process1 . logs_reopened , True ) <EOL> def test_removelogs ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STOPPING ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> group = self . _makeOne ( gconfig ) <EOL> group . processes = { '<STR_LIT>' : process1 } <EOL> group . removelogs ( ) <EOL> self . assertEqual ( process1 . logsremoved , True ) <EOL> def test_cmp ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig1 = DummyPGroupConfig ( options ) <EOL> group1 = self . _makeOne ( gconfig1 ) <EOL> gconfig2 = DummyPGroupConfig ( options ) <EOL> group2 = self . _makeOne ( gconfig2 ) <EOL> group1 . priority = <NUM_LIT:5> <EOL> group2 . priority = <NUM_LIT:1> <EOL> L = [ group1 , group2 ] <EOL> L . sort ( ) <EOL> self . assertEqual ( L , [ group2 , group1 ] ) <EOL> class ProcessGroupTests ( ProcessGroupBaseTests ) : <EOL> def _getTargetClass ( self ) : <EOL> from supervisor . process import ProcessGroup <EOL> return ProcessGroup <EOL> def test_repr ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> group = self . _makeOne ( gconfig ) <EOL> s = repr ( group ) <EOL> self . assertTrue ( s . startswith ( <EOL> '<STR_LIT>' ) , s ) <EOL> self . assertTrue ( s . endswith ( '<STR_LIT>' ) , s ) <EOL> def test_transition ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STOPPING ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> group = self . _makeOne ( gconfig ) <EOL> group . processes = { '<STR_LIT>' : process1 } <EOL> group . transition ( ) <EOL> self . assertEqual ( process1 . transitioned , True ) <EOL> class FastCGIProcessGroupTests ( unittest . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from supervisor . process import FastCGIProcessGroup <EOL> return FastCGIProcessGroup <EOL> def _makeOne ( self , * args , ** kw ) : <EOL> return self . _getTargetClass ( ) ( * args , ** kw ) <EOL> def test_stop_requested_signals_socket_close ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyFCGIGroupConfig ( options ) <EOL> group = self . _makeOne ( gconfig , socketManager = DummySocketManager ) <EOL> group . stop_requested ( ) <EOL> self . assertTrue ( group . socket_manager . request_close_called ) <EOL> class EventListenerPoolTests ( ProcessGroupBaseTests ) : <EOL> def setUp ( self ) : <EOL> from supervisor . events import clear <EOL> clear ( ) <EOL> def tearDown ( self ) : <EOL> from supervisor . events import clear <EOL> clear ( ) <EOL> def _getTargetClass ( self ) : <EOL> from supervisor . process import EventListenerPool <EOL> return EventListenerPool <EOL> def test_ctor ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> class EventType : <EOL> pass <EOL> gconfig . pool_events = ( EventType , ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> from supervisor import events <EOL> self . assertEqual ( len ( events . callbacks ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( events . callbacks [ <NUM_LIT:0> ] , <EOL> ( EventType , pool . _acceptEvent ) ) <EOL> self . assertEqual ( events . callbacks [ <NUM_LIT:1> ] , <EOL> ( events . EventRejectedEvent , pool . handle_rejected ) ) <EOL> self . assertEqual ( pool . serial , - <NUM_LIT:1> ) <EOL> def test__eventEnvelope ( self ) : <EOL> options = DummyOptions ( ) <EOL> options . identifier = '<STR_LIT>' <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> gconfig . name = '<STR_LIT>' <EOL> pool = self . _makeOne ( gconfig ) <EOL> from supervisor import events <EOL> result = pool . _eventEnvelope ( <EOL> events . EventTypes . PROCESS_COMMUNICATION_STDOUT , <NUM_LIT> , <NUM_LIT:20> , '<STR_LIT>' ) <EOL> header , payload = result . split ( '<STR_LIT:\n>' , <NUM_LIT:1> ) <EOL> headers = header . split ( ) <EOL> self . assertEqual ( headers [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ <NUM_LIT:4> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ <NUM_LIT:5> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ <NUM_LIT:6> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( payload , '<STR_LIT>' ) <EOL> def test_handle_rejected_no_overflow ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> pool . processes = { '<STR_LIT>' : process1 } <EOL> pool . event_buffer = [ None , None ] <EOL> class DummyEvent1 : <EOL> serial = '<STR_LIT:abc>' <EOL> class DummyEvent2 : <EOL> process = process1 <EOL> event = DummyEvent1 ( ) <EOL> dummyevent = DummyEvent2 ( ) <EOL> dummyevent . serial = <NUM_LIT:1> <EOL> pool . handle_rejected ( dummyevent ) <EOL> self . assertEqual ( pool . event_buffer , [ dummyevent . event , None , None ] ) <EOL> def test_handle_rejected_event_buffer_overflowed ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> gconfig . buffer_size = <NUM_LIT:3> <EOL> pool = self . _makeOne ( gconfig ) <EOL> pool . processes = { '<STR_LIT>' : process1 } <EOL> class DummyEvent : <EOL> def __init__ ( self , serial ) : <EOL> self . serial = serial <EOL> class DummyRejectedEvent : <EOL> def __init__ ( self , serial ) : <EOL> self . process = process1 <EOL> self . event = DummyEvent ( serial ) <EOL> event_a = DummyEvent ( '<STR_LIT:a>' ) <EOL> event_b = DummyEvent ( '<STR_LIT:b>' ) <EOL> event_c = DummyEvent ( '<STR_LIT:c>' ) <EOL> rej_event = DummyRejectedEvent ( '<STR_LIT>' ) <EOL> pool . event_buffer = [ event_a , event_b , event_c ] <EOL> pool . handle_rejected ( rej_event ) <EOL> serials = [ x . serial for x in pool . event_buffer ] <EOL> self . assertEqual ( serials , [ '<STR_LIT>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ) <EOL> self . assertEqual ( pool . config . options . logger . data [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> def test_dispatch_pipe_error ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> from supervisor . states import EventListenerStates <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> process1 = pool . processes [ '<STR_LIT>' ] <EOL> import errno <EOL> process1 . write_error = errno . EPIPE <EOL> process1 . listener_state = EventListenerStates . READY <EOL> event = DummyEvent ( ) <EOL> pool . _acceptEvent ( event ) <EOL> pool . dispatch ( ) <EOL> self . assertEqual ( process1 . listener_state , EventListenerStates . READY ) <EOL> self . assertEqual ( pool . event_buffer , [ event ] ) <EOL> self . assertEqual ( options . logger . data [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> def test__acceptEvent_attaches_pool_serial_and_serial ( self ) : <EOL> from supervisor . process import GlobalSerial <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> process1 = pool . processes [ '<STR_LIT>' ] <EOL> from supervisor . states import EventListenerStates <EOL> process1 . listener_state = EventListenerStates . READY <EOL> event = DummyEvent ( None ) <EOL> pool . _acceptEvent ( event ) <EOL> self . assertEqual ( event . serial , GlobalSerial . serial ) <EOL> self . assertEqual ( event . pool_serials [ '<STR_LIT>' ] , pool . serial ) <EOL> def test_repr ( self ) : <EOL> options = DummyOptions ( ) <EOL> gconfig = DummyPGroupConfig ( options ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> s = repr ( pool ) <EOL> self . assertTrue ( s . startswith ( <EOL> '<STR_LIT>' ) ) <EOL> self . assertTrue ( s . endswith ( '<STR_LIT>' ) ) <EOL> def test_transition_nobody_ready ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STARTING ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> pool . processes = { '<STR_LIT>' : process1 } <EOL> event = DummyEvent ( ) <EOL> event . serial = '<STR_LIT:a>' <EOL> from supervisor . states import EventListenerStates <EOL> process1 . listener_state = EventListenerStates . BUSY <EOL> pool . _acceptEvent ( event ) <EOL> pool . transition ( ) <EOL> self . assertEqual ( process1 . transitioned , True ) <EOL> self . assertEqual ( pool . event_buffer , [ event ] ) <EOL> data = pool . config . options . logger . data <EOL> def test_transition_event_proc_not_running ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . STARTING ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> pool . processes = { '<STR_LIT>' : process1 } <EOL> event = DummyEvent ( ) <EOL> from supervisor . states import EventListenerStates <EOL> event . serial = <NUM_LIT:1> <EOL> process1 . listener_state = EventListenerStates . READY <EOL> pool . _acceptEvent ( event ) <EOL> pool . transition ( ) <EOL> self . assertEqual ( process1 . transitioned , True ) <EOL> self . assertEqual ( pool . event_buffer , [ event ] ) <EOL> self . assertEqual ( process1 . stdin_buffer , '<STR_LIT>' ) <EOL> self . assertEqual ( process1 . listener_state , EventListenerStates . READY ) <EOL> def test_transition_event_proc_running ( self ) : <EOL> options = DummyOptions ( ) <EOL> from supervisor . states import ProcessStates <EOL> pconfig1 = DummyPConfig ( options , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> process1 = DummyProcess ( pconfig1 , state = ProcessStates . RUNNING ) <EOL> gconfig = DummyPGroupConfig ( options , pconfigs = [ pconfig1 ] ) <EOL> pool = self . _makeOne ( gconfig ) <EOL> pool . processes = { '<STR_LIT>' : process1 } <EOL> event = DummyEvent ( ) <EOL> from supervisor . states import EventListenerStates <EOL> process1 . listener_state = EventListenerStates . READY <EOL> class DummyGroup : <EOL> config = gconfig <EOL> process1 . group = DummyGroup <EOL> pool . _acceptEvent ( event ) <EOL> pool . transition ( ) <EOL> self . assertEqual ( process1 . transitioned , True ) <EOL> self . assertEqual ( pool . event_buffer , [ ] ) <EOL> header , payload = process1 . stdin_buffer . split ( '<STR_LIT:\n>' , <NUM_LIT:1> ) <EOL> self . assertEquals ( payload , '<STR_LIT>' , payload ) <EOL> self . assertEqual ( process1 . listener_state , EventListenerStates . BUSY ) <EOL> self . assertEqual ( process1 . event , event ) <EOL> def test_suite ( ) : <EOL> return unittest . findTestCases ( sys . modules [ __name__ ] ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
