diff --git a/dev/libvirt-bootstrap.sh b/dev/libvirt-bootstrap.sh
index 8a3cdfe..603b189 100644
--- a/dev/libvirt-bootstrap.sh
+++ b/dev/libvirt-bootstrap.sh
@@ -45,7 +45,7 @@ echowarn() {
# DESCRIPTION: Echo debug information to stdout.
#-------------------------------------------------------------------------------
echodebug() {
- if [ $_ECHO_DEBUG -eq $BS_TRUE ]; then
+ if [ "${_ECHO_DEBUG}" -eq "${BS_TRUE}" ]; then
printf "${BC} * DEBUG${EC}: %s\n" "$@";
fi
}
@@ -154,8 +154,7 @@ __gather_linux_system_info() {
DISTRO_VERSION=""
# Let's test if the lsb_release binary is available
- rv=$(lsb_release >/dev/null 2>&1)
- if [ $? -eq 0 ]; then
+ if lsb_release >/dev/null 2>&1; then
DISTRO_NAME=$(lsb_release -si)
if [ "x$(echo "$DISTRO_NAME" | grep RedHat)" != "x" ]; then
# Let's convert CamelCase to Camel Case
@@ -208,7 +207,7 @@ __gather_linux_system_info() {
;;
arch ) n="Arch Linux" ;;
centos ) n="CentOS" ;;
- almalinux ) n="AlmaLinux" ;;
+ almalinux ) n="AlmaLinux" ;;
debian ) n="Debian" ;;
ubuntu ) n="Ubuntu" ;;
fedora ) n="Fedora" ;;
@@ -248,7 +247,7 @@ __gather_linux_system_info() {
;;
esac
;;
- * ) n="${n}" ;
+ * ) ;;
esac
DISTRO_NAME=$n
DISTRO_VERSION=$v
@@ -779,8 +778,7 @@ if [ "$INSTALL_FUNC" = "null" ]; then
exit 1
else
echoinfo "Running ${INSTALL_FUNC}()"
- $INSTALL_FUNC
- if [ $? -ne 0 ]; then
+ if ! $INSTALL_FUNC; then
echoerror "Failed to run ${INSTALL_FUNC}()!!!"
exit 1
fi
@@ -803,8 +801,7 @@ if [ "$POST_INSTALL_FUNC" = "null" ]; then
exit 1
else
echoinfo "Running ${POST_INSTALL_FUNC}()"
- $POST_INSTALL_FUNC
- if [ $? -ne 0 ]; then
+ if ! $POST_INSTALL_FUNC; then
echoerror "Failed to run ${POST_INSTALL_FUNC}()!!!"
exit 1
fi
@@ -827,8 +824,7 @@ if [ "$DAEMONS_RUNNING_FUNC" = "null" ]; then
exit 1
else
echoinfo "Running ${DAEMONS_RUNNING_FUNC}()"
- $DAEMONS_RUNNING_FUNC
- if [ $? -ne 0 ]; then
+ if ! $DAEMONS_RUNNING_FUNC; then
echoerror "Failed to run ${DAEMONS_RUNNING_FUNC}()!!!"
exit 1
fi
diff --git a/gunicorn.conf.py b/gunicorn.conf.py
index 90bd54e..7fab982 100644
--- a/gunicorn.conf.py
+++ b/gunicorn.conf.py
@@ -75,10 +75,7 @@ backlog = 2048
def get_workers():
procs = os.sysconf('SC_NPROCESSORS_ONLN')
- if procs > 0:
- return procs * 2 + 1
- else:
- return 3
+ return procs * 2 + 1 if procs > 0 else 3
workers = get_workers()
diff --git a/vrtManager/IPy.py b/vrtManager/IPy.py
index 3cddf29..099ac90 100644
--- a/vrtManager/IPy.py
+++ b/vrtManager/IPy.py
@@ -181,10 +181,7 @@ class IPint(object):
if isinstance(data, INT_TYPES):
self.ip = int(data)
if ipversion == 0:
- if self.ip <= MAX_IPV4_ADDRESS:
- ipversion = 4
- else:
- ipversion = 6
+ ipversion = 4 if self.ip <= MAX_IPV4_ADDRESS else 6
if ipversion == 4:
if self.ip > MAX_IPV4_ADDRESS:
raise ValueError("IPv4 Address can't be larger than %x: %x" % (MAX_IPV4_ADDRESS, self.ip))
diff --git a/vrtManager/create.py b/vrtManager/create.py
index 61d5040..99ec7e7 100644
--- a/vrtManager/create.py
+++ b/vrtManager/create.py
@@ -1,4 +1,5 @@
import string
+import contextlib
from vrtManager import util
from vrtManager.connection import wvmConnect
@@ -30,19 +31,13 @@ class wvmCreate(wvmConnect):
"""
Function return all images on all storages
"""
- images = list()
+ images = []
storages = self.get_storages(only_actives=True)
for storage in storages:
stg = self.get_storage(storage)
- try:
+ with contextlib.suppress(Exception):
stg.refresh(0)
- except Exception:
- pass
- for img in stg.listVolumes():
- if img.lower().endswith(".iso"):
- pass
- else:
- images.append(img)
+ images.extend(img for img in stg.listVolumes() if not img.lower().endswith(".iso"))
return images
def get_os_type(self):
@@ -58,10 +53,7 @@ class wvmCreate(wvmConnect):
stg = self.get_storage(storage)
storage_type = util.get_xml_path(stg.XMLDesc(0), "/pool/@type")
if storage_type == "dir":
- if image_format in ("qcow", "qcow2"):
- name += "." + image_format
- else:
- name += ".img"
+ name += f".{image_format}" if image_format in ("qcow", "qcow2") else ".img"
alloc = 0
else:
image_format = 'raw'
@@ -87,28 +79,19 @@ class wvmCreate(wvmConnect):
"""
stg.createXML(xml, metadata)
- try:
+
+ with contextlib.suppress(Exception):
stg.refresh(0)
- except:
- pass
vol = stg.storageVolLookupByName(name)
return vol.path()
def get_volume_format_type(self, path):
vol = self.get_volume_by_path(path)
vol_type = util.get_xml_path(vol.XMLDesc(0), "/volume/target/format/@type")
- if vol_type == "unknown" or vol_type == "iso":
- return "raw"
- if vol_type:
- return vol_type
- else:
- return "raw"
+ return "raw" if vol_type in ["unknown", "iso"] else vol_type or "raw"
def get_volume_path(self, volume, pool=None):
- if not pool:
- storages = self.get_storages(only_actives=True)
- else:
- storages = [pool]
+ storages = [pool] if pool else self.get_storages(only_actives=True)
for storage in storages:
stg = self.get_storage(storage)
if stg.info()[0] != 0:
@@ -124,10 +107,7 @@ class wvmCreate(wvmConnect):
def clone_from_template(self, clone, template, storage=None, metadata=False, disk_owner_uid=0, disk_owner_gid=0):
vol = self.get_volume_by_path(template)
- if not storage:
- stg = vol.storagePoolLookupByVolume()
- else:
- stg = self.get_storage(storage)
+ stg = self.get_storage(storage) if storage else vol.storagePoolLookupByVolume()
storage_type = util.get_xml_path(stg.XMLDesc(0), "/pool/@type")
format = util.get_xml_path(vol.XMLDesc(0), "/volume/target/format/@type")
@@ -226,12 +206,8 @@ class wvmCreate(wvmConnect):
if caps["features"]:
xml += """"""
- if "acpi" in caps["features"]:
- xml += """"""
- if "apic" in caps["features"]:
- xml += """"""
- if "pae" in caps["features"]:
- xml += """"""
+ for feat in [x for x in ("acpi", "apic", "pae",) if x in caps["features"]]:
+ xml += f"""<{feat}/>"""
if firmware.get("secure", "no") == "yes":
xml += """"""
xml += """"""
@@ -240,9 +216,7 @@ class wvmCreate(wvmConnect):
xml += """"""
elif vcpu_mode == "host-passthrough":
xml += """"""
- elif vcpu_mode == "":
- pass
- else:
+ elif vcpu_mode != "":
xml += f"""
{vcpu_mode}"""
xml += """"""
@@ -306,7 +280,7 @@ class wvmCreate(wvmConnect):
xml += """""" % (hd_disk_letters.pop(0), volume.get("bus"))
elif volume.get("bus") == "fdc":
xml += """""" % (fd_disk_letters.pop(0), volume.get("bus"))
- elif volume.get("bus") == "sata" or volume.get("bus") == "scsi":
+ elif volume.get("bus") in ["sata", "scsi"]:
xml += """""" % (sd_disk_letters.pop(0), volume.get("bus"))
else:
xml += """""" % sd_disk_letters.pop(0)
@@ -345,14 +319,13 @@ class wvmCreate(wvmConnect):
if console_pass == "random":
console_pass = "passwd='" + util.randomPasswd() + "'"
- else:
- if not console_pass == "":
- console_pass = "passwd='" + console_pass + "'"
+ elif console_pass != "":
+ console_pass = "passwd='" + console_pass + "'"
if "usb" in dom_caps["disk_bus"]:
- xml += """""".format("virtio" if virtio else "usb")
- xml += """""".format("virtio" if virtio else "usb")
- xml += """""".format("virtio" if virtio else "usb")
+ xml += f""""""
+ xml += f""""""
+ xml += f""""""
else:
xml += """"""
xml += """"""
diff --git a/vrtManager/hostdetails.py b/vrtManager/hostdetails.py
index 2996904..6ee97fb 100644
--- a/vrtManager/hostdetails.py
+++ b/vrtManager/hostdetails.py
@@ -21,12 +21,11 @@ class wvmHostDetails(wvmConnect):
freemem = self.wvm.getMemoryStats(-1, 0)
if isinstance(freemem, dict):
free = (freemem["buffers"] + freemem["free"] + freemem["cached"]) * 1024
- percent = abs(100 - ((free * 100) // all_mem))
+ percent = abs(100 - free * 100 // all_mem)
usage = all_mem - free
- mem_usage = {"total": all_mem, "usage": usage, "percent": percent}
+ return {"total": all_mem, "usage": usage, "percent": percent}
else:
- mem_usage = {"total": None, "usage": None, "percent": None}
- return mem_usage
+ return {"total": None, "usage": None, "percent": None}
def get_cpu_usage(self):
"""
@@ -35,30 +34,30 @@ class wvmHostDetails(wvmConnect):
prev_idle = 0
prev_total = 0
cpu = self.wvm.getCPUStats(-1, 0)
- if isinstance(cpu, dict):
- for num in range(2):
- idle = self.wvm.getCPUStats(-1, 0)["idle"]
- total = sum(self.wvm.getCPUStats(-1, 0).values())
- diff_idle = idle - prev_idle
- diff_total = total - prev_total
- diff_usage = (1000 * (diff_total - diff_idle) / diff_total + 5) / 10
- prev_total = total
- prev_idle = idle
- if num == 0:
- time.sleep(1)
- else:
- if diff_usage < 0:
- diff_usage = 0
- else:
+ if not isinstance(cpu, dict):
return {"usage": None}
+
+ for num in range(2):
+ idle = self.wvm.getCPUStats(-1, 0)["idle"]
+ total = sum(self.wvm.getCPUStats(-1, 0).values())
+ diff_idle = idle - prev_idle
+ diff_total = total - prev_total
+ diff_usage = (1000 * (diff_total - diff_idle) /
+ diff_total + 5) / 10
+ prev_total = total
+ prev_idle = idle
+ if num == 0:
+ time.sleep(1)
+ else:
+ diff_usage = max(diff_usage, 0)
+
return {"usage": diff_usage}
def get_node_info(self):
"""
Function return host server information: hostname, cpu, memory, ...
"""
- info = list()
- info.append(self.wvm.getHostname()) # hostname
+ info = [self.wvm.getHostname()] # hostname
info.append(self.wvm.getInfo()[0]) # architecture
info.append(self.wvm.getInfo()[1] * 1048576) # memory
info.append(self.wvm.getInfo()[2]) # cpu core count
diff --git a/vrtManager/interface.py b/vrtManager/interface.py
index 14ddebc..a2eb777 100644
--- a/vrtManager/interface.py
+++ b/vrtManager/interface.py
@@ -57,7 +57,7 @@ class wvmInterface(wvmConnect):
try:
xml = self._XMLDesc(VIR_INTERFACE_XML_INACTIVE)
return util.get_xml_path(xml, "/interface/start/@mode")
- except:
+ except Exception:
return None
def is_active(self):
@@ -65,10 +65,7 @@ class wvmInterface(wvmConnect):
def get_mac(self):
mac = self.iface.MACString()
- if mac:
- return mac
- else:
- return None
+ return mac or None
def get_type(self):
xml = self._XMLDesc()
@@ -78,11 +75,8 @@ class wvmInterface(wvmConnect):
try:
xml = self._XMLDesc(VIR_INTERFACE_XML_INACTIVE)
ipaddr = util.get_xml_path(xml, "/interface/protocol[@family='ipv4']/ip/@address")
- if ipaddr:
- return "static"
- else:
- return "dhcp"
- except:
+ return "static" if ipaddr else "dhcp"
+ except Exception:
return None
def get_ipv4(self):
@@ -92,17 +86,14 @@ class wvmInterface(wvmConnect):
if not int_ipv4_ip or not int_ipv4_mask:
return None
else:
- return int_ipv4_ip + "/" + int_ipv4_mask
+ return f"{int_ipv4_ip}/{int_ipv4_mask}"
def get_ipv6_type(self):
try:
xml = self._XMLDesc(VIR_INTERFACE_XML_INACTIVE)
ipaddr = util.get_xml_path(xml, "/interface/protocol[@family='ipv6']/ip/@address")
- if ipaddr:
- return "static"
- else:
- return "dhcp"
- except:
+ return "static" if ipaddr else "dhcp"
+ except Exception:
return None
def get_ipv6(self):
@@ -112,40 +103,39 @@ class wvmInterface(wvmConnect):
if not int_ipv6_ip or not int_ipv6_mask:
return None
else:
- return int_ipv6_ip + "/" + int_ipv6_mask
+ return f"{int_ipv6_ip}/{int_ipv6_mask}"
def get_bridge(self):
bridge = None
- if self.get_type() == "bridge":
- bridge = util.get_xml_path(self._XMLDesc(), "/interface/bridge/interface/@name")
- for iface in self.get_bridge_slave_ifaces():
- if iface.get("state") == "up" and iface.get("speed") != "unknown":
- bridge = iface.get("name")
- return bridge
- return bridge
- else:
+ if self.get_type() != "bridge":
return None
+ bridge = util.get_xml_path(self._XMLDesc(), "/interface/bridge/interface/@name")
+ for iface in self.get_bridge_slave_ifaces():
+ if iface.get("state") == "up" and iface.get("speed") != "unknown":
+ bridge = iface.get("name")
+ return bridge
+ return bridge
def get_bridge_slave_ifaces(self):
- ifaces = list()
- if self.get_type() == "bridge":
- tree = ElementTree.fromstring(self._XMLDesc())
- for iface in tree.findall("./bridge/"):
- address = state = speed = None
- name = iface.get("name")
- if_type = iface.get("type")
- link = iface.find("link")
- if link is not None:
- state = link.get("state")
- speed = link.get("speed")
- mac = iface.find("mac")
- if mac is not None:
- address = mac.get("address")
- ifaces.append({"name": name, "type": if_type, "state": state, "speed": speed, "mac": address})
- return ifaces
- else:
+ if self.get_type() != "bridge":
return None
+ ifaces = []
+ tree = ElementTree.fromstring(self._XMLDesc())
+ for iface in tree.findall("./bridge/"):
+ address = state = speed = None
+ name = iface.get("name")
+ if_type = iface.get("type")
+ link = iface.find("link")
+ if link is not None:
+ state = link.get("state")
+ speed = link.get("speed")
+ mac = iface.find("mac")
+ if mac is not None:
+ address = mac.get("address")
+ ifaces.append({"name": name, "type": if_type, "state": state, "speed": speed, "mac": address})
+ return ifaces
+
def get_details(self):
mac = self.get_mac()
itype = self.get_type()
diff --git a/vrtManager/network.py b/vrtManager/network.py
index 9a5a149..45467f8 100644
--- a/vrtManager/network.py
+++ b/vrtManager/network.py
@@ -23,10 +23,8 @@ def network_size(subnet, dhcp=None):
if addr.version() == 6:
mask = mask.lstrip("/") if "/" in mask else mask
dhcp_pool = [IP(addr[0].strCompressed() + hex(256)), IP(addr[0].strCompressed() + hex(512 - 1))]
- if dhcp:
- return gateway, mask, dhcp_pool
- else:
- return gateway, mask, None
+
+ return (gateway, mask, dhcp_pool) if dhcp else (gateway, mask, None)
class wvmNetworks(wvmConnect):
@@ -43,6 +41,7 @@ class wvmNetworks(wvmConnect):
net_forward = util.get_xml_path(net.XMLDesc(0), "/network/forward/@mode")
networks.append({"name": network, "status": net_status, "device": net_bridge, "forward": net_forward})
+
return networks
def define_network(self, xml):
@@ -137,7 +136,7 @@ class wvmNetwork(wvmConnect):
def get_bridge_device(self):
try:
return self.net.bridgeName()
- except:
+ except Exception:
return util.get_xml_path(self._XMLDesc(0), "/network/forward/interface/@dev")
def start(self):
@@ -153,7 +152,7 @@ class wvmNetwork(wvmConnect):
return self.net.update(command, section, parentIndex, xml, flags)
def get_ip_networks(self):
- ip_networks = dict()
+ ip_networks = {}
xml = self._XMLDesc(0)
if util.get_xml_path(xml, "/network/ip") is None:
return ip_networks
@@ -164,10 +163,10 @@ class wvmNetwork(wvmConnect):
netmask_str = ip.get("netmask")
prefix = ip.get("prefix")
family = ip.get("family", "ipv4")
- base = 32 if family == "ipv4" else 128
if prefix:
prefix = int(prefix)
- binstr = (prefix * "1") + ((base - prefix) * "0")
+ base = 32 if family == "ipv4" else 128
+ binstr = prefix * "1" + (base - prefix) * "0"
netmask_str = str(IP(int(binstr, base=2)))
if netmask_str:
@@ -183,8 +182,7 @@ class wvmNetwork(wvmConnect):
def get_network_mac(self):
xml = self._XMLDesc(0)
- mac = util.get_xml_path(xml, "/network/mac/@address")
- return mac
+ return util.get_xml_path(xml, "/network/mac/@address")
def get_network_forward(self):
xml = self._XMLDesc(0)
@@ -201,22 +199,15 @@ class wvmNetwork(wvmConnect):
dhcpstart = util.get_xml_path(xml, "/network/ip[@family='ipv6']/dhcp/range[1]/@start")
dhcpend = util.get_xml_path(xml, "/network/ip[@family='ipv6']/dhcp/range[1]/@end")
- if not dhcpstart or not dhcpend:
- return None
-
- return [IP(dhcpstart), IP(dhcpend)]
+ return None if not dhcpstart or not dhcpend else [IP(dhcpstart), IP(dhcpend)]
def get_dhcp_range_start(self, family="ipv4"):
dhcp = self.get_dhcp_range(family)
- if not dhcp:
- return None
- return dhcp[0]
+ return dhcp[0] if dhcp else None
def get_dhcp_range_end(self, family="ipv4"):
dhcp = self.get_dhcp_range(family)
- if not dhcp:
- return None
- return dhcp[1]
+ return dhcp[1] if dhcp else None
def can_pxe(self):
xml = self._XMLDesc(0)
@@ -226,21 +217,19 @@ class wvmNetwork(wvmConnect):
return bool(util.get_xml_path(xml, "/network/ip/dhcp/bootp/@file"))
def get_dhcp_host_addr(self, family="ipv4"):
- result = list()
+ result = []
tree = etree.fromstring(self._XMLDesc(0))
-
for ipdhcp in tree.findall("./ip"):
if family == "ipv4":
- if ipdhcp.get("family") is None:
- hosts = ipdhcp.findall("./dhcp/host")
- for host in hosts:
- host_ip = host.get("ip")
- mac = host.get("mac")
- name = host.get("name", "")
- result.append({"ip": host_ip, "mac": mac, "name": name})
- return result
- else:
+ if ipdhcp.get("family") is not None:
continue
+ hosts = ipdhcp.findall("./dhcp/host")
+ for host in hosts:
+ host_ip = host.get("ip")
+ mac = host.get("mac")
+ name = host.get("name", "")
+ result.append({"ip": host_ip, "mac": mac, "name": name})
+ return result
if family == "ipv6":
hosts = tree.xpath("./ip[@family='ipv6']/dhcp/host")
for host in hosts:
@@ -272,9 +261,9 @@ class wvmNetwork(wvmConnect):
for h in hosts:
if h.get("ip") == ip:
if family == "ipv4":
- new_xml = ''.format(h.get("mac"), h.get("name"), ip)
+ new_xml = f''
if family == "ipv6":
- new_xml = ''.format(h.get("id"), h.get("name"), ip)
+ new_xml = f''
self.update(
VIR_NETWORK_UPDATE_COMMAND_DELETE,
@@ -298,12 +287,7 @@ class wvmNetwork(wvmConnect):
compare_var = "id"
parent_index = self.parent_count - 1
new_host_xml = etree.fromstring(new_xml)
-
- host = None
- for h in hosts:
- if h.get(compare_var) == mac_duid:
- host = h
- break
+ host = next((h for h in hosts if h.get(compare_var) == mac_duid), None)
if host is None:
self.update(
VIR_NETWORK_UPDATE_COMMAND_ADD_LAST,
@@ -326,7 +310,7 @@ class wvmNetwork(wvmConnect):
)
def get_qos(self):
- qos_values = dict()
+ qos_values = {}
tree = etree.fromstring(self._XMLDesc(0))
qos = tree.xpath("/network/bandwidth")
if qos:
@@ -348,13 +332,10 @@ class wvmNetwork(wvmConnect):
return qos_values
def set_qos(self, direction, average, peak, burst):
- if direction == "inbound":
- xml = f""
- elif direction == "outbound":
- xml = f""
- else:
+ if direction not in ("inbound","outbound"):
raise Exception("Direction must be inbound or outbound")
+ xml = f"<{direction} average='{average}' peak='{peak}' burst='{burst}'/>"
tree = etree.fromstring(self._XMLDesc(0))
band = tree.xpath("/network/bandwidth")
@@ -388,7 +369,7 @@ class wvmNetwork(wvmConnect):
self.leases = self.net.DHCPLeases()
except Exception as e:
self.leases = []
- raise "Error getting {} DHCP leases: {}".format(self, e)
+ raise f"Error getting {self} DHCP leases: {e}" from e
def get_dhcp_leases(self):
if self.leases is None:
diff --git a/vrtManager/nwfilters.py b/vrtManager/nwfilters.py
index 627e571..8b8d1b9 100644
--- a/vrtManager/nwfilters.py
+++ b/vrtManager/nwfilters.py
@@ -47,11 +47,8 @@ class wvmNWFilter(wvmConnect):
return ElementTree.tostring(tree).decode()
def get_filter_refs(self):
- refs = []
tree = ElementTree.fromstring(self._XMLDesc(0))
- for ref in tree.findall("./filterref"):
- refs.append(ref.get("filter"))
- return refs
+ return [ref.get("filter") for ref in tree.findall("./filterref")]
def get_rules(self):
rules = []
diff --git a/vrtManager/storage.py b/vrtManager/storage.py
index c7b37f8..d2346ce 100644
--- a/vrtManager/storage.py
+++ b/vrtManager/storage.py
@@ -1,3 +1,5 @@
+import contextlib
+
from vrtManager import util
from vrtManager.connection import wvmConnect
@@ -10,10 +12,7 @@ class wvmStorages(wvmConnect):
stg = self.get_storage(pool)
stg_status = stg.isActive()
stg_type = util.get_xml_path(stg.XMLDesc(0), "/pool/@type")
- if stg_status:
- stg_vol = len(stg.listVolumes())
- else:
- stg_vol = None
+ stg_vol = len(stg.listVolumes()) if stg_status else None
stg_size = stg.info()[1]
storages.append(
{"name": pool, "status": stg_status, "type": stg_type, "volumes": stg_vol, "size": stg_size}
@@ -33,7 +32,7 @@ class wvmStorages(wvmConnect):
"""
if stg_type == "logical":
- target = "/dev/" + name
+ target = f"/dev/{name}"
xml += f"""
{target}
@@ -228,23 +227,12 @@ class wvmStorage(wvmConnect):
self.pool.refresh(0)
def update_volumes(self):
- try:
+ with contextlib.suppress(Exception):
self.refresh()
- except Exception:
- pass
vols = self.get_volumes()
- vol_list = []
-
- for volname in vols:
- vol_list.append(
- {
- "name": volname,
- "size": self.get_volume_size(volname),
- "allocation": self.get_volume_allocation(volname),
- "type": self.get_volume_format_type(volname),
- }
- )
- return vol_list
+ return [{"name": volname, "size": self.get_volume_size(volname),
+ "allocation": self.get_volume_allocation(volname),
+ "type": self.get_volume_format_type(volname)} for volname in vols]
def create_volume(self, name, size, vol_fmt="qcow2", metadata=False, disk_owner_uid=0, disk_owner_gid=0):
size = int(size) * 1073741824
@@ -253,10 +241,7 @@ class wvmStorage(wvmConnect):
if vol_fmt == "unknown":
vol_fmt = "raw"
if storage_type == "dir":
- if vol_fmt in ("qcow", "qcow2"):
- name += "." + vol_fmt
- else:
- name += ".img"
+ name += f".{vol_fmt}" if vol_fmt in ("qcow", "qcow2") else ".img"
alloc = 0
xml = f"""
@@ -300,9 +285,9 @@ class wvmStorage(wvmConnect):
storage_type = self.get_type()
if storage_type == "dir":
if vol_fmt in ["qcow", "qcow2"]:
- target_file += "." + vol_fmt
+ target_file += f".{vol_fmt}"
else:
- suffix = "." + file_suffix
+ suffix = f".{file_suffix}"
target_file += suffix if len(suffix) > 1 else ""
xml = f"""
diff --git a/vrtManager/util.py b/vrtManager/util.py
index 1f15b37..d618896 100644
--- a/vrtManager/util.py
+++ b/vrtManager/util.py
@@ -9,10 +9,7 @@ import lxml.etree as etree
def is_kvm_available(xml):
kvm_domains = get_xml_path(xml, "//domain/@type='kvm'")
- if kvm_domains > 0:
- return True
- else:
- return False
+ return kvm_domains > 0
def randomMAC():
@@ -26,40 +23,40 @@ def randomMAC():
def randomUUID():
"""Generate a random UUID."""
- u = [secrets.randbelow(256) for ignore in range(0, 16)]
- u[6] = (u[6] & 0x0F) | (4 << 4)
- u[8] = (u[8] & 0x3F) | (2 << 6)
+ u = [secrets.randbelow(256) for _ in range(16)]
+ u[6] = u[6] & 15 | 4 << 4
+ u[8] = u[8] & 63 | 2 << 6
return "-".join(["%02x" * 4, "%02x" * 2, "%02x" * 2, "%02x" * 2, "%02x" * 6]) % tuple(u)
def randomPasswd(length=12, alphabet=string.ascii_letters + string.digits):
"""Generate a random password"""
- return "".join([secrets.choice(alphabet) for i in range(length)])
+ return "".join([secrets.choice(alphabet) for _ in range(length)])
-def get_max_vcpus(conn, type=None):
+def get_max_vcpus(conn, guest_type=None):
"""@param conn: libvirt connection to poll for max possible vcpus
- @type type: optional guest type (kvm, etc.)"""
+ @param guest_type: optional guest type (kvm, etc.)"""
if type is None:
- type = conn.getType()
+ guest_type = conn.getType()
try:
- m = conn.getMaxVcpus(type.lower())
+ m = conn.getMaxVcpus(guest_type.lower())
except libvirt.libvirtError:
m = 32
return m
-def xml_escape(str):
+def xml_escape(xml_str):
"""Replaces chars ' " < > & with xml safe counterparts"""
- if str is None:
+ if xml_str is None:
return None
- str = str.replace("&", "&")
- str = str.replace("'", "'")
- str = str.replace('"', """)
- str = str.replace("<", "<")
- str = str.replace(">", ">")
- return str
+ xml_str = xml_str.replace("&", "&")
+ xml_str = xml_str.replace("'", "'")
+ xml_str = xml_str.replace('"', """)
+ xml_str = xml_str.replace("<", "<")
+ xml_str = xml_str.replace(">", ">")
+ return xml_str
def compareMAC(p, q):
@@ -108,10 +105,7 @@ def get_xpath(doc, path):
if ret is not None:
if isinstance(ret, list):
if len(ret) >= 1:
- if hasattr(ret[0], "text"):
- result = ret[0].text
- else:
- result = ret[0]
+ result = ret[0].text if hasattr(ret[0], "text") else ret[0]
else:
result = ret
diff --git a/webvirtcloud/common_tags.py b/webvirtcloud/common_tags.py
index 7197800..4b85df7 100644
--- a/webvirtcloud/common_tags.py
+++ b/webvirtcloud/common_tags.py
@@ -7,28 +7,20 @@ register = template.Library()
@register.simple_tag
def app_active(request, app_name):
- if request.resolver_match.app_name == app_name:
- return "active"
- return ""
+ return "active" if request.resolver_match.app_name == app_name else ""
@register.simple_tag
def view_active(request, view_name):
- if request.resolver_match.view_name == view_name:
- return "active"
- return ""
+ return "active" if request.resolver_match.view_name == view_name else ""
@register.simple_tag
def class_active(request, pattern):
- if re.search(pattern, request.path):
- # Not sure why 'class="active"' returns class=""active""
- return "active"
- return ""
+ # Not sure why 'class="active"' returns class=""active""
+ return "active" if re.search(pattern, request.path) else ""
@register.simple_tag
def has_perm(user, permission_codename):
- if user.has_perm(permission_codename):
- return True
- return False
+ return bool(user.has_perm(permission_codename))
diff --git a/webvirtcloud/ldapbackend.py b/webvirtcloud/ldapbackend.py
index 15c0834..0f7a797 100644
--- a/webvirtcloud/ldapbackend.py
+++ b/webvirtcloud/ldapbackend.py
@@ -42,48 +42,49 @@ try:
def authenticate(self, request, username=None, password=None, **kwargs):
if not settings.LDAP_ENABLED:
- return None
+ return None
print("authenticate_ldap")
# Get the user information from the LDAP if he can be authenticated
isAdmin = False
isStaff = False
isTechnician = False
-
+
requeteLdap = self.get_LDAP_user(username, password, settings.LDAP_SEARCH_GROUP_FILTER_ADMINS)
+ isAdmin = requeteLdap is not None
+ isStaff = requeteLdap is not None
+
if requeteLdap is None:
requeteLdap = self.get_LDAP_user(username, password, settings.LDAP_SEARCH_GROUP_FILTER_STAFF)
- if requeteLdap is None:
- requeteLdap = self.get_LDAP_user(username, password, settings.LDAP_SEARCH_GROUP_FILTER_TECHNICIANS)
- if requeteLdap is None:
- requeteLdap = self.get_LDAP_user(username, password, settings.LDAP_SEARCH_GROUP_FILTER_USERS)
- if requeteLdap is None:
- print("User does not belong to any search group. Check LDAP_SEARCH_GROUP_FILTER in settings.")
- return None
- else:
- isTechnician = True
- else:
- isStaff = True
- else:
- isAdmin = True
- isStaff = True
+ isStaff = requeteLdap is not None
+
+ if requeteLdap is None:
+ requeteLdap = self.get_LDAP_user(username, password, settings.LDAP_SEARCH_GROUP_FILTER_TECHNICIANS)
+ isTechnician = requeteLdap is not None
+
+ if requeteLdap is None:
+ requeteLdap = self.get_LDAP_user(username, password, settings.LDAP_SEARCH_GROUP_FILTER_USERS)
+
+ if requeteLdap is None:
+ print("User does not belong to any search group. Check LDAP_SEARCH_GROUP_FILTER in settings.")
+ return None
techniciansGroup = Group.objects.get(name='Technicians')
-
+
try:
user = User.objects.get(username=username)
attributes = UserAttributes.objects.get(user=user)
user.is_staff = isStaff
user.is_superuser = isAdmin
- if isTechnician is False and user.groups.filter(name='Technicians').exists():
+ if not isTechnician and user.groups.filter(name='Technicians').exists():
user.groups.remove(techniciansGroup)
- elif isTechnician is True and user.groups.filter(name='Technicians').exists() is False:
+ elif isTechnician and not user.groups.filter(name='Technicians').exists():
user.groups.add(techniciansGroup)
else:
print("The user is already in the Technicians group")
user.save()
# TODO VERIFY
except User.DoesNotExist:
- print("authenticate-create new user: {}".format(username))
+ print(f"authenticate-create new user: {username}")
user = User(username=username)
user.first_name = requeteLdap[1]
user.last_name = requeteLdap[2]
@@ -93,7 +94,7 @@ try:
user.is_superuser = isAdmin
user.set_password(uuid.uuid4().hex)
user.save()
- if isTechnician is True:
+ if isTechnician:
user.groups.add(techniciansGroup)
maxInstances = 1
maxCpus = 1