Mirrors/vsphere-automation-sdk-python
1
0
Fork 0
mirror of https://github.com/vmware/vsphere-automation-sdk-python.git synced 2024-11-22 09:39:58 -05:00
Code

fix format to meet pep8 requirements

Browse Source
This commit is contained in:
Tianhao He 2017-02-24 00:41:48 -08:00
parent e24bbb74cd
commit 81ac07e798
42 changed files with 235 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
samples/vsphere/common/__init__.py
View File

@ -2,7 +2,9 @@
# package as multiple distribution
try:
import pkg_resources
pkg_resources.declare_namespace(__name__)
except ImportError:
from pkgutil import extend_path
__path__ = extend_path(__path__, __name__) # @ReservedAssignment

1
samples/vsphere/common/id_generator.py
View File

@ -36,6 +36,7 @@ def main():
print(generate_random_string(5))
print(rand('Simple VM-'))
# Start program
if __name__ == "__main__":
main()

122
samples/vsphere/common/lookup_service_helper.py
View File

@ -16,6 +16,7 @@ __copyright__ = 'Copyright 2013 VMware, Inc. All rights reserved.'
import os
from suds.client import Client
class LookupServiceHelper(object):
def __init__(self, wsdl_url, soap_url, skip_verification):
self.wsdl_url = wsdl_url
@ -50,11 +51,13 @@ class LookupServiceHelper(object):
assert self.client is not None
self.client.set_options(service='LsService', port='LsPort')
self.managedObjectReference = self.client.factory.create('ns0:ManagedObjectReference')
self.managedObjectReference = self.client.factory.create(
'ns0:ManagedObjectReference')
self.managedObjectReference._type = 'LookupServiceInstance'
self.managedObjectReference.value = 'ServiceInstance'
lookupServiceContent = self.client.service.RetrieveServiceContent(self.managedObjectReference)
lookupServiceContent = self.client.service.RetrieveServiceContent(
self.managedObjectReference)
self.serviceRegistration = lookupServiceContent.serviceRegistration
@ -67,9 +70,9 @@ class LookupServiceHelper(object):
:return: list of SSO Service endpoint URLs
"""
return self.__find_platform_service_urls(product='com.vmware.cis',
service='cs.identity',
endpoint='com.vmware.cis.cs.identity.sso',
protocol='wsTrust')
service='cs.identity',
endpoint='com.vmware.cis.cs.identity.sso',
protocol='wsTrust')
def find_sso_url(self):
"""
@ -81,9 +84,9 @@ class LookupServiceHelper(object):
:return: SSO Service endpoint URL
"""
result = self.__find_platform_service_urls(product='com.vmware.cis',
service='cs.identity',
endpoint='com.vmware.cis.cs.identity.sso',
protocol='wsTrust')
service='cs.identity',
endpoint='com.vmware.cis.cs.identity.sso',
protocol='wsTrust')
return result[0]
def find_vapi_urls(self):
@ -95,9 +98,9 @@ class LookupServiceHelper(object):
:return: vapi service endpoint URLs in a dictionary where the key is the node_id and the value is the service URL
"""
return self.__find_service_urls(product='com.vmware.cis',
service='cs.vapi',
endpoint='com.vmware.vapi.endpoint',
protocol='vapi.json.https.public')
service='cs.vapi',
endpoint='com.vmware.vapi.endpoint',
protocol='vapi.json.https.public')
def find_vapi_url(self, node_id):
"""
@ -110,9 +113,9 @@ class LookupServiceHelper(object):
"""
assert node_id is not None
result = self.__find_service_urls(product='com.vmware.cis',
service='cs.vapi',
endpoint='com.vmware.vapi.endpoint',
protocol='vapi.json.https.public')
service='cs.vapi',
endpoint='com.vmware.vapi.endpoint',
protocol='vapi.json.https.public')
assert result is not None
return result.get(node_id)
@ -125,9 +128,9 @@ class LookupServiceHelper(object):
:return: vim service endpoint URLs in a dictionary where the key is the node_id and the value is the service URL
"""
return self.__find_service_urls(product='com.vmware.cis',
service='vcenterserver',
endpoint='com.vmware.vim',
protocol='vmomi')
service='vcenterserver',
endpoint='com.vmware.vim',
protocol='vmomi')
def find_vim_url(self, node_id):
"""
@ -140,9 +143,9 @@ class LookupServiceHelper(object):
"""
assert node_id is not None
result = self.__find_service_urls(product='com.vmware.cis',
service='vcenterserver',
endpoint='com.vmware.vim',
protocol='vmomi')
service='vcenterserver',
endpoint='com.vmware.vim',
protocol='vmomi')
assert result is not None
return result.get(node_id)
@ -155,9 +158,9 @@ class LookupServiceHelper(object):
:return: spbm service endpoint URLs in a dictionary where the key is the node_id and the value is the service URL
"""
return self.__find_service_urls(product='com.vmware.vim.sms',
service='sms',
endpoint='com.vmware.vim.pbm',
protocol='https')
service='sms',
endpoint='com.vmware.vim.pbm',
protocol='https')
def find_vim_pbm_url(self, node_id):
"""
@ -170,9 +173,9 @@ class LookupServiceHelper(object):
"""
assert node_id is not None
result = self.__find_service_urls(product='com.vmware.vim.sms',
service='sms',
endpoint='com.vmware.vim.pbm',
protocol='https')
service='sms',
endpoint='com.vmware.vim.pbm',
protocol='https')
assert result is not None
return result.get(node_id)
@ -184,20 +187,27 @@ class LookupServiceHelper(object):
assert self.client is not None
assert self.serviceRegistration is not None
lookupServiceRegistrationFilter = self.__create_filter_spec(product, service, endpoint, protocol)
lookupServiceRegistrationFilter = self.__create_filter_spec(product,
service,
endpoint,
protocol)
result = self.client.service.List(self.serviceRegistration, lookupServiceRegistrationFilter)
result = self.client.service.List(self.serviceRegistration,
lookupServiceRegistrationFilter)
assert len(result) > 0
# Support for MxN
# return the results in a dictionary where key is NodeId and Value is Service URL
results_dict = {}
for lookupServiceRegistrationInfo in result:
lookupServiceRegistrationEndpoint = lookupServiceRegistrationInfo.serviceEndpoints[0]
lookupServiceRegistrationEndpoint = \
lookupServiceRegistrationInfo.serviceEndpoints[0]
assert lookupServiceRegistrationEndpoint is not None
results_dict[lookupServiceRegistrationInfo.nodeId] = lookupServiceRegistrationEndpoint.url
results_dict[
lookupServiceRegistrationInfo.nodeId] = lookupServiceRegistrationEndpoint.url
return results_dict
def __find_platform_service_urls(self, product, service, endpoint, protocol):
def __find_platform_service_urls(self, product, service, endpoint,
protocol):
"""
Finds the endpoint URLs of a service running on PSCs (Platform Service Controller).
Returns a list of service URLs since there is no node id associated with the PSC.
@ -205,14 +215,19 @@ class LookupServiceHelper(object):
assert self.client is not None
assert self.serviceRegistration is not None
lookupServiceRegistrationFilter = self.__create_filter_spec(product, service, endpoint, protocol)
lookupServiceRegistrationFilter = self.__create_filter_spec(product,
service,
endpoint,
protocol)
result = self.client.service.List(self.serviceRegistration, lookupServiceRegistrationFilter)
result = self.client.service.List(self.serviceRegistration,
lookupServiceRegistrationFilter)
assert len(result) > 0
urls = []
for lookupServiceRegistrationInfo in result:
lookupServiceRegistrationEndpoint = lookupServiceRegistrationInfo.serviceEndpoints[0]
lookupServiceRegistrationEndpoint = \
lookupServiceRegistrationInfo.serviceEndpoints[0]
assert lookupServiceRegistrationEndpoint is not None
urls.append(lookupServiceRegistrationEndpoint.url)
return urls
@ -220,15 +235,18 @@ class LookupServiceHelper(object):
def __create_filter_spec(self, product, service, endpoint, protocol):
assert self.client is not None
lookupServiceRegistrationServiceType = self.client.factory.create('ns0:LookupServiceRegistrationServiceType')
lookupServiceRegistrationServiceType = self.client.factory.create(
'ns0:LookupServiceRegistrationServiceType')
lookupServiceRegistrationServiceType.product = product
lookupServiceRegistrationServiceType.type = service
lookupServiceRegistrationEndpointType = self.client.factory.create('ns0:LookupServiceRegistrationEndpointType')
lookupServiceRegistrationEndpointType = self.client.factory.create(
'ns0:LookupServiceRegistrationEndpointType')
lookupServiceRegistrationEndpointType.type = endpoint
lookupServiceRegistrationEndpointType.protocol = protocol
lookupServiceRegistrationFilter = self.client.factory.create('ns0:LookupServiceRegistrationFilter')
lookupServiceRegistrationFilter = self.client.factory.create(
'ns0:LookupServiceRegistrationFilter')
lookupServiceRegistrationFilter.serviceType = lookupServiceRegistrationServiceType
lookupServiceRegistrationFilter.endpointType = lookupServiceRegistrationEndpointType
return lookupServiceRegistrationFilter
@ -243,26 +261,31 @@ class LookupServiceHelper(object):
assert self.client is not None
assert self.serviceRegistration is not None
lookupServiceRegistrationServiceType = self.client.factory.create('ns0:LookupServiceRegistrationServiceType')
lookupServiceRegistrationServiceType = self.client.factory.create(
'ns0:LookupServiceRegistrationServiceType')
lookupServiceRegistrationServiceType.product = 'com.vmware.cis'
lookupServiceRegistrationServiceType.type = 'vcenterserver'
lookupServiceRegistrationEndpointType = self.client.factory.create('ns0:LookupServiceRegistrationEndpointType')
lookupServiceRegistrationEndpointType = self.client.factory.create(
'ns0:LookupServiceRegistrationEndpointType')
lookupServiceRegistrationEndpointType.type = 'com.vmware.vim'
lookupServiceRegistrationEndpointType.protocol = 'vmomi'
lookupServiceRegistrationFilter = self.client.factory.create('ns0:LookupServiceRegistrationFilter')
lookupServiceRegistrationFilter = self.client.factory.create(
'ns0:LookupServiceRegistrationFilter')
lookupServiceRegistrationFilter.serviceType = lookupServiceRegistrationServiceType
lookupServiceRegistrationFilter.endpointType = lookupServiceRegistrationEndpointType
result = self.client.service.List(self.serviceRegistration, lookupServiceRegistrationFilter)
result = self.client.service.List(self.serviceRegistration,
lookupServiceRegistrationFilter)
assert len(result) > 0
results_dict = {}
for lookupServiceRegistrationInfo in result:
for lookupServiceRegistrationAttribute in lookupServiceRegistrationInfo.serviceAttributes:
if lookupServiceRegistrationAttribute.key == 'com.vmware.vim.vcenter.instanceName':
results_dict[lookupServiceRegistrationAttribute.value] = lookupServiceRegistrationInfo.nodeId
results_dict[
lookupServiceRegistrationAttribute.value] = lookupServiceRegistrationInfo.nodeId
return results_dict
def get_mgmt_node_id(self, instance_name):
@ -295,8 +318,10 @@ class LookupServiceHelper(object):
if len(result) < 1:
raise Exception('No management node found')
if len(result) > 1:
raise MultipleManagementNodeException(MultipleManagementNodeException.format(result))
return list(result.keys())[0], list(result.values())[0] # python 3.x dict.keys() returns a view rather than a list
raise MultipleManagementNodeException(
MultipleManagementNodeException.format(result))
return list(result.keys())[0], list(result.values())[
0] # python 3.x dict.keys() returns a view rather than a list
class MultipleManagementNodeException(Exception):
@ -315,13 +340,15 @@ class MultipleManagementNodeException(Exception):
"""
message = 'Multiple Management Node Found on server'
for k, v in nodes.items():
message = message + os.linesep + 'Node name: {0} uuid: {1}'.format(k, v)
message = message + os.linesep + 'Node name: {0} uuid: {1}'.format(
k, v)
return message
def main():
lookup_service_helper = LookupServiceHelper(wsdl_url='file:///path/to/lookupservice.wsdl',
soap_url='https://server_ip/lookupservice/sdk')
lookup_service_helper = LookupServiceHelper(
wsdl_url='file:///path/to/lookupservice.wsdl',
soap_url='https://server_ip/lookupservice/sdk')
lookup_service_helper.connect()
print(lookup_service_helper.find_sso_url())
print(lookup_service_helper.find_vapi_urls())
@ -329,6 +356,7 @@ def main():
print(lookup_service_helper.find_vim_pbm_urls())
print(lookup_service_helper.find_mgmt_nodes())
# Start program
if __name__ == "__main__":
main()

