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

Merge pull request #30 from tianhao64/master

Browse Source 
Improve sample code
This commit is contained in:
Tianhao He 2017-08-18 04:49:14 -07:00 committed by GitHub
commit 8503fcd9a0
13 changed files with 731 additions and 695 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
sample_template/sample_template.py
View File

@ -19,11 +19,12 @@ __vcenter_version__ = 'TODO: <compatible vcenter versions>'
import atexit
from com.vmware.vcenter_client import VM
from samples.vsphere.common.sample_util import parse_cli_args
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common.service_manager import ServiceManager
class Sample:
class Sample(object):
"""
TODO: Sample description and prerequisites.
e.g. Demonstrates getting list of VMs present in vCenter
@ -33,40 +34,44 @@ class Sample:
"""
def __init__(self):
self.vm_service = None # Service used by the sample code.
self.stub_config = None
self.si = None
self.service_manager = None
self.vm_name = None
self.cleardata = None
def setup(self):
server, username, password, cleardata, skip_verification = \
parse_cli_args()
self.cleardata = cleardata
# Create argument parser for standard inputs:
# server, username, password, cleanup and skipverification
parser = sample_cli.build_arg_parser()
# Add your custom input arguments
parser.add_argument('-n', '--vm_name',
action='store',
default='Sample_Default_VM_for_Simple_Testbed',
help='Name of the testing vm')
args = sample_util.process_cli_args(parser.parse_args())
self.vm_name = args.vm_name
self.cleardata = args.cleardata
# Connect to both Vim and vAPI services
service_manager = ServiceManager(server,
username,
password,
skip_verification)
service_manager.connect()
atexit.register(service_manager.disconnect)
# Get the vAPI stub
self.stub_config = service_manager.stub_config
self.vm_service = VM(self.stub_config)
# Get VIM service instance (pyVmomi)
self.si = service_manager.si
self.service_manager = ServiceManager(args.server,
args.username,
args.password,
args.skipverification)
self.service_manager.connect()
atexit.register(self.service_manager.disconnect)
def run(self):
# TODO add steps to demo your API
# TODO add your sample code here
# Using vAPI services
vms = self.vm_service.list()
# Using REST API service
vm_service = VM(self.service_manager.stub_config)
filter_spec = VM.FilterSpec(names=set([self.vm_name]))
vms = vm_service.list(filter_spec)
print(vms)
# Using vim services
current_time = self.si.CurrentTime()
# Using Vim API service (pyVmomi)
current_time = self.service_manager.si.CurrentTime()
print(current_time)
def cleanup(self):

4
samples/vsphere/common/sample_cli.py
View File

@ -25,7 +25,7 @@ def build_arg_parser():
-u username
-p password
-c cleanup
-v skip_verification
-v skipverification
"""
parser = argparse.ArgumentParser(
@ -43,7 +43,7 @@ def build_arg_parser():
action='store',
help='Password to use when connecting to vc')
parser.add_argument('-c', '--cleanup',
parser.add_argument('-c', '--cleardata',
action='store_true',
help='Clean up after sample run. ')

59
samples/vsphere/common/sample_util.py
View File

@ -28,18 +28,6 @@ def pp(value):
return output.getvalue()
def parse_cli_args():
"""
Parse the server IP and credential used by samples.
Use values from command line arguments if present, otherwise use values
from testbed.py
"""
# parse command line
parser = sample_cli.build_arg_parser()
args = parser.parse_args()
return process_cli_args(args)
def parse_cli_args_vm(vm_name):
"""
Parse the server IP, credential and vm name used by vcenter vm samples.
@ -51,10 +39,7 @@ def parse_cli_args_vm(vm_name):
parser.add_argument('-n', '--vm_name',
action='store',
help='Name of the testing vm')
args = parser.parse_args()
server, username, password, cleardata, skip_verification = \
process_cli_args(args)
args = process_cli_args(parser.parse_args())
if args.vm_name:
vm_name = args.vm_name
@ -64,45 +49,37 @@ def parse_cli_args_vm(vm_name):
raise Exception("vm name is required")
print("vm name = {}".format(vm_name))
return server, username, password, cleardata, skip_verification, vm_name
return args.server, args.username, args.password, args.cleardata, \
args.skipverification, vm_name
def process_cli_args(args):
"""
Process server IP and credential args.
Verify if required inputs (server, username and password) are provided.
If they are not passed through cmd arguments, we will try to get them from
testbed.py. If they are not configured in testbed.py either, we will raise
an exception to remind the user to provide them.
"""
if args.server:
server = args.server
else:
if not args.server:
print("Using vcenter server specified in testbed.py")
server = testbed.config['SERVER']
if not server:
args.server = testbed.config['SERVER']
if not args.server:
raise Exception("vcenter server is required")
print("vcenter server = {}".format(server))
print("vcenter server = {}".format(args.server))
if args.username:
username = args.username
else:
if not args.username:
print("Using vc user specified in testbed.py")
username = testbed.config['USERNAME']
if not username:
args.username = testbed.config['USERNAME']
if not args.username:
raise Exception("vc username is required")
print("vc username = {}".format(username))
print("vc username = {}".format(args.username))
if args.password:
password = args.password
else:
if not args.password:
print("Using vc password specified in testbed.py")
password = testbed.config['PASSWORD']
args.password = testbed.config['PASSWORD']
cleardata = args.cleanup
print("sample cleanup = {}".format(cleardata))
skip_verification = args.skipverification
print("skip server cert verification = {}".format(skip_verification))
return server, username, password, cleardata, skip_verification
return args
class Context(object):

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

