Fix some linting issues.

Also, changing workflow for image downloading fails.

README.rst

@@ -36,10 +36,13 @@ Tested distros
   - 18.04
   - 20.04
   - 22.04
+  - 24.04
 - Fedora
   - 37
   - 38
   - 39
+  - 40
+  - 41
 - Centos Stream
   - 8
   - 9
@@ -47,6 +50,7 @@ Tested distros
   - 10 (buster)
   - 11 (bullseye)
   - 12 (bookworm)
+  - 13 (trixie) - prerelease
 
 There is possibility to use whatever OS image which supports cloud-init. Use
 the ``--image`` param for ``create`` command to pass image filename, although

box.py

@@ -18,12 +18,10 @@ import requests
 import yaml
 
 
-__version__ = "1.9.2"
+__version__ = "1.11.0"
 
 CACHE_DIR = os.environ.get('XDG_CACHE_HOME', os.path.expanduser('~/.cache'))
 CLOUD_IMAGE = "ci.iso"
-FEDORA_RELEASE_MAP = {'37': '1.7', '38': '1.6', '39': '1.5'}
-DEBIAN_CODENAME_MAP = {'12': 'bookworm', '11': 'bullseye', '10': 'buster'}
 TYPE_MAP = {'HardDisk': 'disk', 'DVD': 'dvd', 'Floppy': 'floppy'}
 DISTRO_MAP = {'ubuntu': 'Ubuntu', 'fedora': 'Fedora',
               'centos': 'Centos Stream', 'debian': 'Debian'}
@@ -220,8 +218,22 @@ _boxpy() {
             fi
             ;;
         stop)
+            items=(--poweroff)
             if [[ ${prev} == ${cmd} ]]; then
-                _vms_comp runningvms
+                if [[ ${cmd} = "stop" ]]; then
+                    _vms_comp runningvms
+                else
+                    COMPREPLY=( $(compgen -W "${items[*]}" -- ${cur}) )
+                fi
+            else
+                _get_excluded_items "${items[@]}"
+                COMPREPLY=( $(compgen -W "$result" -- ${cur}) )
+
+                case "${prev}" in
+                    --*)
+                        COMPREPLY=( )
+                        ;;
+                esac
             fi
             ;;
     esac
