<s> import os <EOL> import random <EOL> import ConfigParser <EOL> from unittest import TestCase <EOL> from pysphere import VIServer , VIProperty , MORTypes , VIException , FaultTypes , VMPowerState , ToolsStatus <EOL> class VIServerTest ( TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> config_path = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) <EOL> cls . config = ConfigParser . ConfigParser ( ) <EOL> cls . config . read ( config_path ) <EOL> host = cls . config . get ( "<STR_LIT>" , "<STR_LIT:host>" ) <EOL> user = cls . config . get ( "<STR_LIT>" , "<STR_LIT:user>" ) <EOL> pswd = cls . config . get ( "<STR_LIT>" , "<STR_LIT:password>" ) <EOL> cls . server = VIServer ( ) <EOL> cls . server . connect ( host , user , pswd ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . server . disconnect ( ) <EOL> def test_get_hosts_by_datacenter ( self ) : <EOL> all_hosts = self . server . get_hosts ( ) <EOL> datacenters = self . server . get_datacenters ( ) <EOL> hosts_by_datacenter = [ self . server . get_hosts ( from_mor = dc ) <EOL> for dc in datacenters . keys ( ) ] <EOL> all_hosts2 = { } <EOL> [ all_hosts2 . update ( group ) for group in hosts_by_datacenter ] <EOL> assert all_hosts == all_hosts2 <EOL> def test_get_hosts_by_cluster ( self ) : <EOL> clusters = self . server . get_clusters ( ) <EOL> count_hosts1 = <NUM_LIT:0> <EOL> count_hosts2 = <NUM_LIT:0> <EOL> for cl in clusters . keys ( ) : <EOL> count_hosts1 += len ( self . server . get_hosts ( from_mor = cl ) ) <EOL> prop = VIProperty ( self . server , cl ) <EOL> count_hosts2 += len ( prop . host ) <EOL> assert count_hosts1 == count_hosts2 <EOL> def test_get_datastores_by_datacenter ( self ) : <EOL> all_datastores = self . server . get_datastores ( ) <EOL> datacenters = self . server . get_datacenters ( ) <EOL> datastores_by_datacenter = [ self . server . get_datastores ( from_mor = dc ) <EOL> for dc in datacenters . keys ( ) ] <EOL> all_datastores2 = { } <EOL> [ all_datastores2 . update ( group ) for group in datastores_by_datacenter ] <EOL> assert all_datastores == all_datastores2 <EOL> def test_get_resource_pools ( self ) : <EOL> all_rp = self . server . get_resource_pools ( ) <EOL> rp_by_root_rp = [ ] <EOL> for rp_key , rp_path in all_rp . items ( ) : <EOL> if rp_path . count ( '<STR_LIT:/>' ) == <NUM_LIT:1> : <EOL> rp_by_root_rp . extend ( self . server . get_resource_pools ( <EOL> from_mor = rp_key ) . keys ( ) ) <EOL> assert sorted ( rp_by_root_rp ) == sorted ( all_rp . keys ( ) ) <EOL> def test_get_registered_vms ( self ) : <EOL> all_vms = self . server . get_registered_vms ( ) <EOL> datacenters = self . server . get_datacenters ( ) <EOL> vms_by_datacenter = [ ] <EOL> resource_pools = self . server . get_resource_pools ( ) <EOL> vms_by_root_rp = [ ] <EOL> for dc in datacenters . keys ( ) : <EOL> vms_by_datacenter . extend ( self . server . get_registered_vms ( <EOL> datacenter = dc ) ) <EOL> for rp_key , rp_path in resource_pools . items ( ) : <EOL> if rp_path . count ( '<STR_LIT:/>' ) == <NUM_LIT:1> : <EOL> vms_by_root_rp . extend ( self . server . get_registered_vms ( <EOL> resource_pool = rp_key ) ) <EOL> oc = self . server . _retrieve_properties_traversal ( <EOL> property_names = [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> obj_type = MORTypes . VirtualMachine ) <EOL> templates = [ ] <EOL> for o in oc : <EOL> path = None <EOL> template = False <EOL> if not hasattr ( o , "<STR_LIT>" ) : <EOL> continue <EOL> for p in o . PropSet : <EOL> if p . Name == "<STR_LIT>" and p . Val : <EOL> template = True <EOL> if p . Name == "<STR_LIT>" : <EOL> path = p . Val <EOL> if template : <EOL> templates . append ( path ) <EOL> vms_by_root_rp . extend ( templates ) <EOL> assert sorted ( all_vms ) == sorted ( vms_by_datacenter ) == sorted ( <EOL> vms_by_root_rp ) <EOL> def test_get_registered_vms_by_datacenter ( self ) : <EOL> for dc_key , dc_name in self . server . get_datacenters ( ) . items ( ) : <EOL> vms1 = self . server . get_registered_vms ( datacenter = dc_key ) <EOL> vms2 = self . server . get_registered_vms ( datacenter = dc_name ) <EOL> assert vms1 == vms2 <EOL> def test_get_registered_vms_by_cluster ( self ) : <EOL> for cl_key , cl_name in self . server . get_clusters ( ) . items ( ) : <EOL> vms1 = self . server . get_registered_vms ( cluster = cl_key ) <EOL> vms2 = self . server . get_registered_vms ( cluster = cl_name ) <EOL> assert vms1 == vms2 <EOL> def test_get_registerd_vms_by_status ( self ) : <EOL> vms = self . server . get_registered_vms ( status = '<STR_LIT>' ) <EOL> if vms : <EOL> vm = self . server . get_vm_by_path ( random . choice ( vms ) ) <EOL> assert vm . get_status ( basic_status = True ) == VMPowerState . POWERED_ON <EOL> vms = self . server . get_registered_vms ( status = '<STR_LIT>' ) <EOL> if vms : <EOL> vm = self . server . get_vm_by_path ( random . choice ( vms ) ) <EOL> assert vm . get_status ( basic_status = True ) == VMPowerState . POWERED_OFF <EOL> vms = self . server . get_registered_vms ( status = '<STR_LIT>' ) <EOL> if vms : <EOL> vm = self . server . get_vm_by_path ( random . choice ( vms ) ) <EOL> assert vm . get_status ( basic_status = True ) == VMPowerState . SUSPENDED <EOL> def test_get_registered_vms_advanced_filters ( self ) : <EOL> vms = self . server . get_registered_vms ( advanced_filters = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> } ) <EOL> random . shuffle ( vms ) <EOL> for vm in vms [ : <NUM_LIT:10> ] : <EOL> vm = self . server . get_vm_by_path ( random . choice ( vms ) ) <EOL> assert vm . get_status ( basic_status = True ) == VMPowerState . POWERED_ON <EOL> assert vm . get_tools_status ( ) in [ ToolsStatus . RUNNING , <EOL> ToolsStatus . RUNNING_OLD ] <EOL> def test_is_connected ( self ) : <EOL> assert self . server . is_connected ( ) <EOL> def test_keep_session_alive ( self ) : <EOL> assert self . server . keep_session_alive ( ) <EOL> def test_api_version_and_server ( self ) : <EOL> assert self . server . get_api_version ( ) <EOL> assert self . server . get_server_type ( ) <EOL> def test_existent_vm_path ( self ) : <EOL> vms = self . server . get_registered_vms ( ) <EOL> path = random . choice ( vms ) <EOL> vm = self . server . get_vm_by_path ( path ) <EOL> assert path == vm . properties . config . files . vmPathName <EOL> def test_unexistent_vm_path ( self ) : <EOL> try : <EOL> self . server . get_vm_by_path ( "<STR_LIT>" ) <EOL> except VIException , e : <EOL> assert e . fault == FaultTypes . OBJECT_NOT_FOUND <EOL> else : <EOL> assert False <EOL> def test_existent_vm_path_by_datacenter ( self ) : <EOL> datacenters = self . server . get_datacenters ( ) <EOL> for k , v in datacenters . items ( ) : <EOL> vms = self . server . get_registered_vms ( datacenter = k ) <EOL> path = random . choice ( vms ) <EOL> vm1 = self . server . get_vm_by_path ( path , datacenter = k ) <EOL> vm2 = self . server . get_vm_by_path ( path , datacenter = v ) <EOL> assert vm1 . _mor == vm2 . _mor <EOL> def test_existent_vm_path_in_other_datacenter ( self ) : <EOL> datacenters = self . server . get_datacenters ( ) <EOL> if len ( datacenters ) >= <NUM_LIT:2> : <EOL> dc1 = datacenters . keys ( ) [ <NUM_LIT:0> ] <EOL> dc2 = datacenters . keys ( ) [ <NUM_LIT:1> ] <EOL> vms = self . server . get_registered_vms ( datacenter = dc1 ) <EOL> path = random . choice ( vms ) <EOL> try : <EOL> self . server . get_vm_by_path ( path , datacenter = dc2 ) <EOL> except VIException , e : <EOL> assert e . fault == FaultTypes . OBJECT_NOT_FOUND <EOL> else : <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> def test_get_object_properties ( self ) : <EOL> hosts = self . server . get_hosts ( ) <EOL> props = [ '<STR_LIT:name>' , '<STR_LIT>' ] <EOL> for mor , name in hosts . items ( ) : <EOL> prop = self . server . _get_object_properties ( mor , <EOL> property_names = props ) <EOL> assert prop . Obj == mor <EOL> assert len ( prop . PropSet ) == <NUM_LIT:2> <EOL> pname = None <EOL> pvmotion = None <EOL> for p in prop . PropSet : <EOL> if p . Name == '<STR_LIT:name>' : <EOL> pname = p . Val <EOL> elif p . Name == "<STR_LIT>" : <EOL> pvmotion = p . Val <EOL> else : <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> assert pname == name <EOL> assert pvmotion in ( True , False ) <EOL> def test_get_object_properties_get_all ( self ) : <EOL> datastores = self . server . get_datastores ( ) <EOL> props = [ '<STR_LIT:name>' , '<STR_LIT>' ] <EOL> for mor , name in datastores . items ( ) : <EOL> prop = self . server . _get_object_properties ( mor , <EOL> property_names = props , <EOL> get_all = True ) <EOL> assert prop . Obj == mor <EOL> assert len ( prop . PropSet ) > <NUM_LIT:2> <EOL> for p in prop . PropSet : <EOL> if p . Name == '<STR_LIT:name>' : <EOL> assert p . Val == name <EOL> def test_get_object_properties_bulk ( self ) : <EOL> hosts = self . server . get_hosts ( ) <EOL> datastores = self . server . get_datastores ( ) <EOL> oc = self . server . _retrieve_properties_traversal ( <EOL> property_names = [ '<STR_LIT:name>' ] , <EOL> obj_type = MORTypes . VirtualMachine ) <EOL> random . shuffle ( oc ) <EOL> vms = dict ( [ ( o . Obj , o . PropSet [ <NUM_LIT:0> ] . Val ) for o in oc [ : <NUM_LIT:10> ] ] ) <EOL> all_mors = hosts . keys ( ) + datastores . keys ( ) + vms . keys ( ) <EOL> found = [ ] <EOL> random . shuffle ( all_mors ) <EOL> properties = { MORTypes . VirtualMachine : [ '<STR_LIT:name>' , '<STR_LIT>' ] , <EOL> MORTypes . HostSystem : [ '<STR_LIT:name>' , '<STR_LIT>' ] , <EOL> MORTypes . Datastore : [ ] , <EOL> MORTypes . Folder : [ '<STR_LIT:name>' ] <EOL> } <EOL> res = self . server . _get_object_properties_bulk ( all_mors , properties ) <EOL> for o in res : <EOL> mor = o . Obj <EOL> found . append ( mor ) <EOL> assert mor in all_mors <EOL> if mor . get_attribute_type ( ) == MORTypes . VirtualMachine : <EOL> assert len ( o . PropSet ) == <NUM_LIT:2> <EOL> path = None <EOL> name = None <EOL> for p in o . PropSet : <EOL> if p . Name == "<STR_LIT:name>" : name = p . Val <EOL> elif p . Name == "<STR_LIT>" : path = p . Val <EOL> else : <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % p . Name ) <EOL> assert name == vms [ mor ] <EOL> assert path <EOL> elif mor . get_attribute_type ( ) == MORTypes . Datastore : <EOL> assert len ( o . PropSet ) > <NUM_LIT:5> <EOL> name = None <EOL> for p in o . PropSet : <EOL> if p . Name == "<STR_LIT:name>" : name = p . Val <EOL> assert name == datastores [ mor ] <EOL> elif mor . get_attribute_type ( ) == MORTypes . HostSystem : <EOL> assert len ( o . PropSet ) == <NUM_LIT:2> <EOL> name = None <EOL> vmotion = None <EOL> for p in o . PropSet : <EOL> if p . Name == "<STR_LIT:name>" : name = p . Val <EOL> elif p . Name == "<STR_LIT>" : vmotion = p . Val <EOL> else : <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % p . Name ) <EOL> assert name == hosts [ mor ] <EOL> assert vmotion in [ True , False ] <EOL> else : <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % mor . get_attribute_type ( ) ) <EOL> assert sorted ( all_mors ) == sorted ( found ) </s>
