""" Various utils """ from . import settings import uuid import subprocess import sys import asyncio import functools class NumberedClass: """ A class that counts its current instance number """ next_id = 1 # Count from 1: IP addresses and id 0 are not a good mix. @classmethod def get_id(cls): out = cls.next_id cls.next_id += 1 return out def __init__(self): self.id = self.get_id() class LibvirtObject(NumberedClass): """ A class that has the basic attributes of a libvirt objects: * id * uuid and the applicable restrictions """ class TooMany(Exception): def __init__(self, count): self.count = count def __str__(self): return ( "Limit number reached. The current instance #{} does not fit in IPv4" ).format(self.count) def __init__(self, conn): super().__init__() self.conn = conn if self.id > 250: raise self.TooMany(self.id) self.uuid = uuid.uuid4() class MACAddress: """ A MAC address for a NIC or bridge """ def __init__(self, link_id, dev_id=None): """ link_id: id of the current link dev_id: id of the current NIC. If the device is a bridge, use None """ if dev_id is None: dev_id = 0xFE self.link_id = link_id self.dev_id = dev_id def __str__(self): return "52:54:00:{nwid:02x}:{link_id:02x}:{dev_id:02x}".format( nwid=settings.NETWORK_ID, link_id=self.link_id, dev_id=self.dev_id ) def __repr__(self): return str(self) class Addrv4: """ An address in IPv4 """ def __init__(self, link_id, dev_id=None, host_address=False): """ link_id: id of the current link dev_id: id of the current NIC. Use None to get the base address host_address: if True and dev_id is None, returns the host machine address on this network instead of the base address """ self.host_address = host_address if dev_id is None: if host_address: dev_id = 0xFE else: dev_id = 0 self.link_id = link_id self.dev_id = dev_id self.netmask = "255.255.255.0" self.prefix = 24 def __str__(self): return "{base_range}.{link_id}.{dev_id}".format( base_range=settings.IPV4_RANGE, link_id=self.link_id, dev_id=self.dev_id, ) def __repr__(self): return str(self) class Addrv6: """ An address in IPv6 """ def __init__(self, link_id, dev_id=None, host_address=False): """ link_id: id of the current link dev_id: id of the current NIC. Use None to get the base address host_address: if True and dev_id is None, returns the host machine address on this network instead of the base address """ self.host_address = host_address if dev_id is None: if host_address: dev_id = 0xFE else: dev_id = 0 self.link_id = link_id self.dev_id = dev_id self.prefix = 64 def __str__(self): return "{base_range}:{link_id:04x}::{dev_id:04x}".format( base_range=settings.IPV6_RANGE, link_id=self.link_id, dev_id=self.dev_id, ) def run_cmd_retry(command, *args, **kwargs): rc = subprocess.run(command, *args, **kwargs) while rc.returncode != 0: print("Command failed. Try again:", file=sys.stderr) rc = subprocess.run(command, *args, **kwargs) return rc def run_in_executor(f): """ Decorator: transforms a blocking function into an awaitable one """ @functools.wraps(f) def inner(*args, **kwargs): loop = asyncio.get_running_loop() return loop.run_in_executor(None, functools.partial(f, *args, **kwargs)) return inner