<s> import tempfile <EOL> import time <EOL> import os <EOL> import shutil <EOL> import yaml <EOL> from cloudify_rest_client . exceptions import UserUnauthorizedError <EOL> from nose . plugins . attrib import attr <EOL> from manager_rest . test import base_test <EOL> from manager_rest . test . security . security_test_base import SecurityTestBase <EOL> _ , USERSTORE_FILE = tempfile . mkstemp ( prefix = '<STR_LIT>' ) <EOL> _ , ROLES_CONFIG_FILE = tempfile . mkstemp ( prefix = '<STR_LIT>' ) <EOL> USERSTORE_LOADED_SUCCESSFULLY = '<STR_LIT>' <EOL> USERSTORE_MISSING_USERS = '<STR_LIT>' <EOL> USERSTORE_FILE_INVALID_DICT = '<STR_LIT>' <EOL> USERSTORE_INVALID_YAML = '<STR_LIT>' <EOL> ROLES_CONFIG_FILE_INVALID_DICT = '<STR_LIT>' <EOL> ROLES_CONFIG_LOADED_SUCCESSFULLY = '<STR_LIT>' '<STR_LIT>' <EOL> @ attr ( client_min_version = <NUM_LIT:2> , client_max_version = base_test . LATEST_API_VERSION ) <EOL> class TestUserstoreReloadFile ( SecurityTestBase ) : <EOL> def setUp ( self ) : <EOL> self . create_roles_config_copy ( ) <EOL> self . restore_userstore_file ( ) <EOL> self . restore_roles_config ( ) <EOL> super ( TestUserstoreReloadFile , self ) . setUp ( ) <EOL> def create_configuration ( self ) : <EOL> test_config = super ( TestUserstoreReloadFile , self ) . create_configuration ( ) <EOL> self . security_log_path = test_config . security_audit_log_file <EOL> return test_config <EOL> def create_roles_config_copy ( self ) : <EOL> abs_path = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> file_path = os . path . join ( abs_path , '<STR_LIT>' ) <EOL> shutil . copyfile ( file_path , ROLES_CONFIG_FILE ) <EOL> def get_roles_config_file_path ( self ) : <EOL> return ROLES_CONFIG_FILE <EOL> def _init_test ( self ) : <EOL> self . clear_log_file ( ) <EOL> self . modify_userstore_file ( self . get_users ( ) , <EOL> self . get_groups ( ) , <EOL> USERSTORE_LOADED_SUCCESSFULLY ) <EOL> self . restore_roles_config ( ROLES_CONFIG_LOADED_SUCCESSFULLY ) <EOL> def _create_secured_client ( self ) : <EOL> auth_header = SecurityTestBase . create_auth_header ( username = '<STR_LIT>' , <EOL> password = '<STR_LIT>' ) <EOL> client = self . create_client ( headers = auth_header ) <EOL> return client <EOL> def restore_userstore_file ( self ) : <EOL> self . modify_userstore_file ( self . get_users ( ) , self . get_groups ( ) ) <EOL> def restore_roles_config ( self , message = None ) : <EOL> with open ( self . get_roles_config_file_path ( ) , '<STR_LIT:w>' ) as outfile : <EOL> outfile . write ( yaml . safe_dump ( self . get_valid_roles_config ( ) , <EOL> default_flow_style = True ) ) <EOL> if message : <EOL> self . wait_for_log_message ( message ) <EOL> def test_reload_userstore ( self ) : <EOL> self . _init_test ( ) <EOL> client = self . _create_secured_client ( ) <EOL> client . deployments . list ( ) <EOL> new_file_users = [ <EOL> { <EOL> '<STR_LIT:user>' : '<STR_LIT>' , <EOL> '<STR_LIT:password>' : '<STR_LIT>' <EOL> } <EOL> ] <EOL> self . modify_userstore_file ( new_file_users , <EOL> [ ] , <EOL> USERSTORE_LOADED_SUCCESSFULLY ) <EOL> self . assertRaises ( UserUnauthorizedError , client . deployments . list ) <EOL> def test_reload_invalid_userstore_dict ( self ) : <EOL> self . _test_invalid_userstore ( '<STR_LIT>' , <EOL> USERSTORE_FILE_INVALID_DICT ) <EOL> def test_reload_invalid_userstore_no_users ( self ) : <EOL> self . _test_invalid_userstore ( '<STR_LIT>' , <EOL> USERSTORE_MISSING_USERS ) <EOL> def test_reload_invalid_userstore_invalid_yaml ( self ) : <EOL> self . _test_invalid_userstore ( '<STR_LIT>' , USERSTORE_INVALID_YAML ) <EOL> def _test_invalid_userstore ( self , userstore_file_text , error_msg ) : <EOL> self . _init_test ( ) <EOL> client = self . _create_secured_client ( ) <EOL> client . deployments . list ( ) <EOL> self . corrupt_userstore_file ( userstore_file_text , <EOL> wait_for_message = error_msg ) <EOL> client . deployments . list ( ) <EOL> def test_reload_invalid_roles_config_dict ( self ) : <EOL> self . _init_test ( ) <EOL> client = self . _create_secured_client ( ) <EOL> with open ( self . get_roles_config_file_path ( ) , '<STR_LIT:w>' ) as outfile : <EOL> outfile . write ( '<STR_LIT>' ) <EOL> self . wait_for_log_message ( ROLES_CONFIG_FILE_INVALID_DICT ) <EOL> client . deployments . list ( ) <EOL> def test_reload_valid_roles_config_dict ( self ) : <EOL> self . _init_test ( ) <EOL> client = self . _create_secured_client ( ) <EOL> client . deployments . list ( ) <EOL> roles_config = self . get_valid_roles_config ( ) <EOL> roles_config [ '<STR_LIT>' ] [ '<STR_LIT>' ] = [ '<STR_LIT:GET>' ] <EOL> with open ( self . get_roles_config_file_path ( ) , '<STR_LIT:w>' ) as outfile : <EOL> outfile . write ( yaml . dump ( roles_config , <EOL> default_flow_style = True ) ) <EOL> self . wait_for_log_message ( ROLES_CONFIG_LOADED_SUCCESSFULLY ) <EOL> self . assertRaises ( UserUnauthorizedError , client . deployments . list ) <EOL> def modify_userstore_file ( self , users , groups , message = None ) : <EOL> userstore_settings = { <EOL> '<STR_LIT>' : users , <EOL> '<STR_LIT>' : groups <EOL> } <EOL> with open ( USERSTORE_FILE , '<STR_LIT:w>' ) as outfile : <EOL> outfile . write ( yaml . dump ( userstore_settings , <EOL> default_flow_style = True ) ) <EOL> if message : <EOL> self . wait_for_log_message ( message ) <EOL> def corrupt_userstore_file ( self , text , wait_for_message = None ) : <EOL> with open ( USERSTORE_FILE , '<STR_LIT:w>' ) as outfile : <EOL> outfile . write ( yaml . dump ( text , <EOL> default_flow_style = True ) ) <EOL> if wait_for_message : <EOL> self . wait_for_log_message ( USERSTORE_FILE_INVALID_DICT ) <EOL> def clear_log_file ( self ) : <EOL> with open ( self . security_log_path , "<STR_LIT:w>" ) : <EOL> pass <EOL> def wait_for_log_message ( self , message ) : <EOL> timeout = time . time ( ) + <NUM_LIT> <EOL> while time . time ( ) < timeout : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> if self . log_contains ( message ) : <EOL> return <EOL> self . fail ( '<STR_LIT>' <EOL> . format ( message ) ) <EOL> def log_contains ( self , expected_text ) : <EOL> with open ( self . security_log_path ) as f : <EOL> logged_text = f . read ( ) <EOL> return expected_text in logged_text <EOL> def get_userstore_configuration ( self ) : <EOL> return { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : USERSTORE_FILE <EOL> } <EOL> } <EOL> def get_valid_roles_config ( self ) : <EOL> return { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:*>' : [ '<STR_LIT>' ] <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:*>' : [ '<STR_LIT:*>' ] <EOL> } <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:*>' : [ '<STR_LIT:GET>' ] <EOL> } <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:*>' : [ '<STR_LIT:*>' ] <EOL> } <EOL> } <EOL> } </s>