10
samples/vsphere/common/platform_service_controller.py
View File

@ -24,7 +24,9 @@ class PlatformServiceController(object):
"""
Manages services on the infrastructure node (e.g. lookup service, SSO etc.)
"""
def __init__(self, lswsdlurl, lssoapurl, ssousername, ssopassword, skip_verification):
def __init__(self, lswsdlurl, lssoapurl, ssousername, ssopassword,
skip_verification):
self.lswsdlurl = lswsdlurl
self.lssoapurl = lssoapurl
self.ssousername = ssousername
@ -48,11 +50,13 @@ class PlatformServiceController(object):
self.stsurl = self.lookupservicehelper.find_sso_url()
assert self.stsurl is not None
print('Retrieving a SAML bearer token from STS url : {0}'.format(self.stsurl))
print('Retrieving a SAML bearer token from STS url : {0}'.format(
self.stsurl))
au = sso.SsoAuthenticator(self.stsurl)
context = None
if self.skip_verification:
context = get_unverified_context()
self.bearer_token = au.get_bearer_saml_assertion(
self.ssousername, self.ssopassword, delegatable=True, ssl_context=context)
self.ssousername, self.ssopassword, delegatable=True,
ssl_context=context)
self.sec_ctx = create_saml_bearer_security_context(self.bearer_token)

