1
0
mirror of https://github.com/gryf/boxpy.git synced 2026-02-02 22:25:53 +01:00

3 Commits

Author SHA1 Message Date
eeebab74ad Fix some linting issues.
Also, changing workflow for image downloading fails.
2024-11-17 19:27:37 +01:00
81ab5de7c4 Bump Fedora version 2024-11-15 08:18:46 +01:00
62b86d5f81 Added poweroff option for stopping machine 2024-05-14 19:30:56 +02:00
3 changed files with 120 additions and 63 deletions

View File

@@ -42,6 +42,7 @@ Tested distros
- 38 - 38
- 39 - 39
- 40 - 40
- 41
- Centos Stream - Centos Stream
- 8 - 8
- 9 - 9

159
box.py
View File

@@ -18,7 +18,7 @@ import requests
import yaml import yaml
__version__ = "1.10.0" __version__ = "1.11.0"
CACHE_DIR = os.environ.get('XDG_CACHE_HOME', os.path.expanduser('~/.cache')) CACHE_DIR = os.environ.get('XDG_CACHE_HOME', os.path.expanduser('~/.cache'))
CLOUD_IMAGE = "ci.iso" CLOUD_IMAGE = "ci.iso"
@@ -218,8 +218,22 @@ _boxpy() {
fi fi
;; ;;
stop) stop)
items=(--poweroff)
if [[ ${prev} == ${cmd} ]]; then 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 fi
;; ;;
esac esac
@@ -261,28 +275,28 @@ class Run:
Helper class on subprocess.run() Helper class on subprocess.run()
command is a list with command and its params to execute command is a list with command and its params to execute
""" """
def __init__(self, command, capture_output=True): def __init__(self, command):
result = subprocess.run(command, encoding='utf-8', result = subprocess.run(command, encoding='utf-8', capture_output=True)
capture_output=capture_output)
if result.stdout:
LOG.debug2(result.stdout)
if result.stderr:
LOG.debug2(result.stderr)
self.returncode = result.returncode self.returncode = result.returncode
self.stdout = result.stdout.strip() if result.stdout else '' self.stdout = result.stdout.strip() if result.stdout else ''
self.stderr = result.stderr.strip() if result.stderr 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): class BoxError(Exception):
pass pass
class BoxNotFound(BoxError): class BoxNotFoundError(BoxError):
pass pass
class BoxVBoxFailure(BoxError): class BoxVBoxError(BoxError):
pass pass
@@ -510,7 +524,7 @@ class Config:
def _read_filename(self, fname): def _read_filename(self, fname):
fullpath = os.path.expanduser(os.path.expandvars(fname)) fullpath = os.path.expanduser(os.path.expandvars(fname))
if not os.path.exists(fullpath): if not os.path.exists(fullpath):
return return None
with open(fname) as fobj: with open(fname) as fobj:
return fobj.read() return fobj.read()
@@ -619,11 +633,11 @@ class OsTypes:
def ubuntu(self): def ubuntu(self):
lts = '' 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: if major % 2 == 0 and minor == 4:
lts = '_LTS' lts = '_LTS'
name = "Ubuntu%s%s_64" % (major, lts) name = f"Ubuntu{major}{lts}_64"
if name not in self._ostypes: if name not in self._ostypes:
return 'Ubuntu_64' return 'Ubuntu_64'
@@ -634,7 +648,7 @@ class OsTypes:
return "Fedora_64" return "Fedora_64"
def debian(self): def debian(self):
name = "Debian%s_64" % self._conf.version name = f"Debian{self._conf.version}_64"
if name not in self._ostypes: if name not in self._ostypes:
return 'Debian_64' return 'Debian_64'
@@ -660,7 +674,7 @@ class VBoxManage:
def get_vm_base_path(self): def get_vm_base_path(self):
path = self._get_vm_config() path = self._get_vm_config()
if not path: if not path:
return return None
return os.path.dirname(path) return os.path.dirname(path)
@@ -668,7 +682,7 @@ class VBoxManage:
path = self._get_vm_config() path = self._get_vm_config()
if not path: if not path:
LOG.warning('Configuration for "%s" not found', self.name_or_uuid) LOG.warning('Configuration for "%s" not found', self.name_or_uuid)
return return None
dom = xml.dom.minidom.parse(path) dom = xml.dom.minidom.parse(path)
if len(dom.getElementsByTagName('HardDisk')) != 1: if len(dom.getElementsByTagName('HardDisk')) != 1:
@@ -695,6 +709,7 @@ class VBoxManage:
return line return line
return line.split(' ')[0].strip() return line.split(' ')[0].strip()
return None
def get_vm_info(self): def get_vm_info(self):
out = Run(['vboxmanage', 'showvminfo', self.name_or_uuid]) out = Run(['vboxmanage', 'showvminfo', self.name_or_uuid])
@@ -795,6 +810,7 @@ class VBoxManage:
'--delete']).returncode != 0: '--delete']).returncode != 0:
LOG.fatal('Removing VM "%s" failed', self.name_or_uuid) LOG.fatal('Removing VM "%s" failed', self.name_or_uuid)
return 7 return 7
return None
def create(self, conf): def create(self, conf):
memory = convert_to_mega(conf.memory) memory = convert_to_mega(conf.memory)
@@ -817,7 +833,7 @@ class VBoxManage:
if not self.uuid: if not self.uuid:
msg = f'Cannot create VM "{self.name_or_uuid}".' msg = f'Cannot create VM "{self.name_or_uuid}".'
raise BoxVBoxFailure(msg) raise BoxVBoxError(msg)
port = conf.port if conf.port else self._find_unused_port() port = conf.port if conf.port else self._find_unused_port()
@@ -840,14 +856,15 @@ class VBoxManage:
if Run(cmd).returncode != 0: if Run(cmd).returncode != 0:
LOG.fatal(f'Cannot modify VM "{self.name_or_uuid}"') LOG.fatal(f'Cannot modify VM "{self.name_or_uuid}"')
raise BoxVBoxFailure raise BoxVBoxError
if conf.disable_nested == 'False': if conf.disable_nested == 'False' and Run(['vboxmanage', 'modifyvm',
if Run(['vboxmanage', 'modifyvm', self.name_or_uuid, self.name_or_uuid,
'--nested-hw-virt', 'on']).returncode != 0: '--nested-hw-virt',
LOG.fatal(f'Cannot set nested virtualization for VM ' 'on']).returncode != 0:
f'"{self.name_or_uuid}"') LOG.fatal(f'Cannot set nested virtualization for VM '
raise BoxVBoxFailure f'"{self.name_or_uuid}"')
raise BoxVBoxError
return self.uuid return self.uuid
@@ -883,7 +900,7 @@ class VBoxManage:
if Run(['vboxmanage', 'modifymedium', 'disk', src, '--resize', if Run(['vboxmanage', 'modifymedium', 'disk', src, '--resize',
str(size), '--move', fullpath]).returncode != 0: str(size), '--move', fullpath]).returncode != 0:
LOG.fatal('Resizing and moving image %s has failed', dst) LOG.fatal('Resizing and moving image %s has failed', dst)
raise BoxVBoxFailure raise BoxVBoxError
return fullpath return fullpath
def storageattach(self, controller_name, port, type_, image): def storageattach(self, controller_name, port, type_, image):
@@ -907,7 +924,7 @@ class VBoxManage:
if Run(['vboxmanage', 'startvm', self.name_or_uuid, '--type', if Run(['vboxmanage', 'startvm', self.name_or_uuid, '--type',
type_]).returncode != 0: type_]).returncode != 0:
LOG.fatal('Failed to start: %s', self.name_or_uuid) LOG.fatal('Failed to start: %s', self.name_or_uuid)
raise BoxVBoxFailure raise BoxVBoxError
def setextradata(self, key, val): def setextradata(self, key, val):
res = Run(['vboxmanage', 'setextradata', self.name_or_uuid, key, val]) res = Run(['vboxmanage', 'setextradata', self.name_or_uuid, key, val])
@@ -921,7 +938,7 @@ class VBoxManage:
if Run(['vboxmanage', 'modifyvm', self.name_or_uuid, f'--{nic}', if Run(['vboxmanage', 'modifyvm', self.name_or_uuid, f'--{nic}',
kind]).returncode != 0: kind]).returncode != 0:
LOG.fatal('Cannot modify VM "%s"', self.name_or_uuid) LOG.fatal('Cannot modify VM "%s"', self.name_or_uuid)
raise BoxVBoxFailure raise BoxVBoxError
def is_port_in_use(self, port): def is_port_in_use(self, port):
used_ports = self._get_defined_ports() used_ports = self._get_defined_ports()
@@ -1024,6 +1041,7 @@ class Image:
return False return False
LOG.info('Calculating checksum for "%s"', self._img_fname) 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) fname = os.path.join(self._tmp, self._checksum_file)
expected_sum = self._get_checksum(fname) expected_sum = self._get_checksum(fname)
@@ -1047,7 +1065,10 @@ class Image:
fname = os.path.join(CACHE_DIR, self._img_fname) fname = os.path.join(CACHE_DIR, self._img_fname)
LOG.header('Downloading image %s', self._img_url) LOG.header('Downloading image %s', self._img_url)
Run(['wget', '-q', self._img_url, '-O', fname]) 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(): if not self._checksum():
# TODO: make some retry mechanism? # TODO: make some retry mechanism?
@@ -1121,14 +1142,23 @@ class Fedora(Image):
REVISION = {'37': '1.7', REVISION = {'37': '1.7',
'38': '1.6', '38': '1.6',
'39': '1.5', '39': '1.5',
'40': '1.14'} '40': '1.14',
'41': '1.4'}
def __init__(self, vbox, version, arch, fname=None): def __init__(self, vbox, version, arch, fname=None):
super().__init__(vbox, version, arch) super().__init__(vbox, version, arch)
revision = self.REVISION[version] revision = self.REVISION[version]
if int(version) > 39: if int(version) >= 40:
self.IMG = "Fedora-Cloud-Base-Generic.%s-%s-%s.qcow2" if int(version) == 40:
self._img_fname = self.IMG % (arch, version, revision) # 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: else:
self._img_fname = self.IMG % (version, revision, arch) self._img_fname = self.IMG % (version, revision, arch)
self._img_url = self.URL % (version, arch, self._img_fname) self._img_url = self.URL % (version, arch, self._img_fname)
@@ -1156,8 +1186,7 @@ class CentosStream(Image):
def __init__(self, vbox, version, arch, release, fname=None): def __init__(self, vbox, version, arch, release, fname=None):
super().__init__(vbox, version, arch, release) super().__init__(vbox, version, arch, release)
self._checksum_file = '%s-centos-stream-%s-%s' % (self.CHKS, version, self._checksum_file = f'{self.CHKS}-centos-stream-{version}-{arch}'
arch)
self._checksum_url = self.URL % (version, arch, self.CHKS) self._checksum_url = self.URL % (version, arch, self.CHKS)
# there is assumption, that we always need latest relese for specific # there is assumption, that we always need latest relese for specific
# version and architecture. # version and architecture.
@@ -1184,6 +1213,7 @@ class CentosStream(Image):
images.reverse() images.reverse()
if images: if images:
return images[0] return images[0]
return None
def _get_checksum(self, fname): def _get_checksum(self, fname):
expected_sum = None expected_sum = None
@@ -1215,7 +1245,7 @@ DISTROS = {'ubuntu': {'username': 'ubuntu',
'realname': 'fedora', 'realname': 'fedora',
'img_class': Fedora, 'img_class': Fedora,
'amd64': 'x86_64', 'amd64': 'x86_64',
'default_version': '40'}, 'default_version': '41'},
'centos': {'username': 'centos', 'centos': {'username': 'centos',
'realname': 'centos', 'realname': 'centos',
'img_class': CentosStream, 'img_class': CentosStream,
@@ -1312,9 +1342,8 @@ def vmcreate(args, conf=None):
if not vbox.setextradata(key, getattr(conf, key)): if not vbox.setextradata(key, getattr(conf, key)):
return 5 return 5
if conf.user_data: if conf.user_data and not vbox.setextradata('user_data', conf.user_data):
if not vbox.setextradata('user_data', conf.user_data): return 6
return 6
if not vbox.setextradata('creator', 'boxpy'): if not vbox.setextradata('creator', 'boxpy'):
return 13 return 13
@@ -1420,24 +1449,21 @@ def vmcreate(args, conf=None):
# reread config to update fields # reread config to update fields
conf = Config(args, vbox) 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: if conf.username and conf.username != username:
LOG.info(f'ssh -p {conf.port} -i {conf.ssh_key_path[:-4]} ' username = conf.username
f'{conf.username}@localhost')
else: else:
LOG.info(f'ssh -p {conf.port} -i {conf.ssh_key_path[:-4]} ' username = DISTROS[conf.distro]["username"]
f'{username}@localhost') 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('or simply:')
LOG.info(f'boxpy ssh {conf.name}') LOG.info('boxpy ssh %s', conf.name)
return 0 return 0
def vmdestroy(args): def vmdestroy(args):
if isinstance(args.name, list):
vm_names = args.name vm_names = args.name if isinstance(args.name, list) else [args.name]
else:
vm_names = [args.name]
for name in vm_names: for name in vm_names:
vbox = VBoxManage(name) vbox = VBoxManage(name)
@@ -1531,7 +1557,8 @@ def vminfo(args):
LOG.info(line) LOG.info(line)
if 'user_data' in info: 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): def vmrebuild(args):
@@ -1539,12 +1566,12 @@ def vmrebuild(args):
if not vbox.get_vm_info(): if not vbox.get_vm_info():
LOG.fatal(f'Cannot rebuild VM "{args.name}" - it doesn\'t exists.') LOG.fatal(f'Cannot rebuild VM "{args.name}" - it doesn\'t exists.')
return 20 return 20
else:
LOG.header('Rebuilding VM: %s', args.name) LOG.header('Rebuilding VM: %s', args.name)
try: try:
conf = Config(args, vbox) conf = Config(args, vbox)
except BoxNotFound as ex: except BoxNotFoundError as ex:
LOG.fatal(f'Error with parsing config: {ex}') LOG.fatal(f'Error with parsing config: {ex}')
return 8 return 8
except yaml.YAMLError: except yaml.YAMLError:
@@ -1593,7 +1620,7 @@ def connect(args):
try: try:
conf = Config(args, vbox) conf = Config(args, vbox)
except BoxNotFound: except BoxNotFoundError:
return 11 return 11
except yaml.YAMLError: except yaml.YAMLError:
LOG.fatal(f'Cannot read or parse file `{args.config}` as YAML ' LOG.fatal(f'Cannot read or parse file `{args.config}` as YAML '
@@ -1611,7 +1638,7 @@ def connect(args):
return Run(cmd, False).returncode return Run(cmd, False).returncode
def _set_vmstate(name, state, guitype=None): def _set_vmstate(name, state, guitype=None, poweroff=False):
vbox = VBoxManage(name) vbox = VBoxManage(name)
if not vbox.get_vm_info(): if not vbox.get_vm_info():
@@ -1619,25 +1646,28 @@ def _set_vmstate(name, state, guitype=None):
return 20 return 20
if vbox.running and state == "start": if vbox.running and state == "start":
LOG.info(f'VM "{name}" is already running.') LOG.info('VM "%s" is already running.', name)
return return 1
if not vbox.running and state == "stop": if not vbox.running and state == "stop":
LOG.info(f'VM "{name}" is already stopped.') LOG.info('VM "%s" is already stopped.', name)
return return 1
if state == "start": if state == "start":
vbox.poweron(guitype) vbox.poweron(guitype)
elif poweroff:
vbox.poweroff()
else: else:
vbox.acpipowerbutton() vbox.acpipowerbutton()
return 0
def vmstart(args): def vmstart(args):
_set_vmstate(args.name, 'start', args.type) _set_vmstate(args.name, 'start', guitype=args.type)
def vmstop(args): def vmstop(args):
_set_vmstate(args.name, 'stop') _set_vmstate(args.name, 'stop', poweroff=args.poweroff)
def main(): def main():
@@ -1755,6 +1785,8 @@ def main():
stop = subparsers.add_parser('stop', help='stop VM') stop = subparsers.add_parser('stop', help='stop VM')
stop.add_argument('name', help='name or UUID of the 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) stop.set_defaults(func=vmstop)
completion = subparsers.add_parser('completion', help='generate shell ' completion = subparsers.add_parser('completion', help='generate shell '
@@ -1780,7 +1812,7 @@ def main():
LOG.set_verbose(args.verbose, args.quiet) LOG.set_verbose(args.verbose, args.quiet)
if 'func' not in args and args.version: if 'func' not in args and args.version:
LOG.info(f'boxpy {__version__}') LOG.info('boxpy %s', __version__)
parser.exit() parser.exit()
if hasattr(args, 'func'): if hasattr(args, 'func'):
@@ -1788,6 +1820,7 @@ def main():
parser.print_help() parser.print_help()
parser.exit() parser.exit()
return 23
if __name__ == '__main__': if __name__ == '__main__':

View File

@@ -24,6 +24,7 @@ classifiers = [
"Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3 :: Only" "Programming Language :: Python :: 3 :: Only"
] ]
dependencies = [ dependencies = [
@@ -45,3 +46,25 @@ version = {attr = "box.__version__"}
[tool.distutils.bdist_wheel] [tool.distutils.bdist_wheel]
universal = true 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
]