<s> import os <EOL> import threading <EOL> from uuid import uuid4 <EOL> from opennode . cli . actions import templates <EOL> from opennode . cli . actions import storage <EOL> from opennode . cli . actions import vm <EOL> from opennode . cli . tests import BaseTestCase , signal_when_called <EOL> class TestOpenVZ ( BaseTestCase ) : <EOL> def _setUp ( self ) : <EOL> self . testsp = '<STR_LIT>' <EOL> storage . add_pool ( self . testsp ) <EOL> self . typical_vm_params = { '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : <NUM_LIT:1.0> } <EOL> def _tearDown ( self ) : <EOL> try : <EOL> storage . delete_pool ( self . testsp ) <EOL> except Exception : <EOL> pass <EOL> def test_deploy_vm ( self ) : <EOL> remote_repo = '<STR_LIT>' <EOL> template = templates . get_template_list ( remote_repo ) [ <NUM_LIT:0> ] <EOL> self . _addCleanup ( templates . delete_template , self . testsp , '<STR_LIT>' , template ) <EOL> templates . sync_template ( remote_repo , template , self . testsp , silent = True ) <EOL> vmuuid = str ( uuid4 ( ) ) <EOL> vms = vm . list_vms ( '<STR_LIT>' ) <EOL> self . assertTrue ( vmuuid not in map ( lambda v : v [ '<STR_LIT>' ] , vms ) ) <EOL> self . _addCleanup ( vm . undeploy_vm , '<STR_LIT>' , vmuuid ) <EOL> self . typical_vm_params . update ( { '<STR_LIT>' : vmuuid , <EOL> '<STR_LIT>' : template } ) <EOL> vm . deploy_vm ( '<STR_LIT>' , self . typical_vm_params ) <EOL> vms = vm . list_vms ( '<STR_LIT>' ) <EOL> expected_path = '<STR_LIT>' % ( self . testsp , template ) <EOL> self . assertTrue ( os . path . exists ( expected_path ) , expected_path ) <EOL> self . assertTrue ( vmuuid in map ( lambda v : v [ '<STR_LIT>' ] , vms ) , <EOL> "<STR_LIT:%s>" % map ( lambda v : v [ '<STR_LIT>' ] , vms ) ) <EOL> nvm = filter ( lambda v : v [ '<STR_LIT>' ] == vmuuid , vms ) <EOL> self . assertEqual ( nvm [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_deploy_vm_duplicate_uuid ( self ) : <EOL> remote_repo = '<STR_LIT>' <EOL> template = templates . get_template_list ( remote_repo ) [ <NUM_LIT:0> ] <EOL> self . _addCleanup ( templates . delete_template , self . testsp , '<STR_LIT>' , template ) <EOL> templates . sync_template ( remote_repo , template , self . testsp , silent = True ) <EOL> vmuuid = str ( uuid4 ( ) ) <EOL> vms = vm . list_vms ( '<STR_LIT>' ) <EOL> self . assertTrue ( vmuuid not in map ( lambda v : v [ '<STR_LIT>' ] , vms ) , <EOL> '<STR_LIT>' % vmuuid ) <EOL> self . _addCleanup ( vm . undeploy_vm , '<STR_LIT>' , vmuuid ) <EOL> self . typical_vm_params . update ( { '<STR_LIT>' : vmuuid , <EOL> '<STR_LIT>' : template , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> vm . deploy_vm ( '<STR_LIT>' , self . typical_vm_params ) <EOL> vms = vm . list_vms ( '<STR_LIT>' ) <EOL> expected_path = '<STR_LIT>' % ( self . testsp , template ) <EOL> self . assertTrue ( os . path . exists ( expected_path ) , expected_path ) <EOL> self . assertTrue ( vmuuid in map ( lambda v : v [ '<STR_LIT>' ] , vms ) , <EOL> "<STR_LIT:%s>" % map ( lambda v : v [ '<STR_LIT>' ] , vms ) ) <EOL> nvm = filter ( lambda v : v [ '<STR_LIT>' ] == vmuuid , vms ) <EOL> self . assertTrue ( nvm [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' ) <EOL> self . typical_vm_params . update ( { '<STR_LIT>' : vmuuid , <EOL> '<STR_LIT>' : template , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> vm . deploy_vm ( '<STR_LIT>' , self . typical_vm_params ) <EOL> vms = vm . list_vms ( '<STR_LIT>' ) <EOL> nvm = filter ( lambda v : v [ '<STR_LIT>' ] == vmuuid , vms ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( nvm ) ) <EOL> def test_deploy_vm_while_template_is_downloaded ( self ) : <EOL> remote_repo = '<STR_LIT>' <EOL> template = templates . get_template_list ( remote_repo ) [ <NUM_LIT:0> ] <EOL> good_to_go = threading . Event ( ) <EOL> try : <EOL> orig_unpack_template = templates . unpack_template <EOL> templates . unpack_template = signal_when_called ( good_to_go ) ( templates . unpack_template ) <EOL> sync_thread = threading . Thread ( name = '<STR_LIT>' , <EOL> target = templates . sync_template , <EOL> args = ( remote_repo , template , self . testsp ) , <EOL> kwargs = { '<STR_LIT>' : True } ) <EOL> self . _addCleanup ( templates . delete_template , self . testsp , '<STR_LIT>' , template ) <EOL> sync_thread . start ( ) <EOL> good_to_go . wait ( <NUM_LIT> ) <EOL> self . assertTrue ( good_to_go . is_set ( ) , '<STR_LIT>' ) <EOL> finally : <EOL> templates . unpack_template = orig_unpack_template <EOL> try : <EOL> vmuuid = str ( uuid4 ( ) ) <EOL> vms = vm . list_vms ( '<STR_LIT>' ) <EOL> self . _addCleanup ( vm . undeploy_vm , '<STR_LIT>' , vmuuid ) <EOL> self . typical_vm_params . update ( { '<STR_LIT>' : vmuuid , <EOL> '<STR_LIT>' : template , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> vm . deploy_vm ( '<STR_LIT>' , self . typical_vm_params ) <EOL> vms = vm . list_vms ( '<STR_LIT>' ) <EOL> self . assertFalse ( vmuuid in map ( lambda v : v [ '<STR_LIT>' ] , vms ) , <EOL> "<STR_LIT>" % ( vmuuid , map ( lambda v : v [ '<STR_LIT>' ] , vms ) ) ) <EOL> expected_path = '<STR_LIT>' % ( self . testsp , template ) <EOL> self . assertFalse ( os . path . exists ( expected_path ) , expected_path ) <EOL> finally : <EOL> sync_thread . join ( ) </s>
