<s> '''<STR_LIT>''' <EOL> from __future__ import absolute_import <EOL> import getpass <EOL> import os <EOL> import sys <EOL> import stat <EOL> import shutil <EOL> import resource <EOL> import tempfile <EOL> import socket <EOL> from salttesting import skipIf , TestCase <EOL> from salttesting . helpers import ( <EOL> ensure_in_syspath , <EOL> requires_network , <EOL> TestsLoggingHandler <EOL> ) <EOL> ensure_in_syspath ( '<STR_LIT>' ) <EOL> import salt . utils <EOL> import integration <EOL> from salt . utils . verify import ( <EOL> check_user , <EOL> verify_env , <EOL> verify_socket , <EOL> zmq_version , <EOL> check_max_open_files , <EOL> valid_id <EOL> ) <EOL> from salt . ext . six . moves import range <EOL> class TestVerify ( TestCase ) : <EOL> '''<STR_LIT>''' <EOL> def test_valid_id_exception_handler ( self ) : <EOL> '''<STR_LIT>''' <EOL> opts = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . assertFalse ( valid_id ( opts , None ) ) <EOL> def test_zmq_verify ( self ) : <EOL> self . assertTrue ( zmq_version ( ) ) <EOL> def test_zmq_verify_insufficient ( self ) : <EOL> import zmq <EOL> zmq . __version__ = '<STR_LIT>' <EOL> self . assertFalse ( zmq_version ( ) ) <EOL> def test_user ( self ) : <EOL> self . assertTrue ( check_user ( getpass . getuser ( ) ) ) <EOL> def test_no_user ( self ) : <EOL> class FakeWriter ( object ) : <EOL> def __init__ ( self ) : <EOL> self . output = "<STR_LIT>" <EOL> def write ( self , data ) : <EOL> self . output += data <EOL> stderr = sys . stderr <EOL> writer = FakeWriter ( ) <EOL> sys . stderr = writer <EOL> self . assertFalse ( check_user ( '<STR_LIT>' ) ) <EOL> sys . stderr = stderr <EOL> if writer . output != '<STR_LIT>' : <EOL> sys . stderr . write ( writer . output ) <EOL> @ skipIf ( sys . platform . startswith ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def test_verify_env ( self ) : <EOL> root_dir = tempfile . mkdtemp ( dir = integration . SYS_TMP_DIR ) <EOL> var_dir = os . path . join ( root_dir , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> verify_env ( [ var_dir ] , getpass . getuser ( ) ) <EOL> self . assertTrue ( os . path . exists ( var_dir ) ) <EOL> dir_stat = os . stat ( var_dir ) <EOL> self . assertEqual ( dir_stat . st_uid , os . getuid ( ) ) <EOL> self . assertEqual ( dir_stat . st_mode & stat . S_IRWXU , stat . S_IRWXU ) <EOL> self . assertEqual ( dir_stat . st_mode & stat . S_IRWXG , <NUM_LIT> ) <EOL> self . assertEqual ( dir_stat . st_mode & stat . S_IRWXO , <NUM_LIT:5> ) <EOL> @ requires_network ( only_local_network = True ) <EOL> def test_verify_socket ( self ) : <EOL> self . assertTrue ( verify_socket ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> ) ) <EOL> if socket . has_ipv6 : <EOL> try : <EOL> self . assertTrue ( verify_socket ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> ) ) <EOL> except socket . error as serr : <EOL> pass <EOL> @ skipIf ( True , '<STR_LIT>' ) <EOL> def test_max_open_files ( self ) : <EOL> with TestsLoggingHandler ( ) as handler : <EOL> logmsg_dbg = ( <EOL> '<STR_LIT>' <EOL> ) <EOL> logmsg_chk = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> logmsg_crash = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> mof_s , mof_h = resource . getrlimit ( resource . RLIMIT_NOFILE ) <EOL> tempdir = tempfile . mkdtemp ( prefix = '<STR_LIT>' ) <EOL> keys_dir = os . path . join ( tempdir , '<STR_LIT>' ) <EOL> os . makedirs ( keys_dir ) <EOL> mof_test = <NUM_LIT> <EOL> resource . setrlimit ( resource . RLIMIT_NOFILE , ( mof_test , mof_h ) ) <EOL> try : <EOL> prev = <NUM_LIT:0> <EOL> for newmax , level in ( ( <NUM_LIT> , None ) , ( <NUM_LIT> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , '<STR_LIT>' ) , ( <NUM_LIT> , '<STR_LIT>' ) ) : <EOL> for n in range ( prev , newmax ) : <EOL> kpath = os . path . join ( keys_dir , str ( n ) ) <EOL> with salt . utils . fopen ( kpath , '<STR_LIT:w>' ) as fp_ : <EOL> fp_ . write ( str ( n ) ) <EOL> opts = { <EOL> '<STR_LIT>' : newmax , <EOL> '<STR_LIT>' : tempdir <EOL> } <EOL> check_max_open_files ( opts ) <EOL> if level is None : <EOL> self . assertEqual ( <EOL> [ logmsg_dbg . format ( newmax ) ] , handler . messages <EOL> ) <EOL> else : <EOL> self . assertIn ( <EOL> logmsg_dbg . format ( newmax ) , handler . messages <EOL> ) <EOL> self . assertIn ( <EOL> logmsg_chk . format ( <EOL> level , <EOL> newmax , <EOL> mof_test , <EOL> mof_h - newmax , <EOL> ) , <EOL> handler . messages <EOL> ) <EOL> handler . clear ( ) <EOL> prev = newmax <EOL> newmax = mof_test <EOL> for n in range ( prev , newmax ) : <EOL> kpath = os . path . join ( keys_dir , str ( n ) ) <EOL> with salt . utils . fopen ( kpath , '<STR_LIT:w>' ) as fp_ : <EOL> fp_ . write ( str ( n ) ) <EOL> opts = { <EOL> '<STR_LIT>' : newmax , <EOL> '<STR_LIT>' : tempdir <EOL> } <EOL> check_max_open_files ( opts ) <EOL> self . assertIn ( logmsg_dbg . format ( newmax ) , handler . messages ) <EOL> self . assertIn ( <EOL> logmsg_crash . format ( <EOL> '<STR_LIT>' , <EOL> newmax , <EOL> mof_test , <EOL> mof_h - newmax , <EOL> ) , <EOL> handler . messages <EOL> ) <EOL> handler . clear ( ) <EOL> except IOError as err : <EOL> if err . errno == <NUM_LIT> : <EOL> self . skipTest ( '<STR_LIT>' ) <EOL> raise <EOL> finally : <EOL> shutil . rmtree ( tempdir ) <EOL> resource . setrlimit ( resource . RLIMIT_NOFILE , ( mof_s , mof_h ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from integration import run_tests <EOL> run_tests ( TestVerify , needs_daemon = False ) </s>