@ -22,19 +22,23 @@ try:
except ImportError:
import urllib.request as urllib2
import atexit
from com.vmware.content.library_client import Item
from com.vmware.vcenter.ovf_client import LibraryItem
from pyVmomi import vim
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common.id_generator import generate_random_uuid
from samples.vsphere.common.sample_base import SampleBase
from samples.vsphere.common.service_manager import ServiceManager
from samples.vsphere.common.vim.helpers.vim_utils import (
get_obj, get_obj_by_moId, poweron_vm, poweroff_vm, delete_object)
from samples.vsphere.contentlibrary.lib.cls_api_client import ClsApiClient
from samples.vsphere.contentlibrary.lib.cls_api_helper import ClsApiHelper
class DeployOvfTemplate(SampleBase):
class DeployOvfTemplate:
"""
Demonstrates the workflow to deploy an OVF library item to a resource pool.
Note: the sample needs an existing library item with an OVF template
@ -42,7 +46,6 @@ class DeployOvfTemplate(SampleBase):
"""
def __init__(self):
SampleBase.__init__(self, self.__doc__)
self.servicemanager = None
self.client = None
self.helper = None
@ -51,29 +54,37 @@ class DeployOvfTemplate(SampleBase):
self.vm_obj = None
self.vm_name = None
def _options(self):
self.argparser.add_argument('-clustername',
def setup(self):
parser = sample_cli.build_arg_parser()
parser.add_argument('-n', '--vm_name',
action='store',
help='Name of the testing vm')
parser.add_argument('-clustername',
'--clustername',
help='The name of the cluster to be used.')
self.argparser.add_argument('-libitemname',
parser.add_argument('-libitemname',
'--libitemname',
help='The name of the library item to deploy.'
'The library item should contain an OVF package.')
args = sample_util.process_cli_args(parser.parse_args())
self.lib_item_name = args.libitemname
self.cluster_name = args.clustername
self.vm_name = args.vm_name
def _setup(self):
self.cluster_name = self.args.clustername
assert self.cluster_name is not None
self.servicemanager = ServiceManager(args.server,
args.username,
args.password,
args.skipverification)
self.servicemanager.connect()
atexit.register(self.servicemanager.disconnect)
self.lib_item_name = self.args.libitemname
assert self.lib_item_name is not None
self.servicemanager = self.get_service_manager()
self.client = ClsApiClient(self.servicemanager)
self.helper = ClsApiHelper(self.client, self.skip_verification)
self.helper = ClsApiHelper(self.client, args.skipverification)
# Default VM name
self.vm_name = 'vm-' + str(generate_random_uuid())
def _execute(self):
def execute(self):
# Find the cluster's resource pool moid
cluster_obj = get_obj(self.servicemanager.content,
@ -141,7 +152,7 @@ class DeployOvfTemplate(SampleBase):
for error in result.error.errors:
print('OVF error: {}'.format(error.message))
def _cleanup(self):
def cleanup(self):
if self.vm_obj is not None:
# Power off the VM and wait for the power off operation to complete
poweroff_vm(self.servicemanager.content, self.vm_obj)
@ -151,7 +162,9 @@ class DeployOvfTemplate(SampleBase):
def main():
deploy_ovf_sample = DeployOvfTemplate()
deploy_ovf_sample.main()
deploy_ovf_sample.setup()
deploy_ovf_sample.execute()
deploy_ovf_sample.cleanup()
if __name__ == '__main__':

30
samples/vsphere/sso/embedded_psc_sso_workflow.py
View File

@ -17,20 +17,19 @@ __author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2017 VMware, Inc. All rights reserved.'
__vcenter_version__ = '6.0+'
from pprint import pprint
import requests
from com.vmware.cis.tagging_client import (Category, CategoryModel)
from com.vmware.cis_client import Session
from vmware.vapi.lib.connect import get_requests_connector
from vmware.vapi.security.session import create_session_security_context
from vmware.vapi.security.sso import create_saml_bearer_security_context
from vmware.vapi.stdlib.client.factories import StubConfigurationFactory
from com.vmware.cis.tagging_client import (Category, CategoryModel)
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common import sso
from samples.vsphere.common.ssl_helper import get_unverified_context
from samples.vsphere.common.vapiconnect import create_unverified_session
from samples.vsphere.common.sample_util import parse_cli_args
from samples.vsphere.common import sso
class EmbeddedPscSsoWorkflow(object):
@ -40,18 +39,15 @@ class EmbeddedPscSsoWorkflow(object):
"""
def __init__(self):
self.server = None
self.username = None
self.password = None
self.args = None
self.session = None
self.session_id = None
self.skip_verification = False
self.category_svc = None
self.category_id = None
def setup(self):
self.server, self.username, self.password, _, self.skip_verification = \
parse_cli_args()
parser = sample_cli.build_arg_parser()
self.args = sample_util.process_cli_args(parser.parse_args())
def run(self):
print('\n\n#### Example: Login to vCenter server with '
@ -59,18 +55,18 @@ class EmbeddedPscSsoWorkflow(object):
# Since the platform services controller is embedded, the sso server
# is the same as the vCenter server.
ssoUrl = 'https://{}/sts/STSService'.format(self.server)
ssoUrl = 'https://{}/sts/STSService'.format(self.args.server)
print('\nStep 1: Connect to the Single Sign-On URL and '
'retrieve the SAML bearer token.')
authenticator = sso.SsoAuthenticator(ssoUrl)
context = None
if self.skip_verification:
if self.args.skipverification:
context = get_unverified_context()
bearer_token = authenticator.get_bearer_saml_assertion(
self.username,
self.password,
self.args.username,
self.args.password,
delegatable=True,
ssl_context=context)
@ -81,12 +77,12 @@ class EmbeddedPscSsoWorkflow(object):
# The URL for the stub requests are made against the /api HTTP endpoint
# of the vCenter system.
vapi_url = 'https://{}/api'.format(self.server)
vapi_url = 'https://{}/api'.format(self.args.server)
# Create an authenticated stub configuration object that can be used to
# issue requests against vCenter.
session = requests.Session()
if self.skip_verification:
if self.args.skipverification:
session = create_unverified_session(session)
connector = get_requests_connector(session=session, url=vapi_url)
connector.set_security_context(sec_ctx)

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

@ -30,8 +30,7 @@ from samples.vsphere.vcenter.setup.setup_cli import build_arg_parser
from samples.vsphere.vcenter.setup.testbed_setup import cleanup as testbed_cleanup
from samples.vsphere.vcenter.setup.testbed_setup import setup as testbed_setup
from samples.vsphere.vcenter.setup.testbed_setup import validate as testbed_validate
from samples.vsphere.vcenter.vm.main import cleanup as sample_cleanup
from samples.vsphere.vcenter.vm.main import run as sample_run
from samples.vsphere.vcenter.vm.main import VMSetup
from samples.vsphere.common.ssl_helper import get_unverified_context
@ -91,10 +90,12 @@ print(context.to_option_string())
# Testbed Setup
###############################################################################
vm_setup = VMSetup(context)
# Setup testbed
if context.option['DO_TESTBED_SETUP']:
# Clean up in case of past failures
sample_cleanup(context)
vm_setup.cleanup()
testbed_cleanup(context)
testbed_setup(context)
@ -113,11 +114,12 @@ if (context.option['DO_TESTBED_SETUP'] or
# Run Sample
if context.option['DO_SAMPLES']:
sample_run(context)
vm_setup.setup(context)
vm_setup.run()
# Cleanup after sample run
if context.option['DO_SAMPLES_CLEANUP']:
sample_cleanup(context)
vm_setup.cleanup()
###############################################################################
# Testbed Cleanup
@ -125,5 +127,5 @@ if context.option['DO_SAMPLES_CLEANUP']:
# Teardown testbed.
if context.option['DO_TESTBED_CLEANUP']:
sample_cleanup(context)
vm_setup.cleanup()
testbed_cleanup(context)

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

@ -15,9 +15,9 @@ __author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2016 VMware, Inc. All rights reserved.'
config = {}
config["SERVER"] = ""
config["SERVER"] = "10.192.248.66"
config["USERNAME"] = "administrator@vsphere.local"
config["PASSWORD"] = "VMware1!"
config["PASSWORD"] = "Admin!23"
config["ESX_HOST1"] = ""
config["ESX_HOST2"] = ""

99
samples/vsphere/vcenter/vm/create/create_basic_vm.py
View File

@ -25,15 +25,18 @@ from com.vmware.vcenter.vm.hardware_client import (
from com.vmware.vcenter.vm.hardware_client import ScsiAddressSpec
from com.vmware.vcenter.vm_client import (Power)
from com.vmware.vcenter_client import VM
from samples.vsphere.common import vapiconnect
from samples.vsphere.common.sample_util import parse_cli_args_vm
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common.sample_util import pp
from samples.vsphere.common.service_manager import ServiceManager
from samples.vsphere.vcenter.helper import network_helper
from samples.vsphere.vcenter.helper import vm_placement_helper
from samples.vsphere.vcenter.helper.vm_helper import get_vm
from samples.vsphere.vcenter.setup import testbed
from samples.vsphere.vcenter.helper.vm_helper import get_vm
class CreateBasicVM(object):
"""
Demonstrates how to create a basic VM with following configuration:
2 disks, 1 nic
@ -45,50 +48,53 @@ Sample Prerequisites:
- standard switch network
"""
stub_config = None
cleardata = False
vm_name = None
def __init__(self, stub_config=None, placement_spec=None):
self.context = None
self.service_manager = None
self.stub_config = stub_config
self.placement_spec = placement_spec
self.vm_name = testbed.config['VM_NAME_BASIC']
self.cleardata = None
def setup(self):
parser = sample_cli.build_arg_parser()
parser.add_argument('-n', '--vm_name',
action='store',
help='Name of the testing vm')
args = sample_util.process_cli_args(parser.parse_args())
if args.vm_name:
self.vm_name = args.vm_name
self.cleardata = args.cleardata
def setup(context=None):
global stub_config, cleardata, vm_name
if context:
# Run sample suite via setup script
vm_name = testbed.config['VM_NAME_BASIC']
stub_config = context.stub_config
else:
# Run sample in standalone mode
server, username, password, cleardata, skip_verification, vm_name = \
parse_cli_args_vm(testbed.config['VM_NAME_BASIC'])
stub_config = vapiconnect.connect(server,
username,
password,
skip_verification)
atexit.register(vapiconnect.logout, stub_config)
self.service_manager = ServiceManager(args.server,
args.username,
args.password,
args.skipverification)
self.service_manager.connect()
atexit.register(self.service_manager.disconnect)
self.stub_config = self.service_manager.stub_config
def run():
def run(self):
# Get a placement spec
datacenter_name = testbed.config['VM_DATACENTER_NAME']
vm_folder_name = testbed.config['VM_FOLDER2_NAME']
datastore_name = testbed.config['VM_DATASTORE_NAME']
std_portgroup_name = testbed.config['STDPORTGROUP_NAME']
placement_spec = vm_placement_helper.get_placement_spec_for_resource_pool(
stub_config,
if not self.placement_spec:
self.placement_spec = vm_placement_helper.get_placement_spec_for_resource_pool(
self.stub_config,
datacenter_name,
vm_folder_name,
datastore_name)
# Get a standard network backing
standard_network = network_helper.get_standard_network_backing(
stub_config,
self.stub_config,
std_portgroup_name,
datacenter_name)
create_basic_vm(stub_config, placement_spec, standard_network)
def create_basic_vm(stub_config, placement_spec, standard_network):
"""
Create a basic VM.
@ -115,14 +121,13 @@ def create_basic_vm(stub_config, placement_spec, standard_network):
type=Ethernet.BackingType.STANDARD_PORTGROUP,
network=standard_network))
# TODO Should DISK be put before ETHERNET? Does the BIOS automatically try
# the next device if the DISK is empty?
boot_device_order = [BootDevice.EntryCreateSpec(BootDevice.Type.ETHERNET),
boot_device_order = [
BootDevice.EntryCreateSpec(BootDevice.Type.ETHERNET),
BootDevice.EntryCreateSpec(BootDevice.Type.DISK)]
vm_create_spec = VM.CreateSpec(name=vm_name,
vm_create_spec = VM.CreateSpec(name=self.vm_name,
guest_os=guest_os,
placement=placement_spec,
placement=self.placement_spec,
disks=[boot_disk, data_disk],
nics=[nic],
boot_devices=boot_device_order)
@ -130,38 +135,38 @@ def create_basic_vm(stub_config, placement_spec, standard_network):
print(pp(vm_create_spec))
print('-----')
vm_svc = VM(stub_config)
vm_svc = VM(self.stub_config)
vm = vm_svc.create(vm_create_spec)
print("create_basic_vm: Created VM '{}' ({})".format(vm_name, vm))
print("create_basic_vm: Created VM '{}' ({})".format(self.vm_name, vm))
vm_info = vm_svc.get(vm)
print('vm.get({}) -> {}'.format(vm, pp(vm_info)))
return vm
def cleanup():
vm = get_vm(stub_config, vm_name)
def cleanup(self):
vm = get_vm(self.stub_config, self.vm_name)
if vm:
power_svc = Power(stub_config)
vm_svc = VM(stub_config)
power_svc = Power(self.stub_config)
vm_svc = VM(self.stub_config)
state = power_svc.get(vm)
if state == Power.Info(state=Power.State.POWERED_ON):
power_svc.stop(vm)
elif state == Power.Info(state=Power.State.SUSPENDED):
power_svc.start(vm)
power_svc.stop(vm)
print("Deleting VM '{}' ({})".format(vm_name, vm))
print("Deleting VM '{}' ({})".format(self.vm_name, vm))
vm_svc.delete(vm)
def main():
setup()
cleanup()
run()
if cleardata:
cleanup()
create_basic_vm = CreateBasicVM()
create_basic_vm.setup()
create_basic_vm.cleanup()
create_basic_vm.run()
if create_basic_vm.cleardata:
create_basic_vm.cleanup()
if __name__ == '__main__':

95
samples/vsphere/vcenter/vm/create/create_default_vm.py
View File

@ -19,16 +19,19 @@ __vcenter_version__ = '6.5+'
import atexit
from com.vmware.vcenter.vm_client import Power
from com.vmware.vcenter.vm_client import (Power)
from com.vmware.vcenter_client import VM
from samples.vsphere.common import vapiconnect
from samples.vsphere.common.sample_util import parse_cli_args_vm
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common.sample_util import pp
from samples.vsphere.common.service_manager import ServiceManager
from samples.vsphere.vcenter.helper import vm_placement_helper
from samples.vsphere.vcenter.helper.vm_helper import get_vm
from samples.vsphere.vcenter.setup import testbed
from samples.vsphere.vcenter.helper.vm_helper import get_vm
class CreateDefaultVM(object):
"""
Demonstrates how to create a VM with system provided defaults
@ -38,44 +41,46 @@ Sample Prerequisites:
- datastore
"""
stub_config = None
cleardata = False
vm_name = None
def __init__(self, stub_config=None, placement_spec=None):
self.context = None
self.service_manager = None
self.stub_config = stub_config
self.placement_spec = placement_spec
self.vm_name = testbed.config['VM_NAME_DEFAULT']
self.cleardata = None
def setup(self):
parser = sample_cli.build_arg_parser()
parser.add_argument('-n', '--vm_name',
action='store',
help='Name of the testing vm')
args = sample_util.process_cli_args(parser.parse_args())
if args.vm_name:
self.vm_name = args.vm_name
self.cleardata = args.cleardata
def setup(context=None):
global stub_config, cleardata, vm_name
if context:
# Run sample suite via setup script
vm_name = testbed.config['VM_NAME_DEFAULT']
stub_config = context.stub_config
else:
# Run sample in standalone mode
server, username, password, cleardata, skip_verification, vm_name = \
parse_cli_args_vm(testbed.config['VM_NAME_DEFAULT'])
stub_config = vapiconnect.connect(server,
username,
password,
skip_verification)
atexit.register(vapiconnect.logout, stub_config)
self.service_manager = ServiceManager(args.server,
args.username,
args.password,
args.skipverification)
self.service_manager.connect()
atexit.register(self.service_manager.disconnect)
self.stub_config = self.service_manager.stub_config
def run():
def run(self):
# Get a placement spec
datacenter_name = testbed.config['VM_DATACENTER_NAME']
vm_folder_name = testbed.config['VM_FOLDER2_NAME']
datastore_name = testbed.config['VM_DATASTORE_NAME']
placement_spec = vm_placement_helper.get_placement_spec_for_resource_pool(
stub_config,
if not self.placement_spec:
self.placement_spec = vm_placement_helper.get_placement_spec_for_resource_pool(
self.stub_config,
datacenter_name,
vm_folder_name,
datastore_name)
# Create the VMs
create_default_vm(stub_config, placement_spec)
def create_default_vm(stub_config, placement_spec):
"""
Create a default VM.
@ -83,43 +88,43 @@ def create_default_vm(stub_config, placement_spec):
and provided name. Use all the guest and system provided defaults.
"""
guest_os = testbed.config['VM_GUESTOS']
vm_create_spec = VM.CreateSpec(name=vm_name,
vm_create_spec = VM.CreateSpec(name=self.vm_name,
guest_os=guest_os,
placement=placement_spec)
placement=self.placement_spec)
print('\n# Example: create_default_vm: Creating a VM using spec\n-----')
print(pp(vm_create_spec))
print('-----')
vm_svc = VM(stub_config)
vm_svc = VM(self.stub_config)
vm = vm_svc.create(vm_create_spec)
print("create_default_vm: Created VM '{}' ({})".format(vm_name, vm))
print("create_default_vm: Created VM '{}' ({})".format(self.vm_name, vm))
vm_info = vm_svc.get(vm)
print('vm.get({}) -> {}'.format(vm, pp(vm_info)))
return vm
def cleanup():
vm = get_vm(stub_config, vm_name)
def cleanup(self):
vm = get_vm(self.stub_config, self.vm_name)
if vm:
power_svc = Power(stub_config)
vm_svc = VM(stub_config)
power_svc = Power(self.stub_config)
vm_svc = VM(self.stub_config)
state = power_svc.get(vm)
if state == Power.Info(state=Power.State.POWERED_ON):
power_svc.stop(vm)
elif state == Power.Info(state=Power.State.SUSPENDED):
power_svc.start(vm)
power_svc.stop(vm)
print("Deleting VM '{}' ({})".format(vm_name, vm))
print("Deleting VM '{}' ({})".format(self.vm_name, vm))
vm_svc.delete(vm)
def main():
setup()
cleanup()
run()
if cleardata:
cleanup()
create_default_vm = CreateDefaultVM()
create_default_vm.setup()
create_default_vm.cleanup()
create_default_vm.run()
if create_default_vm.cleardata:
create_default_vm.cleanup()
if __name__ == '__main__':

115
samples/vsphere/vcenter/vm/create/create_exhaustive_vm.py
View File

@ -25,15 +25,18 @@ from com.vmware.vcenter.vm.hardware_client import (
from com.vmware.vcenter.vm.hardware_client import ScsiAddressSpec
from com.vmware.vcenter.vm_client import (Hardware, Power)
from com.vmware.vcenter_client import VM
from samples.vsphere.common import vapiconnect
from samples.vsphere.common.sample_util import parse_cli_args_vm
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common.sample_util import pp
from samples.vsphere.common.service_manager import ServiceManager
from samples.vsphere.vcenter.helper import network_helper
from samples.vsphere.vcenter.helper import vm_placement_helper
from samples.vsphere.vcenter.helper.vm_helper import get_vm
from samples.vsphere.vcenter.setup import testbed
from samples.vsphere.vcenter.helper.vm_helper import get_vm
class CreateExhaustiveVM(object):
"""
Demonstrates how to create a exhaustive VM with the below configuration:
3 disks, 2 nics, 2 vcpu, 2 GB, memory, boot=BIOS, 1 cdrom, 1 serial port,
@ -49,59 +52,65 @@ Sample Prerequisites:
- An iso file on the datastore mentioned above
"""
stub_config = None
cleardata = False
vm_name = None
def __init__(self, stub_config=None, placement_spec=None,
standard_network=None, distributed_network=None):
self.context = None
self.service_manager = None
self.stub_config = stub_config
self.placement_spec = placement_spec
self.standard_network = standard_network
self.distributed_network = distributed_network
self.vm_name = testbed.config['VM_NAME_EXHAUSTIVE']
self.cleardata = None
def setup(self):
parser = sample_cli.build_arg_parser()
parser.add_argument('-n', '--vm_name',
action='store',
help='Name of the testing vm')
args = sample_util.process_cli_args(parser.parse_args())
if args.vm_name:
self.vm_name = args.vm_name
self.cleardata = args.cleardata
def setup(context=None):
global stub_config, cleardata, vm_name
if context:
# Run sample suite via setup script
vm_name = testbed.config['VM_NAME_EXHAUSTIVE']
stub_config = context.stub_config
else:
# Run sample in standalone mode
server, username, password, cleardata, skip_verification, vm_name = \
parse_cli_args_vm(testbed.config['VM_NAME_EXHAUSTIVE'])
stub_config = vapiconnect.connect(server,
username,
password,
skip_verification)
atexit.register(vapiconnect.logout, stub_config)
self.service_manager = ServiceManager(args.server,
args.username,
args.password,
args.skipverification)
self.service_manager.connect()
atexit.register(self.service_manager.disconnect)
self.stub_config = self.service_manager.stub_config
def run():
def run(self):
# Get a placement spec
datacenter_name = testbed.config['VM_DATACENTER_NAME']
vm_folder_name = testbed.config['VM_FOLDER2_NAME']
datastore_name = testbed.config['VM_DATASTORE_NAME']
std_portgroup_name = testbed.config['STDPORTGROUP_NAME']
dv_portgroup_name = testbed.config['VDPORTGROUP1_NAME']
placement_spec = vm_placement_helper.get_placement_spec_for_resource_pool(
stub_config,
if not self.placement_spec:
self.placement_spec = vm_placement_helper.get_placement_spec_for_resource_pool(
self.stub_config,
datacenter_name,
vm_folder_name,
datastore_name)
# Get a standard network backing
standard_network = network_helper.get_standard_network_backing(
stub_config,
if not self.standard_network:
self.standard_network = network_helper.get_standard_network_backing(
self.stub_config,
std_portgroup_name,
datacenter_name)
# Get a distributed network backing
distributed_network = network_helper.get_distributed_network_backing(
stub_config,
if not self.distributed_network:
self.distributed_network = network_helper.get_distributed_network_backing(
self.stub_config,
dv_portgroup_name,
datacenter_name)
create_exhaustive_vm(stub_config, placement_spec, standard_network,
distributed_network)
def create_exhaustive_vm(stub_config, placement_spec, standard_network,
distributed_network):
"""
Create an exhaustive VM.
@ -140,8 +149,8 @@ def create_exhaustive_vm(stub_config, placement_spec, standard_network,
vm_create_spec = VM.CreateSpec(
guest_os=guest_os,
name=vm_name,
placement=placement_spec,
name=self.vm_name,
placement=self.placement_spec,
hardware_version=Hardware.Version.VMX_11,
cpu=Cpu.UpdateSpec(count=2,
cores_per_socket=1,
@ -166,13 +175,13 @@ def create_exhaustive_vm(stub_config, placement_spec, standard_network,
mac_address='11:23:58:13:21:34',
backing=Ethernet.BackingSpec(
type=Ethernet.BackingType.STANDARD_PORTGROUP,
network=standard_network)),
network=self.standard_network)),
Ethernet.CreateSpec(
start_connected=True,
mac_type=Ethernet.MacAddressType.GENERATED,
backing=Ethernet.BackingSpec(
type=Ethernet.BackingType.DISTRIBUTED_PORTGROUP,
network=distributed_network)),
network=self.distributed_network)),
],
cdroms=[
Cdrom.CreateSpec(
@ -214,42 +223,44 @@ def create_exhaustive_vm(stub_config, placement_spec, standard_network,
BootDevice.EntryCreateSpec(BootDevice.Type.ETHERNET)
]
)
print('# Example: create_exhaustive_vm: Creating a VM using spec\n-----')
print(
'# Example: create_exhaustive_vm: Creating a VM using spec\n-----')
print(pp(vm_create_spec))
print('-----')
vm_svc = VM(stub_config)
vm_svc = VM(self.stub_config)
vm = vm_svc.create(vm_create_spec)
print("create_exhaustive_vm: Created VM '{}' ({})".format(vm_name, vm))
print("create_exhaustive_vm: Created VM '{}' ({})".format(self.vm_name,
vm))
vm_info = vm_svc.get(vm)
print('vm.get({}) -> {}'.format(vm, pp(vm_info)))
return vm
def cleanup():
vm = get_vm(stub_config, vm_name)
def cleanup(self):
vm = get_vm(self.stub_config, self.vm_name)
if vm:
power_svc = Power(stub_config)
vm_svc = VM(stub_config)
power_svc = Power(self.stub_config)
vm_svc = VM(self.stub_config)
state = power_svc.get(vm)
if state == Power.Info(state=Power.State.POWERED_ON):
power_svc.stop(vm)
elif state == Power.Info(state=Power.State.SUSPENDED):
power_svc.start(vm)
power_svc.stop(vm)
print("Deleting VM '{}' ({})".format(vm_name, vm))
print("Deleting VM '{}' ({})".format(self.vm_name, vm))
vm_svc.delete(vm)
def main():
setup()
cleanup()
run()
if cleardata:
cleanup()
create_exhaustive_vm = CreateExhaustiveVM()
create_exhaustive_vm.setup()
create_exhaustive_vm.cleanup()
create_exhaustive_vm.run()
if create_exhaustive_vm.cleardata:
create_exhaustive_vm.cleanup()
if __name__ == '__main__':

61
samples/vsphere/vcenter/vm/delete_vm.py
View File

@ -13,45 +13,59 @@
* NON-INFRINGEMENT AND FITNESS FOR A PARTICULAR PURPOSE.
"""
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2017 VMware, Inc. All rights reserved.'
__vcenter_version__ = '6.5+'
import atexit
from samples.vsphere.common import vapiconnect
from samples.vsphere.common.sample_util import parse_cli_args_vm
from com.vmware.vcenter_client import VM
from com.vmware.vcenter.vm_client import Power
from com.vmware.vcenter_client import VM
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common.service_manager import ServiceManager
from samples.vsphere.vcenter.helper.vm_helper import get_vm
class DeleteVM(object):
"""
Demonstrates Deleting User specified Virtual Machine (VM)
Sample Prerequisites:
vCenter/ESX
"""
stub_config = None
vm_name = None
vm = None
def __init__(self):
self.service_manager = None
self.vm_name = None
def setup(self):
parser = sample_cli.build_arg_parser()
parser.add_argument('-n', '--vm_name',
action='store',
default='Sample_Default_VM_for_Simple_Testbed',
help='Name of the testing vm')
args = sample_util.process_cli_args(parser.parse_args())
self.vm_name = args.vm_name
def setup(context=None):
global stub_config, vm_name
server, username, password, cleardata, skip_verification, vm_name = \
parse_cli_args_vm(vm_name)
stub_config = vapiconnect.connect(server, username, password,
skip_verification)
atexit.register(vapiconnect.logout, stub_config)
self.service_manager = ServiceManager(args.server,
args.username,
args.password,
args.skipverification)
self.service_manager.connect()
atexit.register(self.service_manager.disconnect)
def run():
def run(self):
"""
Delete User specified VM from Server
"""
global vm
vm_svc = VM(stub_config)
power_svc = Power(stub_config)
vm = get_vm(stub_config, vm_name)
vm_svc = VM(self.service_manager.stub_config)
power_svc = Power(self.service_manager.stub_config)
vm = get_vm(self.service_manager.stub_config, self.vm_name)
if not vm:
raise Exception('Sample requires an existing vm with name ({}).'
'Please create the vm first.'.format(vm_name))
print("Deleting VM -- '{}-({})')".format(vm_name, vm))
print("Deleting VM -- '{}-({})')".format(self.vm_name, vm))
state = power_svc.get(vm)
if state == Power.Info(state=Power.State.POWERED_ON):
power_svc.stop(vm)
@ -59,12 +73,13 @@ def run():
power_svc.start(vm)
power_svc.stop(vm)
vm_svc.delete(vm)
print("Deleted VM -- '{}-({})".format(vm_name, vm))
print("Deleted VM -- '{}-({})".format(self.vm_name, vm))
def main():
setup()
run()
delete_vm = DeleteVM()
delete_vm.setup()
delete_vm.run()
if __name__ == '__main__':

47
samples/vsphere/vcenter/vm/list_vms.py
View File

@ -13,47 +13,58 @@
* NON-INFRINGEMENT AND FITNESS FOR A PARTICULAR PURPOSE.
"""
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2017 VMware, Inc. All rights reserved.'
__vcenter_version__ = '6.5+'
import atexit
from samples.vsphere.common import vapiconnect
from samples.vsphere.common.sample_util import parse_cli_args
from samples.vsphere.common.sample_util import pp
from com.vmware.vcenter_client import VM
from samples.vsphere.common import sample_cli
from samples.vsphere.common import sample_util
from samples.vsphere.common.service_manager import ServiceManager
from pprint import pprint
class ListVM(object):
"""
Demonstrates getting list of VMs present in vCenter
Sample Prerequisites:
vCenter/ESX
"""
stub_config = None
cleardata = False
def __init__(self):
self.service_manager = None
def setup(self):
parser = sample_cli.build_arg_parser()
args = sample_util.process_cli_args(parser.parse_args())
def setup(context=None):
global stub_config, cleardata
server, username, password, cleardata, skip_verification = parse_cli_args()
stub_config = vapiconnect.connect(server, username, password,
skip_verification)
atexit.register(vapiconnect.logout, stub_config)
self.service_manager = ServiceManager(args.server,
args.username,
args.password,
args.skipverification)
self.service_manager.connect()
atexit.register(self.service_manager.disconnect)
def run():
def run(self):
"""
List VMs present in server
"""
vm_svc = VM(stub_config)
vm_svc = VM(self.service_manager.stub_config)
list_of_vms = vm_svc.list()
print("----------------------------")
print("List Of VMs")
print("----------------------------")
for vm in list_of_vms:
print('{}'.format(vm))
pprint(list_of_vms)
print("----------------------------")
def main():
setup()
run()
list_vm = ListVM()
list_vm.setup()
list_vm.run()
if __name__ == '__main__':

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

@ -21,54 +21,29 @@ import samples.vsphere.vcenter.vm.placement
import samples.vsphere.vcenter.vm.power
from samples.vsphere.common.sample_util import pp
from samples.vsphere.vcenter.setup import testbed_setup
from samples.vsphere.vcenter.vm.create import create_basic_vm
from samples.vsphere.vcenter.vm.create import create_default_vm
from samples.vsphere.vcenter.vm.create import create_exhaustive_vm
from samples.vsphere.vcenter.setup import testbed
from samples.vsphere.vcenter.vm.create.create_default_vm import CreateDefaultVM
from samples.vsphere.vcenter.vm.create.create_basic_vm import CreateBasicVM
from samples.vsphere.vcenter.vm.create.create_exhaustive_vm import \
CreateExhaustiveVM
def setup(context):
class VMSetup(object):
def __init__(self, context=None):
self.context = context
self.basic_vm = None
self.default_vm = None
self.exhaustive_vm = None
def setup(self, context):
print('Setup Samples Started')
create_default_vm.setup(context)
create_basic_vm.setup(context)
create_exhaustive_vm.setup(context)
print('Setup Samples Complete')
def cleanup(context):
setup(context)
print('Cleanup Samples Started')
create_default_vm.cleanup()
create_basic_vm.cleanup()
create_exhaustive_vm.cleanup()
print('Cleanup Samples Complete\n')
def validate(context):
print('Validating and Detecting Resources in vcenter.vm Samples')
r = testbed_setup.validate(context)
if r:
print('==> Samples Setup validated')
return True
else:
print('==> Samples Setup has errors')
return False
def run(context):
# Clean up in case of past failures
cleanup(context)
# Check that sample is ready to run
if context.option['DO_SAMPLES']:
if not validate(context):
exit(0)
self.context = context
###########################################################################
# Getting a PlacementSpec
###########################################################################
placement_spec = samples.vsphere.vcenter.vm.placement \
.get_placement_spec_for_resource_pool(context)
placement_spec = samples.vsphere.vcenter.vm.placement.get_placement_spec_for_resource_pool(context)
print('=' * 79)
print('= Resource selection')
print('=' * 79)
@ -96,6 +71,43 @@ def run(context):
print('=' * 79)
self.default_vm = CreateDefaultVM(context.stub_config,
placement_spec)
self.basic_vm = CreateBasicVM(context.stub_config, placement_spec)
self.exhaustive_vm = CreateExhaustiveVM(context.stub_config,
placement_spec,
standard_network,
distributed_network)
print('Setup Samples Complete')
def cleanup(self):
print('Cleanup Samples Started')
CreateDefaultVM(self.context.stub_config).cleanup()
CreateBasicVM(self.context.stub_config).cleanup()
CreateExhaustiveVM(self.context.stub_config).cleanup()
print('Cleanup Samples Complete\n')
def validate(self):
print('Validating and Detecting Resources in vcenter.vm Samples')
r = testbed_setup.validate(self.context)
if r:
print('==> Samples Setup validated')
return True
else:
print('==> Samples Setup has errors')
return False
def run(self):
# Clean up in case of past failures
self.cleanup()
# Check that sample is ready to run
if self.context.option['DO_SAMPLES']:
if not self.validate():
exit(0)
###########################################################################
# Create VM samples
#
@ -106,37 +118,21 @@ def run(context):
# 1 serial port, 1 parallel port, 1 floppy,
# boot_devices= [CDROM, DISK, ETHERNET])
###########################################################################
create_default_vm.create_default_vm(context.stub_config, placement_spec)
create_basic_vm.create_basic_vm(context.stub_config,
placement_spec,
standard_network)
create_exhaustive_vm.create_exhaustive_vm(context.stub_config,
placement_spec,
standard_network,
distributed_network)
###########################################################################
# Power operation samples
#
# Runs through the power lifecycle for the VM: start, suspend,
# resume (start), stop
#
###########################################################################
samples.vsphere.vcenter.vm.power.setup(context)
samples.vsphere.vcenter.vm.power.run()
samples.vsphere.vcenter.vm.power.cleanup()
self.default_vm.run()
self.basic_vm.run()
self.exhaustive_vm.run()
###########################################################################
# Incremental device CRUDE + connect/disconnect samples
#
###########################################################################
if context.option['DO_SAMPLES_INCREMENTAL']:
samples.vsphere.vcenter.vm.hardware.main.setup(context)
samples.vsphere.vcenter.vm.hardware.main.validate(context)
if self.context.option['DO_SAMPLES_INCREMENTAL']:
samples.vsphere.vcenter.vm.hardware.main.setup(self.context)
samples.vsphere.vcenter.vm.hardware.main.validate(self.context)
samples.vsphere.vcenter.vm.hardware.main.run()
if context.option['DO_SAMPLES_CLEANUP']:
if self.context.option['DO_SAMPLES_CLEANUP']:
samples.vsphere.vcenter.vm.hardware.main.cleanup()
# Sample cleanup
if context.option['DO_SAMPLES_CLEANUP']:
cleanup(context)
if self.context.option['DO_SAMPLES_CLEANUP']:
self.cleanup()