13
samples/vsphere/common/sample_base.py
View File

@ -13,7 +13,6 @@
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2013, 2016 VMware, Inc. All rights reserved.'
import argparse
import traceback
from samples.vsphere.common.service_manager_factory import ServiceManagerFactory
@ -26,12 +25,16 @@ class SampleBase(object):
self.description = description
# setup the argument parser
self.argparser = argparse.ArgumentParser(description=description)
self.argparser.add_argument('-s', '--server', help='Hostname of vCenter Server')
self.argparser.add_argument('-u', '--username', help='Username to login to the vCenter Server')
self.argparser.add_argument('-p', '--password', help='Password to login to the vCenter Server')
self.argparser.add_argument('-s', '--server',
help='Hostname of vCenter Server')
self.argparser.add_argument('-u', '--username',
help='Username to login to the vCenter Server')
self.argparser.add_argument('-p', '--password',
help='Password to login to the vCenter Server')
self.argparser.add_argument('-c', '--cleardata', action='store_true',
help='Clears the sample data on server after running')
self.argparser.add_argument('-v', '--skipverification', action='store_true',
self.argparser.add_argument('-v', '--skipverification',
action='store_true',
help='Do not verify server certificate')
self.args = None
self.server = None

1
samples/vsphere/common/service_manager.py
View File

@ -64,4 +64,3 @@ class ServiceManager(object):
print('disconnecting the session')
vapiconnect.logout(self.stub_config)
Disconnect(self.si)

1
samples/vsphere/common/service_manager_factory.py
View File

@ -36,5 +36,6 @@ class ServiceManagerFactory(object):
if cls.service_manager:
cls.service_manager.disconnect()
import atexit
atexit.register(ServiceManagerFactory.disconnect)

1
samples/vsphere/common/ssl_helper.py
View File

@ -15,6 +15,7 @@ __copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
import ssl
def get_unverified_context():
"""
Get an unverified ssl context. Used to disable the server certificate

2
samples/vsphere/common/vim/__init__.py
View File

@ -2,7 +2,9 @@
# package as multiple distribution
try:
import pkg_resources
pkg_resources.declare_namespace(__name__)
except ImportError:
from pkgutil import extend_path
__path__ = extend_path(__path__, __name__) # @ReservedAssignment

10
samples/vsphere/common/vim/datastore_file.py
View File

@ -182,7 +182,7 @@ class File(object):
match_pattern = None
dirname = None
if path != None:
if path is not None:
# Determine the dirname and the basename making sure only the
# basename is passed in the match_pattern
paths = path.split('/')
@ -239,13 +239,13 @@ class File(object):
stub = self._datastore_mo._stub
cookie = self._make_cookie(stub)
f = None
if src_file != None:
if src_file is not None:
f = src_file
elif src_url != None:
elif src_url is not None:
f = requests.get(src_url, stream=True)
elif src_path != None:
elif src_path is not None:
f = open(src_file, 'wb')
elif content == None:
elif content is None:
raise Exception('No input provided for put')
if f:

2
samples/vsphere/common/vim/helpers/__init__.py
View File

@ -2,7 +2,9 @@
# package as multiple distribution
try:
import pkg_resources
pkg_resources.declare_namespace(__name__)
except ImportError:
from pkgutil import extend_path
__path__ = extend_path(__path__, __name__) # @ReservedAssignment

7
samples/vsphere/common/vim/helpers/get_cluster_by_name.py
View File

@ -23,6 +23,7 @@ class GetClusterByName(SampleBase):
"""
Retrieves the given cluster MOID from VC using container view
"""
def __init__(self):
SampleBase.__init__(self, self.__doc__)
self.cluster_name = None
@ -30,7 +31,8 @@ class GetClusterByName(SampleBase):
self.servicemanager = None
def _options(self):
self.argparser.add_argument('-clustername', '--clustername', help='Name of the cluster to be queried')
self.argparser.add_argument('-clustername', '--clustername',
help='Name of the cluster to be queried')
def _setup(self):
if self.cluster_name is None:
@ -41,7 +43,8 @@ class GetClusterByName(SampleBase):
def _execute(self):
content = self.servicemanager.content
cluster_obj = get_obj(content, [vim.ClusterComputeResource], self.cluster_name)
cluster_obj = get_obj(content, [vim.ClusterComputeResource],
self.cluster_name)
if cluster_obj is not None:
self.mo_id = cluster_obj._GetMoId()
print('Cluster MoId: {0}'.format(self.mo_id))

4
samples/vsphere/common/vim/helpers/get_datastore_by_name.py
View File

@ -23,6 +23,7 @@ class GetDatastoreByName(SampleBase):
"""
Retrieves the given datastore MOID from VC using container view
"""
def __init__(self):
SampleBase.__init__(self, self.__doc__)
self.datastore_name = None
@ -30,7 +31,8 @@ class GetDatastoreByName(SampleBase):
self.servicemanager = None
def _options(self):
self.argparser.add_argument('-datastorename', '--datastorename', help='Name of the datastore to be queried')
self.argparser.add_argument('-datastorename', '--datastorename',
help='Name of the datastore to be queried')
def _setup(self):
if self.datastore_name is None:

24
samples/vsphere/common/vim/helpers/vim_utils.py
View File

@ -15,7 +15,6 @@ __copyright__ = 'Copyright 2013 VMware, Inc. All rights reserved.'
from pyVmomi import vim, vmodl
_views = [] # list of container views
@ -24,7 +23,8 @@ def get_obj(content, vimtype, name):
Get the vsphere managed object associated with a given text name
"""
obj = None
container = content.viewManager.CreateContainerView(content.rootFolder, vimtype, True)
container = content.viewManager.CreateContainerView(content.rootFolder,
vimtype, True)
_views.append(container)
for c in container.view:
if c.name == name:
@ -38,7 +38,8 @@ def get_obj_by_moId(content, vimtype, moid):
Get the vsphere managed object by moid value
"""
obj = None
container = content.viewManager.CreateContainerView(content.rootFolder, vimtype, True)
container = content.viewManager.CreateContainerView(content.rootFolder,
vimtype, True)
_views.append(container)
for c in container.view:
if c._GetMoId() == moid:
@ -55,8 +56,9 @@ def delete_object(content, mo):
try:
wait_for_tasks(content, [mo.Destroy()])
print('Deleted {0}'.format(mo._GetMoId()))
except:
print('Unexpected error while deleting managed object {0}'.format(mo._GetMoId()))
except Exception:
print('Unexpected error while deleting managed object {0}'.format(
mo._GetMoId()))
return False
return True
@ -72,7 +74,7 @@ def poweron_vm(content, mo):
try:
wait_for_tasks(content, [mo.PowerOn()])
print('{0} powered on successfully'.format(mo._GetMoId()))
except:
except Exception:
print('Unexpected error while powering on vm {0}'.format(mo._GetMoId()))
return False
return True
@ -89,8 +91,9 @@ def poweroff_vm(content, mo):
try:
wait_for_tasks(content, [mo.PowerOff()])
print('{0} powered off successfully'.format(mo._GetMoId()))
except:
print('Unexpected error while powering off vm {0}'.format(mo._GetMoId()))
except Exception:
print(
'Unexpected error while powering off vm {0}'.format(mo._GetMoId()))
return False
return True
@ -102,7 +105,8 @@ def wait_for_tasks(content, tasks):
taskList = [str(task) for task in tasks]
# Create filter
objSpecs = [vmodl.query.PropertyCollector.ObjectSpec(obj=task) for task in tasks]
objSpecs = [vmodl.query.PropertyCollector.ObjectSpec(obj=task) for task in
tasks]
propSpec = vmodl.query.PropertyCollector.PropertySpec(type=vim.Task,
pathSet=[], all=True)
filterSpec = vmodl.query.PropertyCollector.FilterSpec()
@ -149,5 +153,7 @@ def __destroy_container_views():
except vmodl.fault.ManagedObjectNotFound:
pass # silently bypass the exception if the objects are already deleted/not found on the server
import atexit
atexit.register(__destroy_container_views)

2
samples/vsphere/common/vim/inventory.py
View File

@ -35,7 +35,7 @@ def get_datastore_mo(stub_config, soap_stub,
# property collector query but it's a little more complicated
def get_datacenter_for_datastore(datastore_mo):
datacenter_mo = datastore_mo.parent
while datacenter_mo != None:
while datacenter_mo is not None:
if isinstance(datacenter_mo, vim.Datacenter):
return datacenter_mo
datacenter_mo = datacenter_mo.parent

2
samples/vsphere/common/vim/vmdk.py
View File

@ -33,12 +33,14 @@ def create_vmdk(service_instance, datacenter_mo, datastore_path):
format(datastore_path, datacenter_mo.name))
return task.info.result
def delete_vmdk(service_instance, datacenter_mo, datastore_path):
"""Delete vmdk from specific datastore"""
vdm = service_instance.content.virtualDiskManager
task = vdm.DeleteVirtualDisk(datastore_path, datacenter_mo)
pyVim.task.WaitForTask(task)
def detect_vmdk(stub_config, soap_stub, datacenter_name, datastore_name,
datastore_path):
"""Find vmdk in specific datastore"""

1
samples/vsphere/contentlibrary/crud/library_crud.py
View File

@ -110,4 +110,3 @@ def main():
if __name__ == '__main__':
main()

24
samples/vsphere/contentlibrary/isomount/iso_mount.py
View File

@ -22,6 +22,7 @@ from samples.vsphere.contentlibrary.lib.cls_api_helper import ClsApiHelper
from samples.vsphere.vcenter.helper.vm_helper import get_vm
class IsoMount(SampleBase):
"""
Demonstrates the content library ISO item mount and
@ -64,12 +65,16 @@ class IsoMount(SampleBase):
self.helper = ClsApiHelper(self.client, self.skip_verification)
def _execute(self):
storage_backings = self.helper.create_storage_backings(self.servicemanager, self.datastore_name)
storage_backings = self.helper.create_storage_backings(
self.servicemanager, self.datastore_name)
library_id = self.helper.create_local_library(storage_backings, self.lib_name)
library_id = self.helper.create_local_library(storage_backings,
self.lib_name)
self.local_library = self.client.local_library_service.get(library_id)
library_item_id = self.helper.create_iso_library_item(library_id,self.iso_item_name, self.ISO_FILENAME)
library_item_id = self.helper.create_iso_library_item(library_id,
self.iso_item_name,
self.ISO_FILENAME)
vm_id = get_vm(self.servicemanager.stub_config, self.vm_name)
assert vm_id is not None
@ -77,18 +82,17 @@ class IsoMount(SampleBase):
# Mount the iso item as a CDROM device
device_id = self.client.iso_service.mount(library_item_id, vm_id)
assert device_id is not None
print('Mounted library item {0}'
' on vm {1} at device {2}'.format(self.iso_item_name, self.vm_name, device_id))
print('Mounted library item {0} on vm {1} at device {2}'.
format(self.iso_item_name, self.vm_name, device_id))
# Unmount the CDROM
self.client.iso_service.unmount(vm_id, device_id)
print('Unmounted library item {0}'
'from vm {1} mounted at device {2}'.format(self.iso_item_name,
self.vm_name, device_id))
print('Unmounted library item {0} from vm {1} mounted at device {2}'.
format(self.iso_item_name, self.vm_name, device_id))
def _cleanup(self):
if self.local_library:
self.client.local_library_service.delete(library_id=self.local_library.id)
self.client.local_library_service.delete(
library_id=self.local_library.id)
print('Deleted Library Id: {0}'.format(self.local_library.id))

4
samples/vsphere/contentlibrary/lib/__init__.py
View File

@ -10,16 +10,16 @@
* NON-INFRINGEMENT AND FITNESS FOR A PARTICULAR PURPOSE.
"""
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
# Required to distribute different parts of this
# package as multiple distribution
try:
import pkg_resources
pkg_resources.declare_namespace(__name__)
except ImportError:
from pkgutil import extend_path
__path__ = extend_path(__path__, __name__) # @ReservedAssignment

3
samples/vsphere/contentlibrary/lib/cls_api_helper.py
View File

@ -126,8 +126,7 @@ class ClsApiHelper(object):
def get_iso_file_map(self, item_filename, disk_filename):
iso_files_map = {}
iso_file_path = os.path.abspath(os.path.join(os.path.dirname(os.path.realpath(__file__)),
self.ISO_FILE_RELATIVE_DIR
+ disk_filename))
self.ISO_FILE_RELATIVE_DIR + disk_filename))
iso_files_map[item_filename] = iso_file_path
return iso_files_map

16
samples/vsphere/contentlibrary/lib/cls_sync_helper.py
View File

@ -16,6 +16,7 @@ __vcenter_version__ = '6.0+'
import time
class ClsSyncHelper:
"""
Helper class to wait for the subscribed libraries and items to be
@ -54,7 +55,8 @@ class ClsSyncHelper:
"""
self.start_time = time.time()
is_synced = False
pub_item_id = self.client.library_item_service.get(sub_item_id).source_id
pub_item_id = self.client.library_item_service.get(
sub_item_id).source_id
pub_item = self.client.library_item_service.get(pub_item_id)
while self.not_timed_out():
@ -94,10 +96,11 @@ class ClsSyncHelper:
while self.not_timed_out():
# Get the subscribed library's updated information from server.
refreshed_sub_lib = self.client.subscribed_library_service.get(sub_lib.id)
if refreshed_sub_lib.last_sync_time != None:
if (sub_lib.last_sync_time == None or
refreshed_sub_lib.last_sync_time > sub_lib.last_sync_time):
refreshed_sub_lib = self.client.subscribed_library_service.get(
sub_lib.id)
if refreshed_sub_lib.last_sync_time is not None:
if (sub_lib.last_sync_time is None or
refreshed_sub_lib.last_sync_time > sub_lib.last_sync_time):
is_synced = True
break
time.sleep(self.wait_interval_sec)
@ -113,7 +116,8 @@ class ClsSyncHelper:
return False
synced_item_ids = []
for sub_item_id in sub_item_ids:
source_id = self.client.library_item_service.get(sub_item_id).source_id
source_id = self.client.library_item_service.get(
sub_item_id).source_id
if source_id not in synced_item_ids and source_id in pub_item_ids:
synced_item_ids.append(sub_item_id)

5
samples/vsphere/contentlibrary/ovfdeploy/deploy_ovf_template.py
View File

@ -33,7 +33,6 @@ from samples.vsphere.contentlibrary.lib.cls_api_client import ClsApiClient
from samples.vsphere.contentlibrary.lib.cls_api_helper import ClsApiHelper
class DeployOvfTemplate(SampleBase):
"""
Demonstrates the workflow to deploy an OVF library item to a resource pool.
@ -71,7 +70,7 @@ class DeployOvfTemplate(SampleBase):
self.client = ClsApiClient(self.servicemanager)
self.helper = ClsApiHelper(self.client, self.skip_verification)
# Default VM name
self.vm_name = 'vm-'+str(generate_random_uuid())
self.vm_name = 'vm-' + str(generate_random_uuid())
def _execute(self):
@ -123,7 +122,7 @@ class DeployOvfTemplate(SampleBase):
# The type and ID of the target deployment is available in the deployment result.
if result.succeeded:
print('Deployment successful. Result resource: {0}, ID: {1}'
.format(result.resource_id.type, result.resource_id.id))
.format(result.resource_id.type, result.resource_id.id))
self.vm_id = result.resource_id.id
error = result.error
if error is not None:

