diff --git a/README.md b/README.md index f521902b..21d2be3a 100644 --- a/README.md +++ b/README.md @@ -6,4 +6,4 @@ https://www.qubes-os.org/doc/admin-api/ for protocol specification. Compatibility ============= -This package requires Python >= 3.5 +This package requires Python >= 3.11 diff --git a/qubesadmin/__init__.py b/qubesadmin/__init__.py index 2cbcfc7e..cd87b410 100644 --- a/qubesadmin/__init__.py +++ b/qubesadmin/__init__.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/app.py b/qubesadmin/app.py index dae03869..4688b6fe 100644 --- a/qubesadmin/app.py +++ b/qubesadmin/app.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -256,8 +255,8 @@ def add_pool(self, name, driver, **kwargs): see :py:meth:`pool_driver_parameters` for a list """ # sort parameters only to ease testing, not required by API - payload = "name={}\n".format(name) + "".join( - "{}={}\n".format(key, value) + payload = f"name={name}\n" + "".join( + f"{key}={value}\n" for key, value in sorted(kwargs.items()) ) self.qubesd_call( @@ -356,13 +355,13 @@ def add_new_vm( raise ValueError("only one of pool= and pools= can be used") method_prefix = "admin.vm.Create." - payload = "name={} label={}".format(name, label) + payload = f"name={name} label={label}" if pool: - payload += " pool={}".format(str(pool)) + payload += f" pool={str(pool)}" method_prefix = "admin.vm.CreateInPool." if pools: payload += "".join( - " pool:{}={}".format(vol, str(pool)) + f" pool:{vol}={str(pool)}" for vol, pool in sorted(pools.items()) ) method_prefix = "admin.vm.CreateInPool." @@ -445,13 +444,13 @@ def clone_vm( pools[volume.name] = volume.pool method_prefix = "admin.vm.Create." - payload = "name={} label={}".format(new_name, label) + payload = f"name={new_name} label={label}" if pool: - payload += " pool={}".format(str(pool)) + payload += f" pool={str(pool)}" method_prefix = "admin.vm.CreateInPool." if pools: payload += "".join( - " pool:{}={}".format(vol, str(pool)) + f" pool:{vol}={str(pool)}" for vol, pool in sorted(pools.items()) ) method_prefix = "admin.vm.CreateInPool." @@ -483,7 +482,7 @@ def clone_vm( pass except qubesadmin.exc.QubesException as e: dst_vm.log.error( - "Failed to set {!s} property: {!s}".format(prop, e) + f"Failed to set {prop!s} property: {e!s}" ) if not ignore_errors: raise @@ -495,7 +494,7 @@ def clone_vm( dst_vm.tags.add(tag) except qubesadmin.exc.QubesException as e: dst_vm.log.error( - "Failed to add {!s} tag: {!s}".format(tag, e) + f"Failed to add {tag!s} tag: {e!s}" ) if not ignore_errors: raise @@ -510,7 +509,7 @@ def clone_vm( dst_vm.features[feature] = value except qubesadmin.exc.QubesException as e: dst_vm.log.error( - "Failed to set {!s} feature: {!s}".format(feature, e) + f"Failed to set {feature!s} feature: {e!s}" ) if not ignore_errors: raise @@ -521,7 +520,7 @@ def clone_vm( dst_vm.set_notes(vm_notes) except qubesadmin.exc.QubesException as e: dst_vm.log.error( - 'Failed to clone qube notes: {!s}'.format(e)) + f'Failed to clone qube notes: {e!s}') if not ignore_errors: raise @@ -596,7 +595,7 @@ def clone_vm( if ignore_volumes and dst_volume.name in ignore_volumes: continue src_volume = src_vm.volumes[dst_volume.name] - dst_vm.log.info("Cloning {} volume".format(dst_volume.name)) + dst_vm.log.info(f"Cloning {dst_volume.name} volume") dst_volume.clone(src_volume) except qubesadmin.exc.QubesException: @@ -840,7 +839,7 @@ def qubesd_call( ) if not os.path.exists(method_path): raise qubesadmin.exc.QubesDaemonCommunicationError( - "{} not found".format(method_path) + f"{method_path} not found" ) command = [ "env", @@ -859,7 +858,7 @@ def qubesd_call( try: client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) client_socket.connect(qubesadmin.config.QUBESD_SOCKET) - except (IOError, OSError) as e: + except OSError as e: raise qubesadmin.exc.QubesDaemonCommunicationError( "Failed to connect to qubesd service: %s", str(e) ) diff --git a/qubesadmin/backup/__init__.py b/qubesadmin/backup/__init__.py index 6bceab9c..ffd3b1aa 100644 --- a/qubesadmin/backup/__init__.py +++ b/qubesadmin/backup/__init__.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/backup/core2.py b/qubesadmin/backup/core2.py index 10c50c32..44821f42 100644 --- a/qubesadmin/backup/core2.py +++ b/qubesadmin/backup/core2.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -229,7 +228,7 @@ def set_netvm_dependency(self, element: _Element) -> None: if dispvm_netvm != self.globals['default_netvm']: if dispvm_netvm: - dispvm_tpl_name = 'disp-{}'.format(dispvm_netvm) + dispvm_tpl_name = f'disp-{dispvm_netvm}' else: dispvm_tpl_name = 'disp-no-netvm' @@ -303,7 +302,7 @@ def import_core2_vm(self, element: _Element) -> None: if value and value.lower() == "none": value = None value_is_default = element.get( - "uses_default_{}".format(attr)) + f"uses_default_{attr}") if value_is_default and value_is_default.lower() != \ "true": vm.properties[attr] = value @@ -354,8 +353,7 @@ def load(self) -> bool | None: try: # pylint: disable=no-member tree = lxml.etree.parse(fh) - except (EnvironmentError, # pylint: disable=broad-except - xml.parsers.expat.ExpatError) as err: + except (OSError, xml.parsers.expat.ExpatError) as err: self.log.error(err) return False diff --git a/qubesadmin/backup/core3.py b/qubesadmin/backup/core3.py index 00aaf40f..77728128 100644 --- a/qubesadmin/backup/core3.py +++ b/qubesadmin/backup/core3.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -80,7 +79,7 @@ def get_property(xml_obj: _Element, prop: str) -> str | None: Object can be any PropertyHolder serialized to XML - in practice :py:class:`BaseVM` or :py:class:`Qubes`. ''' - xml_prop = xml_obj.findall('./property[@name=\'{}\']'.format(prop)) + xml_prop = xml_obj.findall(f'./property[@name=\'{prop}\']') if not xml_prop: raise KeyError(prop) return xml_prop[0].text @@ -165,8 +164,7 @@ def load(self) -> bool | None: try: # pylint: disable=no-member tree = lxml.etree.parse(fh) - except (EnvironmentError, # pylint: disable=broad-except - xml.parsers.expat.ExpatError) as err: + except (OSError, xml.parsers.expat.ExpatError) as err: self.log.error(err) return False diff --git a/qubesadmin/backup/dispvm.py b/qubesadmin/backup/dispvm.py index eeb6edd1..715f00b5 100644 --- a/qubesadmin/backup/dispvm.py +++ b/qubesadmin/backup/dispvm.py @@ -335,7 +335,7 @@ def run(self) -> bytes | None: ) if exit_code != 0: raise qubesadmin.exc.BackupRestoreError( - 'qvm-backup-restore failed with {}'.format(exit_code), + f'qvm-backup-restore failed with {exit_code}', backup_log=backup_log) return backup_log except subprocess.CalledProcessError as e: @@ -350,7 +350,7 @@ def run(self) -> bytes | None: except: # pylint: disable=bare-except backup_log = None raise qubesadmin.exc.BackupRestoreError( - 'qvm-backup-restore failed with {}'.format(e.returncode), + f'qvm-backup-restore failed with {e.returncode}', backup_log=backup_log) finally: if self.dispvm is not None: diff --git a/qubesadmin/backup/restore.py b/qubesadmin/backup/restore.py index 9310dd54..0fb65b0c 100644 --- a/qubesadmin/backup/restore.py +++ b/qubesadmin/backup/restore.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -47,7 +46,8 @@ import collections from subprocess import Popen -from typing import Callable, TypeVar, Iterable, IO, Generator +from typing import TypeVar, IO +from collections.abc import Callable, Iterable, Generator import qubesadmin import qubesadmin.vm @@ -208,7 +208,7 @@ def load(self, untrusted_header_text: bytes) -> None: continue header = self.known_headers[key] if key in seen: - raise QubesException("Duplicated header line: {}".format(key)) + raise QubesException(f"Duplicated header line: {key}") seen.add(key) if getattr(self, header.field, None) is not None: # ignore options already set (potentially forced values) @@ -234,7 +234,7 @@ def load(self, untrusted_header_text: bytes) -> None: "Unusual compression filter '{f}' found. Use " "--compression-filter={f} to use it anyway.".format( f=value)) - raise QubesException("Invalid value for header: {}".format(key)) + raise QubesException(f"Invalid value for header: {key}") setattr(self, header.field, value) self.validate() @@ -256,23 +256,23 @@ def validate(self) -> None: for key in expected_attrs: if getattr(self, key) is None: raise QubesException( - "Backup header lack '{}' info".format(key)) + f"Backup header lack '{key}' info") else: raise QubesException( - "Unsupported backup version {}".format(self.version)) + f"Unsupported backup version {self.version}") def save(self, filename: str) -> None: '''Save backup header into a file''' with open(filename, "w", encoding='utf-8') as f_header: # make sure 'version' is the first key - f_header.write('version={}\n'.format(self.version)) + f_header.write(f'version={self.version}\n') for key, header in self.known_headers.items(): if key == 'version': continue attr = header.field if getattr(self, attr) is None: continue - f_header.write("{!s}={!s}\n".format(key, getattr(self, attr))) + f_header.write(f"{key!s}={getattr(self, attr)!s}\n") def launch_proc_with_pty(args: list[str], stdin: int | None=None, stdout: int | None=None, stderr: int | None=None, @@ -327,7 +327,7 @@ def launch_scrypt(action: str, input_name: str, output_name: str, actual_prompt = typing.cast(IO, p.stderr).read(len(prompt)) if actual_prompt != prompt: raise QubesException( - 'Unexpected prompt from scrypt: {}'.format(actual_prompt)) + f'Unexpected prompt from scrypt: {actual_prompt}') pty.write(passphrase.encode('utf-8') + b'\n') pty.flush() # save it here, so garbage collector would not close it (which would kill @@ -451,7 +451,7 @@ def collect_tar_output(self) -> None: try: new_lines = self.tar2_process.stderr \ .read(MAX_STDERR_BYTES).splitlines() - except IOError as e: + except OSError as e: if e.errno == errno.EAGAIN: return raise @@ -1128,7 +1128,7 @@ def load_hmac(hmac_text: str) -> str: ''' if any(ord(x) not in range(128) for x in hmac_text): raise QubesException( - "Invalid content of {}".format(hmacfile)) + f"Invalid content of {hmacfile}") hmac_text_list = hmac_text.strip().split("=") if len(hmac_text_list) > 1: hmac_text = hmac_text_list[1].strip() @@ -1161,7 +1161,7 @@ def load_hmac(hmac_text: str) -> str: with open(f_name + '.dec', 'rb') as f_two: if f_one.read() != f_two.read(): raise QubesException( - 'Invalid hmac on {}'.format(filename)) + f'Invalid hmac on {filename}') return True with open(os.path.join(self.tmpdir, filename), 'rb') as f_input: @@ -1174,10 +1174,10 @@ def load_hmac(hmac_text: str) -> str: if hmac_stderr: raise QubesException( - "ERROR: verify file {0}: {1}".format(filename, hmac_stderr)) + f"ERROR: verify file {filename}: {hmac_stderr}") self.log.debug("Loading hmac for file %s", filename) try: - with open(os.path.join(self.tmpdir, hmacfile), 'r', + with open(os.path.join(self.tmpdir, hmacfile), encoding='ascii') as f_hmac: hmac = load_hmac(f_hmac.read()) except UnicodeDecodeError as err: @@ -1627,7 +1627,7 @@ def new_name_for_conflicting_vm(self, orig_name: str, while (new_name in restore_info.keys() or new_name in [x.name for x in restore_info.values()] or new_name in self.app.domains): - new_name = str('{}-{}'.format(orig_name, number)) + new_name = str(f'{orig_name}-{number}') number += 1 if number == 100: # give up @@ -1902,7 +1902,7 @@ def _handle_dom0(self, stream: BytesIO) -> None: except KeyError: home_dir = os.path.expanduser('~') local_user = getpass.getuser() - restore_home_backupdir = "home-restore-{0}".format( + restore_home_backupdir = "home-restore-{}".format( time.strftime("%Y-%m-%d-%H%M%S")) self.log.info("Restoring home of user '%s' to '%s' directory...", diff --git a/qubesadmin/base.py b/qubesadmin/base.py index b07af14c..ef5f6341 100644 --- a/qubesadmin/base.py +++ b/qubesadmin/base.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -281,7 +280,7 @@ def _parse_type_value(self, prop_type, value): prop_type = prop_type.decode('ascii') if not prop_type.startswith('type='): raise qubesadmin.exc.QubesDaemonCommunicationError( - 'Invalid type prefix received: {}'.format(prop_type)) + f'Invalid type prefix received: {prop_type}') (_, prop_type) = prop_type.split('=', 1) value = value.decode() if prop_type == 'str': @@ -303,7 +302,7 @@ def _parse_type_value(self, prop_type, value): return None return self.app.labels.get_blind(value) raise qubesadmin.exc.QubesDaemonCommunicationError( - 'Received invalid value type: {}'.format(prop_type)) + f'Received invalid value type: {prop_type}') def _fetch_all_properties(self): """ diff --git a/qubesadmin/config.py b/qubesadmin/config.py index d348bf6b..57d83fe4 100644 --- a/qubesadmin/config.py +++ b/qubesadmin/config.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/device_protocol.py b/qubesadmin/device_protocol.py index 29203fbb..243ddd92 100644 --- a/qubesadmin/device_protocol.py +++ b/qubesadmin/device_protocol.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -30,25 +29,27 @@ should be moved to one place. """ +from __future__ import annotations import string import sys +import typing from enum import Enum -from typing import Optional, Dict, Any, List, Union, Tuple, Callable +from typing import Any +from collections.abc import Callable import qubesadmin.exc from qubesadmin.exc import QubesValueError +if typing.TYPE_CHECKING: + from qubesadmin.vm import QubesVM + class ProtocolError(AssertionError): """ Raised when something is wrong with data received. """ - -QubesVM = 'qubesadmin.vm.QubesVM' - - class UnexpectedDeviceProperty(qubesadmin.exc.QubesException, ValueError): """ Device has unexpected property such as backend_domain, devclass etc. @@ -71,7 +72,7 @@ def qbool(value): if lcvalue in ("1", "yes", "true", "on"): return True raise QubesValueError( - "Invalid literal for boolean property: {!r}".format(value) + f"Invalid literal for boolean property: {value!r}" ) return bool(value) @@ -90,7 +91,7 @@ class DeviceSerializer: @classmethod def unpack_properties( cls, untrusted_serialization: bytes - ) -> Tuple[Dict, Dict]: + ) -> tuple[dict, dict]: """ Unpacks basic port properties from a serialized encoded string. @@ -106,8 +107,8 @@ def unpack_properties( "ascii", errors="strict" ).strip() - properties: Dict[str, str] = {} - options: Dict[str, str] = {} + properties: dict[str, str] = {} + options: dict[str, str] = {} if not ut_decoded: return properties, options @@ -155,7 +156,7 @@ def unpack_properties( return properties, options @classmethod - def pack_property(cls, key: str, value: Optional[str]): + def pack_property(cls, key: str, value: str | None): """ Add property `key=value` to serialization. """ @@ -175,7 +176,7 @@ def pack_property(cls, key: str, value: Optional[str]): @staticmethod def parse_basic_device_properties( - expected_device: "VirtualDevice", properties: Dict[str, Any] + expected_device: VirtualDevice, properties: dict[str, Any] ): """ Validates properties against an expected port configuration. @@ -245,7 +246,7 @@ def deserialize_str(value: str) -> str: def sanitize_str( untrusted_value: str, allowed_chars: set, - replace_char: Optional[str] = None, + replace_char: str | None = None, error_message: str = "", ) -> str: """ @@ -281,9 +282,9 @@ class Port: def __init__( self, - backend_domain: Optional[QubesVM], - port_id: Optional[str], - devclass: Optional[str], + backend_domain: QubesVM | None, + port_id: str | None, + devclass: str | None, ): self.__backend_domain = backend_domain self.__port_id = port_id @@ -329,7 +330,7 @@ def backend_name(self) -> str: @classmethod def from_qarg( cls, representation: str, devclass, domains, blind=False - ) -> "Port": + ) -> Port: """ Parse qrexec argument + to retrieve Port. """ @@ -342,7 +343,7 @@ def from_qarg( @classmethod def from_str( cls, representation: str, devclass, domains, blind=False - ) -> "Port": + ) -> Port: """ Parse string : to retrieve Port. """ @@ -355,7 +356,7 @@ def from_str( @classmethod def _parse( cls, representation: str, devclass: str, get_domain: Callable, sep: str - ) -> "Port": + ) -> Port: """ Parse string representation and return instance of Port. """ @@ -375,7 +376,7 @@ def port_id(self) -> str: return "*" @property - def backend_domain(self) -> Optional[QubesVM]: + def backend_domain(self) -> QubesVM | None: """Which domain exposed this port. (immutable)""" return self.__backend_domain @@ -419,18 +420,18 @@ class VirtualDevice: def __init__( self, - port: Optional[Port] = None, - device_id: Optional[str] = None, + port: Port | None = None, + device_id: str | None = None, ): assert not isinstance(port, AnyPort) or device_id is not None - self.port: Optional[Port] = port # type: ignore + self.port: Port | None = port # type: ignore self._device_id = device_id - def clone(self, **kwargs) -> "VirtualDevice": + def clone(self, **kwargs) -> VirtualDevice: """ Clone object and substitute attributes with explicitly given. """ - attr: Dict[str, Any] = { + attr: dict[str, Any] = { "port": self.port, "device_id": self.device_id, } @@ -443,7 +444,7 @@ def port(self) -> Port: return self._port @port.setter - def port(self, value: Union[Port, str, None]): + def port(self, value: Port | str | None): # pylint: disable=missing-function-docstring if isinstance(value, Port): self._port = value @@ -469,7 +470,7 @@ def is_device_id_set(self) -> bool: return self._device_id is not None @property - def backend_domain(self) -> Optional[QubesVM]: + def backend_domain(self) -> QubesVM | None: # pylint: disable=missing-function-docstring return self.port.backend_domain @@ -562,11 +563,11 @@ def __str__(self): def from_qarg( cls, representation: str, - devclass: Optional[str], + devclass: str | None, domains, blind: bool = False, - backend: Optional[QubesVM] = None, - ) -> "VirtualDevice": + backend: QubesVM | None = None, + ) -> VirtualDevice: """ Parse qrexec argument +: to get device info """ @@ -583,11 +584,11 @@ def from_qarg( def from_str( cls, representation: str, - devclass: Optional[str], + devclass: str | None, domains, blind: bool = False, - backend: Optional[QubesVM] = None, - ) -> "VirtualDevice": + backend: QubesVM | None = None, + ) -> VirtualDevice: """ Parse string +: to get device info """ @@ -604,11 +605,11 @@ def from_str( def _parse( cls, representation: str, - devclass: Optional[str], + devclass: str | None, get_domain: Callable, - backend: Optional[QubesVM], + backend: QubesVM | None, sep: str, - ) -> "VirtualDevice": + ) -> VirtualDevice: """ Parse string representation and return instance of VirtualDevice. """ @@ -690,7 +691,7 @@ class DeviceCategory(Enum): PCI_USB = ("p0c03**",) @staticmethod - def from_str(interface_encoding: str) -> "DeviceCategory": + def from_str(interface_encoding: str) -> DeviceCategory: """ Returns `DeviceCategory` from data encoded in string. """ @@ -721,7 +722,7 @@ class DeviceInterface: Peripheral device interface wrapper. """ - def __init__(self, interface_encoding: str, devclass: Optional[str] = None): + def __init__(self, interface_encoding: str, devclass: str | None = None): ifc_padded = interface_encoding.ljust(6, "*") if devclass: if len(ifc_padded) > 6: @@ -762,7 +763,7 @@ def __init__(self, interface_encoding: str, devclass: Optional[str] = None): self._category = DeviceCategory.from_str(self._interface_encoding) @property - def devclass(self) -> Optional[str]: + def devclass(self) -> str | None: """Immutable Device class such like: 'usb', 'pci' etc.""" return self._devclass @@ -772,12 +773,12 @@ def category(self) -> DeviceCategory: return self._category @classmethod - def unknown(cls) -> "DeviceInterface": + def unknown(cls) -> DeviceInterface: """Value for unknown device interface.""" return cls("?******") @staticmethod - def from_str_bulk(interfaces: Optional[str]) -> List["DeviceInterface"]: + def from_str_bulk(interfaces: str | None) -> list[DeviceInterface]: """Interprets string of interfaces as list of `DeviceInterface`. Examples: @@ -884,7 +885,7 @@ def _load_classes(bus: str): return result - def matches(self, other: "DeviceInterface") -> bool: + def matches(self, other: DeviceInterface) -> bool: """ Check if this `DeviceInterface` (pattern) matches given one. @@ -913,15 +914,15 @@ def __init__( self, port: Port, *, - vendor: Optional[str] = None, - product: Optional[str] = None, - manufacturer: Optional[str] = None, - name: Optional[str] = None, - serial: Optional[str] = None, - interfaces: Optional[List[DeviceInterface]] = None, - parent: Optional["DeviceInfo"] = None, - attachment: Optional[QubesVM] = None, - device_id: Optional[str] = None, + vendor: str | None = None, + product: str | None = None, + manufacturer: str | None = None, + name: str | None = None, + serial: str | None = None, + interfaces: list[DeviceInterface] | None = None, + parent: DeviceInfo | None = None, + attachment: QubesVM | None = None, + device_id: str | None = None, **kwargs, ): super().__init__(port, device_id) @@ -1044,7 +1045,7 @@ def description(self) -> str: return f"{cat}: {vendor} {prod}" @property - def interfaces(self) -> List[DeviceInterface]: + def interfaces(self) -> list[DeviceInterface]: """ Non-empty list of device interfaces. @@ -1055,7 +1056,7 @@ def interfaces(self) -> List[DeviceInterface]: return self._interfaces @property - def parent_device(self) -> Optional[VirtualDevice]: + def parent_device(self) -> VirtualDevice | None: """ The parent device, if any. @@ -1065,7 +1066,7 @@ def parent_device(self) -> Optional[VirtualDevice]: return self._parent @property - def subdevices(self) -> List[VirtualDevice]: + def subdevices(self) -> list[VirtualDevice]: """ The list of children devices if any. @@ -1082,7 +1083,7 @@ def subdevices(self) -> List[VirtualDevice]: ] @property - def attachment(self) -> Optional[QubesVM]: + def attachment(self) -> QubesVM | None: """ VM to which device is attached (frontend domain). """ @@ -1135,8 +1136,8 @@ def deserialize( cls, serialization: bytes, expected_backend_domain: QubesVM, - expected_devclass: Optional[str] = None, - ) -> "DeviceInfo": + expected_devclass: str | None = None, + ) -> DeviceInfo: """ Recovers a serialized object, see: :py:meth:`serialize`. """ @@ -1160,7 +1161,7 @@ def deserialize( @classmethod def _deserialize( cls, untrusted_serialization: bytes, expected_device: VirtualDevice - ) -> "DeviceInfo": + ) -> DeviceInfo: """ Actually deserializes the object. """ @@ -1227,7 +1228,7 @@ class UnknownDevice(DeviceInfo): """Unknown device - for example, exposed by domain not running currently""" @staticmethod - def from_device(device: VirtualDevice) -> "UnknownDevice": + def from_device(device: VirtualDevice) -> UnknownDevice: """ Return `UnknownDevice` based on any virtual device. """ @@ -1255,7 +1256,7 @@ def __init__( device: VirtualDevice, frontend_domain=None, options=None, - mode: Union[str, AssignmentMode] = "manual", + mode: str | AssignmentMode = "manual", ): if isinstance(device, DeviceInfo): device = VirtualDevice(device.port, device.device_id) @@ -1273,12 +1274,12 @@ def new( backend_domain: QubesVM, port_id: str, devclass: str, - device_id: Optional[str] = None, + device_id: str | None = None, *, - frontend_domain: Optional[QubesVM] = None, + frontend_domain: QubesVM | None = None, options=None, - mode: Union[str, AssignmentMode] = "manual", - ) -> "DeviceAssignment": + mode: str | AssignmentMode = "manual", + ) -> DeviceAssignment: """Helper method to create a DeviceAssignment object.""" return cls( VirtualDevice(Port(backend_domain, port_id, devclass), device_id), @@ -1333,7 +1334,7 @@ def __lt__(self, other): ) @property - def backend_domain(self) -> Optional[QubesVM]: + def backend_domain(self) -> QubesVM | None: # pylint: disable=missing-function-docstring return self.virtual_device.backend_domain @@ -1358,9 +1359,9 @@ def device_id(self) -> str: return self.virtual_device.device_id @property - def devices(self) -> List[DeviceInfo]: + def devices(self) -> list[DeviceInfo]: """Get DeviceInfo objects corresponding to this DeviceAssignment""" - result: List[DeviceInfo] = [] + result: list[DeviceInfo] = [] if not self.backend_domain: return result if self.port_id != "*": @@ -1400,17 +1401,17 @@ def port(self) -> Port: return Port(self.backend_domain, self.port_id, self.devclass) @property - def frontend_domain(self) -> Optional[QubesVM]: + def frontend_domain(self) -> QubesVM | None: """Which domain the device is attached/assigned to.""" return self.__frontend_domain @frontend_domain.setter - def frontend_domain(self, frontend_domain: Optional[Union[str, QubesVM]]): + def frontend_domain(self, frontend_domain: str | QubesVM | None): """Which domain the device is attached/assigned to.""" if isinstance(frontend_domain, str): if not self.backend_domain: raise ProtocolError("Cannot determine backend domain") - self.__frontend_domain: Optional[QubesVM] = ( + self.__frontend_domain: QubesVM | None = ( self.backend_domain.app.domains[frontend_domain] ) else: @@ -1449,12 +1450,12 @@ def attach_automatically(self) -> bool: ) @property - def options(self) -> Dict[str, Any]: + def options(self) -> dict[str, Any]: """Device options (same as in the legacy API).""" return self.__options @options.setter - def options(self, options: Optional[Dict[str, Any]]): + def options(self, options: dict[str, Any] | None): """Device options (same as in the legacy API).""" self.__options = options or {} @@ -1483,7 +1484,7 @@ def deserialize( cls, serialization: bytes, expected_device: VirtualDevice, - ) -> "DeviceAssignment": + ) -> DeviceAssignment: """ Recovers a serialized object, see: :py:meth:`serialize`. """ @@ -1498,7 +1499,7 @@ def _deserialize( cls, untrusted_serialization: bytes, expected_device: VirtualDevice, - ) -> "DeviceAssignment": + ) -> DeviceAssignment: """ Actually deserializes the object. """ diff --git a/qubesadmin/devices.py b/qubesadmin/devices.py index 228969d1..5df99124 100644 --- a/qubesadmin/devices.py +++ b/qubesadmin/devices.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -32,7 +31,7 @@ class is implemented by an extension. is :py:class:`str`. """ import itertools -from typing import Iterable +from collections.abc import Iterable import qubesadmin.exc from qubesadmin.device_protocol import ( @@ -202,7 +201,7 @@ def get_attached_devices(self) -> Iterable[DeviceAssignment]: return new_cache = [] assignments_str = self._vm.qubesd_call( - None, "admin.vm.device.{}.Attached".format(self._class) + None, f"admin.vm.device.{self._class}.Attached" ).decode() for assignment_str in assignments_str.splitlines(): head, _, untrusted_rest = assignment_str.partition(" ") @@ -234,7 +233,7 @@ def get_assigned_devices( return new_cache = [] assignments_str = self._vm.qubesd_call( - None, "admin.vm.device.{}.Assigned".format(self._class) + None, f"admin.vm.device.{self._class}.Assigned" ).decode() for assignment_str in assignments_str.splitlines(): head, _, untrusted_rest = assignment_str.partition(" ") @@ -257,7 +256,7 @@ def get_exposed_devices(self) -> Iterable[DeviceInfo]: List devices exposed by this vm. """ devices: bytes = self._vm.qubesd_call( - None, "admin.vm.device.{}.Available".format(self._class) + None, f"admin.vm.device.{self._class}.Available" ) for dev_serialized in devices.splitlines(): yield DeviceInfo.deserialize( @@ -280,7 +279,7 @@ def update_assignment( """ self._vm.qubesd_call( None, - "admin.vm.device.{}.Set.assignment".format(self._class), + f"admin.vm.device.{self._class}.Set.assignment", device.repr_for_qarg, required.value.encode("utf-8"), ) diff --git a/qubesadmin/events/__init__.py b/qubesadmin/events/__init__.py index af76d446..4b579341 100644 --- a/qubesadmin/events/__init__.py +++ b/qubesadmin/events/__init__.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/events/utils.py b/qubesadmin/events/utils.py index 41cb50ca..f2c16e2c 100644 --- a/qubesadmin/events/utils.py +++ b/qubesadmin/events/utils.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/exc.py b/qubesadmin/exc.py index 737bd6d9..afa5c704 100644 --- a/qubesadmin/exc.py +++ b/qubesadmin/exc.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/features.py b/qubesadmin/features.py index 8287b677..f7b6768b 100644 --- a/qubesadmin/features.py +++ b/qubesadmin/features.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/firewall.py b/qubesadmin/firewall.py index 5298a8aa..0001ff14 100644 --- a/qubesadmin/firewall.py +++ b/qubesadmin/firewall.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # pylint: disable=too-few-public-methods # # The Qubes OS Project, http://www.qubes-os.org @@ -104,7 +103,7 @@ def __init__(self, value, prefixlen=None): 'netmask for IPv6 must be between 0 and 128') value += '/' + str(self.prefixlen) self.type = 'dst6' - except socket.error: + except OSError: try: socket.inet_pton(socket.AF_INET, value) if value.count('.') != 3: @@ -119,7 +118,7 @@ def __init__(self, value, prefixlen=None): 'netmask for IPv4 must be between 0 and 32') value += '/' + str(self.prefixlen) self.type = 'dst4' - except socket.error: + except OSError: self.type = 'dsthost' self.prefixlen = 0 safe_set = string.ascii_lowercase + string.digits + '-._' @@ -136,7 +135,7 @@ def __init__(self, value, prefixlen=None): if prefixlen > 128: raise ValueError('netmask for IPv6 must be <= 128') self.type = 'dst6' - except socket.error: + except OSError: try: socket.inet_pton(socket.AF_INET, host) if prefixlen > 32: @@ -145,7 +144,7 @@ def __init__(self, value, prefixlen=None): if host.count('.') != 3: raise ValueError( 'Invalid number of dots in IPv4 address') - except socket.error: + except OSError: raise ValueError('Invalid IP address: ' + host) super().__init__(value) @@ -230,7 +229,7 @@ def pretty_value(self): '''Human readable representation''' now = datetime.datetime.utcnow() duration = (self.datetime - now).total_seconds() - return "{:+.0f}s".format(duration) + return f"{duration:+.0f}s" class Comment(RuleOption): @@ -408,7 +407,7 @@ def __eq__(self, other): return NotImplemented def __repr__(self): - return 'Rule(\'{}\')'.format(self.rule) + return f'Rule(\'{self.rule}\')' class Firewall: @@ -453,7 +452,7 @@ def save_rules(self, rules=None): if rules is None: rules = self._rules self.vm.qubesd_call(None, 'admin.vm.firewall.Set', - payload=(''.join('{}\n'.format(rule.rule) + payload=(''.join(f'{rule.rule}\n' for rule in rules)).encode('ascii')) @property diff --git a/qubesadmin/label.py b/qubesadmin/label.py index c615bff3..b31c1156 100644 --- a/qubesadmin/label.py +++ b/qubesadmin/label.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/spinner.py b/qubesadmin/spinner.py index 27c4eb9a..76da546e 100644 --- a/qubesadmin/spinner.py +++ b/qubesadmin/spinner.py @@ -1,5 +1,3 @@ -# vim: fileencoding=utf-8 - # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -97,7 +95,7 @@ def __init__(self, *args, **kwargs): def show(self, prompt): self.hidelen = len(prompt) + 2 - self.stream.write('{} {}'.format(prompt, next(self.charset))) + self.stream.write(f'{prompt} {next(self.charset)}') self.stream.flush() def hide(self): diff --git a/qubesadmin/storage.py b/qubesadmin/storage.py index e7924b32..054eab94 100644 --- a/qubesadmin/storage.py +++ b/qubesadmin/storage.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tags.py b/qubesadmin/tags.py index 6460ddb2..d58b51f3 100644 --- a/qubesadmin/tags.py +++ b/qubesadmin/tags.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/__init__.py b/qubesadmin/tests/__init__.py index 07a24ff8..172585a4 100644 --- a/qubesadmin/tests/__init__.py +++ b/qubesadmin/tests/__init__.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -172,13 +171,13 @@ def qubesd_call(self, dest, method, arg=None, payload=None, call_key = (dest, method, arg, payload) self.actual_calls.append(call_key) if call_key not in self.expected_calls: - raise AssertionError('Unexpected call {!r}'.format(call_key)) + raise AssertionError(f'Unexpected call {call_key!r}') return_data = self.expected_calls[call_key] if isinstance(return_data, list): try: return_data = return_data.pop(0) except IndexError: - raise AssertionError('Extra call {!r}'.format(call_key)) + raise AssertionError(f'Extra call {call_key!r}') return self._parse_qubesd_response(return_data) def run_service(self, dest, service, **kwargs): diff --git a/qubesadmin/tests/app.py b/qubesadmin/tests/app.py index 3b8796cd..972a170a 100644 --- a/qubesadmin/tests/app.py +++ b/qubesadmin/tests/app.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -87,7 +86,7 @@ def test_005_keys(self): b'0\x00test-vm class=AppVM state=Running\n' \ b'test-vm2 class=AppVM state=Running\n' self.assertEqual(set(self.app.domains.keys()), - set(['test-vm', 'test-vm2'])) + {'test-vm', 'test-vm2'}) self.assertAllCalled() def test_006_values(self): @@ -97,8 +96,8 @@ def test_006_values(self): values = self.app.domains.values() for obj in values: self.assertIsInstance(obj, qubesadmin.vm.QubesVM) - self.assertEqual(set([vm.name for vm in values]), - set(['test-vm', 'test-vm2'])) + self.assertEqual({vm.name for vm in values}, + {'test-vm', 'test-vm2'}) self.assertAllCalled() def test_007_getitem_blind_mode(self): diff --git a/qubesadmin/tests/backup/__init__.py b/qubesadmin/tests/backup/__init__.py index dfc73fd0..ad5d215a 100644 --- a/qubesadmin/tests/backup/__init__.py +++ b/qubesadmin/tests/backup/__init__.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/backup/backupcompatibility.py b/qubesadmin/tests/backup/backupcompatibility.py index acd48fa4..f005db95 100644 --- a/qubesadmin/tests/backup/backupcompatibility.py +++ b/qubesadmin/tests/backup/backupcompatibility.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -35,10 +34,7 @@ import contextlib -try: - from unittest import mock -except ImportError: - import mock +from unittest import mock import re import multiprocessing @@ -1039,11 +1035,11 @@ def create_v4_files(self): # normal AppVMs for vm in ('test-work', 'test-d8test', 'test-proxy', 'test-custom-template-appvm', 'test-net'): - os.mkdir(self.fullpath('appvms/{}'.format(vm))) + os.mkdir(self.fullpath(f'appvms/{vm}')) self.create_whitelisted_appmenus(self.fullpath( - 'appvms/{}/whitelisted-appmenus.list'.format(vm))) + f'appvms/{vm}/whitelisted-appmenus.list')) self.create_private_img(self.fullpath( - 'appvms/{}/private.img'.format(vm))) + f'appvms/{vm}/private.img')) # setup firewall only on one VM with open(self.fullpath("appvms/test-work/firewall.xml"), "wb") \ @@ -1062,16 +1058,16 @@ def create_v4_files(self): # StandaloneVMs for vm in ('test-standalonevm', 'test-hvm'): - os.mkdir(self.fullpath('appvms/{}'.format(vm))) + os.mkdir(self.fullpath(f'appvms/{vm}')) self.create_whitelisted_appmenus(self.fullpath( - 'appvms/{}/whitelisted-appmenus.list'.format(vm))) + f'appvms/{vm}/whitelisted-appmenus.list')) self.create_private_img(self.fullpath( - 'appvms/{}/private.img'.format(vm))) + f'appvms/{vm}/private.img')) self.create_sparse( - self.fullpath('appvms/{}/root.img'.format(vm)), 10*2**30) - self.fill_image(self.fullpath('appvms/{}/root.img'.format(vm)), + self.fullpath(f'appvms/{vm}/root.img'), 10*2**30) + self.fill_image(self.fullpath(f'appvms/{vm}/root.img'), 1024*1024, True, - signature='{}/root'.format(vm).encode()) + signature=f'{vm}/root'.encode()) # only for Linux one os.mkdir(self.fullpath('appvms/test-standalonevm/apps.templates')) @@ -1120,7 +1116,7 @@ def scrypt_encrypt(self, f_name, output_name=None, password='qubes', if f_name == 'backup-header': scrypt_pass = 'backup-header!' + password else: - scrypt_pass = '20161020T123455-1234!{}!{}'.format(f_name, password) + scrypt_pass = f'20161020T123455-1234!{f_name}!{password}' with subprocess.Popen(['scrypt', 'enc', '-P', '-t', '0.1', os.path.join(basedir, f_name), os.path.join(basedir, output_name)], @@ -1130,7 +1126,7 @@ def scrypt_encrypt(self, f_name, output_name=None, password='qubes', return output_name def calculate_hmac(self, f_name, algorithm="sha512", password="qubes"): - with open(self.fullpath(f_name), "r", encoding="utf-8") as f_data: + with open(self.fullpath(f_name), encoding="utf-8") as f_data: with open(self.fullpath(f_name+".hmac"), "w", encoding="utf-8") as f_hmac: subprocess.check_call( @@ -1149,7 +1145,7 @@ def handle_v3_file(self, f_name, subdir, stream, compressed=True, # create inner archive tar_cmdline = ["tar", "-Pc", '--sparse', '-C', self.fullpath(os.path.dirname(f_name)), - '--xform', 's:^%s:%s\\0:' % ( + '--xform', 's:^{}:{}\\0:'.format( os.path.basename(f_name), subdir), os.path.basename(f_name) @@ -1185,7 +1181,7 @@ def handle_v3_file(self, f_name, subdir, stream, compressed=True, for part in sorted(os.listdir(stage1_dir)): if not re.match( - r"^{}.[0-9][0-9][0-9]$".format(os.path.basename(f_name)), + fr"^{os.path.basename(f_name)}.[0-9][0-9][0-9]$", part): continue part_with_dir = os.path.join(subdir, part) @@ -1199,7 +1195,7 @@ def handle_v4_file(self, f_name, subdir, stream, compressed="gzip"): # create inner archive tar_cmdline = ["tar", "-Pc", '--sparse', '-C', self.fullpath(os.path.dirname(f_name)), - '--xform', 's:^%s:%s\\0:' % ( + '--xform', 's:^{}:{}\\0:'.format( os.path.basename(f_name), subdir), os.path.basename(f_name) @@ -1224,7 +1220,7 @@ def handle_v4_file(self, f_name, subdir, stream, compressed="gzip"): for part in sorted(os.listdir(stage1_dir)): if not re.match( - r"^{}.[0-9][0-9][0-9]$".format(os.path.basename(f_name)), + fr"^{os.path.basename(f_name)}.[0-9][0-9][0-9]$", part): continue part_with_dir = os.path.join(subdir, part) @@ -1259,7 +1255,7 @@ def create_v3_backup(self, encrypted=True, compressed=True): qubesxml = xml_path.read_bytes() if encrypted: for vmname, subdir in MANGLED_SUBDIRS_R2.items(): - qubesxml = re.sub(r"[a-z-]*/{}".format(vmname).encode(), + qubesxml = re.sub(fr"[a-z-]*/{vmname}".encode(), subdir.encode(), qubesxml) f_qubesxml.write(qubesxml) else: @@ -1327,7 +1323,7 @@ def create_v4_backup(self, compressed="gzip", big=False): qubesxml = xml_path.read_bytes() for vmname, subdir in MANGLED_SUBDIRS_R4.items(): qubesxml = re.sub( - r'backup-path">[a-z-]*/{}'.format(vmname).encode(), + fr'backup-path">[a-z-]*/{vmname}'.encode(), ('backup-path">' + subdir).encode(), qubesxml) f_qubesxml.write(qubesxml) @@ -1481,8 +1477,8 @@ def setup_expected_calls(self, parsed_qubes_xml, templates_map=None): f"backend_domain='{backend_domain}' mode='required' " f"frontend_domain='{name}'".encode()) self.app.expected_calls[ - (name, 'admin.vm.device.{}.Assign'.format(bus), - '{}+{}+_'.format(backend_domain, port_id), + (name, f'admin.vm.device.{bus}.Assign', + f'{backend_domain}+{port_id}+_', encoded_options)] = b'0\0' for feature, value in vm['features'].items(): @@ -1533,7 +1529,7 @@ def create_limited_tmpdir(self, size): # pylint: disable=consider-using-with tmpdir = tempfile.TemporaryDirectory() subprocess.run(['sudo', 'mount', '-t', 'tmpfs', 'none', - tmpdir.name, '-o', 'size={}'.format(size)], + tmpdir.name, '-o', f'size={size}'], check=True) self.addCleanup(self.cleanup_tmpdir, tmpdir) return tmpdir.name diff --git a/qubesadmin/tests/backup/dispvm.py b/qubesadmin/tests/backup/dispvm.py index 9fd70c69..6ef854ec 100644 --- a/qubesadmin/tests/backup/dispvm.py +++ b/qubesadmin/tests/backup/dispvm.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/devices.py b/qubesadmin/tests/devices.py index 6b923856..8898dff6 100644 --- a/qubesadmin/tests/devices.py +++ b/qubesadmin/tests/devices.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/errors.py b/qubesadmin/tests/errors.py index c4157471..449725e9 100644 --- a/qubesadmin/tests/errors.py +++ b/qubesadmin/tests/errors.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -48,7 +47,7 @@ def test_002_exception_with_numbers(self): with self.assertRaises(qubesadmin.exc.QubesException) as context: list(self.app.domains) except TypeError as e: - self.fail('TypeError: {!s}'.format(e)) + self.fail(f'TypeError: {e!s}') self.assertEqual(str(context.exception), 'An error occurred: 1, 2') def test_010_empty(self): diff --git a/qubesadmin/tests/events.py b/qubesadmin/tests/events.py index 8900f8d0..c4bd6174 100644 --- a/qubesadmin/tests/events.py +++ b/qubesadmin/tests/events.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/features.py b/qubesadmin/tests/features.py index cf980a69..dcc8652f 100644 --- a/qubesadmin/tests/features.py +++ b/qubesadmin/tests/features.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/firewall.py b/qubesadmin/tests/firewall.py index e6d788e9..baf62d3e 100644 --- a/qubesadmin/tests/firewall.py +++ b/qubesadmin/tests/firewall.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/label.py b/qubesadmin/tests/label.py index f5b2a4ab..8d6b42cd 100644 --- a/qubesadmin/tests/label.py +++ b/qubesadmin/tests/label.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -33,7 +32,7 @@ def test_000_list(self): for label in self.app.labels.values(): self.assertNotIn(label.name, seen) seen.add(label.name) - self.assertEqual(seen, set(['green', 'red', 'black'])) + self.assertEqual(seen, {'green', 'red', 'black'}) def test_001_list_names(self): self.app.expected_calls[ @@ -43,7 +42,7 @@ def test_001_list_names(self): for label in self.app.labels: self.assertNotIn(label, seen) seen.add(label) - self.assertEqual(seen, set(['green', 'red', 'black'])) + self.assertEqual(seen, {'green', 'red', 'black'}) def test_002_list_keys(self): self.app.expected_calls[ @@ -53,7 +52,7 @@ def test_002_list_keys(self): for label in self.app.labels.keys(): self.assertNotIn(label, seen) seen.add(label) - self.assertEqual(seen, set(['green', 'red', 'black'])) + self.assertEqual(seen, {'green', 'red', 'black'}) def test_003_list_items(self): self.app.expected_calls[ @@ -64,7 +63,7 @@ def test_003_list_items(self): self.assertEqual(name, label.name) self.assertNotIn(name, seen) seen.add(name) - self.assertEqual(seen, set(['green', 'red', 'black'])) + self.assertEqual(seen, {'green', 'red', 'black'}) def test_010_get(self): self.app.expected_calls[ diff --git a/qubesadmin/tests/mock_app.py b/qubesadmin/tests/mock_app.py index 1a707bde..89df66ba 100644 --- a/qubesadmin/tests/mock_app.py +++ b/qubesadmin/tests/mock_app.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -65,7 +64,7 @@ from unittest import mock from copy import deepcopy -from typing import List, Optional, Dict, Tuple, Any +from typing import Any import qubesadmin.events from qubesadmin.tests import QubesTest @@ -263,7 +262,7 @@ def __init__( qube_name: str, volume_type: str, outdated: bool = False, - usage_perc: Optional[float] = 0.0, + usage_perc: float | None = 0.0, save_on_stop: bool = False, ): """ @@ -341,11 +340,11 @@ def __init__( qapp: QubesTest, klass: str = "AppVM", running: bool = False, - features: Optional[Dict] = None, - usage: Optional[float] = 0.0, - tags: Optional[List] = None, - firewall_rules: Optional[List[Dict[str, str]]] = None, - override_default_props: Optional[Dict] = None, + features: dict | None = None, + usage: float | None = 0.0, + tags: list | None = None, + firewall_rules: list[dict[str, str]] | None = None, + override_default_props: dict | None = None, **kwargs, ): """ @@ -645,7 +644,7 @@ def setup_device_calls(self): (self.name, "admin.vm.device.mic.Attached", None, None) ] = b"0\x00" - def setup_firewall_rules(self, rule_list: List[Dict[str, str]]): + def setup_firewall_rules(self, rule_list: list[dict[str, str]]): """setup firewall with provided rules: rules should be provided as list of dictionaries. Typical keys are action, proto, dsthost, dstports, icmptype, specialtarget, expire and comment""" @@ -693,8 +692,8 @@ def __init__( port: str, product: str, vendor: str, - attached: Optional[str] = None, - assigned: Optional[List[Tuple[str, str, list[Any] | None]]] = None, + attached: str | None = None, + assigned: list[tuple[str, str, list[Any] | None]] | None = None, ): """ :param qapp: QubesTest object @@ -826,7 +825,7 @@ def __init__(self): self._devices = [] self.app.qubesd_connection_type = "qrexec" - self._qubes: Dict[str, MockQube] = {"dom0": MockAdminVM(self)} + self._qubes: dict[str, MockQube] = {"dom0": MockAdminVM(self)} labels = [ "red", @@ -905,7 +904,7 @@ def populate_feature_calls(self): + b"\x00" ) - def set_global_property(self, property_name: str, value: Optional[str]): + def set_global_property(self, property_name: str, value: str | None): self._global_properties[property_name].value = value def update_global_properties(self): @@ -1254,7 +1253,7 @@ def __init__( self, event_subject: str, event_name: str, - additional_keys: Optional[List[Tuple[str, str]]] = None, + additional_keys: list[tuple[str, str]] | None = None, ): """ event subject - the name of the object that fired the event diff --git a/qubesadmin/tests/storage.py b/qubesadmin/tests/storage.py index 76d168ec..f2cd6f03 100644 --- a/qubesadmin/tests/storage.py +++ b/qubesadmin/tests/storage.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -286,7 +285,7 @@ def test_000_list(self): self.assertNotIn(pool.name, seen) seen.add(pool.name) - self.assertEqual(seen, set(['file', 'lvm'])) + self.assertEqual(seen, {'file', 'lvm'}) self.assertAllCalled() def test_010_config(self): @@ -354,7 +353,7 @@ def test_020_volumes(self): self.assertNotIn(volume.vid, seen) seen.add(volume.vid) - self.assertEqual(seen, set(['vol1', 'vol2'])) + self.assertEqual(seen, {'vol1', 'vol2'}) self.assertAllCalled() def test_030_pool_drivers(self): @@ -362,9 +361,9 @@ def test_030_pool_drivers(self): ('dom0', 'admin.pool.ListDrivers', None, None)] = \ b'0\x00file dir_path revisions_to_keep\n' \ b'lvm volume_group thin_pool revisions_to_keep\n' - self.assertEqual(set(self.app.pool_drivers), set(['file', 'lvm'])) + self.assertEqual(set(self.app.pool_drivers), {'file', 'lvm'}) self.assertEqual(set(self.app.pool_driver_parameters('file')), - set(['dir_path', 'revisions_to_keep'])) + {'dir_path', 'revisions_to_keep'}) self.assertAllCalled() def test_040_add(self): diff --git a/qubesadmin/tests/tags.py b/qubesadmin/tests/tags.py index f7e251bb..d1ea9662 100644 --- a/qubesadmin/tests/tags.py +++ b/qubesadmin/tests/tags.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/__init__.py b/qubesadmin/tests/tools/__init__.py index 83883150..64d43e7f 100644 --- a/qubesadmin/tests/tools/__init__.py +++ b/qubesadmin/tests/tools/__init__.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -29,10 +28,7 @@ class StdoutBuffer: def __init__(self): self.orig_stdout = None - if sys.version_info[0] >= 3: - self.stdout = io.StringIO() - else: - self.stdout = io.BytesIO() + self.stdout = io.StringIO() def __enter__(self): self.orig_stdout = sys.stdout @@ -47,10 +43,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): class StderrBuffer: def __init__(self): self.orig_stderr = None - if sys.version_info[0] >= 3: - self.stderr = io.StringIO() - else: - self.stderr = io.BytesIO() + self.stderr = io.StringIO() def __enter__(self): self.orig_stderr = sys.stderr diff --git a/qubesadmin/tests/tools/qubes_prefs.py b/qubesadmin/tests/tools/qubes_prefs.py index 48112d5d..97fdc7b0 100644 --- a/qubesadmin/tests/tools/qubes_prefs.py +++ b/qubesadmin/tests/tools/qubes_prefs.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_backup.py b/qubesadmin/tests/tools/qvm_backup.py index 24d57d4e..430cd000 100644 --- a/qubesadmin/tests/tools/qvm_backup.py +++ b/qubesadmin/tests/tools/qvm_backup.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_backup_restore.py b/qubesadmin/tests/tools/qvm_backup_restore.py index 3e845969..e71d13ba 100644 --- a/qubesadmin/tests/tools/qvm_backup_restore.py +++ b/qubesadmin/tests/tools/qvm_backup_restore.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_check.py b/qubesadmin/tests/tools/qvm_check.py index ad59f038..0104168c 100644 --- a/qubesadmin/tests/tools/qvm_check.py +++ b/qubesadmin/tests/tools/qvm_check.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_clone.py b/qubesadmin/tests/tools/qvm_clone.py index 63063374..20c35add 100644 --- a/qubesadmin/tests/tools/qvm_clone.py +++ b/qubesadmin/tests/tools/qvm_clone.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_create.py b/qubesadmin/tests/tools/qvm_create.py index e1f29c67..7c846211 100644 --- a/qubesadmin/tests/tools/qvm_create.py +++ b/qubesadmin/tests/tools/qvm_create.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_features.py b/qubesadmin/tests/tools/qvm_features.py index 4a6bad8e..8e92c7dd 100644 --- a/qubesadmin/tests/tools/qvm_features.py +++ b/qubesadmin/tests/tools/qvm_features.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_firewall.py b/qubesadmin/tests/tools/qvm_firewall.py index b6d6ef34..00d0009e 100644 --- a/qubesadmin/tests/tools/qvm_firewall.py +++ b/qubesadmin/tests/tools/qvm_firewall.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, http://www.qubes-os.org # @@ -172,7 +171,7 @@ def test_002_list_expire(self): r'0 accept qubes-os.org tcp 443 - ' r' - \+(.{4})s -', line) - self.assertTrue(match, "no match for: {!r}".format(line)) + self.assertTrue(match, f"no match for: {line!r}") duration = int(match.group(1)) self.assertTrue(3590 < duration <= 3600) diff --git a/qubesadmin/tests/tools/qvm_kill.py b/qubesadmin/tests/tools/qvm_kill.py index 081fd83e..732a9338 100644 --- a/qubesadmin/tests/tools/qvm_kill.py +++ b/qubesadmin/tests/tools/qvm_kill.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_ls.py b/qubesadmin/tests/tools/qvm_ls.py index 23a1d6f0..0da09c1c 100644 --- a/qubesadmin/tests/tools/qvm_ls.py +++ b/qubesadmin/tests/tools/qvm_ls.py @@ -345,7 +345,7 @@ def test_100_list_with_status(self): self.app.expected_calls[ ('vm1', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() # setup sys-net @@ -353,7 +353,7 @@ def test_100_list_with_status(self): self.app.expected_calls[ ('sys-net', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() # setup template1 @@ -362,7 +362,7 @@ def test_100_list_with_status(self): self.app.expected_calls[ ('template1', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() with qubesadmin.tests.tools.StdoutBuffer() as stdout: @@ -395,7 +395,7 @@ def test_101_list_selected(self): self.app.expected_calls[ ('vm1', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() # setup sys-net @@ -403,7 +403,7 @@ def test_101_list_selected(self): self.app.expected_calls[ ('sys-net', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() with qubesadmin.tests.tools.StdoutBuffer() as stdout: @@ -468,7 +468,7 @@ def test_111_filter_class(self): self.app.expected_calls[ ('template1', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() with qubesadmin.tests.tools.StdoutBuffer() as stdout: @@ -495,7 +495,7 @@ def test_112_filter_label(self): self.app.expected_calls[ ('sys-net', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() # setup template1 @@ -504,7 +504,7 @@ def test_112_filter_label(self): self.app.expected_calls[ ('template1', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() with qubesadmin.tests.tools.StdoutBuffer() as stdout: @@ -530,7 +530,7 @@ def test_113_filter_template_source(self): self.app.expected_calls[ ('sys-net', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() # setup template1 @@ -539,7 +539,7 @@ def test_113_filter_template_source(self): self.app.expected_calls[ ('template1', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() with qubesadmin.tests.tools.StdoutBuffer() as stdout: @@ -566,7 +566,7 @@ def test_114_filter_netvm_is(self): self.app.expected_calls[ ('sys-net', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() # setup template1 @@ -575,7 +575,7 @@ def test_114_filter_netvm_is(self): self.app.expected_calls[ ('template1', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() with qubesadmin.tests.tools.StdoutBuffer() as stdout: @@ -603,7 +603,7 @@ def test_115_internal_servicevm_pending_updates(self): self.app.expected_calls[ ('internalvm', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() self.app.expected_calls[ @@ -623,7 +623,7 @@ def test_115_internal_servicevm_pending_updates(self): self.app.expected_calls[ ('service-vm', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() self.app.expected_calls[ @@ -681,7 +681,7 @@ def test_116_features(self): self.app.expected_calls[ ('internalvm', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() self.app.expected_calls[ @@ -693,7 +693,7 @@ def test_116_features(self): self.app.expected_calls[ ('service-vm', 'admin.vm.property.GetAll', None, None)] = \ b'0\x00' + ''.join( - '{} default=True {}\n'.format(key, value) + f'{key} default=True {value}\n' for key, value in props.items()).encode() self.app.expected_calls[ diff --git a/qubesadmin/tests/tools/qvm_notes.py b/qubesadmin/tests/tools/qvm_notes.py index 20956a00..9abf6729 100644 --- a/qubesadmin/tests/tools/qvm_notes.py +++ b/qubesadmin/tests/tools/qvm_notes.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_pause.py b/qubesadmin/tests/tools/qvm_pause.py index 88f89283..da9b1865 100644 --- a/qubesadmin/tests/tools/qvm_pause.py +++ b/qubesadmin/tests/tools/qvm_pause.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_pool.py b/qubesadmin/tests/tools/qvm_pool.py index cce5e7a5..1e999b35 100644 --- a/qubesadmin/tests/tools/qvm_pool.py +++ b/qubesadmin/tests/tools/qvm_pool.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_pool_legacy.py b/qubesadmin/tests/tools/qvm_pool_legacy.py index ef6ed578..6e47fada 100644 --- a/qubesadmin/tests/tools/qvm_pool_legacy.py +++ b/qubesadmin/tests/tools/qvm_pool_legacy.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_prefs.py b/qubesadmin/tests/tools/qvm_prefs.py index 97754eab..1280bab2 100644 --- a/qubesadmin/tests/tools/qvm_prefs.py +++ b/qubesadmin/tests/tools/qvm_prefs.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -20,8 +19,6 @@ # pylint: disable=missing-docstring -import sys - import qubesadmin.tests import qubesadmin.tests.tools import qubesadmin.tools.qvm_prefs @@ -56,11 +53,8 @@ def test_001_no_vm(self): with self.assertRaises(SystemExit): with qubesadmin.tests.tools.StderrBuffer() as stderr: qubesadmin.tools.qvm_prefs.main([], app=self.app) - if sys.version_info[0] == 2: - self.assertIn('too few arguments', stderr.getvalue()) - else: - self.assertIn('error: the following arguments are required: VMNAME', - stderr.getvalue()) + self.assertIn('error: the following arguments are required: VMNAME', + stderr.getvalue()) self.assertAllCalled() def test_002_set_property(self): diff --git a/qubesadmin/tests/tools/qvm_remove.py b/qubesadmin/tests/tools/qvm_remove.py index 6f87038a..b7c787b4 100644 --- a/qubesadmin/tests/tools/qvm_remove.py +++ b/qubesadmin/tests/tools/qvm_remove.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_run.py b/qubesadmin/tests/tools/qvm_run.py index 15044e42..86ac68c8 100644 --- a/qubesadmin/tests/tools/qvm_run.py +++ b/qubesadmin/tests/tools/qvm_run.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_service.py b/qubesadmin/tests/tools/qvm_service.py index 3e94f175..72713c74 100644 --- a/qubesadmin/tests/tools/qvm_service.py +++ b/qubesadmin/tests/tools/qvm_service.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_shutdown.py b/qubesadmin/tests/tools/qvm_shutdown.py index e4257967..caa93c49 100644 --- a/qubesadmin/tests/tools/qvm_shutdown.py +++ b/qubesadmin/tests/tools/qvm_shutdown.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_start.py b/qubesadmin/tests/tools/qvm_start.py index b6ab478a..3c430ff1 100644 --- a/qubesadmin/tests/tools/qvm_start.py +++ b/qubesadmin/tests/tools/qvm_start.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_start_daemon.py b/qubesadmin/tests/tools/qvm_start_daemon.py index 666175da..79d35114 100644 --- a/qubesadmin/tests/tools/qvm_start_daemon.py +++ b/qubesadmin/tests/tools/qvm_start_daemon.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_tags.py b/qubesadmin/tests/tools/qvm_tags.py index b151d61f..2c135541 100644 --- a/qubesadmin/tests/tools/qvm_tags.py +++ b/qubesadmin/tests/tools/qvm_tags.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_template.py b/qubesadmin/tests/tools/qvm_template.py index 12d503ad..3798fa50 100644 --- a/qubesadmin/tests/tools/qvm_template.py +++ b/qubesadmin/tests/tools/qvm_template.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -434,7 +433,7 @@ def test_100_install_local_success( mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) mock_time.now.return_value = \ - datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) + datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.UTC) with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', '/tmp/test.lock'), \ mock.patch('datetime.datetime', new=mock_time), \ @@ -549,7 +548,7 @@ def test_101_install_local_postprocargs_success( mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) mock_time.now.return_value = \ - datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) + datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.UTC) with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', '/tmp/test.lock'), \ mock.patch('datetime.datetime', new=mock_time), \ @@ -916,7 +915,7 @@ def test_107_install_download_success( mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) mock_time.now.return_value = \ - datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) + datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.UTC) with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', '/tmp/test.lock'), \ mock.patch('datetime.datetime', new=mock_time), \ @@ -1019,7 +1018,7 @@ def test_108_install_download_fail_exists( mock_dl_list.return_value = dl_list mock_time = mock.Mock(wraps=datetime.datetime) mock_time.now.return_value = \ - datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) + datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.UTC) with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', '/tmp/test.lock'), \ mock.patch('datetime.datetime', new=mock_time), \ @@ -1097,7 +1096,7 @@ def test_109_install_fail_extract( mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) mock_time.now.return_value = \ - datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) + datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.UTC) # Extraction error mock_extract.return_value = False with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', @@ -1504,7 +1503,7 @@ def test_120_b_qrexec_repoquery_success_dnf5(self, mock_payload): 'qubes-templates-itl', 228848654, datetime.datetime(2024, 5, 27, 21, 57, 22, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPLv3+', 'http://www.qubes-os.org', 'Qubes OS template for debian-12-minimal', @@ -2843,7 +2842,7 @@ def test_152_list_templates_extras_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for fedora-31', @@ -2932,7 +2931,7 @@ def test_153_list_templates_upgrades_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for fedora-31', @@ -2946,7 +2945,7 @@ def test_153_list_templates_upgrades_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for fedora-31', @@ -2960,7 +2959,7 @@ def test_153_list_templates_upgrades_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for fedora-31', @@ -3036,7 +3035,7 @@ def __test_list_templates_all_success(self, operation, 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for fedora-31', @@ -3247,7 +3246,7 @@ def test_170_search_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for fedora-31', @@ -3261,7 +3260,7 @@ def test_170_search_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Older Qubes template for fedora-31', @@ -3275,7 +3274,7 @@ def test_170_search_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org/test-vm', 'Qubes template for fedora-31', @@ -3332,7 +3331,7 @@ def test_171_search_summary_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for test-vm :)', @@ -3346,7 +3345,7 @@ def test_171_search_summary_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm', @@ -3360,7 +3359,7 @@ def test_171_search_summary_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for test-vm', @@ -3374,7 +3373,7 @@ def test_171_search_summary_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for test-vm-2', @@ -3414,7 +3413,7 @@ def test_172_search_namesummaryexact_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm', @@ -3428,7 +3427,7 @@ def test_172_search_namesummaryexact_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm', @@ -3442,7 +3441,7 @@ def test_172_search_namesummaryexact_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm', @@ -3481,7 +3480,7 @@ def test_173_search_multiquery_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm', @@ -3495,7 +3494,7 @@ def test_173_search_multiquery_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm', @@ -3509,7 +3508,7 @@ def test_173_search_multiquery_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Summary', @@ -3544,7 +3543,7 @@ def test_174_search_multiquery_exact_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'summary', @@ -3558,7 +3557,7 @@ def test_174_search_multiquery_exact_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm Summary', @@ -3595,7 +3594,7 @@ def test_175_search_all_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org/keyword-url', 'summary', @@ -3609,7 +3608,7 @@ def test_175_search_all_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm Summary', @@ -3623,7 +3622,7 @@ def test_175_search_all_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'test-vm-exac2', 'test-vm Summary', @@ -3637,7 +3636,7 @@ def test_175_search_all_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'test-vm Summary', @@ -3651,7 +3650,7 @@ def test_175_search_all_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Summary', @@ -3693,7 +3692,7 @@ def test_176_search_wildcard_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org', 'Qubes template for fedora-31', @@ -3707,7 +3706,7 @@ def test_176_search_wildcard_success(self, mock_query): 'qubes-templates-itl', 1048576, datetime.datetime(2020, 9, 1, 14, 30, - tzinfo=datetime.timezone.utc), + tzinfo=datetime.UTC), 'GPL', 'https://qubes-os.org/test-vm', 'Qubes template for fedora-31', @@ -3735,7 +3734,7 @@ def _mock_qrexec_download(self, args, app, spec, path, key, dlsize=None, refresh=False): # pylint: disable=unused-argument self.assertFalse(os.path.exists(path), - '{} should not exist before'.format(path)) + f'{path} should not exist before') # just create an empty file with open(path, 'wb') as f_template: f_template.truncate(dlsize) @@ -4216,7 +4215,7 @@ def _mock_qrexec_download_short(self, args, app, spec, path, key, dlsize=None, refresh=False): # pylint: disable=unused-argument self.assertFalse(os.path.exists(path), - '{} should not exist before'.format(path)) + f'{path} should not exist before') # just create an empty file with open(path, 'wb') as f_pkg: f_pkg.truncate(dlsize // 2) @@ -4292,7 +4291,7 @@ def test_200_reinstall_success( mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) mock_time.now.return_value = \ - datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) + datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.UTC) selector = qubesadmin.tools.qvm_template.VersionSelector.REINSTALL with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', '/tmp/test.lock'), \ mock.patch('datetime.datetime', new=mock_time), \ @@ -4510,7 +4509,7 @@ def test_202_reinstall_local_success( mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) mock_time.now.return_value = \ - datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) + datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.UTC) selector = qubesadmin.tools.qvm_template.VersionSelector.REINSTALL with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', '/tmp/test.lock'), \ mock.patch('datetime.datetime', new=mock_time), \ @@ -5475,7 +5474,7 @@ def test_240_qubes_release(self, mock_exists): self.assertEqual(mock_exists.mock_calls, [ mock.call('/usr/share/qubes/marker-vm') ]) - mock_open.assert_called_with('/etc/os-release', 'r', + mock_open.assert_called_with('/etc/os-release', encoding='ascii') self.assertAllCalled() @@ -5500,7 +5499,7 @@ def test_241_qubes_release_quotes(self, mock_exists): self.assertEqual(mock_exists.mock_calls, [ mock.call('/usr/share/qubes/marker-vm') ]) - mock_open.assert_called_with('/etc/os-release', 'r', + mock_open.assert_called_with('/etc/os-release', encoding='ascii') self.assertAllCalled() @@ -5520,7 +5519,7 @@ def test_242_qubes_release_quotes(self, mock_exists): self.assertEqual(mock_exists.mock_calls, [ mock.call('/usr/share/qubes/marker-vm') ]) - mock_open.assert_called_with('/usr/share/qubes/marker-vm', 'r', + mock_open.assert_called_with('/usr/share/qubes/marker-vm', encoding='ascii') self.assertAllCalled() diff --git a/qubesadmin/tests/tools/qvm_template_postprocess.py b/qubesadmin/tests/tools/qvm_template_postprocess.py index 26cdf832..0826718e 100644 --- a/qubesadmin/tests/tools/qvm_template_postprocess.py +++ b/qubesadmin/tests/tools/qvm_template_postprocess.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_unpause.py b/qubesadmin/tests/tools/qvm_unpause.py index b09c89d3..c45d23b9 100644 --- a/qubesadmin/tests/tools/qvm_unpause.py +++ b/qubesadmin/tests/tools/qvm_unpause.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/tools/qvm_volume.py b/qubesadmin/tests/tools/qvm_volume.py index 762dd1d4..077556dc 100644 --- a/qubesadmin/tests/tools/qvm_volume.py +++ b/qubesadmin/tests/tools/qvm_volume.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/utils.py b/qubesadmin/tests/utils.py index 36f7434c..8c484976 100644 --- a/qubesadmin/tests/utils.py +++ b/qubesadmin/tests/utils.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -90,10 +89,10 @@ def test_01_only_vm(self): self.assertSetEqual( set(result), - set([(vm, prop) for vm in self.vms for prop in self.vm_properties + {(vm, prop) for vm in self.vms for prop in self.vm_properties if (not vm.startswith('template') or not prop.startswith( - 'template')) and vm != 'template1']), + 'template')) and vm != 'template1'}, "Incorrect VM properties listed.") def test_02_empty(self): diff --git a/qubesadmin/tests/vm/__init__.py b/qubesadmin/tests/vm/__init__.py index ba0a10ac..3c9c761c 100644 --- a/qubesadmin/tests/vm/__init__.py +++ b/qubesadmin/tests/vm/__init__.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/vm/actions.py b/qubesadmin/tests/vm/actions.py index 570bd87b..16f090ab 100644 --- a/qubesadmin/tests/vm/actions.py +++ b/qubesadmin/tests/vm/actions.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/vm/dispvm.py b/qubesadmin/tests/vm/dispvm.py index 02d8030a..4372487d 100644 --- a/qubesadmin/tests/vm/dispvm.py +++ b/qubesadmin/tests/vm/dispvm.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/vm/properties.py b/qubesadmin/tests/vm/properties.py index 126f2d6d..8e16388a 100644 --- a/qubesadmin/tests/vm/properties.py +++ b/qubesadmin/tests/vm/properties.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tests/vm/storage.py b/qubesadmin/tests/vm/storage.py index f25f2f78..af7dce87 100644 --- a/qubesadmin/tests/vm/storage.py +++ b/qubesadmin/tests/vm/storage.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -29,7 +28,7 @@ def test_000_list_volumes(self): ('test-vm', 'admin.vm.volume.List', None, None)] = \ b'0\x00root\nprivate\nvolatile\nmodules\n' self.assertEqual(set(self.vm.volumes.keys()), - set(['root', 'private', 'volatile', 'modules'])) + {'root', 'private', 'volatile', 'modules'}) self.assertAllCalled() def test_001_volume_get(self): diff --git a/qubesadmin/tools/__init__.py b/qubesadmin/tools/__init__.py index b2f7f17b..18e11206 100644 --- a/qubesadmin/tools/__init__.py +++ b/qubesadmin/tools/__init__.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -21,7 +20,6 @@ '''Qubes' command line tools ''' -from __future__ import print_function import argparse import fnmatch @@ -71,7 +69,7 @@ def __call__(self, parser, namespace, values, option_string=None): try: prop, value = values.split('=', 1) except ValueError: - parser.error('invalid property token: {!r}'.format(values)) + parser.error(f'invalid property token: {values!r}') properties = getattr(namespace, self.dest) # copy it, to not modify _mutable_ self.default @@ -95,9 +93,9 @@ def __init__(self, required=False, help=None): if help is None: - help = 'set {!r} property to a value'.format(dest) + help = f'set {dest!r} property to a value' if const is not None: - help += ' {!r}'.format(const) + help += f' {const!r}' if const is not None: nargs = 0 @@ -176,7 +174,7 @@ def parse_qubes_app(self, parser, namespace): try: namespace.domains += [app.domains[vm_name]] except KeyError: - parser.error('no such domain: {!r}'.format(vm_name)) + parser.error(f'no such domain: {vm_name!r}') else: destinations = set() for destination in getattr(namespace, self.dest): @@ -191,7 +189,7 @@ def parse_qubes_app(self, parser, namespace): try: namespace.domains += [app.domains[vm_name]] except KeyError: - parser.error('no such domain: {!r}'.format(vm_name)) + parser.error(f'no such domain: {vm_name!r}') class RunningVmNameAction(VmNameAction): @@ -263,7 +261,7 @@ def parse_qubes_app(self, parser, namespace): else: setattr(namespace, self.dest, volume[0]) except KeyError: - parser.error_runtime('no pool {!r}'.format(pool_name)) + parser.error_runtime(f'no pool {pool_name!r}') except ValueError: parser.error('expected a pool & volume id combination like foo:bar') @@ -302,7 +300,7 @@ def parse_qubes_app(self, parser, namespace): parser.error_runtime('vm {!r} has no volume {!r}'.format( vm_name, vol_name)) except KeyError: - parser.error_runtime('no vm {!r}'.format(vm_name)) + parser.error_runtime(f'no vm {vm_name!r}') except ValueError: parser.error('expected a vm & volume combination like foo:bar') @@ -448,7 +446,7 @@ def error_runtime(self, message, exit_code=1): :param str message: message to show ''' - self.exit(exit_code, '{}: error: {}\n'.format(self.prog, message)) + self.exit(exit_code, f'{self.prog}: error: {message}\n') @staticmethod @@ -501,7 +499,7 @@ def __call__(self, parser, namespace, values, option_string=None): for pseudo_action in subparsers_action._choices_actions: choice = pseudo_action.dest.split(' ', 1)[0] subparser = subparsers_action.choices[choice] - print("\nCommand '{}':".format(choice)) + print(f"\nCommand '{choice}':") choice_help = subparser.format_usage() choice_help = self._indent(2, choice_help) print(choice_help) diff --git a/qubesadmin/tools/dochelpers.py b/qubesadmin/tools/dochelpers.py index d9d2ddde..3d2f38ce 100644 --- a/qubesadmin/tools/dochelpers.py +++ b/qubesadmin/tools/dochelpers.py @@ -54,16 +54,16 @@ def make_rst_section(heading, char): """Format a section header in rst""" - return '{}\n{}\n\n'.format(heading, char[0] * len(heading)) + return f'{heading}\n{char[0] * len(heading)}\n\n' def prepare_manpage(command): """Build a man page skeleton""" parser = qubesadmin.tools.get_parser_for_command(command) stream = io.StringIO() - stream.write('.. program:: {}\n\n'.format(command)) + stream.write(f'.. program:: {command}\n\n') stream.write(make_rst_section( - ':program:`{}` -- {}'.format(command, parser.description), '=')) + f':program:`{command}` -- {parser.description}', '=')) stream.write(""".. warning:: This page was autogenerated from command-line parser. It shouldn't be 1:1 @@ -81,7 +81,7 @@ def prepare_manpage(command): # replace METAVARS with *METAVARS* usage = re.sub(r'\b([A-Z]{2,})\b', r'*\1*', usage) - stream.write(':command:`{}` {}\n\n'.format(command, usage)) + stream.write(f':command:`{command}` {usage}\n\n') stream.write(make_rst_section('Options', '-')) @@ -94,7 +94,7 @@ def prepare_manpage(command): sorted(action.option_strings))) else: stream.write(', '.join(sorted(action.option_strings))) - stream.write('\n\n {}\n\n'.format(action.help)) + stream.write(f'\n\n {action.help}\n\n') stream.write(make_rst_section('Authors', '-')) stream.write("""\ @@ -134,7 +134,7 @@ def visit_desc_name(self, node): self.args.remove(arg) except KeyError: raise sphinx.errors.SphinxError( - 'No such argument for {!r}: {!r}'.format(self.command, arg)) + f'No such argument for {self.command!r}: {arg!r}') def check_undocumented_arguments(self, ignored_options=None): """ Call this to check if any undocumented arguments are left. @@ -193,7 +193,7 @@ def visit_section(self, node): del self.sub_commands[cmd] except KeyError: raise sphinx.errors.SphinxError( - 'No such sub-command {!r}'.format(sub_cmd)) + f'No such sub-command {sub_cmd!r}') def visit_Text(self, node): """ If the visited text node starts with 'alias: ', all the provided @@ -291,7 +291,7 @@ def check_man_args(app, doctree, docname): if os.path.basename(dirname) != 'manpages': return - msg = 'Checking arguments for {!r}'.format(command) + msg = f'Checking arguments for {command!r}' if log: log.info(msg) doctree.walk(ManpageCheckVisitor(app, command, doctree)) diff --git a/qubesadmin/tools/qubes_prefs.py b/qubesadmin/tools/qubes_prefs.py index 2e354c34..f95d60ff 100644 --- a/qubesadmin/tools/qubes_prefs.py +++ b/qubesadmin/tools/qubes_prefs.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, http://www.qubes-os.org # @@ -20,7 +19,6 @@ ''' Manipulate global properties.''' -from __future__ import print_function import sys diff --git a/qubesadmin/tools/qvm_backup.py b/qubesadmin/tools/qvm_backup.py index 29b5182f..ad060295 100644 --- a/qubesadmin/tools/qvm_backup.py +++ b/qubesadmin/tools/qvm_backup.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -121,7 +120,7 @@ def print_progress(expected_profile, _subject, _event, backup_profile, '''Event handler for reporting backup progress''' if backup_profile != expected_profile: return - sys.stderr.write('\rMaking a backup... {:.02f}%'.format(float(progress))) + sys.stderr.write(f'\rMaking a backup... {float(progress):.02f}%') def main(args=None, app=None): '''Main function of qvm-backup tool''' @@ -139,7 +138,7 @@ def main(args=None, app=None): # don't care about collisions because only the user in dom0 can # trigger this, and qrexec policy should not allow random VM # to execute the same backup in the meantime - profile_name = 'backup-run-{}'.format(os.getpid()) + profile_name = f'backup-run-{os.getpid()}' # first write the backup profile without passphrase, to display # summary profile_path = os.path.join( @@ -178,7 +177,7 @@ def main(args=None, app=None): 'dom0', 'admin.backup.Info', profile_name) print(backup_summary.decode()) except QubesException as err: - print('\nBackup preparation error: {}'.format(err), file=sys.stderr) + print(f'\nBackup preparation error: {err}', file=sys.stderr) return 1 if not args.yes: @@ -220,7 +219,7 @@ def main(args=None, app=None): loop.run_until_complete(loop.run_in_executor(None, args.app.qubesd_call, 'dom0', 'admin.backup.Execute', profile_name)) except QubesException as err: - print('\nBackup error: {}'.format(err), file=sys.stderr) + print(f'\nBackup error: {err}', file=sys.stderr) return 1 finally: if have_events: diff --git a/qubesadmin/tools/qvm_backup_restore.py b/qubesadmin/tools/qvm_backup_restore.py index a1506a46..50829a9c 100644 --- a/qubesadmin/tools/qvm_backup_restore.py +++ b/qubesadmin/tools/qvm_backup_restore.py @@ -233,7 +233,7 @@ def main(args=None, app=None): try: appvm = args.app.domains[args.appvm] except KeyError: - parser.error('no such domain: {!r}'.format(args.appvm)) + parser.error(f'no such domain: {args.appvm!r}') if args.location_is_service and not args.appvm: parser.error('--location-is-service option requires -d') diff --git a/qubesadmin/tools/qvm_create.py b/qubesadmin/tools/qvm_create.py index 6f36f18a..caacb18b 100644 --- a/qubesadmin/tools/qvm_create.py +++ b/qubesadmin/tools/qvm_create.py @@ -27,7 +27,6 @@ # TODO list labels (maybe in qvm-prefs) # TODO features, devices, tags -from __future__ import print_function import argparse import os @@ -154,7 +153,7 @@ def main(args=None, app=None): try: args.app.get_vm_class(args.cls) except KeyError: - parser.error('no such domain class: {!r}'.format(args.cls)) + parser.error(f'no such domain class: {args.cls!r}') try: if args.cls == 'StandaloneVM' and 'template' in args.properties: @@ -174,7 +173,7 @@ def main(args=None, app=None): pool=pool, pools=pools) except qubesadmin.exc.QubesException as e: - args.app.log.error('Error creating VM: {!s}'.format(e)) + args.app.log.error(f'Error creating VM: {e!s}') return 1 retcode = 0 diff --git a/qubesadmin/tools/qvm_device.py b/qubesadmin/tools/qvm_device.py index eb9ffb62..de931039 100644 --- a/qubesadmin/tools/qvm_device.py +++ b/qubesadmin/tools/qvm_device.py @@ -1,5 +1,3 @@ -# encoding=utf-8 - # # The Qubes OS Project, http://www.qubes-os.org # @@ -90,7 +88,7 @@ class Line: # pylint: disable=too-few-public-methods def __init__(self, device: DeviceInfo, assignment=False): - self.ident = "{!s}:{!s}".format(device.backend_domain, device.port_id) + self.ident = f"{device.backend_domain!s}:{device.port_id!s}" self.description = device.description self.assignment = assignment self.frontends = [] @@ -209,7 +207,7 @@ def _frontend_desc(vm, assignment, virtual=False): vm, mode, ", ".join( - "{}={}".format(key, value) + f"{key}={value}" for key, value in assignment.options.items() ), ) diff --git a/qubesadmin/tools/qvm_features.py b/qubesadmin/tools/qvm_features.py index 65e5344f..56012e1f 100644 --- a/qubesadmin/tools/qvm_features.py +++ b/qubesadmin/tools/qvm_features.py @@ -1,4 +1,3 @@ -# coding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -24,7 +23,6 @@ '''qvm-features - Manage domain's features''' -from __future__ import print_function import sys diff --git a/qubesadmin/tools/qvm_firewall.py b/qubesadmin/tools/qvm_firewall.py index 979fd6be..253993ba 100644 --- a/qubesadmin/tools/qvm_firewall.py +++ b/qubesadmin/tools/qvm_firewall.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, http://www.qubes-os.org # @@ -50,7 +49,7 @@ def __call__(self, _parser, namespace, values, option_string=None): for opt in values: if opt[-1] == '=': raise argparse.ArgumentError( - None, 'invalid rule description: {}'.format(opt)) + None, f'invalid rule description: {opt}') opt_elements = opt.split('=') if len(opt_elements) == 2: key, value = opt_elements @@ -58,12 +57,12 @@ def __call__(self, _parser, namespace, values, option_string=None): key, value = assumed_order[0], opt else: raise argparse.ArgumentError(None, - 'invalid rule description: {}'.format(opt)) + f'invalid rule description: {opt}') if key in ['dst4', 'dst6']: key = 'dsthost' if key not in allowed_opts: raise argparse.ArgumentError(None, - 'Invalid rule element: {}'.format(opt)) + f'Invalid rule element: {opt}') if key == 'expire' and value.startswith('+'): value = (datetime.datetime.now() + datetime.timedelta(seconds=int(value[1:]))).\ diff --git a/qubesadmin/tools/qvm_kill.py b/qubesadmin/tools/qvm_kill.py index 88dd6d5a..d362aaf3 100644 --- a/qubesadmin/tools/qvm_kill.py +++ b/qubesadmin/tools/qvm_kill.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -43,7 +42,7 @@ def main(args=None, app=None): domain.kill() except qubesadmin.exc.QubesVMNotStartedError: pass - except (IOError, OSError, qubesadmin.exc.QubesException) as e: + except (OSError, qubesadmin.exc.QubesException) as e: exit_code = 1 parser.print_error("Failed to kill '{}': {}".format( domain.name, e)) diff --git a/qubesadmin/tools/qvm_ls.py b/qubesadmin/tools/qvm_ls.py index abfb5f52..2574caa6 100644 --- a/qubesadmin/tools/qvm_ls.py +++ b/qubesadmin/tools/qvm_ls.py @@ -26,7 +26,6 @@ '''qvm-ls - List available domains''' -from __future__ import print_function import argparse import collections.abc @@ -327,7 +326,7 @@ def calc_used(vm, volume_name): if size == 0: return 0 usage = calc_usage(vm, volume_name) - return '{}%'.format(usage * 100 // size) + return f'{usage * 100 // size}%' # todo maxmem @@ -691,8 +690,8 @@ def get_parser(): help='exclude qubes having specific tag(s)') for pwstate in DOMAIN_POWER_STATES: - parser_filter.add_argument('--{}'.format(pwstate), action='store_true', - help='show {} VMs'.format(pwstate)) + parser_filter.add_argument(f'--{pwstate}', action='store_true', + help=f'show {pwstate} VMs') parser_filter.add_argument('--template-source', nargs='+', metavar='TEMPLATE', action='store', @@ -876,9 +875,9 @@ def main(args=None, app=None): try: key, value = feature.split('=', 1) except ValueError: - parser.error("Invalid argument: --features {}".format(feature)) + parser.error(f"Invalid argument: --features {feature}") if not key: - parser.error("Invalid argument: --features {}".format(feature)) + parser.error(f"Invalid argument: --features {feature}") if value == '': value = None elif value in ['\'\'', '""']: @@ -891,9 +890,9 @@ def main(args=None, app=None): try: key, value = pref.split('=', 1) except ValueError: - parser.error("Invalid argument: --prefs {}".format(pref)) + parser.error(f"Invalid argument: --prefs {pref}") if not key: - parser.error("Invalid argument: --prefs {}".format(pref)) + parser.error(f"Invalid argument: --prefs {pref}") if value == '': value = None elif value in ['\'\'', '""']: diff --git a/qubesadmin/tools/qvm_notes.py b/qubesadmin/tools/qvm_notes.py index 70f7d8a1..3d2c97cf 100644 --- a/qubesadmin/tools/qvm_notes.py +++ b/qubesadmin/tools/qvm_notes.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/tools/qvm_pause.py b/qubesadmin/tools/qvm_pause.py index 7a4b777e..378cb982 100644 --- a/qubesadmin/tools/qvm_pause.py +++ b/qubesadmin/tools/qvm_pause.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -51,7 +50,7 @@ def main(args=None, app=None): domain.suspend() else: domain.pause() - except (IOError, OSError, qubesadmin.exc.QubesException) as e: + except (OSError, qubesadmin.exc.QubesException) as e: exit_code = 1 parser.print_error(str(e)) diff --git a/qubesadmin/tools/qvm_pool.py b/qubesadmin/tools/qvm_pool.py index 971ce115..b2f1b016 100644 --- a/qubesadmin/tools/qvm_pool.py +++ b/qubesadmin/tools/qvm_pool.py @@ -20,7 +20,6 @@ '''Manages Qubes pools and their options''' -from __future__ import print_function import argparse import sys @@ -79,7 +78,7 @@ def remove_pools(args): errors.append('No such pool %s\n' % pool_name) except qubesadmin.exc.QubesException as e: errors.append( - 'Failed to remove pool %s: %s\n' % (pool_name, str(e))) + f'Failed to remove pool {pool_name}: {str(e)}\n') if errors: raise qubesadmin.exc.QubesException('\n'.join(errors)) @@ -92,12 +91,12 @@ def set_pool(args): for opt, value in options: if not hasattr(type(pool), opt): errors.append( - 'Setting option %s is not supported for pool %s\n' % ( + 'Setting option {} is not supported for pool {}\n'.format( opt, pool.name)) try: setattr(pool, opt, value) except qubesadmin.exc.QubesException as e: - errors.append('Failed to set option %s for pool %s: %s\n' % ( + errors.append('Failed to set option {} for pool {}: {}\n'.format( opt, pool.name, str(e))) if errors: raise qubesadmin.exc.QubesException('\n'.join(errors)) diff --git a/qubesadmin/tools/qvm_pool_legacy.py b/qubesadmin/tools/qvm_pool_legacy.py index d3968989..4fadf387 100644 --- a/qubesadmin/tools/qvm_pool_legacy.py +++ b/qubesadmin/tools/qvm_pool_legacy.py @@ -20,7 +20,6 @@ '''Manages Qubes pools and their options''' -from __future__ import print_function import argparse import sys @@ -179,14 +178,14 @@ def main(args=None, app=None): args.app.add_pool(name=args.name, driver=args.driver, **args.options) except qubesadmin.exc.QubesException as e: - parser.error('failed to add pool %s: %s\n' % (args.name, str(e))) + parser.error(f'failed to add pool {args.name}: {str(e)}\n') elif args.command == 'remove': try: args.app.remove_pool(args.name) except KeyError: parser.print_error('no such pool %s\n' % args.name) except qubesadmin.exc.QubesException as e: - parser.error('failed to remove pool %s: %s\n' % (args.name, str(e))) + parser.error(f'failed to remove pool {args.name}: {str(e)}\n') elif args.command == 'info': for pool in args.pools: pool_info(pool) @@ -199,7 +198,7 @@ def main(args=None, app=None): try: setattr(pool, opt, value) except qubesadmin.exc.QubesException as e: - parser.error('failed to set pool %s option %s: %s\n' % ( + parser.error('failed to set pool {} option {}: {}\n'.format( pool.name, opt, str(e))) return 0 diff --git a/qubesadmin/tools/qvm_prefs.py b/qubesadmin/tools/qvm_prefs.py index dcd0c972..5c87ddbd 100644 --- a/qubesadmin/tools/qvm_prefs.py +++ b/qubesadmin/tools/qvm_prefs.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, http://www.qubes-os.org # @@ -20,7 +19,6 @@ ''' Manipulate VM properties.''' -from __future__ import print_function import sys import textwrap @@ -126,7 +124,7 @@ def process_actions(parser, args, target): try: setattr(target, args.property, args.value) except qubesadmin.exc.QubesNoSuchPropertyError: - parser.error('no such property: {!r}'.format(args.property)) + parser.error(f'no such property: {args.property!r}') except qubesadmin.exc.QubesException as e: parser.error_runtime(e) return 0 @@ -135,7 +133,7 @@ def process_actions(parser, args, target): try: delattr(target, args.property) except qubesadmin.exc.QubesNoSuchPropertyError: - parser.error('no such property: {!r}'.format(args.property)) + parser.error(f'no such property: {args.property!r}') except qubesadmin.exc.QubesException as e: parser.error_runtime(e) return 0 @@ -148,7 +146,7 @@ def process_actions(parser, args, target): if value is not None: print(str(value)) except qubesadmin.exc.QubesNoSuchPropertyError: - parser.error('no such property: {!r}'.format(args.property)) + parser.error(f'no such property: {args.property!r}') except qubesadmin.exc.QubesException as e: parser.error_runtime(e) diff --git a/qubesadmin/tools/qvm_remove.py b/qubesadmin/tools/qvm_remove.py index 8326bf63..8f5c473b 100644 --- a/qubesadmin/tools/qvm_remove.py +++ b/qubesadmin/tools/qvm_remove.py @@ -96,12 +96,12 @@ def main(args=None, app=None): # pylint: disable=missing-docstring for holder, prop in dependencies: if holder: print( - " - {} for {}".format(prop, holder.name), + f" - {prop} for {holder.name}", file=sys.stderr, ) else: print( - " - global property {}".format(prop), + f" - global property {prop}", file=sys.stderr, ) # Display the original message as well diff --git a/qubesadmin/tools/qvm_run.py b/qubesadmin/tools/qvm_run.py index cc69f664..21d7dc5c 100644 --- a/qubesadmin/tools/qvm_run.py +++ b/qubesadmin/tools/qvm_run.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -218,7 +217,7 @@ def print_no_color(msg, file, color): Namely reset to base color first, print a message, then restore color. """ if color: - print("\033[0m{}\033[0;{}m".format(msg, color), file=file) + print(f"\033[0m{msg}\033[0;{color}m", file=file) else: print(msg, file=file) @@ -287,7 +286,7 @@ def run_command_single(args, vm): elif args.user == "root": service = "qubes.VMRootShell" args.user = None - shell_cmd = " ".join(shlex.quote(arg) for arg in all_args) + shell_cmd = shlex.join(all_args) else: service = "qubes.VMShell" if args.gui and args.dispvm: @@ -394,10 +393,10 @@ def main(args=None, app=None): assert len(domains) == 1 args.gui = has_gui(domains[0]) if args.color_output: - sys.stdout.write("\033[0;{}m".format(args.color_output)) + sys.stdout.write(f"\033[0;{args.color_output}m") sys.stdout.flush() if args.color_stderr: - sys.stderr.write("\033[0;{}m".format(args.color_stderr)) + sys.stderr.write(f"\033[0;{args.color_stderr}m") sys.stderr.flush() copy_proc = None try: @@ -406,7 +405,7 @@ def main(args=None, app=None): if not args.autostart and not vm.is_running(): if verbose > 0: print_no_color( - "Qube '{}' not started".format(vm.name), + f"Qube '{vm.name}' not started", file=sys.stderr, color=args.color_stderr, ) @@ -416,7 +415,7 @@ def main(args=None, app=None): if not args.autostart: if verbose > 0: print_no_color( - "Qube '{}' is paused".format(vm.name), + f"Qube '{vm.name}' is paused", file=sys.stderr, color=args.color_stderr, ) @@ -427,7 +426,7 @@ def main(args=None, app=None): except qubesadmin.exc.QubesException: if verbose > 0: print_no_color( - "Qube '{}' cannot be unpaused".format(vm.name), + f"Qube '{vm.name}' cannot be unpaused", file=sys.stderr, color=args.color_stderr, ) @@ -436,7 +435,7 @@ def main(args=None, app=None): try: if verbose > 0: print_no_color( - "Running '{}' on {}".format(args.cmd, vm.name), + f"Running '{args.cmd}' on {vm.name}", file=sys.stderr, color=args.color_stderr, ) diff --git a/qubesadmin/tools/qvm_service.py b/qubesadmin/tools/qvm_service.py index b2f5d34d..2826884c 100644 --- a/qubesadmin/tools/qvm_service.py +++ b/qubesadmin/tools/qvm_service.py @@ -1,4 +1,3 @@ -# coding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -24,7 +23,6 @@ '''qvm-service - Manage domain's services''' -from __future__ import print_function import argparse import sys @@ -80,7 +78,7 @@ def parse_bool(value): if lcvalue in ('1', 'yes', 'true', 'on'): return True raise qubesadmin.exc.QubesValueError( - 'Invalid literal for boolean value: {!r}'.format(value)) + f'Invalid literal for boolean value: {value!r}') return bool(value) diff --git a/qubesadmin/tools/qvm_shutdown.py b/qubesadmin/tools/qvm_shutdown.py index 765b1039..14599b99 100644 --- a/qubesadmin/tools/qvm_shutdown.py +++ b/qubesadmin/tools/qvm_shutdown.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, http://www.qubes-os.org # @@ -22,7 +21,6 @@ ''' Shutdown a qube ''' -from __future__ import print_function import sys import time @@ -91,7 +89,7 @@ def main(args=None, app=None): # pylint: disable=missing-docstring pass except qubesadmin.exc.QubesException as e: if not args.wait: - vm.log.error('Shutdown error: {}'.format(e)) + vm.log.error(f'Shutdown error: {e}') else: remaining_domains.add(vm) if not args.wait: @@ -112,7 +110,7 @@ def main(args=None, app=None): # pylint: disable=missing-docstring qubesadmin.events.utils.wait_for_domain_shutdown( this_round_domains), args.timeout)) - except asyncio.TimeoutError: + except TimeoutError: if not args.dry_run: for vm in this_round_domains: try: diff --git a/qubesadmin/tools/qvm_start.py b/qubesadmin/tools/qvm_start.py index 1158e963..3ab19376 100644 --- a/qubesadmin/tools/qvm_start.py +++ b/qubesadmin/tools/qvm_start.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -181,7 +180,7 @@ def main(args=None, app=None): continue exit_code = 1 parser.print_error( - 'domain {} is already running'.format(domain.name)) + f'domain {domain.name} is already running') return exit_code drive_assignment = None try: @@ -198,8 +197,7 @@ def main(args=None, app=None): if drive_assignment: # don't reconnect this device after VM reboot domain.devices['block'].unassign(drive_assignment) - except (IOError, OSError, qubesadmin.exc.QubesException, - ValueError) as e: + except (OSError, qubesadmin.exc.QubesException, ValueError) as e: if drive_assignment: try: domain.devices['block'].detach(drive_assignment) diff --git a/qubesadmin/tools/qvm_start_daemon.py b/qubesadmin/tools/qvm_start_daemon.py index a6ff78df..330ce49c 100644 --- a/qubesadmin/tools/qvm_start_daemon.py +++ b/qubesadmin/tools/qvm_start_daemon.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -287,7 +286,7 @@ def serialize_gui_daemon_options(options): ) continue - lines.append(" {} = {};".format(name, serialized)) + lines.append(f" {name} = {serialized};") lines.append("}") lines.append("") return "\n".join(lines) @@ -310,7 +309,7 @@ def escape_config_string(value): assert not NON_ASCII_RE.match( value - ), "expected an ASCII string: {!r}".format(value) + ), f"expected an ASCII string: {value!r}" value = ( value.replace("\\", "\\\\") @@ -321,7 +320,7 @@ def escape_config_string(value): .replace("\t", r"\t") ) value = UNPRINTABLE_CHARACTER_RE.sub( - lambda m: r"\x{:02x}".format(ord(m.group(0))), value + lambda m: fr"\x{ord(m.group(0)):02x}", value ) return '"' + value + '"' @@ -635,7 +634,7 @@ def kde_guid_args(vm): session_owner = line.split(b":")[2].decode() if session_owner is not None: data_dir = os.path.expanduser( - "~{}/.local/share".format(session_owner) + f"~{session_owner}/.local/share" ) else: # fallback to current user @@ -692,17 +691,17 @@ def write_guid_config(config_path, config): @staticmethod def guid_pidfile(xid): """Helper function to construct a GUI pidfile path""" - return "/var/run/qubes/guid-running.{}".format(xid) + return f"/var/run/qubes/guid-running.{xid}" @staticmethod def guid_config_file(xid): """Helper function to construct a GUI configuration file path""" - return "/var/run/qubes/guid-conf.{}".format(xid) + return f"/var/run/qubes/guid-conf.{xid}" @staticmethod def pacat_pidfile(xid): """Helper function to construct an AUDIO pidfile path""" - return "/var/run/qubes/pacat.{}".format(xid) + return f"/var/run/qubes/pacat.{xid}" @staticmethod def pacat_domid(vm): @@ -736,7 +735,7 @@ async def start_gui_for_vm(self, vm, monitor_layout=None): if not vm.debug and os.path.exists(stubdom_guid_pidfile): # Terminate stubdom guid once "real" gui agent connects with open( - stubdom_guid_pidfile, "r", encoding="ascii" + stubdom_guid_pidfile, encoding="ascii" ) as pidfile: stubdom_guid_pid = pidfile.read().strip() guid_cmd += ["-K", stubdom_guid_pid] @@ -811,7 +810,7 @@ async def start_gui(self, vm, force_stubdom=False, monitor_layout=None): return guivm = getattr(vm, "guivm", None) if guivm != vm.app.local_name: - vm.log.info("GUI connected to {}. Skipping.".format(guivm)) + vm.log.info(f"GUI connected to {guivm}. Skipping.") return if vm.virt_mode == "hvm": @@ -834,7 +833,7 @@ async def start_audio(self, vm): return audiovm = getattr(vm, "audiovm", None) if audiovm != vm.app.local_name: - vm.log.info("AUDIO connected to {}. Skipping.".format(audiovm)) + vm.log.info(f"AUDIO connected to {audiovm}. Skipping.") return if not vm.features.check_with_template("audio", True): @@ -1178,7 +1177,7 @@ def main(): loop.close() elif args.notify_monitor_layout: try: - with open(pidfile_path, "r", encoding="ascii") as pidfile: + with open(pidfile_path, encoding="ascii") as pidfile: pid = int(pidfile.read().strip()) os.kill(pid, signal.SIGHUP) except (FileNotFoundError, ValueError) as e: diff --git a/qubesadmin/tools/qvm_tags.py b/qubesadmin/tools/qvm_tags.py index 799105a7..4f770946 100644 --- a/qubesadmin/tools/qvm_tags.py +++ b/qubesadmin/tools/qvm_tags.py @@ -21,7 +21,6 @@ '''qvm-tags - Manage domain's tags''' -from __future__ import print_function import sys diff --git a/qubesadmin/tools/qvm_template.py b/qubesadmin/tools/qvm_template.py index 4503b17c..b20e094e 100644 --- a/qubesadmin/tools/qvm_template.py +++ b/qubesadmin/tools/qvm_template.py @@ -63,7 +63,7 @@ WRAPPER_PAYLOAD_BEGIN = "###!Q!BEGIN-QUBES-WRAPPER!Q!###" WRAPPER_PAYLOAD_END = "###!Q!END-QUBES-WRAPPER!Q!###" -UPDATEVM = str('global UpdateVM') +UPDATEVM = 'global UpdateVM' class AlreadyRunning(Exception): @@ -77,14 +77,14 @@ class SignatureVerificationError(Exception): def qubes_release() -> str: """Return the Qubes release.""" if os.path.exists('/usr/share/qubes/marker-vm'): - with open('/usr/share/qubes/marker-vm', 'r', encoding='ascii') as fd: + with open('/usr/share/qubes/marker-vm', encoding='ascii') as fd: # Get the first non-comment line release = [l.strip() for l in fd.readlines() if l.strip() and not l.startswith('#')] # sanity check if release and release[0] and release[0][0].isdigit(): return release[0] - with open('/etc/os-release', 'r', encoding='ascii') as fd: + with open('/etc/os-release', encoding='ascii') as fd: release = None distro_id = None for line in fd: @@ -327,7 +327,7 @@ def evr(self): class DlEntry(typing.NamedTuple): """Information about a template to be downloaded.""" - evr: typing.Tuple[str, str, str] + evr: tuple[str, str, str] reponame: str dlsize: int @@ -335,14 +335,14 @@ class DlEntry(typing.NamedTuple): # pylint: enable=too-few-public-methods,inherit-non-class -def build_version_str(evr: typing.Tuple[str, str, str]) -> str: +def build_version_str(evr: tuple[str, str, str]) -> str: """Return version string described by ``evr``, which is in (epoch, version, release) format.""" return '%s:%s-%s' % evr def is_match_spec(name: str, epoch: str, version: str, release: str, spec: str - ) -> typing.Tuple[bool, float]: + ) -> tuple[bool, float]: """Check whether (name, epoch, version, release) matches the spec string. For the algorithm, refer to section "NEVRA Matching" in the DNF @@ -393,7 +393,7 @@ def query_local(vm: qubesadmin.vm.QubesVM) -> Template: vm.features['template-description'].replace('|', '\n')) -def query_local_evr(vm: qubesadmin.vm.QubesVM) -> typing.Tuple[str, str, str]: +def query_local_evr(vm: qubesadmin.vm.QubesVM) -> tuple[str, str, str]: """Return the (epoch, version, release) of ``vm``. Requires the VM to be managed by qvm-template. @@ -421,7 +421,7 @@ def get_managed_template_vm(app: qubesadmin.app.QubesBase, name: str return vm -def confirm_action(msg: str, affected: typing.List[str]) -> None: +def confirm_action(msg: str, affected: list[str]) -> None: """Confirm user action.""" print(msg) for name in affected: @@ -439,7 +439,7 @@ def qrexec_popen( args: argparse.Namespace, app: qubesadmin.app.QubesBase, service: str, - stdout: typing.Union[int, typing.IO] = subprocess.PIPE, + stdout: int | typing.IO = subprocess.PIPE, filter_esc: bool = True) -> subprocess.Popen: """Return ``Popen`` object that communicates with the given qrexec call in ``args.updatevm``. @@ -557,7 +557,7 @@ def check_newline(string, name): repo_config = "" for path in args.repo_files: - with open(path, 'r', encoding='utf-8') as fd: + with open(path, encoding='utf-8') as fd: repo_config += fd.read() + '\n' payload += repo_config @@ -569,7 +569,7 @@ def qrexec_repoquery( args: argparse.Namespace, app: qubesadmin.app.QubesBase, spec: str = '*', - refresh: bool = False) -> typing.List[Template]: + refresh: bool = False) -> list[Template]: """Query template information from repositories. :param args: Arguments received by the application. Specifically, @@ -639,7 +639,7 @@ def qrexec_repoquery( elif buildtime.isnumeric(): # DNF5 provides seconds since epoch buildtime = datetime.datetime.fromtimestamp(int(buildtime), - tz=datetime.timezone.utc) + tz=datetime.UTC) else: raise ValueError # XXX: Perhaps whitelist licenses directly? @@ -665,7 +665,7 @@ def qrexec_download( spec: str, path: str, key: str, - dlsize: typing.Optional[int] = None, + dlsize: int | None = None, refresh: bool = False) -> None: """Download a template from repositories. @@ -726,8 +726,8 @@ def qrexec_download( raise ConnectionError("rpmcanon failed") -def get_keys_for_repos(repo_files: typing.List[str], - releasever: str) -> typing.Dict[str, str]: +def get_keys_for_repos(repo_files: list[str], + releasever: str) -> dict[str, str]: """List gpg keys Returns a dict reponame -> key path @@ -749,7 +749,7 @@ def get_keys_for_repos(repo_files: typing.List[str], def verify_rpm(path: str, key: str, *, nogpgcheck: bool = False, - template_name: typing.Optional[str] = None) -> rpm.hdr: + template_name: str | None = None) -> rpm.hdr: """Verify the digest and signature of a RPM package and return the package header. @@ -851,7 +851,7 @@ def filter_version( version_selector: VersionSelector = VersionSelector.LATEST): """Select only one version for given template name""" # We only select one package for each distinct package name - results: typing.Dict[str, Template] = {} + results: dict[str, Template] = {} for entry in query_res: evr = (entry.epoch, entry.version, entry.release) @@ -893,7 +893,7 @@ def get_dl_list( args: argparse.Namespace, app: qubesadmin.app.QubesBase, version_selector: VersionSelector = VersionSelector.LATEST -) -> typing.Dict[str, DlEntry]: +) -> dict[str, DlEntry]: """Return list of templates that needs to be downloaded. :param args: Arguments received by the application. @@ -904,7 +904,7 @@ def get_dl_list( :return: Dictionary that maps to ``DlEntry`` the names of templates that needs to be downloaded """ - full_candid: typing.Dict[str, DlEntry] = {} + full_candid: dict[str, DlEntry] = {} for template in args.templates: # Skip local RPMs if template.endswith('.rpm'): @@ -944,10 +944,10 @@ def get_dl_list( def download( args: argparse.Namespace, app: qubesadmin.app.QubesBase, - path_override: typing.Optional[str] = None, - dl_list: typing.Optional[typing.Dict[str, DlEntry]] = None, + path_override: str | None = None, + dl_list: dict[str, DlEntry] | None = None, version_selector: VersionSelector = VersionSelector.LATEST) \ - -> typing.Dict[str, rpm.hdr]: + -> dict[str, rpm.hdr]: """Command that downloads template packages. :param args: Arguments received by the application. @@ -1214,11 +1214,11 @@ def verify(rpmfile, reponame, package_hdr=None): tpl.features['template-buildtime'] = \ datetime.datetime.fromtimestamp( int(package_hdr[rpm.RPMTAG_BUILDTIME]), - tz=datetime.timezone.utc) \ + tz=datetime.UTC) \ .strftime(DATE_FMT) tpl.features['template-installtime'] = \ datetime.datetime.now( - tz=datetime.timezone.utc).strftime(DATE_FMT) + tz=datetime.UTC).strftime(DATE_FMT) tpl.features['template-license'] = \ package_hdr[rpm.RPMTAG_LICENSE] tpl.features['template-url'] = \ @@ -1344,7 +1344,7 @@ def check_append(name, evr): if args.all or args.available or args.extras or args.upgrades: if args.templates: - query_res_set: typing.Set[Template] = set() + query_res_set: set[Template] = set() for spec in args.templates: query_res_set |= set(qrexec_repoquery( args, app, PACKAGE_NAME_PREFIX + spec)) @@ -1452,7 +1452,7 @@ def compare(lhs, rhs): (WEIGHT_URL, 'URL')] search_res_by_idx: \ - typing.Dict[int, typing.List[typing.Tuple[int, str, bool]]] = \ + dict[int, list[tuple[int, str, bool]]] = \ collections.defaultdict(list) for keyword in args.templates: for idx, entry in enumerate(query_res): @@ -1472,7 +1472,7 @@ def compare(lhs, rhs): keywords = set(args.templates) idxs = list(search_res_by_idx.keys()) for idx in idxs: - if keywords != set(x[1] for x in search_res_by_idx[idx]): + if keywords != {x[1] for x in search_res_by_idx[idx]}: del search_res_by_idx[idx] def key_func(x): @@ -1486,13 +1486,13 @@ def key_func(x): def gen_header(needles): fields = [] - weight_types = set(x[0] for x in needles) + weight_types = {x[0] for x in needles} for weight, field in WEIGHT_TO_FIELD: if weight in weight_types: fields.append(field) exact = all(x[-1] for x in needles) match = 'Exactly Matched' if exact else 'Matched' - keywords = sorted(list(set(x[1] for x in needles))) + keywords = sorted(list({x[1] for x in needles})) return ' & '.join(fields) + ' ' + match + ': ' + ', '.join(keywords) last_header = '' @@ -1640,7 +1640,7 @@ def repolist(args: argparse.Namespace, app: qubesadmin.app.QubesBase) -> None: # Filter (name, operation) from args.repos repoid = [] enable_disable_repos = [] - repos: typing.List[dnf.repo.Repo] = [] + repos: list[dnf.repo.Repo] = [] if args.repos: for repo in args.repos: operation, name = repo @@ -1712,11 +1712,11 @@ def migrate_from_rpmdb(app): vm.features['template-reponame'] = '@commandline' vm.features['template-buildtime'] = \ datetime.datetime.fromtimestamp( - pkg[rpm.RPMTAG_BUILDTIME], tz=datetime.timezone.utc).\ + pkg[rpm.RPMTAG_BUILDTIME], tz=datetime.UTC).\ strftime(DATE_FMT) vm.features['template-installtime'] = \ datetime.datetime.fromtimestamp( - pkg[rpm.RPMTAG_INSTALLTIME], tz=datetime.timezone.utc).\ + pkg[rpm.RPMTAG_INSTALLTIME], tz=datetime.UTC).\ strftime(DATE_FMT) vm.features['template-license'] = pkg[rpm.RPMTAG_LICENSE] vm.features['template-url'] = pkg[rpm.RPMTAG_URL] @@ -1725,7 +1725,7 @@ def migrate_from_rpmdb(app): pkg[rpm.RPMTAG_DESCRIPTION].replace('\n', '|') vm.installed_by_rpm = False except Exception as e: # pylint: disable=broad-except - print('Failed to set template {} metadata: {}'.format(vm.name, e)) + print(f'Failed to set template {vm.name} metadata: {e}') continue pkgs_to_remove.append(pkg) subprocess.check_call( @@ -1733,8 +1733,8 @@ def migrate_from_rpmdb(app): [p[rpm.RPMTAG_NAME] for p in pkgs_to_remove]) -def main(args: typing.Optional[typing.Sequence[str]] = None, - app: typing.Optional[qubesadmin.app.QubesBase] = None) -> int: +def main(args: typing.Sequence[str] | None = None, + app: qubesadmin.app.QubesBase | None = None) -> int: """Main routine of **qvm-template**. :param args: Override arguments received by the application. Optional diff --git a/qubesadmin/tools/qvm_template_postprocess.py b/qubesadmin/tools/qvm_template_postprocess.py index 52ada236..03d5aba0 100644 --- a/qubesadmin/tools/qvm_template_postprocess.py +++ b/qubesadmin/tools/qvm_template_postprocess.py @@ -183,7 +183,7 @@ def import_appmenus(vm, source_dir, skip_generate=True): # name according to the FreeDesktop spec source_dir = pathlib.Path(source_dir) try: - with open(source_dir / 'vm-whitelisted-appmenus.list', 'r', + with open(source_dir / 'vm-whitelisted-appmenus.list', encoding='ascii') as fd: vm.features['default-menu-items'] = \ ' '.join([x.rstrip() for x in fd]) @@ -191,14 +191,14 @@ def import_appmenus(vm, source_dir, skip_generate=True): vm.log.warning('Cannot set default-menu-items, %s not found', e.filename) try: - with open(source_dir / 'whitelisted-appmenus.list', 'r', + with open(source_dir / 'whitelisted-appmenus.list', encoding='ascii') as fd: vm.features['menu-items'] = ' '.join([x.rstrip() for x in fd]) except FileNotFoundError as e: vm.log.warning('Cannot set menu-items, %s not found', e.filename) try: - with open(source_dir / 'netvm-whitelisted-appmenus.list', 'r', + with open(source_dir / 'netvm-whitelisted-appmenus.list', encoding='ascii') as fd: vm.features['netvm-menu-items'] = ' '.join([x.rstrip() for x in fd]) except FileNotFoundError as e: @@ -222,7 +222,7 @@ def import_appmenus(vm, source_dir, skip_generate=True): def parse_template_config(path): '''Parse template.conf from template package. (KEY=VALUE format)''' - with open(path, 'r', encoding='ascii') as fd: + with open(path, encoding='ascii') as fd: return dict(line.rstrip('\n').split('=', 1) for line in fd) async def call_postinstall_service(vm): @@ -251,7 +251,7 @@ async def call_postinstall_service(vm): await asyncio.wait_for( qubesadmin.events.utils.wait_for_domain_shutdown([vm]), qubesadmin.config.defaults['shutdown_timeout']) - except asyncio.TimeoutError: + except TimeoutError: try: vm.kill() except qubesadmin.exc.QubesVMNotStartedError: @@ -416,7 +416,7 @@ def pre_remove(args): for appvm in tpl.appvms: dependant_vms = True app.log.error( - 'ERROR! Qube "{}" uses this template'.format(appvm.name)) + f'ERROR! Qube "{appvm.name}" uses this template') if dependant_vms: app.log.warning( 'WARNING! Do not use dnf to uninstall templates!') @@ -437,7 +437,7 @@ def is_chroot(): return ( stat_root.st_dev != stat_init_root.st_dev or stat_root.st_ino != stat_init_root.st_ino) - except IOError: + except OSError: return False diff --git a/qubesadmin/tools/qvm_unpause.py b/qubesadmin/tools/qvm_unpause.py index fca9eceb..f7c50d2b 100644 --- a/qubesadmin/tools/qvm_unpause.py +++ b/qubesadmin/tools/qvm_unpause.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -51,7 +50,7 @@ def main(args=None, app=None): domain.resume() else: domain.unpause() - except (IOError, OSError, qubesadmin.exc.QubesException) as e: + except (OSError, qubesadmin.exc.QubesException) as e: exit_code = 1 parser.print_error(str(e)) diff --git a/qubesadmin/tools/qvm_volume.py b/qubesadmin/tools/qvm_volume.py index 21072b81..d872998f 100644 --- a/qubesadmin/tools/qvm_volume.py +++ b/qubesadmin/tools/qvm_volume.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, http://www.qubes-os.org # @@ -22,7 +21,6 @@ """Qubes volume management""" -from __future__ import print_function import argparse import os @@ -86,7 +84,7 @@ def __lt__(self, other): return (self.pool, self.vid) < (other.pool, other.vid) def __str__(self): - return "{!s}:{!s}".format(self.pool, self.vid) + return f"{self.pool!s}:{self.vid!s}" def info_volume(args): @@ -109,7 +107,7 @@ def info_volume(args): print(value) else: raise qubesadmin.exc.StoragePoolException( - 'No such property: {}'.format(args.property)) + f'No such property: {args.property}') else: info = collections.OrderedDict() for item in info_items: @@ -136,7 +134,7 @@ def config_volume(args): volume = args.volume if args.property not in ('rw', 'revisions_to_keep', 'ephemeral'): raise qubesadmin.exc.QubesNoSuchPropertyError( - 'Invalid property: {}'.format(args.property)) + f'Invalid property: {args.property}') setattr(volume, args.property, args.value) diff --git a/qubesadmin/tools/xcffibhelpers.py b/qubesadmin/tools/xcffibhelpers.py index 9474f2b4..14864bea 100644 --- a/qubesadmin/tools/xcffibhelpers.py +++ b/qubesadmin/tools/xcffibhelpers.py @@ -1,4 +1,3 @@ -# -*- encoding: utf8 -*- # # The Qubes OS Project, http://www.qubes-os.org # diff --git a/qubesadmin/utils.py b/qubesadmin/utils.py index 68b89af5..954aafba 100644 --- a/qubesadmin/utils.py +++ b/qubesadmin/utils.py @@ -1,4 +1,3 @@ -# encoding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # @@ -51,7 +50,7 @@ def parse_size(size): size = size[:-len(unit)].strip() return int(size) * multiplier - raise qubesadmin.exc.QubesException("Invalid size: {0}.".format(size)) + raise qubesadmin.exc.QubesException(f"Invalid size: {size}.") def mbytes_to_kmg(size): @@ -176,7 +175,7 @@ def encode_for_vmexec(args): def encode(part): if part.group(0) == b'-': return b'--' - return '-{:02X}'.format(ord(part.group(0))).encode('ascii') + return f'-{ord(part.group(0)):02X}'.encode('ascii') parts = [] for arg in args: diff --git a/qubesadmin/vm/__init__.py b/qubesadmin/vm/__init__.py index e5869996..2dea408a 100644 --- a/qubesadmin/vm/__init__.py +++ b/qubesadmin/vm/__init__.py @@ -1,4 +1,3 @@ -# -*- encoding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # @@ -392,7 +391,7 @@ def run_with_args(self, *args, **kwargs): e.cmd = str(args) raise e - return self.run(" ".join(shlex.quote(arg) for arg in args), **kwargs) + return self.run(shlex.join(args), **kwargs) @property def appvms(self):