<s> import os <EOL> from mock import patch <EOL> from nose . plugins . attrib import attr <EOL> from cloudify_rest_client import exceptions <EOL> from manager_rest . test import base_test <EOL> from base_test import BaseServerTestCase <EOL> from manager_rest import ( storage_manager , <EOL> models , <EOL> utils ) <EOL> from manager_rest . constants import ( <EOL> MAINTENANCE_MODE_ACTIVATED , <EOL> MAINTENANCE_MODE_ACTIVATING , <EOL> MAINTENANCE_MODE_DEACTIVATED , <EOL> MAINTENANCE_MODE_STATUS_FILE ) <EOL> @ attr ( client_min_version = <NUM_LIT> , client_max_version = base_test . LATEST_API_VERSION ) <EOL> class MaintenanceModeTest ( BaseServerTestCase ) : <EOL> def test_maintenance_mode_inactive ( self ) : <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_DEACTIVATED , response . status ) <EOL> self . assertFalse ( response . activation_requested_at ) <EOL> self . client . blueprints . list ( ) <EOL> def test_maintenance_activation ( self ) : <EOL> response = self . client . maintenance_mode . activate ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_ACTIVATING , response . status ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_ACTIVATED , response . status ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_ACTIVATED , response . status ) <EOL> def test_any_cmd_activates_maintenance_mode ( self ) : <EOL> response = self . client . maintenance_mode . activate ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_ACTIVATING , response . status ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActiveError , <EOL> self . client . blueprints . upload , <EOL> blueprint_path = self . get_mock_blueprint_path ( ) , <EOL> blueprint_id = '<STR_LIT>' ) <EOL> self . client . maintenance_mode . deactivate ( ) <EOL> response = self . client . maintenance_mode . activate ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_ACTIVATING , response . status ) <EOL> self . client . manager . get_version ( ) <EOL> maintenance_file = os . path . join ( self . maintenance_mode_dir , <EOL> MAINTENANCE_MODE_STATUS_FILE ) <EOL> state = utils . read_json_file ( maintenance_file ) <EOL> self . assertEqual ( state [ '<STR_LIT:status>' ] , MAINTENANCE_MODE_ACTIVATED ) <EOL> def test_request_denial_in_maintenance_mode ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActiveError , <EOL> self . client . blueprints . list ) <EOL> def test_request_approval_in_maintenance_mode ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> self . client . maintenance_mode . status ( ) <EOL> self . client . manager . get_version ( ) <EOL> self . client . manager . get_status ( ) <EOL> def test_internal_request_denial_in_maintenance_mode ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> with patch ( '<STR_LIT>' , <EOL> new = self . _get_remote_addr ) : <EOL> with patch ( '<STR_LIT>' , <EOL> new = self . _get_host ) : <EOL> self . assertRaises ( exceptions . MaintenanceModeActiveError , <EOL> self . client . blueprints . list ) <EOL> def test_external_request_approval_in_maintenance_mode ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> internal_request_bypass_maintenance_client = self . create_client ( <EOL> headers = { '<STR_LIT>' : '<STR_LIT:true>' } ) <EOL> with patch ( '<STR_LIT>' , <EOL> new = self . _get_remote_addr ) : <EOL> with patch ( '<STR_LIT>' , <EOL> new = self . _get_host ) : <EOL> internal_request_bypass_maintenance_client . blueprints . list ( ) <EOL> def test_bypass_maintenance_denial_in_maintenance_mode ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> internal_request_client = self . create_client ( <EOL> headers = { '<STR_LIT>' : '<STR_LIT:true>' } ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActiveError , <EOL> internal_request_client . blueprints . list ) <EOL> def test_multiple_maintenance_mode_activations ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> try : <EOL> self . _activate_maintenance_mode ( ) <EOL> self . fail ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> except exceptions . NotModifiedError as e : <EOL> self . assertEqual ( <NUM_LIT> , e . status_code ) <EOL> self . assertIn ( '<STR_LIT>' , e . message ) <EOL> def test_transition_to_active ( self ) : <EOL> execution = self . _start_maintenance_transition_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( response . status , MAINTENANCE_MODE_ACTIVATING ) <EOL> self . _terminate_execution ( execution . id ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( response . status , MAINTENANCE_MODE_ACTIVATED ) <EOL> def test_deployment_denial_in_maintenance_transition_mode ( self ) : <EOL> self . _start_maintenance_transition_mode ( ) <EOL> self . client . blueprints . upload ( <EOL> self . get_mock_blueprint_path ( ) , <EOL> blueprint_id = '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActivatingError , <EOL> self . client . deployments . create , <EOL> blueprint_id = '<STR_LIT>' , <EOL> deployment_id = '<STR_LIT>' ) <EOL> def test_deployment_modification_denial_maintenance_transition_mode ( self ) : <EOL> self . put_deployment ( '<STR_LIT>' , blueprint_id = '<STR_LIT>' ) <EOL> self . _start_maintenance_transition_mode ( ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActivatingError , <EOL> self . client . deployment_modifications . start , <EOL> deployment_id = '<STR_LIT>' , <EOL> nodes = { } ) <EOL> def test_snapshot_creation_denial_in_maintenance_transition_mode ( self ) : <EOL> self . _start_maintenance_transition_mode ( ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActivatingError , <EOL> self . client . snapshots . create , <EOL> snapshot_id = '<STR_LIT>' , <EOL> include_metrics = False , <EOL> include_credentials = False ) <EOL> def test_snapshot_restoration_denial_in_maintenance_transition_mode ( self ) : <EOL> self . _start_maintenance_transition_mode ( ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActivatingError , <EOL> self . client . snapshots . restore , <EOL> snapshot_id = '<STR_LIT>' ) <EOL> def test_executions_denial_in_maintenance_transition_mode ( self ) : <EOL> self . _start_maintenance_transition_mode ( ) <EOL> self . client . blueprints . upload ( <EOL> self . get_mock_blueprint_path ( ) , <EOL> blueprint_id = '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActivatingError , <EOL> self . client . executions . start , <EOL> deployment_id = '<STR_LIT>' , <EOL> workflow_id = '<STR_LIT>' ) <EOL> def test_request_approval_in_maintenance_transition_mode ( self ) : <EOL> self . _start_maintenance_transition_mode ( ) <EOL> try : <EOL> self . client . blueprints . list ( ) <EOL> self . client . manager . get_version ( ) <EOL> except exceptions . CloudifyClientError : <EOL> self . fail ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_execution_amount_maintenance_activated ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertIsNone ( response . remaining_executions ) <EOL> def test_execution_amount_maintenance_deactivated ( self ) : <EOL> self . _activate_and_deactivate_maintenance_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertIsNone ( response . remaining_executions ) <EOL> def test_execution_amount_maintenance_activating ( self ) : <EOL> execution = self . _start_maintenance_transition_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( response . remaining_executions ) ) <EOL> self . assertEqual ( execution . id , <EOL> response . remaining_executions [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( execution . deployment_id , <EOL> response . remaining_executions [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( execution . workflow_id , <EOL> response . remaining_executions [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( models . Execution . STARTED , <EOL> response . remaining_executions [ <NUM_LIT:0> ] [ '<STR_LIT:status>' ] ) <EOL> def test_no_user_in_unsecured_maintenance_activation ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertFalse ( response . requested_by ) <EOL> def test_trigger_time_maintenance_activated ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertTrue ( response . activated_at ) <EOL> def test_trigger_time_maintenance_deactivated ( self ) : <EOL> self . _activate_and_deactivate_maintenance_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertFalse ( response . activated_at ) <EOL> def test_trigger_time_maintenance_activating ( self ) : <EOL> self . _start_maintenance_transition_mode ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertTrue ( response . activation_requested_at ) <EOL> def test_deactivate_maintenance_mode ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> response = self . client . maintenance_mode . deactivate ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_DEACTIVATED , response . status ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_DEACTIVATED , response . status ) <EOL> def test_request_approval_after_maintenance_mode_deactivation ( self ) : <EOL> self . _activate_and_deactivate_maintenance_mode ( ) <EOL> self . client . blueprints . upload ( <EOL> self . get_mock_blueprint_path ( ) , <EOL> blueprint_id = '<STR_LIT>' ) <EOL> self . client . deployments . create ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_multiple_maintenance_mode_deactivations ( self ) : <EOL> self . _activate_and_deactivate_maintenance_mode ( ) <EOL> try : <EOL> self . client . maintenance_mode . deactivate ( ) <EOL> self . fail ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> except exceptions . NotModifiedError as e : <EOL> self . assertEqual ( <NUM_LIT> , e . status_code ) <EOL> self . assertTrue ( '<STR_LIT>' in e . message ) <EOL> def test_maintenance_mode_activated_error_raised ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActiveError , <EOL> self . client . blueprints . list ) <EOL> def test_running_execution_maintenance_activating_error_raised ( self ) : <EOL> self . _test_different_execution_status_in_activating_mode ( ) <EOL> def test_pending_execution_maintenance_activating_error_raised ( self ) : <EOL> self . _test_different_execution_status_in_activating_mode ( <EOL> models . Execution . PENDING ) <EOL> def test_cancelling_execution_maintenance_activating_error_raised ( self ) : <EOL> self . _test_different_execution_status_in_activating_mode ( <EOL> models . Execution . CANCELLING ) <EOL> def test_force_cancelling_execution_maintenance_activating_error_raised ( <EOL> self ) : <EOL> self . _test_different_execution_status_in_activating_mode ( <EOL> models . Execution . FORCE_CANCELLING ) <EOL> def _test_different_execution_status_in_activating_mode ( <EOL> self , <EOL> execution_status = None ) : <EOL> self . client . blueprints . upload ( <EOL> self . get_mock_blueprint_path ( ) , <EOL> blueprint_id = '<STR_LIT>' ) <EOL> self . _start_maintenance_transition_mode ( <EOL> execution_status = execution_status ) <EOL> self . assertRaises ( exceptions . MaintenanceModeActivatingError , <EOL> self . client . deployments . create , <EOL> blueprint_id = '<STR_LIT>' , <EOL> deployment_id = '<STR_LIT>' ) <EOL> def _activate_maintenance_mode ( self ) : <EOL> self . client . maintenance_mode . activate ( ) <EOL> self . client . maintenance_mode . status ( ) <EOL> def _start_maintenance_transition_mode ( <EOL> self , <EOL> execution_status = models . Execution . STARTED ) : <EOL> ( blueprint_id , deployment_id , blueprint_response , <EOL> deployment_response ) = self . put_deployment ( '<STR_LIT>' ) <EOL> execution = self . client . executions . start ( deployment_id , '<STR_LIT>' ) <EOL> execution = self . client . executions . get ( execution . id ) <EOL> self . assertEquals ( '<STR_LIT>' , execution . status ) <EOL> storage_manager . _get_instance ( ) . update_execution_status ( <EOL> execution . id , execution_status , error = '<STR_LIT>' ) <EOL> self . client . maintenance_mode . activate ( ) <EOL> response = self . client . maintenance_mode . status ( ) <EOL> self . assertEqual ( MAINTENANCE_MODE_ACTIVATING , response . status ) <EOL> return execution <EOL> def _terminate_execution ( self , execution_id ) : <EOL> storage_manager . _get_instance ( ) . update_execution_status ( <EOL> execution_id , models . Execution . TERMINATED , error = '<STR_LIT>' ) <EOL> def _activate_and_deactivate_maintenance_mode ( self ) : <EOL> self . _activate_maintenance_mode ( ) <EOL> self . client . maintenance_mode . deactivate ( ) <EOL> def _get_remote_addr ( self ) : <EOL> return '<STR_LIT>' <EOL> def _get_host ( self ) : <EOL> return '<STR_LIT>' </s>