15
samples/vsphere/contentlibrary/publishsubscribe/library_publish_subscribe.py
View File

@ -56,6 +56,7 @@ class LibraryPublishSubscribe(SampleBase):
self.argparser.add_argument('-datastorename',
'--datastorename',
help='The name of the datastore.')
def _setup(self):
self.datastore_name = self.args.datastorename
assert self.datastore_name is not None
@ -65,8 +66,9 @@ class LibraryPublishSubscribe(SampleBase):
self.helper = ClsApiHelper(self.client, self.skip_verification)
def _execute(self):
storage_backings = self.helper.create_storage_backings(self.servicemanager,
self.datastore_name)
storage_backings = self.helper.create_storage_backings(
self.servicemanager,
self.datastore_name)
# Create a published library backed the VC datastore using vAPIs
self.pub_lib_id = self.create_published_library(storage_backings)
@ -78,7 +80,9 @@ class LibraryPublishSubscribe(SampleBase):
print('Publish URL : {0}'.format(pub_lib_url))
# Create a library item in the published library
pub_lib_item_id = self.helper.create_iso_library_item(self.pub_lib_id, 'item_1', self.DEMO_FILENAME)
pub_lib_item_id = self.helper.create_iso_library_item(self.pub_lib_id,
'item_1',
self.DEMO_FILENAME)
assert self.client.library_item_service.get(pub_lib_item_id) is not None
# Create the subscribed library
@ -97,7 +101,8 @@ class LibraryPublishSubscribe(SampleBase):
assert len(sub_item_ids) == 1, 'Subscribed library must have one item'
# Add another item to the published library
self.helper.create_iso_library_item(self.pub_lib_id, 'item_2', self.DEMO_FILENAME)
self.helper.create_iso_library_item(self.pub_lib_id, 'item_2',
self.DEMO_FILENAME)
# Manually synchronize the subscribed library to get the latest changes immediately.
self.client.subscribed_library_service.sync(self.sub_lib_id)
@ -110,7 +115,7 @@ class LibraryPublishSubscribe(SampleBase):
# List the subscribed items.
sub_item_ids = self.client.library_item_service.list(self.sub_lib_id)
assert len(sub_item_ids) == 2, 'Subscribed library must have two items'
for sub_item_id in sub_item_ids :
for sub_item_id in sub_item_ids:
sub_item = self.client.library_item_service.get(sub_item_id)
print('Subscribed item : {0}'.format(sub_item.name))

