pve_tests/functions.py

109 lines
2.7 KiB
Python

import logging
from typing import Optional, Tuple
logger = logging.getLogger(__name__)
def create_vm(
proxmox,
node: Optional[str] = None,
vmid: Optional[int] = None,
name: Optional[str] = None,
prefix: str = 'stepchenkoas',
cores: int = 2,
memory: int = 2048,
disk_size: int = 16,
start: int = 1,
):
if node is None:
nodes = proxmox.nodes.get()
nodes = list(filter(lambda node: node['status'] == 'online', nodes))
node = min(nodes, key=lambda node: node['disk'])['node']
if vmid is None:
vmid = proxmox.cluster.get('nextid')
if name is None:
name = f'{prefix}-{vmid}'
proxmox.nodes(node).qemu.post(
node=node,
vmid=vmid,
description='This VM was created automatically through Proxmox VE API',
cores=cores,
cpu='host',
memory=memory,
name=name,
net0='virtio,bridge=vmbr0,tag=103,firewall=1',
ostype='l26',
pool='Virt_LAB',
#alt-server-v-10.1-rc3-x86_64.iso
sata2='templates:iso/alt-server-v-10.1-rc3-x86_64.iso,media=cdrom',
scsi0=f'rbd-storage:{disk_size},discard=on',
scsihw='virtio-scsi-pci',
sockets=1,
start=start,
)
logger.info('VM %s was created successfully!', vmid)
def delete_vm(
proxmox,
node,
vmid,
):
proxmox.nodes(node).qemu(vmid).delete(
node=node,
vmid=vmid,
)
logger.info('VM %s was deleted successfully!', vmid)
def clone_template(
proxmox,
node: str,
vmid: int,
newid: Optional[int] = None,
name: Optional[str] = None,
prefix: str ='stepchenkoas',
) -> Tuple[int, str]:
if newid is None:
newid = int(proxmox.cluster.get('nextid'))
if name is None:
name = f'{prefix}-{newid}'
assert type(newid) is int
assert type(name) is str
upid: str = proxmox.nodes(node).qemu(vmid).clone.post(
newid=newid,
node=node,
vmid=vmid,
format='raw',
full=1,
name=name,
pool='Virt_LAB',
storage='rbd-storage',
target=node,
)
# TODO: check task status
logger.debug("UPID '%s'", upid)
logger.info('VM %s was cloned to %s successfully!', vmid, newid)
return (newid, upid)
def get_vm_ip(
proxmox,
node: 'str',
vmid: int,
):
res = proxmox.nodes(node).qemu(vmid).agent('network-get-interfaces').get()
vm_ip_address = None
for net in res['result']:
if net['name'] == 'eth0' or net['name'] == 'ens19':
for ip in net['ip-addresses']:
if ip['ip-address-type'] == 'ipv4':
vm_ip_address = ip['ip-address']
return vm_ip_address