Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def deallocate_vm(self, vm_id):
url_params = azure_helpers.parse_url(VM_RESOURCE_ID,
vm_id)
vm_name = url_params.get(VM_NAME)
self.compute_client. \
virtual_machines.deallocate(self.resource_group,
vm_name).wait()
def restart_vm(self, vm_id):
url_params = azure_helpers.parse_url(VM_RESOURCE_ID,
vm_id)
vm_name = url_params.get(VM_NAME)
return self.compute_client.virtual_machines.restart(
self.resource_group, vm_name).wait()
def delete_vm(self, vm_id):
url_params = azure_helpers.parse_url(VM_RESOURCE_ID,
vm_id)
vm_name = url_params.get(VM_NAME)
return self.compute_client.virtual_machines.delete(
self.resource_group, vm_name).wait()
def delete_vm_firewall_rule(self, fw_rule_id, vm_firewall):
url_params = azure_helpers.parse_url(VM_FIREWALL_RULE_RESOURCE_ID,
fw_rule_id)
name = url_params.get(VM_FIREWALL_RULE_NAME)
return self.network_management_client.security_rules. \
delete(self.resource_group, vm_firewall, name).result()
def create_subnet(self, network_id, subnet_name, params):
url_params = azure_helpers.parse_url(NETWORK_RESOURCE_ID, network_id)
network_name = url_params.get(NETWORK_NAME)
result_create = self.network_management_client \
.subnets.create_or_update(
self.resource_group,
network_name,
subnet_name,
params
)
subnet_info = result_create.result()
return subnet_info
def update_snapshot_tags(self, snapshot_id, tags):
url_params = azure_helpers.parse_url(SNAPSHOT_RESOURCE_ID,
snapshot_id)
snapshot_name = url_params.get(SNAPSHOT_NAME)
return self.compute_client.snapshots.update(
self.resource_group,
snapshot_name,
{'tags': tags},
raw=True
)
def get_network(self, network_id):
url_params = azure_helpers.parse_url(NETWORK_RESOURCE_ID,
network_id)
network_name = url_params.get(NETWORK_NAME)
return self.network_management_client.virtual_networks.get(
self.resource_group, network_name)
def update_disk_tags(self, disk_id, tags):
url_params = azure_helpers.parse_url(VOLUME_RESOURCE_ID,
disk_id)
disk_name = url_params.get(VOLUME_NAME)
return self.compute_client.disks.update(
self.resource_group,
disk_name,
{'tags': tags},
raw=True
)
def get_vm_firewall(self, fw_id):
url_params = azure_helpers.parse_url(VM_FIREWALL_RESOURCE_ID,
fw_id)
fw_name = url_params.get(VM_FIREWALL_NAME)
return self.network_management_client.network_security_groups. \
get(self.resource_group, fw_name)
def create_vm_firewall_rule(self, fw_id,
rule_name, parameters):
url_params = azure_helpers.parse_url(VM_FIREWALL_RESOURCE_ID,
fw_id)
vm_firewall_name = url_params.get(VM_FIREWALL_NAME)
return self.network_management_client.security_rules. \
create_or_update(self.resource_group, vm_firewall_name,
rule_name, parameters).result()