Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def modify_static_ip_pool(self, ip_range_param, new_ip_range_param):
"""Modify static IP pool of org vdc network.
:param str ip_range_param: ip range. For ex: 2.3.3.2-2.3.3.10
:param str new_ip_range_param: new ip range. For ex: 2.3.3.11-2.3.3.20
:return: object containing EntityType.TASK XML data representing the
asynchronous task.
:rtype: lxml.objectify.ObjectifiedElement
"""
vdc_network = self.get_resource()
ip_scope = vdc_network.Configuration.IpScopes.IpScope
if not hasattr(ip_scope, 'IpRanges'):
raise OperationNotSupportedException('No IP range found.')
ip_ranges_list = ip_scope.IpRanges.IpRange
for ip_range in ip_ranges_list:
ip_range_arr = ip_range_param.split('-')
new_ip_range_arr = new_ip_range_param.split('-')
if len(ip_range_arr) <= 1 or len(new_ip_range_arr) <= 1:
raise InvalidParameterException('Input params should be in '
'x.x.x.x-y.y.y.y')
if (ip_range.StartAddress + '-' + ip_range.EndAddress) != \
ip_range_param:
continue
new_start_address = new_ip_range_arr[0]
new_end_address = new_ip_range_arr[1]
ip_range.StartAddress = E.StartAddress(new_start_address)
ip_range.EndAddress = E.StartAddress(new_end_address)
def list_compute_policies(self):
"""List VdcComputePolicy references.
:return: list of VdcComputePolicyReference XML elements each of which
refers to VcdComputePolicy.
:rtype: list of lxml.objectify.StringElement
:raises: OperationNotSupportedException: if the api version is not
supported.
"""
if float(self.client.get_api_version()) < \
float(ApiVersion.VERSION_32.value):
raise OperationNotSupportedException("Unsupported API version")
policy_references = self._fetch_compute_policies()
policy_list = []
for policy_reference in policy_references.VdcComputePolicyReference:
policy_list.append(policy_reference)
return policy_list
def delete_nic(self):
vm = self.get_vm()
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 str service_id: the id of the extension service.
:param format QueryResultFormat: dictates whether id or href should be
part of the returned record. By default id is returned.
:return: API filters registered for the API extension.
:rtype: generator object
"""
try:
records = self.client.get_typed_query(
ResourceType.API_FILTER.value,
equality_filter=('service', service_id),
query_result_format=format).execute()
except OperationNotSupportedException:
msg = 'User doesn\'t have permission to view api filters.'
raise OperationNotSupportedException(msg)
return records
def add_compute_policy(self, href):
"""Add a VdcComputePolicy.
:param str href: URI of the compute policy
:return: an object containing VdcComputePolicyReferences XML element
that refers to individual VdcComputePolicies.
:rtype: lxml.objectify.ObjectifiedElement
:raises: OperationNotSupportedException: if the api version is not
supported.
"""
if float(self.client.get_api_version()) < \
float(ApiVersion.VERSION_32.value):
raise OperationNotSupportedException("Unsupported API version")
policy_references = self._fetch_compute_policies()
policy_id = retrieve_compute_policy_id_from_href(href)
policy_reference_element = E.VdcComputePolicyReference()
policy_reference_element.set('href', href)
policy_reference_element.set('id', policy_id)
policy_references.append(policy_reference_element)
return self.client.put_linked_resource(
self.resource, RelationType.DOWN,
EntityType.VDC_COMPUTE_POLICY_REFERENCES.value,
policy_references)
:return: requests.models.Response response
:rtype: dict
:raises ClusterNotFoundError, CseDuplicateClusterError
"""
filters = client_utils.construct_filters(org=org, vdc=vdc)
entity_svc = def_entity_svc.DefEntityService(self._cloudapi_client)
def_entities = entity_svc.get_entities_by_name(entity_name=cluster_name, filters=filters) # noqa: E501
if len(def_entities) > 1:
raise cse_exceptions.CseDuplicateClusterError(DUPLICATE_CLUSTER_ERROR_MSG) # noqa: E501
if len(def_entities) == 0:
raise cse_exceptions.ClusterNotFoundError(f"Cluster '{cluster_name}' not found.") # noqa: E501
def_entity = def_entities[0]
if def_entity.entity.kind == ClusterEntityKind.NATIVE.value:
return self._nativeCluster.get_upgrade_plan_by_cluster_id(def_entity.id) # noqa: E501
else:
raise vcd_exceptions.OperationNotSupportedException(f"upgrade-plan is not supported for k8-runtime:{def_entity.entity.kind}") # noqa: E501
:return: an object containing EntityType.TASK XML data which represents
the asynchronous task that is tracking the power operation on the
vm.
:rtype: lxml.objectify.ObjectifiedElement
:raises OperationNotSupportedException: if the power operation can't be
performed on the vm.
"""
vm_resource = self.get_resource()
try:
return self.client.post_linked_resource(vm_resource, rel,
media_type, contents)
except OperationNotSupportedException:
power_state = self.get_power_state(vm_resource)
raise OperationNotSupportedException(
'Can\'t {0} vm. Current state of vm: {1}.'.format(
operation_name, VCLOUD_STATUS_MAP[power_state]))
def delete_linked_resource(self, resource, rel, media_type):
"""Deletes the resource referenced by the link.
Deletes the resource referenced by the link with the specified rel and
mediaType in the specified resource.
:raises: OperationNotSupportedException: if the operation fails due to
the link being not visible to the logged in user of the client.
"""
try:
return self.delete_resource(
find_link(resource, rel, media_type).href)
except MissingLinkException as e:
raise OperationNotSupportedException(
"Operation is not supported").with_traceback(e.__traceback__)