Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, client, name=None, href=None, resource=None):
"""Constructor for External Network objects.
:param pyvcloud.vcd.client.Client client: the client that will be used
to make REST calls to vCD.
:param str name: name of the entity.
:param str href: URI of the entity.
:param lxml.objectify.ObjectifiedElement resource: object containing
EntityType.EXTERNAL_NETWORK XML data representing the external
network.
"""
self.client = client
self.name = name
if href is None and resource is None:
raise InvalidParameterException(
"External network initialization failed as arguments are "
"either invalid or None")
self.href = href
self.resource = resource
if resource is not None:
self.name = resource.get('name')
self.href = resource.get('href')
self.href_admin = get_admin_href(self.href)
"""Detach a portgroup from an external network.
:param str vim_server_name: name of vim server where
portgroup is present.
:param str port_group_name: name of the portgroup to be detached from
external network.
return: object containing vmext:VMWExternalNetwork XML element that
representing the external network.
:rtype: lxml.objectify.ObjectifiedElement
"""
ext_net = self.get_resource()
platform = Platform(self.client)
if not vim_server_name or not port_group_name:
raise InvalidParameterException(
"Either vCenter Server name is none or portgroup name is none")
vc_record = platform.get_vcenter(vim_server_name)
vc_href = vc_record.get('href')
if hasattr(ext_net, 'VimPortGroupRefs'):
pg_moref_types = \
platform.get_port_group_moref_types(vim_server_name,
port_group_name)
else:
raise \
InvalidParameterException("External network"
" has only one port group")
vim_port_group_refs = ext_net.VimPortGroupRefs
vim_obj_refs = vim_port_group_refs.VimObjectRef
for vim_obj_ref in vim_obj_refs:
"""Return information for an independent disk.
:param str name: name of the disk.
:param str disk_id: id of the disk.
:return: an object containing EntityType.DISK XML data which represents
the disk.
:rtype: lxml.objectify.ObjectifiedElement
:raises: InvalidParameterException: if neither name nor the disk_id
param is specified.
:raises: EntityNotFoundException: if the named disk cannot be located.
"""
if name is None and disk_id is None:
raise InvalidParameterException(
'Unable to idendify disk without name or id.')
self.get_resource()
disks = self.get_disks()
result = None
if disk_id is not None:
if not disk_id.startswith('urn:vcloud:disk:'):
disk_id = 'urn:vcloud:disk:' + disk_id
for disk in disks:
if disk.get('id') == disk_id:
result = disk
# disk-id's are unique so it is ok to break the loop
# and stop looking further.
break
:param str gateway_name: name of the gateway entity.
:param str rule_id: nat rule id.
:param str nat_href: nat rule href.
:param lxml.objectify.ObjectifiedElement resource: object containing
EntityType.NAT XML data representing the nat rule.
"""
self.client = client
self.gateway_name = gateway_name
self.rule_id = rule_id
if gateway_name is not None and \
rule_id is not None and \
nat_href is None and \
resource is None:
self.__build_self_href()
if nat_href is None and resource is None and self.href is None:
raise InvalidParameterException(
"NatRule initialization failed as arguments are either "
"invalid or None")
if nat_href is not None:
self.rule_id = self.__extract_rule_id(nat_href)
self.href = nat_href
self.resource = resource
def __init__(self, client, href=None, resource=None):
"""Constructor for a PVDC object.
:param pyvcloud.vcd.client.Client client: the client that will be used
to make REST calls to vCD.
:param str href: URI of the entity.
:param lxml.objectify.ObjectifiedElement resource: object containing
EntityType.PROVIDER_VDC XML data representing the provider vdc.
"""
self.client = client
if href is None and resource is None:
raise InvalidParameterException(
"PVDC initialization failed as arguments are either invalid "
"or None")
self.href = href
self.resource = resource
if resource is not None:
self.name = resource.get('name')
self.href = resource.get('href')
self.admin_resource = get_admin_href(self.href)
:param client.MetadataDomain domain: domain where the new entries would
be put.
:param client.MetadataVisibility visibility: visibility of the metadata
entries.
:param client.MetadataValueType metadata_value_type:
:param bool use_admin_endpoint: if True, will use the /api/admin
endpoint to add new entries to the metadata object else will use
the vanilla /api endpoint.
:return: an object of type EntityType.TASK XML which represents
the asynchronous task that is updating the metadata entries.
:rtype: lxml.objectify.ObjectifiedElement
"""
if not isinstance(domain, MetadataDomain):
raise InvalidParameterException('Invalid domain.')
if not isinstance(visibility, MetadataVisibility):
raise InvalidParameterException('Invalid visibility.')
if not isinstance(metadata_value_type, MetadataValueType):
raise InvalidParameterException('Invalid type of value.')
metadata = self.get_all_metadata(use_admin_endpoint)
new_metadata = E.Metadata()
for k, v in key_value_dict.items():
entry = E.MetadataEntry(
{'type': 'xs:string'},
E.Domain(domain.value, visibility=visibility.value),
E.Key(k),
E.TypedValue(
{'{' + NSMAP['xsi'] + '}type': metadata_value_type.value},
E.Value(v)))
new_metadata.append(entry)
def _populate_services(self, firewall_rule_temp, services):
"""Populates service elements.
:param firewall_rule_temp: Firewall rule
:param [] services: protocol to port mapping.
e.g., [{'tcp' : {'any' : any}}]
"""
if services:
for service in services:
protocol = [k for k in service.keys()][0]
if protocol not in FirewallRule.__PROTOCOL_LIST:
valid_protocols = ', '.join(FirewallRule.__PROTOCOL_LIST)
raise InvalidParameterException(
protocol + " is not valid. It should be from " +
valid_protocols)
value = service.get(protocol)
source_port = [port for port in value.keys()][0]
destination_port = value.get(source_port)
self.__populate_protocol_elements(firewall_rule_temp, protocol,
source_port,
destination_port)
:rtype: lxml.objectify.ObjectifiedElement
"""
access_settings_params = E.AccessSettings()
for access_setting in access_settings_list:
if access_setting["type"] == 'user':
org_href = self.get_org_href()
subject_href = self.client.get_user_in_org(
access_setting['name'], org_href).get('href')
subject_type = EntityType.USER.value
elif access_setting["type"] == 'org':
subject_href = get_admin_href(
self.client.get_org_by_name(
access_setting['name']).get('href'))
subject_type = EntityType.ADMIN_ORG.value
else:
raise InvalidParameterException("Invalid subject type")
subject_name = access_setting['name']
# Make 'ReadOnly' the default access_level if it is not specified.
if 'access_level' in access_setting:
access_level = access_setting['access_level']
else:
access_level = 'ReadOnly'
access_setting_params = E.AccessSetting(
E.Subject(
name=subject_name, href=subject_href, type=subject_type),
E.AccessLevel(access_level))
access_settings_params.append(access_setting_params)
return access_settings_params
nic_ids = self.params.get('nic_ids')
vm_name = self.params.get('vm_name')
response = dict()
response['changed'] = False
# VM should be powered off before deleting a nic
if not vm.is_powered_off():
msg = "VM {0} is powered on. Cant remove nics in the current state"
raise OperationNotSupportedException(msg.format(vm_name))
for nic_id in nic_ids:
try:
delete_nic_task = vm.delete_nic(nic_id)
self.execute_task(delete_nic_task)
response['changed'] = True
except InvalidParameterException as error:
response['msg'] = error.__str__()
else:
response['msg'] = 'VM nic(s) has been deleted'
return response
:param int external_port: external port of NAT rule if nat_type is
portForwarding.
:param int internal_port: internal port of NAT rule if nat_type is
portForwarding.
:param str protocol: protocol of NAT rule if nat_type is
portForwarding.
:return: an object containing EntityType.TASK XML data which represents
the asynchronous task that is updating the vApp network.
:rtype: lxml.objectify.ObjectifiedElement
:raises: InvalidParameterException: Enable NAT service failed as
given network's connection is not routed
"""
self._get_resource()
fence_mode = self.resource.Configuration.FenceMode
if fence_mode != 'natRouted':
raise InvalidParameterException(
"Enable NAT service failed as given network's connection "
"is not routed")
features = self.resource.Configuration.Features
if not hasattr(features, 'NatService'):
VappNat._makeNatServiceAttr(features)
nat_service = features.NatService
if nat_service.NatType != nat_type:
VappNat._delete_all_nat_rule(nat_service)
nat_service.NatType = E.NatType(nat_type)
nat_rule = E.NatRule()
if nat_type == 'ipTranslation':
one_to_vm_rule = E.OneToOneVmRule()
one_to_vm_rule.append(E.MappingMode(mapping_mode))
if mapping_mode == 'manual':
one_to_vm_rule.append(E.ExternalIpAddress(external_ip_address))
one_to_vm_rule.append(E.VAppScopedVmId(vapp_scoped_vm_id))