17
samples/vsphere/contentlibrary/vmcapture/vm_template_capture.py
View File

@ -29,7 +29,6 @@ from samples.vsphere.contentlibrary.lib.cls_api_helper import ClsApiHelper
from samples.vsphere.vcenter.helper.vm_helper import get_vm
class CaptureVMTemplateToContentLibrary(SampleBase):
"""
Demonstrates the workflow to capture a virtual machine into a content library.
@ -54,7 +53,7 @@ class CaptureVMTemplateToContentLibrary(SampleBase):
def _options(self):
self.argparser.add_argument('-datastorename', '--datastorename',
help='The name of the datastore for'
' content library backing (of type vmfs)')
' content library backing (of type vmfs)')
self.argparser.add_argument('-vmname', '--vmname',
help='Name of the VM to be captured')
@ -72,8 +71,9 @@ class CaptureVMTemplateToContentLibrary(SampleBase):
self.helper = ClsApiHelper(self.client, self.skip_verification)
def _execute(self):
storage_backings = self.helper.create_storage_backings(self.servicemanager,
self.datastore_name)
storage_backings = self.helper.create_storage_backings(
self.servicemanager,
self.datastore_name)
print('Creating Content Library')
# Create a content library
@ -94,7 +94,8 @@ class CaptureVMTemplateToContentLibrary(SampleBase):
(vm_id, self.library_item_id))
def capture_source_vm(self, vm_id, param):
source = LibraryItem.DeployableIdentity(self.deployable_resource_type, vm_id)
source = LibraryItem.DeployableIdentity(self.deployable_resource_type,
vm_id)
result = self.client.ovf_lib_item_service.create(source,
param["target"],
param["spec"],
@ -111,15 +112,17 @@ class CaptureVMTemplateToContentLibrary(SampleBase):
def _cleanup(self):
# delete the local library.
if self.content_library is not None:
self.client.local_library_service.delete(library_id=self.content_library.id)
self.client.local_library_service.delete(
library_id=self.content_library.id)
print('Deleted Library Id: {0}'.format
(self.content_library.id))
(self.content_library.id))
def main():
vm_template_capture = CaptureVMTemplateToContentLibrary()
vm_template_capture.main()
# Start program
if __name__ == '__main__':
main()

2
samples/vsphere/sso/__init__.py
View File

@ -17,7 +17,9 @@ __copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
# package as multiple distribution
try:
import pkg_resources
pkg_resources.declare_namespace(__name__)
except ImportError:
from pkgutil import extend_path
__path__ = extend_path(__path__, __name__) # @ReservedAssignment

1
samples/vsphere/tagging/tagging_workflow.py
View File

@ -205,6 +205,7 @@ def main():
tagging_workflow = TaggingWorkflow()
tagging_workflow.main()
# Start program
if __name__ == '__main__':
main()

3
samples/vsphere/vcenter/setup/datacenter.py
View File

@ -13,7 +13,6 @@
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
from com.vmware.vcenter_client import (Datacenter, Folder)
@ -64,7 +63,7 @@ def cleanup_datacenters(context):
datacenter_summaries = datacenter_svc.list(
Datacenter.FilterSpec(names=names))
print("Found {} Datacenters matching names {}".
format(len(datacenter_summaries),", ".
format(len(datacenter_summaries), ", ".
join(["'{}'".format(n) for n in names])))
for datacenter_summary in datacenter_summaries:

13
samples/vsphere/vcenter/setup/datastore.py
View File

@ -13,7 +13,6 @@
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
import pyVim.task
from com.vmware.vcenter_client import Host
from pyVmomi import vim
@ -39,12 +38,12 @@ def detect_nfs_datastore_on_host(context, host_name):
datastore = datastore_mo._moId
print("Detected NFS Volume '{}' as {} on Host '{}' ({})".
format(datastore_name, datastore, host_name, host))
context.testbed.entities['HOST_NFS_DATASTORE_IDS'][host_name]\
context.testbed.entities['HOST_NFS_DATASTORE_IDS'][host_name] \
= datastore
return True
print("NFS Volume '{}' missing on Host '{}'".
format(datastore_name,host_name))
format(datastore_name, host_name))
return False
@ -81,7 +80,7 @@ def cleanup_nfs_datastore(context):
datastore_system.RemoveDatastore(datastore_mo)
print("Removed NFS Volume '{}' ({}) from Host '{}' ({})".
format(datastore_name, datastore_mo._moId,
host_mo.name,host_mo._moId))
host_mo.name, host_mo._moId))
# Remote NFS Datastore at the vCenter level
# TODO Do we need to do this?
@ -133,7 +132,7 @@ def setup_nfs_datastore_on_host(context, host_name):
if info.name == local_path:
print("Found NFS Volume '{}' ({}) on Host '{}' ({})".
format(local_path, datastore_mo._moId,
host_name,host_mo._moId))
host_name, host_mo._moId))
return datastore_mo._moId
else:
print("Found NFS remote host '{}' and path '{}' on Host '{}' ({}) as '{}'".
@ -168,7 +167,7 @@ def detect_vmfs_datastore(context, host_name, datastore_name):
datastore = datastore_mo._moId
print("Detected VMFS Volume '{}' as {} on Host '{}' ({})".
format(datastore_name, datastore, host_name, host))
context.testbed.entities['HOST_VMFS_DATASTORE_IDS'][host_name]\
context.testbed.entities['HOST_VMFS_DATASTORE_IDS'][host_name] \
= datastore
return True
@ -216,7 +215,7 @@ def setup_vmfs_datastore(context, host_name, datastore_name):
datastore = vmfs_datastores[datastore_name]._moId
print("Detected VMFS Volume '{}' as {} on Host '{}' ({})".
format(datastore_name, datastore, host_name, host))
context.testbed.entities['HOST_VMFS_DATASTORE_IDS'][host_name]\
context.testbed.entities['HOST_VMFS_DATASTORE_IDS'][host_name] \
= datastore
return True

2
samples/vsphere/vcenter/setup/floppy_image.py
View File

@ -15,7 +15,7 @@ __copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
from samples.vsphere.common.vim.file import (detect_file, delete_file,
parse_datastore_path)
parse_datastore_path)
from samples.vsphere.common.vim.inventory import get_datastore_mo
from samples.vsphere.common.vim import datastore_file

2
samples/vsphere/vcenter/setup/iso_image.py
View File

@ -15,7 +15,7 @@ __copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
from samples.vsphere.common.vim.file import (detect_file, delete_file,
parse_datastore_path)
parse_datastore_path)
from samples.vsphere.common.vim.inventory import get_datastore_mo
from samples.vsphere.common.vim import datastore_file

1
samples/vsphere/vcenter/setup/main.py
View File

@ -109,4 +109,3 @@ if context.option['DO_SAMPLES_CLEANUP']:
if context.option['DO_TESTBED_CLEANUP']:
sample_cleanup(context)
testbed_cleanup(context)

18
samples/vsphere/vcenter/setup/network.py
View File

@ -13,7 +13,6 @@
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
import pyVim.task
from com.vmware.vcenter_client import Host
from pyVmomi import vim
@ -80,7 +79,7 @@ def cleanup_vdswitch(context):
def find_vdswitch(context, datacenter_name, vdswitch_name):
""" Retrieve an existing Distributed Switch"""
# TODO Ugly deep nesting.
for datacenter_mo in context.service_instance.content.rootFolder\
for datacenter_mo in context.service_instance.content.rootFolder \
.childEntity:
if (isinstance(datacenter_mo, vim.Datacenter) and
datacenter_mo.name == datacenter_name):
@ -102,9 +101,10 @@ def find_vdportgroup(context, datacenter_name, vdswitch_name, vdportgroup_name):
vdswitch_mo = find_vdswitch(context, datacenter_name, vdswitch_name)
for vdportgroup_mo in vdswitch_mo.portgroup:
if vdportgroup_mo.name == vdportgroup_name:
print("Found Distributed Portgroup '{}' ({}) on Distributed Switch '{}' ({})".
print(
"Found Distributed Portgroup '{}' ({}) on Distributed Switch '{}' ({})".
format(vdportgroup_name, vdportgroup_mo._moId,
vdswitch_name,vdswitch_mo._moId))
vdswitch_name, vdswitch_mo._moId))
return vdportgroup_mo
return None
@ -146,8 +146,9 @@ def create_vdportgroup(context, vdswitch_name, vdportgroup_name):
for vdportgroup_mo in vdswitch_mo.portgroup:
if vdportgroup_mo.name == vdportgroup_name:
vdportgroup = vdportgroup_mo._moId
print("Created Distributed Portgroup '{}' ({}) on Distributed Switch '{}' ({})".
format(vdportgroup_name, vdportgroup, vdswitch_name, vdswitch))
print(
"Created Distributed Portgroup '{}' ({}) on Distributed Switch '{}' ({})".
format(vdportgroup_name, vdportgroup, vdswitch_name, vdswitch))
return vdportgroup
@ -245,8 +246,9 @@ def detect_stdportgroup(context, host_name, network_name):
if (type(network_mo) == vim.Network and
network_mo.name == network_name):
network = network_mo._moId
print("Detected Standard Portgroup '{}' as {} on Host '{}' ({})".
format(network_name, network, host_name, host))
print(
"Detected Standard Portgroup '{}' as {} on Host '{}' ({})".
format(network_name, network, host_name, host))
context.testbed.entities['HOST_STANDARD_SWITCH_IDS'][
host_name] = network
return True

4
samples/vsphere/vcenter/setup/testbed.py
View File

@ -78,13 +78,14 @@ config["SERIAL_PORT_NETWORK_PROXY"] = None
config["PARALLEL_PORT_DATACENTER_NAME"] = config["VM_DATACENTER_NAME"]
config["PARALLEL_PORT_DATASTORE_ROOT_PATH"] = config["BACKENDS_DATASTORE_ROOT_PATH"] + "/parallel"
config["PARALLEL_PORT_DATASTORE_PATH"] = config["PARALLEL_PORT_DATASTORE_ROOT_PATH"] + "/parallel.log"
config["PARALLEL_PORT_DATASTORE_PATH"] = config["PARALLEL_PORT_DATASTORE_ROOT_PATH"] + "/parallel.log"
config["FLOPPY_SRC_URL"] = "http://www.ibiblio.org/pub/micro/pc-stuff/freedos/files/distributions/1.0/fdboot.img"
config["FLOPPY_DATACENTER_NAME"] = config["VM_DATACENTER_NAME"]
config["FLOPPY_DATASTORE_ROOT_PATH"] = config["BACKENDS_DATASTORE_ROOT_PATH"] + "/floppy"
config["FLOPPY_DATASTORE_PATH"] = config["FLOPPY_DATASTORE_ROOT_PATH"] + "/fdboot.img"
class Testbed(object):
def __init__(self):
self.config = {}
@ -134,5 +135,6 @@ class Testbed(object):
_testbed = Testbed()
_testbed.config.update(config)
def get():
return _testbed

2
samples/vsphere/vcenter/vm/hardware/adapter/sata.py
View File

@ -39,6 +39,7 @@ cleardata = False
satas_to_delete = []
orig_sata_summaries = None
def setup(context=None):
global vm, vm_name, stub_config, cleardata
if context:
@ -54,6 +55,7 @@ def setup(context=None):
password,
skip_verification)
def run():
global vm
vm = get_vm(stub_config, vm_name)

3
samples/vsphere/vcenter/vm/hardware/adapter/scsi.py
View File

@ -39,6 +39,7 @@ cleardata = False
scsis_to_delete = []
orig_scsi_summaries = None
def setup(context=None):
global vm, vm_name, stub_config, cleardata
if context:
@ -54,6 +55,7 @@ def setup(context=None):
password,
skip_verification)
def run():
global vm
vm = get_vm(stub_config, vm_name)
@ -132,6 +134,7 @@ def cleanup():
print('vm.hardware.adapter.Scsi WARNING: '
'Final SCSI adapters info does not match original')
def main():
try:
setup()

2
samples/vsphere/vcenter/vm/hardware/disk.py
View File

@ -262,7 +262,7 @@ def run():
print('\n# Example: Attach an existing VMDK using the default bus and unit')
datastore_path = datastore_root_path + '/attach-defaults.vmdk'
delete_vmdk_if_exist(stub_config, service_instance._stub, datacenter_name,
datastore_name, datastore_path)
datastore_name, datastore_path)
create_vmdk(service_instance, datacenter_mo, datastore_path)
disk_create_spec = Disk.CreateSpec(
backing=Disk.BackingSpec(type=Disk.BackingType.VMDK_FILE,

4
samples/vsphere/vcenter/vm/hardware/floppy.py
View File

@ -39,6 +39,7 @@ stub_config = None
cleardata = False
orig_floppy_summaries = None
def setup(context=None):
global vm, vm_name, stub_config, cleardata
if context:
@ -54,6 +55,7 @@ def setup(context=None):
password,
skip_verification)
def run():
# * Floppy images must be pre-existing. This API does not expose
# a way to create new floppy images.
@ -114,7 +116,7 @@ def run():
print('\n# Example: Create Floppy with IMAGE_FILE backing, '
'start_connected=True,')
print( ' allow_guest_control=True')
print(' allow_guest_control=True')
floppy_create_spec = Floppy.CreateSpec(
allow_guest_control=True,
start_connected=True,

2
samples/vsphere/vcenter/vm/hardware/memory.py
View File

@ -54,6 +54,7 @@ def setup(context=None):
password,
skip_verification)
def run():
global vm
vm = get_vm(stub_config, vm_name)
@ -120,5 +121,6 @@ def main():
if stub_config:
vapiconnect.logout(stub_config)
if __name__ == '__main__':
main()

6
samples/vsphere/vcenter/vm/main.py
View File

@ -55,7 +55,6 @@ def validate(context):
def run(context):
# Clean up in case of past failures
cleanup(context)
@ -80,14 +79,14 @@ def run(context):
# 1. STANDARD_PORTGROUP on DATACENTER2
# 2. DISTRIBUTED_PORTGROUP on DATACENTER2
###########################################################################
standard_network = samples.vsphere.vcenter.helper\
standard_network = samples.vsphere.vcenter.helper \
.network_helper.get_standard_network_backing(
context.stub_config,
context.testbed.config['STDPORTGROUP_NAME'],
context.testbed.config['VM_DATACENTER_NAME'])
print('standard_network={}'.format(standard_network))
distributed_network = samples.vsphere.vcenter.helper\
distributed_network = samples.vsphere.vcenter.helper \
.network_helper.get_distributed_network_backing(
context.stub_config,
context.testbed.config['VDPORTGROUP1_NAME'],
@ -137,7 +136,6 @@ def run(context):
if context.option['DO_SAMPLES_CLEANUP']:
samples.vsphere.vcenter.vm.hardware.main.cleanup()
# Sample cleanup
if context.option['DO_SAMPLES_CLEANUP']:
cleanup(context)

3
samples/vsphere/vcenter/vm/placement.py
View File

@ -17,10 +17,9 @@ __vcenter_version__ = '6.5+'
from com.vmware.vcenter_client import (Cluster, Datastore, Folder, ResourcePool,
VM)
from samples.vsphere.vcenter.helper import vm_placement_helper
#####################################################################
# Placement samples: How to get a valid PlacementSpec to create a VM
#####################################################################

3
setup.cfg
View File

@ -1,3 +1,2 @@
[pycodestyle]
ignore = E402
max-line-length = 120
ignore = E402, E501, E122, E126, E127, E128, E129, E131