<s> import os <EOL> import subprocess <EOL> import json <EOL> import shlex <EOL> from string import Template <EOL> from ConfigParser import SafeConfigParser <EOL> import yaml <EOL> import pyeapi <EOL> nodes = dict ( ) <EOL> testcases = list ( ) <EOL> config = SafeConfigParser ( ) <EOL> here = os . path . abspath ( os . path . dirname ( __file__ ) ) <EOL> class TestCase ( object ) : <EOL> def __init__ ( self , ** kwargs ) : <EOL> self . name = kwargs [ '<STR_LIT:name>' ] <EOL> self . module = kwargs [ '<STR_LIT>' ] <EOL> self . module_path = kwargs [ '<STR_LIT>' ] <EOL> self . host = None <EOL> self . inventory = kwargs . get ( '<STR_LIT>' ) <EOL> self . exitcode = kwargs . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . idempotent = kwargs . get ( '<STR_LIT>' , True ) <EOL> self . changed = kwargs . get ( '<STR_LIT>' , True ) <EOL> self . arguments = kwargs . get ( '<STR_LIT>' , list ( ) ) <EOL> self . variables = dict ( ) <EOL> self . setup = kwargs . get ( '<STR_LIT>' , list ( ) ) <EOL> self . teardown = kwargs . get ( '<STR_LIT>' , list ( ) ) <EOL> self . _command = None <EOL> def __str__ ( self ) : <EOL> return self . name <EOL> @ property <EOL> def command ( self ) : <EOL> if self . _command : <EOL> return self . _command <EOL> self . _command = self . _build_command ( ) <EOL> return self . _command <EOL> def set_host ( self , value ) : <EOL> self . host = value <EOL> def add_variable ( self , name , value ) : <EOL> self . variables [ name ] = value <EOL> def _build_command ( self ) : <EOL> command = [ '<STR_LIT>' ] <EOL> if self . inventory : <EOL> command . extend ( [ '<STR_LIT>' , self . inventory ] ) <EOL> modpath = '<STR_LIT:{}>' . format ( self . module_path ) <EOL> command . extend ( [ '<STR_LIT>' , modpath ] ) <EOL> command . extend ( [ '<STR_LIT>' , self . module ] ) <EOL> command . extend ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> command . extend ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> arguments = '<STR_LIT>' <EOL> for arg in self . arguments : <EOL> arguments += '<STR_LIT>' % ( arg [ '<STR_LIT:name>' ] , arg [ '<STR_LIT:value>' ] ) <EOL> command . extend ( [ '<STR_LIT>' , '<STR_LIT>' % arguments . strip ( ) ] ) <EOL> command . append ( self . host ) <EOL> return command <EOL> class TestModule ( object ) : <EOL> def __init__ ( self , testcase , node ) : <EOL> self . testcase = testcase <EOL> self . node = node <EOL> self . description = '<STR_LIT>' % ( testcase . module , testcase . name ) <EOL> def __call__ ( self ) : <EOL> self . output ( '<STR_LIT>' ) <EOL> response = self . run_module ( ) <EOL> assert response [ '<STR_LIT>' ] == self . testcase . changed <EOL> if self . testcase . idempotent : <EOL> self . output ( '<STR_LIT>' ) <EOL> response = self . run_module ( ) <EOL> assert not response [ '<STR_LIT>' ] <EOL> def setUp ( self ) : <EOL> if self . testcase . setup : <EOL> try : <EOL> self . output ( '<STR_LIT>' ) <EOL> self . output ( self . testcase . setup ) <EOL> self . node . config ( self . testcase . setup ) <EOL> except pyeapi . eapilib . CommandError as exc : <EOL> print exc <EOL> print self . testcase . setup <EOL> raise <EOL> def tearDown ( self ) : <EOL> if self . testcase . teardown : <EOL> self . output ( '<STR_LIT>' ) <EOL> self . output ( self . testcase . teardown ) <EOL> self . node . config ( self . testcase . teardown ) <EOL> def output ( self , text ) : <EOL> print '<STR_LIT>' , str ( text ) <EOL> def run_module ( self ) : <EOL> ( retcode , out , err ) = self . execute_module ( ) <EOL> assert ( retcode == self . testcase . exitcode ) <EOL> return self . parse_response ( out ) <EOL> def parse_response ( self , output ) : <EOL> try : <EOL> resp = json . loads ( str ( output ) . split ( '<STR_LIT>' ) [ <NUM_LIT:1> ] ) <EOL> except : <EOL> resp = json . loads ( str ( output ) . split ( '<STR_LIT>' ) [ <NUM_LIT:1> ] ) <EOL> return resp <EOL> def compile_command ( self ) : <EOL> command = '<STR_LIT:U+0020>' . join ( self . testcase . command ) <EOL> command = Template ( command ) . safe_substitute ( self . testcase . variables ) <EOL> return shlex . split ( command ) <EOL> def execute_module ( self ) : <EOL> command = self . compile_command ( ) <EOL> self . output ( '<STR_LIT>' % '<STR_LIT:U+0020>' . join ( command ) ) <EOL> stdout = subprocess . PIPE <EOL> stderr = subprocess . PIPE <EOL> proc = subprocess . Popen ( command , stdout = stdout , stderr = stderr ) <EOL> out , err = proc . communicate ( ) <EOL> self . output ( out ) <EOL> self . output ( err ) <EOL> return ( proc . returncode , out , err ) <EOL> def filter_modules ( modules , filenames ) : <EOL> if modules : <EOL> modules = [ '<STR_LIT>' . format ( s ) for s in modules . split ( '<STR_LIT:U+002C>' ) ] <EOL> return list ( set ( modules ) . intersection ( filenames ) ) <EOL> return filenames <EOL> def setup ( ) : <EOL> pyeapi . load_config ( os . path . join ( here , '<STR_LIT>' ) ) <EOL> for name in pyeapi . client . config . connections : <EOL> if name != '<STR_LIT:localhost>' : <EOL> nodes [ name ] = pyeapi . connect_to ( name ) <EOL> assert len ( nodes ) > <NUM_LIT:0> , '<STR_LIT>' <EOL> modules = os . environ . get ( '<STR_LIT>' ) <EOL> testcases_home = os . path . join ( here , '<STR_LIT>' ) <EOL> filenames = os . listdir ( testcases_home ) <EOL> for module in filter_modules ( modules , filenames ) : <EOL> path = os . path . join ( testcases_home , module ) <EOL> definition = yaml . load ( open ( path ) ) <EOL> defaults = definition . get ( '<STR_LIT>' , { } ) <EOL> for testcase in definition [ '<STR_LIT>' ] : <EOL> kwargs = defaults . copy ( ) <EOL> kwargs . update ( testcase ) <EOL> testcases . append ( TestCase ( ** kwargs ) ) <EOL> def test_module ( ) : <EOL> for name , node in nodes . items ( ) : <EOL> for testcase in testcases : <EOL> testcase . set_host ( name ) <EOL> testcase . add_variable ( '<STR_LIT:host>' , name ) <EOL> yield TestModule ( testcase , node ) </s>
