<s> import unittest <EOL> from wsgid . commands import * <EOL> from wsgid . test import FakeOptions , fullpath <EOL> import os <EOL> import simplejson <EOL> from mock import patch , Mock <EOL> import signal <EOL> ROOT_PATH = fullpath ( __file__ ) <EOL> FIXTURES_PATH = os . path . join ( ROOT_PATH , '<STR_LIT>' ) <EOL> APP_PATH = os . path . join ( FIXTURES_PATH , "<STR_LIT>" ) <EOL> def new_app_path ( app_name ) : <EOL> return os . path . join ( FIXTURES_PATH , app_name ) <EOL> class CommandInitTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . init = CommandInit ( ) <EOL> self . opt = FakeOptions ( app_path = APP_PATH ) <EOL> '''<STR_LIT>''' <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_create_root_folter_if_not_exists ( self , * args ) : <EOL> self . init . run ( self . opt ) <EOL> self . assertTrue ( os . path . exists ( APP_PATH ) , "<STR_LIT>" ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_create_pid_structure ( self , * args ) : <EOL> self . init . run ( self . opt ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) , "<STR_LIT>" ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) , "<STR_LIT>" ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) , "<STR_LIT>" ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_create_log_dir ( self , * args ) : <EOL> self . init . run ( self . opt ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) , "<STR_LIT>" ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_create_app_dir ( self , * args ) : <EOL> self . init . run ( self . opt ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) , "<STR_LIT>" ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_create_plugins_dir ( self , * args ) : <EOL> self . init . run ( self . opt ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) , "<STR_LIT>" ) <EOL> '''<STR_LIT>''' <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_init_an_already_inited_path ( self , * args ) : <EOL> self . init . run ( FakeOptions ( app_path = APP_PATH ) ) <EOL> os . system ( "<STR_LIT>" . format ( os . path . join ( APP_PATH , '<STR_LIT>' ) ) ) <EOL> self . init . run ( FakeOptions ( app_path = APP_PATH ) ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( APP_PATH , "<STR_LIT>" ) ) , "<STR_LIT>" ) <EOL> class CommandConfigTest ( unittest . TestCase ) : <EOL> @ patch ( '<STR_LIT>' ) <EOL> def setUp ( self , * args ) : <EOL> self . config = CommandConfig ( ) <EOL> self . init = CommandInit ( ) <EOL> self . CLEAN_PATH = os . path . join ( FIXTURES_PATH , '<STR_LIT>' ) <EOL> self . opt = FakeOptions ( app_path = self . CLEAN_PATH , wsgi_app = "<STR_LIT>" , debug = True , no_daemon = True , workers = <NUM_LIT:8> , keep_alive = True , chroot = True , recv = "<STR_LIT>" , send = "<STR_LIT>" , no_debug = False , no_chroot = False , no_keep_alive = False , mongrel2_chroot = '<STR_LIT>' , <EOL> django = False ) <EOL> self . init . run ( self . opt ) <EOL> '''<STR_LIT>''' <EOL> def test_create_json_if_not_exist ( self ) : <EOL> self . config . run ( self . opt ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( self . CLEAN_PATH , "<STR_LIT>" ) ) ) <EOL> '''<STR_LIT>''' <EOL> def test_override_option ( self ) : <EOL> f = file ( os . path . join ( self . CLEAN_PATH , "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> simplejson . dump ( { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT:True>" , "<STR_LIT>" : <NUM_LIT:8> , "<STR_LIT>" : "<STR_LIT:True>" } , f ) <EOL> f . close ( ) <EOL> self . opt . recv = "<STR_LIT>" <EOL> self . opt . workers = <NUM_LIT:8> <EOL> self . opt . chroot = None <EOL> self . config . run ( self . opt ) <EOL> h = simplejson . loads ( file ( os . path . join ( self . CLEAN_PATH , "<STR_LIT>" ) , "<STR_LIT>" ) . read ( ) ) <EOL> self . assertEquals ( "<STR_LIT>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:True>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( <NUM_LIT:8> , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:True>" , h [ '<STR_LIT>' ] ) <EOL> def test_create_all_options ( self ) : <EOL> open ( os . path . join ( self . CLEAN_PATH , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> opt = FakeOptions ( app_path = self . CLEAN_PATH , wsgi_app = "<STR_LIT>" , debug = True , no_daemon = True , workers = <NUM_LIT:8> , keep_alive = True , chroot = True , recv = "<STR_LIT>" , send = "<STR_LIT>" , no_debug = False , no_chroot = False , no_keep_alive = False , mongrel2_chroot = '<STR_LIT>' , <EOL> django = True ) <EOL> self . config . run ( opt ) <EOL> h = simplejson . loads ( file ( os . path . join ( self . CLEAN_PATH , "<STR_LIT>" ) , "<STR_LIT>" ) . read ( ) ) <EOL> self . assertEquals ( "<STR_LIT>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:True>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( <NUM_LIT:8> , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:True>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:True>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:True>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:True>" , h [ '<STR_LIT>' ] ) <EOL> def test_validate_recv ( self ) : <EOL> open ( os . path . join ( self . CLEAN_PATH , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> opt = FakeOptions ( app_path = self . CLEAN_PATH , wsgi_app = "<STR_LIT>" , debug = True , no_daemon = True , workers = <NUM_LIT:8> , keep_alive = True , chroot = True , recv = "<STR_LIT>" , send = "<STR_LIT>" , no_debug = False , no_chroot = False , no_keep_alive = False , mongrel2_chroot = '<STR_LIT>' , <EOL> django = True ) <EOL> self . assertRaises ( Exception , self . config . run , opt ) <EOL> def test_validate_send ( self ) : <EOL> open ( os . path . join ( self . CLEAN_PATH , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> opt = FakeOptions ( app_path = self . CLEAN_PATH , wsgi_app = "<STR_LIT>" , debug = True , no_daemon = True , workers = <NUM_LIT:8> , keep_alive = True , chroot = True , recv = "<STR_LIT>" , send = "<STR_LIT>" , no_debug = False , no_chroot = False , no_keep_alive = False , mongrel2_chroot = '<STR_LIT>' , <EOL> django = True ) <EOL> self . assertRaises ( Exception , self . config . run , opt ) <EOL> '''<STR_LIT>''' <EOL> def test_disable_boolean_option ( self ) : <EOL> opt = FakeOptions ( app_path = self . CLEAN_PATH , wsgi_app = "<STR_LIT>" , no_debug = True , debug = True , workers = <NUM_LIT:9> , keep_alive = True , chroot = True , recv = "<STR_LIT>" , send = "<STR_LIT>" , <EOL> no_chroot = False , no_keep_alive = False , no_daemon = False , mongrel2_chroot = '<STR_LIT>' , <EOL> django = False ) <EOL> self . config . run ( opt ) <EOL> h = simplejson . loads ( file ( os . path . join ( self . CLEAN_PATH , "<STR_LIT>" ) , "<STR_LIT>" ) . read ( ) ) <EOL> self . assertEquals ( "<STR_LIT>" , h [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( "<STR_LIT:False>" , h [ '<STR_LIT>' ] ) <EOL> class CommandManageTest ( unittest . TestCase ) : <EOL> @ patch ( '<STR_LIT>' ) <EOL> def setUp ( self , * args ) : <EOL> self . init = CommandInit ( ) <EOL> self . manage = CommandManage ( ) <EOL> self . opt = FakeOptions ( app_path = APP_PATH , send_signal = signal . SIGTERM ) <EOL> self . init . run ( self . opt ) <EOL> def test_match_command_names_matches ( self ) : <EOL> self . assertTrue ( self . manage . name_matches ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( self . manage . name_matches ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( self . manage . name_matches ( '<STR_LIT:start>' ) ) <EOL> def test_command_name ( self ) : <EOL> self . assertEquals ( '<STR_LIT>' , self . manage . command_name ( ) ) <EOL> def test_run_stop_command ( self ) : <EOL> open ( os . path . join ( APP_PATH , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( APP_PATH , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> self . manage . run ( self . opt , command_name = '<STR_LIT>' ) <EOL> self . assertEquals ( <NUM_LIT:2> , os . kill . call_count ) <EOL> self . assertTrue ( ( ( <NUM_LIT> , <NUM_LIT:15> ) , { } ) in os . kill . call_args_list ) <EOL> self . assertTrue ( ( ( <NUM_LIT> , <NUM_LIT:15> ) , { } ) in os . kill . call_args_list ) <EOL> def test_run_restart_command ( self ) : <EOL> open ( os . path . join ( APP_PATH , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( APP_PATH , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> self . manage . run ( self . opt , command_name = '<STR_LIT>' ) <EOL> self . assertEquals ( <NUM_LIT:2> , os . kill . call_count ) <EOL> self . assertTrue ( ( ( <NUM_LIT> , <NUM_LIT:15> ) , { } ) in os . kill . call_args_list ) <EOL> self . assertTrue ( ( ( <NUM_LIT> , <NUM_LIT:15> ) , { } ) in os . kill . call_args_list ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_send_custom_signal ( self , * args ) : <EOL> apppath = new_app_path ( '<STR_LIT>' ) <EOL> opts = FakeOptions ( app_path = apppath , send_signal = <NUM_LIT:9> ) <EOL> self . init . run ( opts ) <EOL> open ( os . path . join ( apppath , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( apppath , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> self . manage . run ( opts , command_name = '<STR_LIT>' ) <EOL> self . manage . run ( opts , command_name = '<STR_LIT>' ) <EOL> self . assertEquals ( <NUM_LIT:2> , os . kill . call_count ) <EOL> self . assertEquals ( [ ( ( <NUM_LIT> , <NUM_LIT:9> ) , { } ) , ( ( <NUM_LIT> , <NUM_LIT:9> ) , { } ) ] , os . kill . call_args_list ) <EOL> def test_kill_already_dead_pid ( self ) : <EOL> open ( os . path . join ( APP_PATH , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( APP_PATH , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> os . kill = Mock ( side_effect = OSError ( "<STR_LIT>" ) ) <EOL> self . manage . run ( self . opt , command_name = '<STR_LIT>' ) <EOL> self . assertEquals ( <NUM_LIT:2> , os . kill . call_count ) <EOL> self . assertTrue ( ( ( <NUM_LIT> , <NUM_LIT:15> ) , { } ) in os . kill . call_args_list ) <EOL> self . assertTrue ( ( ( <NUM_LIT> , <NUM_LIT:15> ) , { } ) in os . kill . call_args_list ) <EOL> '''<STR_LIT>''' <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_invalid_pid_files ( self , * args ) : <EOL> new_path = os . path . join ( FIXTURES_PATH , '<STR_LIT>' ) <EOL> opts = FakeOptions ( app_path = new_path , send_signal = signal . SIGTERM ) <EOL> self . init . run ( opts ) <EOL> open ( os . path . join ( new_path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> self . manage . run ( opts , command_name = '<STR_LIT>' ) <EOL> self . assertEquals ( <NUM_LIT:0> , os . kill . call_count ) <EOL> class CommandStatusTest ( unittest . TestCase ) : <EOL> @ patch ( '<STR_LIT>' ) <EOL> def setUp ( self , * args ) : <EOL> self . init = CommandInit ( ) <EOL> self . manage = CommandManage ( ) <EOL> self . new_path = os . path . join ( FIXTURES_PATH , '<STR_LIT>' ) <EOL> self . opt = FakeOptions ( app_path = self . new_path , send_signal = signal . SIGTERM ) <EOL> self . init . run ( self . opt ) <EOL> def test_command_name ( self ) : <EOL> self . assertEquals ( '<STR_LIT:status>' , CommandStatus ( ) . command_name ( ) ) <EOL> def test_command_name_matches ( self ) : <EOL> self . assertTrue ( CommandStatus ( ) . name_matches ( '<STR_LIT:status>' ) ) <EOL> self . assertFalse ( CommandStatus ( ) . name_matches ( '<STR_LIT>' ) ) <EOL> def test_list_master_pids ( self ) : <EOL> with patch ( '<STR_LIT>' ) : <EOL> with patch ( '<STR_LIT>' ) : <EOL> open ( os . path . join ( self . new_path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( self . new_path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> CommandStatus ( ) . run ( self . opt ) <EOL> self . assertEquals ( <NUM_LIT:3> , sys . stdout . write . call_count ) <EOL> self . assertEquals ( [ ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) ] , sys . stdout . write . call_args_list ) <EOL> def test_list_worker_pids ( self ) : <EOL> path = os . path . join ( FIXTURES_PATH , '<STR_LIT>' ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> CommandInit ( ) . run ( FakeOptions ( app_path = path ) ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> with patch ( '<STR_LIT>' ) : <EOL> open ( os . path . join ( path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> CommandStatus ( ) . run ( FakeOptions ( app_path = path ) ) <EOL> self . assertEquals ( <NUM_LIT:3> , sys . stdout . write . call_count ) <EOL> self . assertEquals ( [ ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) ] , sys . stdout . write . call_args_list ) <EOL> def test_status_when_no_worker_pids ( self ) : <EOL> path = os . path . join ( FIXTURES_PATH , '<STR_LIT>' ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> CommandInit ( ) . run ( FakeOptions ( app_path = path ) ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> open ( os . path . join ( path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> CommandStatus ( ) . run ( FakeOptions ( app_path = path ) ) <EOL> self . assertEquals ( <NUM_LIT:3> , sys . stdout . write . call_count ) <EOL> self . assertEquals ( [ ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) ] , sys . stdout . write . call_args_list ) <EOL> def test_check_pids_are_running ( self ) : <EOL> path = os . path . join ( FIXTURES_PATH , '<STR_LIT>' ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> CommandInit ( ) . run ( FakeOptions ( app_path = path ) ) <EOL> with patch ( '<STR_LIT>' ) : <EOL> with patch ( '<STR_LIT>' ) : <EOL> os . kill . side_effect = OSError ( "<STR_LIT>" ) <EOL> open ( os . path . join ( path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> open ( os . path . join ( path , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> CommandStatus ( ) . run ( FakeOptions ( app_path = path ) ) <EOL> self . assertEquals ( <NUM_LIT:3> , sys . stdout . write . call_count ) <EOL> self . assertEquals ( [ ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) , <EOL> ( ( "<STR_LIT>" , ) , { } ) ] , sys . stdout . write . call_args_list ) </s>