@@ -263,28 +275,28 @@ class Run:
     Helper class on subprocess.run()
     command is a list with command and its params to execute
     """
-    def __init__(self, command, capture_output=True):
-        result = subprocess.run(command, encoding='utf-8',
-                                capture_output=capture_output)
-        if result.stdout:
-            LOG.debug2(result.stdout)
-        if result.stderr:
-            LOG.debug2(result.stderr)
+    def __init__(self, command):
+        result = subprocess.run(command, encoding='utf-8', capture_output=True)
 
         self.returncode = result.returncode
         self.stdout = result.stdout.strip() if result.stdout else ''
         self.stderr = result.stderr.strip() if result.stderr else ''
 
+        if self.stdout:
+            LOG.debug2(self.stdout)
+        if self.stderr:
+            LOG.debug2(self.stderr)
+
 
 class BoxError(Exception):
     pass
 
 
-class BoxNotFound(BoxError):
+class BoxNotFoundError(BoxError):
     pass
 
 
-class BoxVBoxFailure(BoxError):
+class BoxVBoxError(BoxError):
     pass
 
 
@@ -512,7 +524,7 @@ class Config:
     def _read_filename(self, fname):
         fullpath = os.path.expanduser(os.path.expandvars(fname))
         if not os.path.exists(fullpath):
-            return
+            return None
 
         with open(fname) as fobj:
             return fobj.read()
@@ -621,11 +633,11 @@ class OsTypes:
 
     def ubuntu(self):
         lts = ''
-        major, minor = [int(x) for x in self._conf.version.split('.')]
+        major, minor = (int(x) for x in self._conf.version.split('.'))
 
         if major % 2 == 0 and minor == 4:
             lts = '_LTS'
-        name = "Ubuntu%s%s_64" % (major, lts)
+        name = f"Ubuntu{major}{lts}_64"
 
         if name not in self._ostypes:
             return 'Ubuntu_64'
@@ -636,7 +648,7 @@ class OsTypes:
         return "Fedora_64"
 
     def debian(self):
-        name = "Debian%s_64" % self._conf.version
+        name = f"Debian{self._conf.version}_64"
         if name not in self._ostypes:
             return 'Debian_64'
 
@@ -662,7 +674,7 @@ class VBoxManage:
     def get_vm_base_path(self):
         path = self._get_vm_config()
         if not path:
-            return
+            return None
 
         return os.path.dirname(path)
 
@@ -670,12 +682,12 @@ class VBoxManage:
         path = self._get_vm_config()
         if not path:
             LOG.warning('Configuration for "%s" not found', self.name_or_uuid)
-            return
+            return None
 
         dom = xml.dom.minidom.parse(path)
         if len(dom.getElementsByTagName('HardDisk')) != 1:
             # don't know what to do with multiple discs
-            raise BoxError()
+            raise BoxError
 
         disk = dom.getElementsByTagName('HardDisk')[0]
         location = disk.getAttribute('location')
@@ -697,6 +709,7 @@ class VBoxManage:
                     return line
 
                 return line.split(' ')[0].strip()
+        return None
 
     def get_vm_info(self):
         out = Run(['vboxmanage', 'showvminfo', self.name_or_uuid])
@@ -797,6 +810,7 @@ class VBoxManage:
                 '--delete']).returncode != 0:
             LOG.fatal('Removing VM "%s" failed', self.name_or_uuid)
             return 7
+        return None
 
     def create(self, conf):
         memory = convert_to_mega(conf.memory)
@@ -818,7 +832,8 @@ class VBoxManage:
                 self.uuid = line.split('UUID:')[1].strip()
 
         if not self.uuid:
-            raise BoxVBoxFailure(f'Cannot create VM "{self.name_or_uuid}".')
+            msg = f'Cannot create VM "{self.name_or_uuid}".'
+            raise BoxVBoxError(msg)
 
         port = conf.port if conf.port else self._find_unused_port()
 
@@ -841,14 +856,15 @@ class VBoxManage:
 
         if Run(cmd).returncode != 0:
             LOG.fatal(f'Cannot modify VM "{self.name_or_uuid}"')
-            raise BoxVBoxFailure()
+            raise BoxVBoxError
 
-        if conf.disable_nested == 'False':
-            if Run(['vboxmanage', 'modifyvm', self.name_or_uuid,
-                    '--nested-hw-virt', 'on']).returncode != 0:
-                LOG.fatal(f'Cannot set nested virtualization for VM '
-                          f'"{self.name_or_uuid}"')
-                raise BoxVBoxFailure()
+        if conf.disable_nested == 'False' and Run(['vboxmanage', 'modifyvm',
+                                                   self.name_or_uuid,
+                                                   '--nested-hw-virt',
+                                                   'on']).returncode != 0:
+            LOG.fatal(f'Cannot set nested virtualization for VM '
+                      f'"{self.name_or_uuid}"')
+            raise BoxVBoxError
 
         return self.uuid
 
@@ -884,7 +900,7 @@ class VBoxManage:
         if Run(['vboxmanage', 'modifymedium', 'disk', src, '--resize',
                 str(size), '--move', fullpath]).returncode != 0:
             LOG.fatal('Resizing and moving image %s has failed', dst)
-            raise BoxVBoxFailure()
+            raise BoxVBoxError
         return fullpath
 
     def storageattach(self, controller_name, port, type_, image):
@@ -908,7 +924,7 @@ class VBoxManage:
         if Run(['vboxmanage', 'startvm', self.name_or_uuid, '--type',
                 type_]).returncode != 0:
             LOG.fatal('Failed to start: %s', self.name_or_uuid)
-            raise BoxVBoxFailure()
+            raise BoxVBoxError
 
     def setextradata(self, key, val):
         res = Run(['vboxmanage', 'setextradata', self.name_or_uuid, key, val])
@@ -922,7 +938,7 @@ class VBoxManage:
         if Run(['vboxmanage', 'modifyvm', self.name_or_uuid, f'--{nic}',
                 kind]).returncode != 0:
             LOG.fatal('Cannot modify VM "%s"', self.name_or_uuid)
-            raise BoxVBoxFailure()
+            raise BoxVBoxError
 
     def is_port_in_use(self, port):
         used_ports = self._get_defined_ports()
@@ -984,7 +1000,7 @@ class Image:
     IMG = ""
     CHECKSUMTOOL = 'sha256sum'
 
-    def __init__(self, vbox, version, arch, release, fname=None):
+    def __init__(self, vbox, version, arch, fname=None):
         self.vbox = vbox
         self._tmp = tempfile.mkdtemp(prefix='boxpy_')
         self._img_fname = fname
@@ -1025,6 +1041,7 @@ class Image:
             return False
 
         LOG.info('Calculating checksum for "%s"', self._img_fname)
+        LOG.debug('Checksum file: "%s"', self._checksum_file)
         fname = os.path.join(self._tmp, self._checksum_file)
         expected_sum = self._get_checksum(fname)
 
@@ -1047,9 +1064,11 @@ class Image:
             return True
 
         fname = os.path.join(CACHE_DIR, self._img_fname)
-        LOG.header('Downloading image %s from %s', self._img_fname,
-                   self._img_url)
-        Run(['wget', '-q', self._img_url, '-O', fname])
+        LOG.header('Downloading image %s', self._img_url)
+        result = Run(['wget', '-q', self._img_url, '-O', fname])
+        if result.returncode != 0:
+            LOG.fatal("Error downloading image %s", self._img_url)
+            return False
 
         if not self._checksum():
             # TODO: make some retry mechanism?
@@ -1060,15 +1079,15 @@ class Image:
         return True
 
     def _get_checksum(self, fname):
-        raise NotImplementedError()
+        raise NotImplementedError
 
 
 class Ubuntu(Image):
     URL = "https://cloud-images.ubuntu.com/releases/%s/release/%s"
     IMG = "ubuntu-%s-server-cloudimg-%s.img"
 
-    def __init__(self, vbox, version, arch, release, fname=None):
-        super().__init__(vbox, version, arch, release)
+    def __init__(self, vbox, version, arch, fname=None):
+        super().__init__(vbox, version, arch)
         self._img_fname = self.IMG % (version, arch)
         self._img_url = self.URL % (version, self._img_fname)
         self._checksum_file = 'SHA256SUMS'
@@ -1090,13 +1109,18 @@ class Debian(Image):
     URL = "https://cloud.debian.org/images/cloud/%s/daily/latest/%s"
     IMG = "debian-%s-generic-%s-daily.qcow2"
     CHECKSUMTOOL = 'sha512sum'
+    CODENAME_MAP = {'13': 'trixie',
+                    '12': 'bookworm',
+                    '11': 'bullseye',
+                    '10': 'buster'}
 
-    def __init__(self, vbox, version, arch, release, fname=None):
-        super().__init__(vbox, version, arch, release)
+    def __init__(self, vbox, version, arch, fname=None):
+        super().__init__(vbox, version, arch)
+        codename = self.CODENAME_MAP[version]
         self._img_fname = self.IMG % (version, arch)
-        self._img_url = self.URL % (release, self._img_fname)
+        self._img_url = self.URL % (codename, self._img_fname)
         self._checksum_file = 'SHA512SUMS'
-        self._checksum_url = self.URL % (release, self._checksum_file)
+        self._checksum_url = self.URL % (codename, self._checksum_file)
 
     def _get_checksum(self, fname):
         expected_sum = None
@@ -1115,12 +1139,30 @@ class Fedora(Image):
            "Cloud/%s/images/%s")
     IMG = "Fedora-Cloud-Base-%s-%s.%s.qcow2"
     CHKS = "Fedora-Cloud-%s-%s-%s-CHECKSUM"
+    REVISION = {'37': '1.7',
+                '38': '1.6',
+                '39': '1.5',
+                '40': '1.14',
+                '41': '1.4'}
 
-    def __init__(self, vbox, version, arch, release, fname=None):
-        super().__init__(vbox, version, arch, release)
-        self._img_fname = self.IMG % (version, release, arch)
+    def __init__(self, vbox, version, arch, fname=None):
+        super().__init__(vbox, version, arch)
+        revision = self.REVISION[version]
+        if int(version) >= 40:
+            if int(version) == 40:
+                # Started from Fedora 40 there is "Generic" in the image names.
+                self.IMG = "Fedora-Cloud-Base-Generic.%s-%s-%s.qcow2"
+                # Fedora 40 have messed up position of the items in filename.
+                self._img_fname = self.IMG % (arch, version, revision)
+            else:
+                # But in Fedora 41 there is no dot between Generic and
+                # version, but between version and arch.
+                self.IMG = "Fedora-Cloud-Base-Generic-%s-%s.%s.qcow2"
+                self._img_fname = self.IMG % (version, revision, arch)
+        else:
+            self._img_fname = self.IMG % (version, revision, arch)
         self._img_url = self.URL % (version, arch, self._img_fname)
-        self._checksum_file = self.CHKS % (version, release, arch)
+        self._checksum_file = self.CHKS % (version, revision, arch)
         self._checksum_url = self.URL % (version, arch, self._checksum_file)
 
     def _get_checksum(self, fname):
@@ -1144,8 +1186,7 @@ class CentosStream(Image):
 
     def __init__(self, vbox, version, arch, release, fname=None):
         super().__init__(vbox, version, arch, release)
-        self._checksum_file = '%s-centos-stream-%s-%s' % (self.CHKS, version,
-                                                          arch)
+        self._checksum_file = f'{self.CHKS}-centos-stream-{version}-{arch}'
         self._checksum_url = self.URL % (version, arch, self.CHKS)
         # there is assumption, that we always need latest relese for specific
         # version and architecture.
@@ -1172,6 +1213,7 @@ class CentosStream(Image):
         images.reverse()
         if images:
             return images[0]
+        return None
 
     def _get_checksum(self, fname):
         expected_sum = None
@@ -1198,12 +1240,12 @@ DISTROS = {'ubuntu': {'username': 'ubuntu',
                       'realname': 'ubuntu',
                       'img_class': Ubuntu,
                       'amd64': 'amd64',
-                      'default_version': '22.04'},
+                      'default_version': '24.04'},
            'fedora': {'username': 'fedora',
                       'realname': 'fedora',
                       'img_class': Fedora,
                       'amd64': 'x86_64',
-                      'default_version': '39'},
+                      'default_version': '41'},
            'centos': {'username': 'centos',
                       'realname': 'centos',
                       'img_class': CentosStream,
@@ -1217,13 +1259,8 @@ DISTROS = {'ubuntu': {'username': 'ubuntu',
 
 
 def get_image_object(vbox, version, image='ubuntu', arch='amd64'):
-    release = None
-    if image == 'fedora':
-        release = FEDORA_RELEASE_MAP[version]
-    if image == 'debian':
-        release = DEBIAN_CODENAME_MAP[version]
     return DISTROS[image]['img_class'](vbox, version, DISTROS[image]['amd64'],
-                                       release, DISTROS[image].get('image'))
+                                       DISTROS[image].get('image'))
 
 
 class IsoImage:
@@ -1305,9 +1342,8 @@ def vmcreate(args, conf=None):
         if not vbox.setextradata(key, getattr(conf, key)):
             return 5
 
-    if conf.user_data:
-        if not vbox.setextradata('user_data', conf.user_data):
-            return 6
+    if conf.user_data and not vbox.setextradata('user_data', conf.user_data):
+        return 6
 
     if not vbox.setextradata('creator', 'boxpy'):
         return 13
@@ -1413,24 +1449,21 @@ def vmcreate(args, conf=None):
 
     # reread config to update fields
     conf = Config(args, vbox)
-    username = DISTROS[conf.distro]["username"]
-    LOG.info('You can access your VM by issuing:')
     if conf.username and conf.username != username:
-        LOG.info(f'ssh -p {conf.port} -i {conf.ssh_key_path[:-4]} '
-                 f'{conf.username}@localhost')
+        username = conf.username
     else:
-        LOG.info(f'ssh -p {conf.port} -i {conf.ssh_key_path[:-4]} '
-                 f'{username}@localhost')
+        username = DISTROS[conf.distro]["username"]
+    LOG.info('You can access your VM by issuing:')
+    LOG.info('ssh -p %s -i %s %s@localhost', conf.port, conf.ssh_key_path[:-4],
+             username)
     LOG.info('or simply:')
-    LOG.info(f'boxpy ssh {conf.name}')
+    LOG.info('boxpy ssh %s', conf.name)
     return 0
 
 
 def vmdestroy(args):
-    if isinstance(args.name, list):
-        vm_names = args.name
-    else:
-        vm_names = [args.name]
+
+    vm_names = args.name if isinstance(args.name, list) else [args.name]
 
     for name in vm_names:
         vbox = VBoxManage(name)
@@ -1524,7 +1557,8 @@ def vminfo(args):
             LOG.info(line)
 
     if 'user_data' in info:
-        LOG.info(f'User data file path:\t{info["user_data"]}')
+        LOG.info('User data file path:\t%s', info['user_data'])
+    return 0
 
 
 def vmrebuild(args):
@@ -1532,12 +1566,12 @@ def vmrebuild(args):
     if not vbox.get_vm_info():
         LOG.fatal(f'Cannot rebuild VM "{args.name}" - it doesn\'t exists.')
         return 20
-    else:
-        LOG.header('Rebuilding VM: %s', args.name)
+
+    LOG.header('Rebuilding VM: %s', args.name)
 
     try:
         conf = Config(args, vbox)
-    except BoxNotFound as ex:
+    except BoxNotFoundError as ex:
         LOG.fatal(f'Error with parsing config: {ex}')
         return 8
     except yaml.YAMLError:
@@ -1586,7 +1620,7 @@ def connect(args):
 
     try:
         conf = Config(args, vbox)
-    except BoxNotFound:
+    except BoxNotFoundError:
         return 11
     except yaml.YAMLError:
         LOG.fatal(f'Cannot read or parse file `{args.config}` as YAML '
@@ -1604,7 +1638,7 @@ def connect(args):
     return Run(cmd, False).returncode
 
 
-def _set_vmstate(name, state, guitype=None):
+def _set_vmstate(name, state, guitype=None, poweroff=False):
 
     vbox = VBoxManage(name)
     if not vbox.get_vm_info():
@@ -1612,25 +1646,28 @@ def _set_vmstate(name, state, guitype=None):
         return 20
 
     if vbox.running and state == "start":
-        LOG.info(f'VM "{name}" is already running.')
-        return
+        LOG.info('VM "%s" is already running.', name)
+        return 1
 
     if not vbox.running and state == "stop":
-        LOG.info(f'VM "{name}" is already stopped.')
-        return
+        LOG.info('VM "%s" is already stopped.', name)
+        return 1
 
     if state == "start":
         vbox.poweron(guitype)
+    elif poweroff:
+        vbox.poweroff()
     else:
         vbox.acpipowerbutton()
+    return 0
 
 
 def vmstart(args):
-    _set_vmstate(args.name, 'start', args.type)
+    _set_vmstate(args.name, 'start', guitype=args.type)
 
 
 def vmstop(args):
-    _set_vmstate(args.name, 'stop')
+    _set_vmstate(args.name, 'stop', poweroff=args.poweroff)
 
 
 def main():
@@ -1748,6 +1785,8 @@ def main():
 
     stop = subparsers.add_parser('stop', help='stop VM')
     stop.add_argument('name', help='name or UUID of the VM')
+    stop.add_argument('-p', '--poweroff', action='store_true', help='poweroff '
+                      'machine instead of using ACPI power signal')
     stop.set_defaults(func=vmstop)
 
     completion = subparsers.add_parser('completion', help='generate shell '
@@ -1773,7 +1812,7 @@ def main():
     LOG.set_verbose(args.verbose, args.quiet)
 
     if 'func' not in args and args.version:
-        LOG.info(f'boxpy {__version__}')
+        LOG.info('boxpy %s', __version__)
         parser.exit()
 
     if hasattr(args, 'func'):
@@ -1781,6 +1820,7 @@ def main():
 
     parser.print_help()
     parser.exit()
+    return 23
 
 
 if __name__ == '__main__':

pyproject.toml

@@ -24,6 +24,7 @@ classifiers = [
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3 :: Only"
 ]
 dependencies = [
@@ -45,3 +46,25 @@ version = {attr = "box.__version__"}
 
 [tool.distutils.bdist_wheel]
 universal = true
+
+[tool.ruff]
+line-length = 79
+indent-width = 4
+
+[tool.ruff.lint]
+extend-select = [
+    "B",  # flake8-bugbear
+    "C4",  # flake8-comprehensions
+    "E",  # pycodestyle
+    "F",  # pyflakes
+    "FA",  # flake8-future-annotations
+    "G",  # flake8-logging-format
+    "N",  # pep8-naming
+    "PGH",  # pygrep-hooks
+    "PIE",  # flake8-pie
+    "RET",  # flake8-return
+    "SIM",  # flake8-simplify
+    "UP",  # pyupgrade
+    "W",  # pycodestyle
+    "YTT",  # flake8-2020
+]