From 3e74712309888a4c6f6e83defb4ec16b21b908e9 Mon Sep 17 00:00:00 2001 From: host Date: Fri, 24 Jul 2026 00:43:08 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=20proxmox=5Ftools.py=20-=20=D0=B8=D0=BD=D1=81=D1=82?= =?UTF-8?q?=D1=80=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D1=8B=20=D0=B4=D0=BB=D1=8F?= =?UTF-8?q?=20=D1=83=D0=BF=D1=80=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D1=8F?= =?UTF-8?q?=20Proxmox=20VE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- proxmox_tools.py | 810 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 810 insertions(+) create mode 100644 proxmox_tools.py diff --git a/proxmox_tools.py b/proxmox_tools.py new file mode 100644 index 0000000..3437866 --- /dev/null +++ b/proxmox_tools.py @@ -0,0 +1,810 @@ +#!/usr/bin/env python3 +""" +proxmox_tools.py - Proxmox API Tools +Description: Full toolset for Proxmox VE management: + - LXC (create, start, stop, delete) + - VM (clone from template, create from ISO, manage) + - Monitoring (nodes, storage, resources, tasks) +Usage: + from proxmox_tools import Tools + tools = Tools() + print(tools.connection()) + print(tools.vm_create(vmid=200, name="ubuntu", ciuser="ubuntu", cipassword="1234567")) +""" + +import json +import re +import time +import requests + +# ============================================================ +# CONFIG - change as needed +# ============================================================ +PROXMOX_HOST = "https://192.168.31.2:8006" +PROXMOX_USER = "root@pam" +PROXMOX_TOKEN_NAME = "openwebui-token" +PROXMOX_TOKEN_VALUE = "7aba7743-0133-494b-9dd7-9715f55e5dca" +VERIFY_SSL = False +DEFAULT_NODE = "pve" +DEFAULT_STORAGE = "local" +DEFAULT_BRIDGE = "vmbr0" + +# ============================================================ +# INTERNAL HELPERS +# ============================================================ +_base_url = PROXMOX_HOST.rstrip("/") + "/api2/json" +_auth_header = { + "Authorization": f"PVEAPIToken={PROXMOX_USER}!{PROXMOX_TOKEN_NAME}={PROXMOX_TOKEN_VALUE}" +} + + +def _api(method, path, params=None, data=None): + """Execute a Proxmox API request.""" + url = _base_url + path + resp = requests.request( + method=method, + url=url, + headers=_auth_header, + params=params, + json=data, + verify=VERIFY_SSL, + timeout=30, + ) + resp.raise_for_status() + return resp.json().get("data", {}) + + +def _fmt(b): + """Format bytes to human-readable.""" + if b is None: + return "N/A" + b = float(b) + for u in ["B", "KB", "MB", "GB", "TB"]: + if b < 1024: + return f"{b:.1f} {u}" + b /= 1024 + return f"{b:.1f} PB" + + +def _pct(u, t): + """Format used/total as '1.2 GB / 4.0 GB (30.0%)'.""" + if u is None or t is None or t == 0: + return "N/A" + return f"{_fmt(u)} / {_fmt(t)} ({u/t*100:.1f}%)" + + +def _uptime(s): + """Format uptime seconds to '1d 2h 3m 4s'.""" + if s is None: + return "N/A" + d, r = divmod(int(s), 86400) + h, r = divmod(r, 3600) + m, s2 = divmod(r, 60) + parts = [] + if d: + parts.append(f"{d}d") + if h: + parts.append(f"{h}h") + if m: + parts.append(f"{m}m") + parts.append(f"{s2}s") + return " ".join(parts) + + +def _wait_for_task(upid, timeout=60): + """ + Wait for a Proxmox task to complete by UPID. + Returns (ok: bool, message: str). + """ + if not upid or not isinstance(upid, str) or not upid.startswith("UPID:"): + return True, "no UPID to wait" + node = DEFAULT_NODE + start = time.time() + while time.time() - start < timeout: + try: + tasks = _api("GET", f"/nodes/{node}/tasks") + for t in tasks: + if t.get("upid") == upid: + status = t.get("status", "") + if status == "stopped": + if t.get("exitstatus") == "OK": + return True, "OK" + else: + return False, str(t) + break + except Exception: + pass + time.sleep(2) + return False, "timeout waiting for task" + + +# ============================================================ +# MAIN TOOLS CLASS +# ============================================================ +class Tools: + """Proxmox VE management toolset.""" + + # ---- CONNECTION & NODES -------------------------------------------- + + @staticmethod + def connection(): + """Check Proxmox connection.""" + try: + v = _api("GET", "/version") + return ( + "OK Proxmox VE " + + v.get("version", "?") + + " | node: pve | IP: 192.168.31.2" + ) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def nodes(): + """Show all cluster nodes.""" + try: + items = _api("GET", "/nodes") + out = ["Nodes:"] + for n in items: + out.append(f"- {n.get('node', '?')} ({n.get('status', '?')})") + out.append( + f" CPU: {n.get('cpu', 0)*100:.1f}% | " + f"RAM: {_pct(n.get('mem'), n.get('maxmem'))}" + ) + out.append(f" Uptime: {_uptime(n.get('uptime'))}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def node_status(node=DEFAULT_NODE): + """Show detailed node status.""" + try: + s = _api("GET", f"/nodes/{node}/status") + out = [f"Node {node}:"] + ci = s.get("cpuinfo", {}) + out.append(f"CPU: {ci.get('model', '?')} ({s.get('cpus', 0)} cores)") + out.append(f"Load: {s.get('cpu', 0)*100:.1f}%") + la = s.get("loadavg", []) + if la: + out.append(f"Loadavg: {' / '.join(f'{float(v):.2f}' for v in la)}") + m = s.get("memory", {}) + out.append(f"RAM: {_pct(m.get('used'), m.get('total'))}") + sw = s.get("swap", {}) + if sw.get("total", 0): + out.append(f"SWAP: {_pct(sw.get('used'), sw.get('total'))}") + r = s.get("rootfs", {}) + if r: + out.append(f"Disk: {_pct(r.get('used'), r.get('total'))}") + out.append(f"Uptime: {_uptime(s.get('uptime'))}") + out.append(f"Kernel: {s.get('kversion', '?')}") + if s.get("pveversion"): + out.append(f"PVE: {s['pveversion']}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def cluster(): + """Show cluster status (quorum, members).""" + try: + items = _api("GET", "/cluster/status") + out = ["Cluster:"] + has_cluster_info = False + for item in items: + t = item.get("type", "") + if t == "cluster": + has_cluster_info = True + out.append( + f"Cluster: {item.get('name', '?')} | " + f"Quorum: {'OK' if item.get('quorate') else 'FAIL'}" + ) + elif t == "node": + out.append( + f" {item.get('name', '?')} - {item.get('status', '?')} " + f"({item.get('ip', '-')})" + ) + if not has_cluster_info: + node_count = sum(1 for i in items if i.get("type") == "node") + out.append("Cluster: single-node | Quorum: OK (N/A)") + out.append(f" Nodes: {node_count}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + # ---- RESOURCES & STORAGE ------------------------------------------ + + @staticmethod + def resources(ftype=""): + """Show all cluster resources (VMs, LXC, nodes, storage).""" + try: + p = {} + if ftype: + p["type"] = ftype + items = _api("GET", "/cluster/resources", params=p) + out = ["Resources:"] + g = {} + for r in items: + g.setdefault(r.get("type", "?"), []).append(r) + labels = { + "node": "Nodes", + "qemu": "QEMU", + "lxc": "LXC", + "storage": "Storage", + } + for t, lst in g.items(): + out.append(f"{labels.get(t, t)}: {len(lst)}") + for r in lst[:25]: + n = r.get("name") or r.get("id") or "?" + st = r.get("status", "?") + line = f" {n} - {st}" + if t in ("qemu", "lxc"): + line += f" (ID:{r.get('vmid', '?')})" + if r.get("maxmem"): + line += f" RAM:{_pct(r.get('mem'), r.get('maxmem'))}" + if r.get("cpu"): + line += f" CPU:{r.get('cpu')*100:.1f}%" + out.append(line) + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def storage(): + """Show all storages.""" + try: + items = _api("GET", "/storage") + if not items: + return "No storage" + out = ["Storage:"] + for s in items: + name = s.get("storage", "?") + out.append(f"- {name} ({s.get('type', '?')})") + out.append(f" Content: {s.get('content', '?')}") + if s.get("total"): + out.append(f" Used: {_pct(s.get('used'), s.get('total'))}") + out.append(f" Free: {_fmt(s.get('avail'))}") + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def tasks(limit=10, node=DEFAULT_NODE): + """Show recent tasks on node.""" + try: + items = _api("GET", f"/nodes/{node}/tasks", params={"limit": limit}) + if not items: + return f"No tasks on {node}" + out = [f"Tasks on {node}:"] + for t in items: + start = time.ctime(t.get("starttime", 0)) if t.get("starttime") else "?" + out.append( + f"- {t.get('type', '?')} | {t.get('status', '?')} | " + f"{t.get('user', '?')} | {start}" + ) + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + # ---- LXC CONTAINERS ----------------------------------------------- + + @staticmethod + def lxc_list(node=DEFAULT_NODE): + """Show LXC container list.""" + try: + items = _api("GET", f"/nodes/{node}/lxc") + if not items: + return f"No LXC on {node}" + out = [f"LXC on {node}:"] + for c in items: + out.append( + f"- {c.get('name', '?')} (ID:{c.get('vmid', '?')}) - " + f"{c.get('status', '?')}" + ) + if c.get("cpu"): + out.append(f" CPU:{c['cpu']*100:.1f}%") + if c.get("maxmem"): + out.append(f" RAM:{_pct(c.get('mem'), c.get('maxmem'))}") + if c.get("maxdisk"): + out.append(f" Disk:{_pct(c.get('disk'), c.get('maxdisk'))}") + if c.get("uptime"): + out.append(f" Uptime:{_uptime(c['uptime'])}") + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def lxc_detail(vmid, node=DEFAULT_NODE): + """Show detailed LXC info.""" + try: + s = _api("GET", f"/nodes/{node}/lxc/{vmid}/status/current") + c = _api("GET", f"/nodes/{node}/lxc/{vmid}/config") + name = s.get("name", c.get("hostname", "?")) + out = [f"LXC #{vmid}: {name} - {s.get('status', '?')}"] + out.append( + f"OS: {c.get('ostype', '?')} | CPU: {c.get('cores', '?')} | " + f"RAM: {c.get('memory', '?')}MB" + ) + out.append(f"Disk: {c.get('rootfs', '?')} | SWAP: {c.get('swap', '?')}MB") + out.append(f"Hostname: {c.get('hostname', '?')}") + nets = [f" {k}: {v}" for k, v in c.items() if k.startswith("net")] + if nets: + out.append("Network:\n" + "\n".join(nets)) + if s.get("maxmem"): + out.append(f"RAM use: {_pct(s.get('mem'), s.get('maxmem'))}") + if s.get("cpu"): + out.append(f"CPU: {s['cpu']*100:.1f}%") + if s.get("uptime"): + out.append(f"Uptime: {_uptime(s['uptime'])}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def lxc_action(vmid, action, node=DEFAULT_NODE): + """ + Manage LXC: start, stop, shutdown, restart, suspend, resume, delete. + """ + try: + acts = { + "start": "start", + "stop": "stop", + "shutdown": "shutdown", + "restart": "reboot", + "suspend": "suspend", + "resume": "resume", + "delete": "del", + } + if action not in acts: + return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" + if action == "delete": + result = _api("DELETE", f"/nodes/{node}/lxc/{vmid}") + else: + result = _api("POST", f"/nodes/{node}/lxc/{vmid}/status/{acts[action]}") + return f"OK LXC #{vmid} {action}\n{result}" + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def templates(storage="local", node=DEFAULT_NODE): + """Show available LXC templates in storage.""" + try: + content = _api("GET", f"/nodes/{node}/storage/{storage}/content") + tmpl = [x for x in content if x.get("content") == "vztmpl"] + if not tmpl: + return f"No templates in {storage}" + out = [f"Templates in {storage}:"] + for t in tmpl: + out.append(f" {t.get('volid', '?')} - {_fmt(t.get('size'))}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def lxc_create( + vmid, + hostname, + ostemplate, + password="", + sshkey="", + storage=DEFAULT_STORAGE, + cores=1, + memory=512, + swap=0, + disk="8G", + bridge=DEFAULT_BRIDGE, + ip="dhcp", + netmask=24, + gw="", + dns="", + domain="", + node=DEFAULT_NODE, + unpriv=True, + start_now=True, + ): + """Create a new LXC container.""" + try: + if not hostname: + return "Error: hostname required" + if not ostemplate: + return "Error: ostemplate required" + if not password and not sshkey: + return "Error: password or sshkey required" + exist = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + exist += [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] + if str(vmid) in exist: + return f"Error: VMID {vmid} already exists" + params = { + "vmid": vmid, + "hostname": hostname, + "ostemplate": ostemplate, + "storage": storage, + "cores": cores, + "memory": memory, + "swap": swap, + "unprivileged": 1 if unpriv else 0, + } + params["rootfs"] = disk if ":" in disk else f"{storage}:{disk}" + net = f"name=eth0,bridge={bridge}" + if ip and ip.lower() != "dhcp": + net += f",ip={ip}/{netmask}" + if gw: + net += f",gw={gw}" + else: + net += ",ip=dhcp" + params["net0"] = net + if password: + params["password"] = password + if sshkey: + params["ssh-public-keys"] = sshkey + if dns: + params["nameserver"] = dns + if domain: + params["searchdomain"] = domain + if start_now: + params["start"] = 1 + result = _api("POST", f"/nodes/{node}/lxc", data=params) + out = [f"OK LXC #{vmid} creating!"] + out.append(f" Name: {hostname}") + out.append(f" Template: {ostemplate}") + out.append(f" CPU: {cores} cores, RAM: {memory}MB, Disk: {disk}") + if ip and ip.lower() != "dhcp": + out.append(f" Net: {bridge}, IP: {ip}/{netmask}") + else: + out.append(f" Net: {bridge}, IP: DHCP") + if isinstance(result, str) and result: + out.append(f"UPID: {result}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + # ---- VIRTUAL MACHINES (QEMU) ------------------------------------- + + @staticmethod + def vm_list(node=DEFAULT_NODE): + """Show all VM (QEMU) list.""" + try: + items = _api("GET", f"/nodes/{node}/qemu") + if not items: + return f"No VMs on {node}" + out = [f"VMs on {node}:"] + for v in items: + out.append( + f"- {v.get('name', '?')} (ID:{v.get('vmid', '?')}) - " + f"{v.get('status', '?')}" + ) + if v.get("cpu"): + out.append(f" CPU:{v['cpu']*100:.1f}%") + if v.get("maxmem"): + out.append(f" RAM:{_pct(v.get('mem'), v.get('maxmem'))}") + if v.get("uptime"): + out.append(f" Uptime:{_uptime(v['uptime'])}") + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def vm_detail(vmid, node=DEFAULT_NODE): + """Show detailed VM info.""" + try: + s = _api("GET", f"/nodes/{node}/qemu/{vmid}/status/current") + out = [f"VM #{vmid}: {s.get('name', '?')}"] + out.append(f"Status: {s.get('status', '?')}") + if s.get("maxmem"): + out.append(f"RAM: {_pct(s.get('mem'), s.get('maxmem'))}") + if s.get("cpu"): + out.append(f"CPU: {s['cpu']*100:.1f}%") + if s.get("maxdisk") and s.get("maxdisk", 0) > 0: + out.append(f"Disk: {_pct(s.get('disk'), s.get('maxdisk'))}") + if s.get("uptime"): + out.append(f"Uptime: {_uptime(s['uptime'])}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def vm_action(vmid, action, node=DEFAULT_NODE): + """ + Manage VM: start, stop, shutdown, restart, suspend, resume, delete. + """ + try: + acts = { + "start": "start", + "stop": "stop", + "shutdown": "shutdown", + "restart": "reboot", + "suspend": "suspend", + "resume": "resume", + "delete": "del", + } + if action not in acts: + return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" + if action == "delete": + result = _api("DELETE", f"/nodes/{node}/qemu/{vmid}") + else: + result = _api( + "POST", f"/nodes/{node}/qemu/{vmid}/status/{acts[action]}" + ) + return f"OK VM #{vmid} {action}\n{result}" + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def vm_create_from_template( + vmid, + name, + template_vmid=999, + full=True, + memory=None, + cores=None, + storage=None, + target_node=None, + pool=None, + start_now=True, + ciuser="", + cipassword="", + sshkeys="", + node=DEFAULT_NODE, + ): + """ + Create a new VM by cloning from a template. + Default template: VM #999 (ubuntu-2404-ci). + + Cloud-init params (for guest OS user setup): + ciuser: username in the guest (e.g. "ubuntu") + cipassword: password for the user + sshkeys: SSH public keys (newline-separated string) + + FIXED: "start" parameter is NOT sent to /clone - + startup is done via separate API call after clone completes. + """ + try: + if not vmid: + return "Error: vmid required" + if not name: + return "Error: name required" + + # Check VMID availability + exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] + exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + if str(vmid) in exist_vm + exist_lxc: + return f"Error: VMID {vmid} already exists" + + # Step 1: Clone (WITHOUT "start" parameter!) + params = { + "newid": vmid, + "name": name, + "full": 1 if full else 0, + } + if storage: + params["storage"] = storage + if target_node: + params["target"] = target_node + if pool: + params["pool"] = pool + + result = _api( + "POST", f"/nodes/{node}/qemu/{template_vmid}/clone", data=params + ) + + out = [f"OK VM #{vmid}: {name} cloning from template #{template_vmid}!"] + out.append(f" Type: {'full' if full else 'linked'} clone") + if storage: + out.append(f" Storage: {storage}") + + upid = result if isinstance(result, str) else "" + if upid: + out.append(f"UPID: {upid}") + + # Step 2: Wait for clone to finish + out.append("Waiting for clone to finish...") + ok, msg = _wait_for_task(upid, timeout=120) + if not ok: + out.append(f" WARN Clone may not have finished: {msg}") + else: + out.append(" Clone done") + + # Step 3: Apply settings (resources + cloud-init) + config_params = {} + if memory is not None: + config_params["memory"] = memory + if cores is not None: + config_params["cores"] = cores + if ciuser: + config_params["ciuser"] = ciuser + if cipassword: + config_params["cipassword"] = cipassword + if sshkeys: + config_params["sshkeys"] = sshkeys + + if config_params: + out.append("Applying settings...") + try: + _api( + "PUT", + f"/nodes/{node}/qemu/{vmid}/config", + data=config_params, + ) + if memory is not None: + out.append(f" RAM -> {memory} MB") + if cores is not None: + out.append(f" CPU -> {cores} cores") + if ciuser: + out.append(f" Cloud-init user: {ciuser}") + if cipassword: + out.append(" Cloud-init password: (set)") + if sshkeys: + out.append(" Cloud-init SSH keys: (set)") + out.append(" Settings applied") + except Exception as e2: + out.append(f" WARN Could not apply settings: {e2}") + + # Step 4: Start VM (separate call!) + if start_now: + out.append("Starting VM...") + try: + _api("POST", f"/nodes/{node}/qemu/{vmid}/status/start") + out.append(f" VM #{vmid} started!") + except Exception as e3: + out.append(f" WARN Could not start: {e3}") + else: + out.append(f" VM #{vmid} created (not started)") + + return "\n".join(out) + + except Exception as e: + return "ERR " + str(e) + + @staticmethod + def vm_create( + vmid, + name, + template_vmid=999, + memory=4096, + cores=2, + disk_size="32G", + disk_storage="local-lvm", + iso="", + bridge=DEFAULT_BRIDGE, + ip="dhcp", + netmask=24, + gw="", + ostype="l26", + agent=1, + ciuser="", + cipassword="", + sshkeys="", + node=DEFAULT_NODE, + start_now=True, + ): + """ + Universal VM creation. + + Modes: + 1. From template (default): clones VM #999 (ubuntu-2404-ci) + 2. From ISO: template_vmid=None, iso="local:iso/image.iso" + + Cloud-init params (for mode 1): + ciuser: guest OS username + cipassword: guest OS password + sshkeys: SSH public keys + + Examples: + # Clone from template with cloud-init user: + tools.vm_create( + vmid=200, + name="web-server", + memory=8192, + cores=4, + ciuser="ubuntu", + cipassword="mypassword", + start_now=True, + ) + + # Create from ISO: + tools.vm_create( + vmid=202, + name="debian-server", + template_vmid=None, + iso="local:iso/debian-12.iso", + ip="192.168.31.60", + gw="192.168.31.1", + ) + """ + try: + # Mode 1: clone from template + if template_vmid: + return Tools.vm_create_from_template( + vmid=vmid, + name=name, + template_vmid=template_vmid, + full=True, + memory=memory, + cores=cores, + storage=disk_storage, + start_now=start_now, + ciuser=ciuser, + cipassword=cipassword, + sshkeys=sshkeys, + node=node, + ) + + # Mode 2: create from scratch (ISO) + if not name: + return "Error: name required" + if not vmid: + return "Error: vmid required" + + exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] + exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + if str(vmid) in exist_vm + exist_lxc: + return f"Error: VMID {vmid} already exists" + + params = { + "vmid": vmid, + "name": name, + "memory": memory, + "cores": cores, + "sockets": 1, + "ostype": ostype, + "agent": str(agent), + } + + # Disk + if ":" in disk_size: + params["virtio0"] = disk_size + else: + params["virtio0"] = f"{disk_storage}:{disk_size}" + + # ISO + if iso: + params["ide2"] = f"{iso},media=cdrom" + params["boot"] = "order=ide2;virtio0" + + # Network + net = f"name=eth0,bridge={bridge}" + if ip and ip.lower() != "dhcp": + net += f",ip={ip}/{netmask}" + if gw: + net += f",gw={gw}" + else: + net += ",ip=dhcp" + params["net0"] = net + + if start_now: + params["start"] = 1 + + result = _api("POST", f"/nodes/{node}/qemu", data=params) + + out = [f"OK VM #{vmid}: {name} creating!"] + out.append(f" CPU: {cores} cores, RAM: {memory} MB") + out.append(f" Disk: {params['virtio0']}") + if iso: + out.append(f" ISO: {iso}") + out.append( + f" Net: {bridge}, " + f"IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}" + ) + out.append(f" QEMU Agent: {'on' if agent else 'off'}") + if isinstance(result, str) and result: + out.append(f"UPID: {result}") + return "\n".join(out) + + except Exception as e: + return "ERR " + str(e) + + +# ============================================================ +# TEST +# ============================================================ +if __name__ == "__main__": + tools = Tools() + print(tools.connection()) + print() + print(tools.vm_list()) + print() + print(tools.storage())