diff --git a/qubes_menu/tests/test_vmmanager.py b/qubes_menu/tests/test_vmmanager.py index d0307df..76dc8f6 100644 --- a/qubes_menu/tests/test_vmmanager.py +++ b/qubes_menu/tests/test_vmmanager.py @@ -28,67 +28,100 @@ def test_vm_manager(test_qapp): dispatcher = qubesadmin.events.EventsDispatcher(test_qapp) vm_manager = VMManager(test_qapp, dispatcher) + vm_name = "test-vm" + template_name = "fedora-36" - entry_test = vm_manager.load_vm_from_name('test-vm') + entry_test = vm_manager.load_vm_from_name(vm_name) assert entry_test - assert entry_test.vm_name == 'test-vm' - assert entry_test.vm_icon_name == 'appvm-green' - entry_template = vm_manager.load_vm_from_name('fedora-36') + assert entry_test.vm_name == vm_name + assert entry_test.vm_icon_name == "appvm-green" + entry_template = vm_manager.load_vm_from_name(template_name) assert entry_template assert not entry_template.has_network + assert not entry_template.internal assert not entry_test.is_dispvm_template assert not entry_test.service_vm - - test_qapp._qubes['fedora-36'].properties['netvm'] = \ - Property('sys-firewall', 'vm', False) - test_qapp._qubes['fedora-36'].update_calls() - - vm_manager._update_domain_property('fedora-36', - 'property-set:netvm', name='netvm', - newvalue='sys-firewall', - oldvalue=None) + assert not entry_test.internal + + test_qapp._qubes["fedora-36"].properties["netvm"] = Property( + "sys-firewall", "vm", False + ) + test_qapp._qubes["fedora-36"].update_calls() + + vm_manager._update_domain_property( + "fedora-36", + "property-set:netvm", + name="netvm", + newvalue="sys-firewall", + oldvalue=None, + ) assert entry_template.has_network - test_qapp._qubes['test-vm'].properties['label'] = \ - Property('red', 'label', False) - test_qapp._qubes['test-vm'].properties['icon'] = \ - Property('appvm-red', 'str', False) - test_qapp._qubes['test-vm'].update_calls() - - vm_manager._update_domain_property('test-vm', - 'property-set:label', name='label', - newvalue='red', - oldvalue='blue') - assert entry_test.vm_icon_name == 'appvm-red' - - test_qapp._qubes['test-vm'].properties['template_for_dispvms'] = \ - Property('True', 'bool', False) - test_qapp._qubes['test-vm'].update_calls() - - vm_manager._update_domain_property('test-vm', - 'property-set:template_for_dispvms', - name='template_for_dispvms', - newvalue=True) + test_qapp._qubes[vm_name].properties["label"] = Property( + "red", "label", False + ) + test_qapp._qubes[vm_name].properties["icon"] = Property( + "appvm-red", "str", False + ) + test_qapp._qubes[vm_name].update_calls() + vm_manager._update_domain_property( + vm_name, + "property-set:label", + name="label", + newvalue="red", + oldvalue="blue", + ) + assert entry_test.vm_icon_name == "appvm-red" + + test_qapp._qubes[vm_name].properties["template_for_dispvms"] = Property( + "True", "bool", False + ) + test_qapp._qubes[vm_name].update_calls() + vm_manager._update_domain_property( + vm_name, + "property-set:template_for_dispvms", + name="template_for_dispvms", + newvalue=True, + ) assert entry_test.is_dispvm_template - test_qapp._qubes['test-vm'].features['servicevm'] = 1 - test_qapp._qubes['test-vm'].update_calls() - - vm_manager._update_domain_feature('test-vm', - 'feature-set:servicevm', - feature='servicevm', - value=1) + test_qapp._qubes[vm_name].features["servicevm"] = 1 + test_qapp._qubes[vm_name].update_calls() + vm_manager._update_domain_feature( + vm_name, "feature-set:servicevm", feature="servicevm", value=1 + ) assert entry_test.service_vm + test_qapp._qubes[vm_name].features["internal"] = 1 + test_qapp._qubes[vm_name].update_calls() + vm_manager._update_domain_feature( + vm_name, "feature-set:internal", feature="internal", value=1 + ) + assert entry_test.internal + + del test_qapp._qubes[vm_name].features["internal"] + test_qapp._qubes[vm_name].update_calls() + vm_manager._update_domain_feature( + vm_name, "feature-delete:internal", feature="internal" + ) + assert not entry_test.internal + + test_qapp._qubes[template_name].features["internal"] = 1 + test_qapp._qubes[vm_name].update_calls() + vm_manager._update_domain_feature( + vm_name, "feature-set:internal", feature="internal", value=1 + ) + assert entry_test.internal + def test_filter(test_qapp): dispatcher = qubesadmin.events.EventsDispatcher(test_qapp) vm_manager = VMManager(test_qapp, dispatcher) - entry_test = vm_manager.load_vm_from_name('test-vm') - entry_template = vm_manager.load_vm_from_name('fedora-36') - entry_service = vm_manager.load_vm_from_name('sys-net') - entry_dvm_template = vm_manager.load_vm_from_name('default-dvm') + entry_test = vm_manager.load_vm_from_name("test-vm") + entry_template = vm_manager.load_vm_from_name("fedora-36") + entry_service = vm_manager.load_vm_from_name("sys-net") + entry_dvm_template = vm_manager.load_vm_from_name("default-dvm") assert entry_test assert entry_template assert entry_service diff --git a/qubes_menu/vm_manager.py b/qubes_menu/vm_manager.py index ad11689..98953bd 100644 --- a/qubes_menu/vm_manager.py +++ b/qubes_menu/vm_manager.py @@ -21,6 +21,7 @@ Helper class that manages all events related to VMs. """ import qubesadmin.events +import qubesadmin.exc from qubesadmin.vm import QubesVM from typing import Optional, Dict, List, Callable @@ -33,35 +34,51 @@ class VMEntry: much data as possible and update it on events, sending also information to all related menu entries to update themselves. """ + def __init__(self, vm: QubesVM): self.vm = vm self.vm_name = str(vm) self.vm_klass = vm.klass - if self.vm.klass == 'DispVM' and self.vm.auto_cleanup: + if self.vm.klass == "DispVM" and self.vm.auto_cleanup: self.parent_vm = self.vm.template - self.sort_name = \ - f'{str(self.parent_vm.name).lower()} :{self.vm_name.lower()}' + self.sort_name = ( + f"{str(self.parent_vm.name).lower()} :{self.vm_name.lower()}" + ) else: self.parent_vm = None # the space here is to assure correct sorting for dispvm children self.sort_name = self.vm_name.lower() + " " + try: + self._internal = bool( + self.vm.features.check_with_template("internal", False) + ) + except qubesadmin.exc.QubesDaemonAccessError: + self._internal = False self._servicevm = bool(self.vm.features.get("servicevm", False)) self._is_dispvm_template = getattr( - self.vm, 'template_for_dispvms', False) - self._has_network = \ - self.vm.is_networked() if vm.klass != 'AdminVM' else False - self._vm_icon_name = getattr(self.vm, 'icon', - getattr(self.vm.label, 'icon', None)) + self.vm, "template_for_dispvms", False + ) + self._has_network = ( + self.vm.is_networked() if vm.klass != "AdminVM" else False + ) + self._vm_icon_name = getattr( + self.vm, "icon", getattr(self.vm.label, "icon", None) + ) self._power_state = self.vm.get_power_state() self.show_dispvm_template_in_apps = bool( - vm.features.get('appmenus-dispvm', False)) + vm.features.get("appmenus-dispvm", False) + ) self.entries: List = [] - def update_entries(self, update_power_state=False, - update_label=False, - update_has_network=False, update_type=False): + def update_entries( + self, + update_power_state=False, + update_label=False, + update_has_network=False, + update_type=False, + ): """ Update all related menu entries. :param update_power_state: did power state change? @@ -70,8 +87,12 @@ def update_entries(self, update_power_state=False, :param update_type: did type change? """ for entry in self.entries: - entry.update_contents(update_power_state, update_label, - update_has_network, update_type) + entry.update_contents( + update_power_state, + update_label, + update_has_network, + update_type, + ) @property def power_state(self): @@ -95,7 +116,7 @@ def vm_icon_name(self): @vm_icon_name.setter def vm_icon_name(self, _new_value): - self._vm_icon_name = getattr(self.vm, 'icon', self.vm.label.icon) + self._vm_icon_name = getattr(self.vm, "icon", self.vm.label.icon) self.update_entries(update_label=True) @property @@ -119,6 +140,16 @@ def is_dispvm_template(self, new_value): self._is_dispvm_template = new_value self.update_entries(update_type=True) + @property + def internal(self): + """Is the VM internal""" + return self._internal + + @internal.setter + def internal(self, new_value): + self._internal = new_value + self.update_entries(update_type=True) + @property def service_vm(self): """Does the VM provide network""" @@ -132,11 +163,13 @@ def service_vm(self, new_value): @property def show_in_apps(self): """Should this qube be shown in the Apps section of the menu?""" + if self.internal: + return False if self.service_vm: return False - if self.vm_klass == 'TemplateVM': + if self.vm_klass == "TemplateVM": return False - if self.vm_klass == 'AdminVM': + if self.vm_klass == "AdminVM": return False if self.is_dispvm_template and not self.show_dispvm_template_in_apps: return False @@ -146,28 +179,32 @@ def show_in_apps(self): def _escaped_name(self) -> str: """Name escaped according to rules from desktop-linux-common package""" - return (self.vm_name.replace('_', '_u'). - replace('-', '_d').replace('.', '_p')) + return ( + self.vm_name.replace("_", "_u") + .replace("-", "_d") + .replace(".", "_p") + ) @property def settings_desktop_file_name(self) -> str: """ Name of relevant .desktop vm settings file. """ - return ('org.qubes-os.qubes-vm-settings._' + - self._escaped_name + '.desktop') + return ( + "org.qubes-os.qubes-vm-settings._" + self._escaped_name + ".desktop" + ) @property def start_vm_desktop_file_name(self) -> str: """ Name of relevant .desktop start vm file. """ - return ('org.qubes-os.vm._' + self._escaped_name + - '.qubes-start.desktop') + return "org.qubes-os.vm._" + self._escaped_name + ".qubes-start.desktop" class VMManager: """A helper class that watches for VM-related events""" + def __init__(self, qapp: qubesadmin.Qubes, dispatcher): self.qapp = qapp self.dispatcher = dispatcher @@ -194,8 +231,11 @@ def load_vm_from_name(self, vm_name: str) -> Optional[VMEntry]: vm: QubesVM = self.qapp.domains[vm_name] except KeyError: return None - if vm.features.get('internal', False): - return None + try: + if vm.features.check_with_template("internal", False): + return None + except qubesadmin.exc.QubesDaemonAccessError: + pass return self._add_vm(vm) @@ -235,43 +275,53 @@ def _update_domain_state(self, vm_name, event, **_kwargs): state = constants.STATE_DICTIONARY[event] vm_entry.power_state = state - def _update_domain_property(self, vm_name, event, newvalue, - *_args, **_kwargs): + def _update_domain_property( + self, vm_name, event, newvalue, *_args, **_kwargs + ): vm_entry = self.load_vm_from_name(vm_name) if not vm_entry: return - if newvalue == 'False': + if newvalue == "False": newvalue = False try: - if event == 'property-set:label': + if event == "property-set:label": vm_entry.vm_icon_name = newvalue - elif event == 'property-set:netvm': + elif event == "property-set:netvm": vm_entry.has_network = vm_entry.vm.is_networked() - elif event == 'property-set:template_for_dispvms': + elif event == "property-set:template_for_dispvms": vm_entry.is_dispvm_template = newvalue except Exception: # pylint: disable=broad-except # dispatcher functions cannot raise any Exception, because # it will disable any future event handling pass - def _update_domain_feature(self, vm, _event, feature=None, value=None, - **_kwargs): + def _update_domain_feature( + self, vm, _event, feature=None, value=None, **_kwargs + ): vm_entry = self.load_vm_from_name(vm) if not vm_entry: return - if value == 'False': + if value == "False": value = False value = bool(value) try: - if feature == 'servicevm': + if feature == "internal": + vm_entry.internal = value + for derived in self.qapp.domains: + if not getattr(derived, "template", None) == vm: + continue + derived_vm_entry = self.load_vm_from_name(derived) + if derived_vm_entry: + derived_vm_entry.internal = value + if feature == "servicevm": vm_entry.service_vm = value - if feature == 'appmenus-dispvm': + if feature == "appmenus-dispvm": vm_entry.show_dispvm_template_in_apps = value except Exception: # pylint: disable=broad-except # dispatcher functions cannot raise any Exception, because @@ -289,35 +339,54 @@ def _update_domain_feature(self, vm, _event, feature=None, value=None, def register_events(self): """Register handlers for all relevant VM events.""" - self.dispatcher.add_handler('domain-pre-start', - self._update_domain_state) - self.dispatcher.add_handler('domain-start', self._update_domain_state) - self.dispatcher.add_handler('domain-start-failed', - self._update_domain_state) - self.dispatcher.add_handler('domain-paused', self._update_domain_state) - self.dispatcher.add_handler('domain-unpaused', - self._update_domain_state) - self.dispatcher.add_handler('domain-shutdown', - self._update_domain_state) - self.dispatcher.add_handler('domain-pre-shutdown', - self._update_domain_state) - self.dispatcher.add_handler('domain-shutdown-failed', - self._update_domain_state) - - self.dispatcher.add_handler('domain-add', self._add_domain) - self.dispatcher.add_handler('domain-delete', self._remove_domain) - - self.dispatcher.add_handler('property-set:netvm', - self._update_domain_property) - self.dispatcher.add_handler('property-set:label', - self._update_domain_property) - self.dispatcher.add_handler('property-set:template_for_dispvms', - self._update_domain_property) - self.dispatcher.add_handler('domain-feature-set:servicevm', - self._update_domain_feature) - self.dispatcher.add_handler('domain-feature-delete:servicevm', - self._update_domain_feature) - self.dispatcher.add_handler('domain-feature-set:appmenus-dispvm', - self._update_domain_feature) - self.dispatcher.add_handler('domain-feature-delete:appmenus-dispvm', - self._update_domain_feature) + self.dispatcher.add_handler( + "domain-pre-start", self._update_domain_state + ) + self.dispatcher.add_handler("domain-start", self._update_domain_state) + self.dispatcher.add_handler( + "domain-start-failed", self._update_domain_state + ) + self.dispatcher.add_handler("domain-paused", self._update_domain_state) + self.dispatcher.add_handler( + "domain-unpaused", self._update_domain_state + ) + self.dispatcher.add_handler( + "domain-shutdown", self._update_domain_state + ) + self.dispatcher.add_handler( + "domain-pre-shutdown", self._update_domain_state + ) + self.dispatcher.add_handler( + "domain-shutdown-failed", self._update_domain_state + ) + + self.dispatcher.add_handler("domain-add", self._add_domain) + self.dispatcher.add_handler("domain-delete", self._remove_domain) + + self.dispatcher.add_handler( + "property-set:netvm", self._update_domain_property + ) + self.dispatcher.add_handler( + "property-set:label", self._update_domain_property + ) + self.dispatcher.add_handler( + "property-set:template_for_dispvms", self._update_domain_property + ) + self.dispatcher.add_handler( + "domain-feature-set:servicevm", self._update_domain_feature + ) + self.dispatcher.add_handler( + "domain-feature-delete:servicevm", self._update_domain_feature + ) + self.dispatcher.add_handler( + "domain-feature-set:appmenus-dispvm", self._update_domain_feature + ) + self.dispatcher.add_handler( + "domain-feature-delete:appmenus-dispvm", self._update_domain_feature + ) + self.dispatcher.add_handler( + "domain-feature-set:internal", self._update_domain_feature + ) + self.dispatcher.add_handler( + "domain-feature-delete:internal", self._update_domain_feature + )