[ui] fix: upgrading: only requesting the root password if required | [flatpak] fix -> crashing when trying to downgrade

This commit is contained in:
Vinicius Moreira
2020-12-10 19:15:34 -03:00
parent d435480165
commit bef465a1ea
19 changed files with 213 additions and 132 deletions

View File

@@ -7,7 +7,7 @@ from threading import Thread
from typing import List, Set, Type, Tuple, Dict
from bauh.api.abstract.controller import SoftwareManager, SearchResult, ApplicationContext, UpgradeRequirements, \
UpgradeRequirement, TransactionResult
UpgradeRequirement, TransactionResult, SoftwareAction
from bauh.api.abstract.disk import DiskCacheLoader
from bauh.api.abstract.handler import ProcessWatcher, TaskManager
from bauh.api.abstract.model import SoftwarePackage, PackageUpdate, PackageHistory, PackageSuggestion, \
@@ -151,14 +151,14 @@ class GenericSoftwareManager(SoftwareManager):
res.installed.extend(apps_found.installed)
res.new.extend(apps_found.new)
def search(self, word: str, disk_loader: DiskCacheLoader = None, limit: int = -1, is_url: bool = False) -> SearchResult:
def search(self, words: str, disk_loader: DiskCacheLoader = None, limit: int = -1, is_url: bool = False) -> SearchResult:
ti = time.time()
self._wait_to_be_ready()
res = SearchResult([], [], 0)
if self.context.is_internet_available():
norm_word = word.strip().lower()
norm_word = words.strip().lower()
url_words = RE_IS_URL.match(norm_word)
disk_loader = self.disk_loader_factory.new()
@@ -385,7 +385,7 @@ class GenericSoftwareManager(SoftwareManager):
if man:
return man.cache_to_disk(pkg, icon_bytes=icon_bytes, only_icon=only_icon)
def requires_root(self, action: str, app: SoftwarePackage) -> bool:
def requires_root(self, action: SoftwareAction, app: SoftwarePackage) -> bool:
if app is None:
if self.managers:
for man in self.managers:

View File

@@ -10,7 +10,7 @@ from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QSizePolicy, QTableWid
from bauh import __app_name__
from bauh.api.abstract.context import ApplicationContext
from bauh.api.abstract.controller import SoftwareManager
from bauh.api.abstract.controller import SoftwareManager, SoftwareAction
from bauh.api.abstract.handler import TaskManager
from bauh.view.qt.components import new_spacer, QCustomToolbar
from bauh.view.qt.qt_utils import centralize
@@ -51,7 +51,7 @@ class Prepare(QThread, TaskManager):
def run(self):
root_pwd = None
if self.manager.requires_root('prepare', None):
if self.manager.requires_root(SoftwareAction.PREPARE, None):
ok, root_pwd = self.ask_password()
if not ok:

View File

@@ -14,7 +14,7 @@ from PyQt5.QtWidgets import QWidget
from bauh import LOGS_PATH
from bauh.api.abstract.cache import MemoryCache
from bauh.api.abstract.controller import SoftwareManager, UpgradeRequirement, UpgradeRequirements
from bauh.api.abstract.controller import SoftwareManager, UpgradeRequirement, UpgradeRequirements, SoftwareAction
from bauh.api.abstract.handler import ProcessWatcher
from bauh.api.abstract.model import PackageStatus, SoftwarePackage, CustomSoftwareAction
from bauh.api.abstract.view import MessageType, MultipleSelectComponent, InputOption, TextComponent, \
@@ -29,7 +29,7 @@ from bauh.view.qt import commons
from bauh.view.qt.view_model import PackageView, PackageViewStatus
from bauh.view.util.translation import I18n
RE_VERSION_IN_NAME = re.compile(r'\s+version\s+[\w\.]+\s*$')
RE_VERSION_IN_NAME = re.compile(r'\s+version\s+[\w.]+\s*$')
class AsyncAction(QThread, ProcessWatcher):
@@ -48,7 +48,7 @@ class AsyncAction(QThread, ProcessWatcher):
super(AsyncAction, self).__init__()
self.wait_confirmation = False
self.confirmation_res = None
self.root_password = None
self.root_password = root_password
self.stop = False
def request_confirmation(self, title: str, body: str, components: List[ViewComponent] = None,
@@ -123,64 +123,78 @@ class AsyncAction(QThread, ProcessWatcher):
return False
def _generate_backup(self, app_config: dict, i18n: I18n, root_password: str) -> bool:
if timeshift.is_available():
if app_config['backup']['mode'] not in ('only_one', 'incremental'):
self.show_message(title=self.i18n['error'].capitalize(),
body='{}: {}'.format(self.i18n['action.backup.invalid_mode'], bold(app_config['backup']['mode'])),
type_=MessageType.ERROR)
return False
def _generate_backup(self, app_config: dict, i18n: I18n, root_password: Optional[str]) -> bool:
if app_config['backup']['mode'] not in ('only_one', 'incremental'):
self.show_message(title=self.i18n['error'].capitalize(),
body='{}: {}'.format(self.i18n['action.backup.invalid_mode'],bold(app_config['backup']['mode'])),
type_=MessageType.ERROR)
self.change_substatus('')
return False
if not user.is_root() and not root_password:
root_pwd, valid = self.request_root_password()
else:
root_pwd, valid = root_password, True
handler = ProcessHandler(self)
if app_config['backup']['mode'] == 'only_one':
self.change_substatus('[{}] {}'.format(i18n['core.config.tab.backup'].lower(), i18n['action.backup.substatus.delete']))
deleted, _ = handler.handle_simple(timeshift.delete_all_snapshots(root_password))
if not valid:
return False
handler = ProcessHandler(self)
if app_config['backup']['mode'] == 'only_one':
self.change_substatus('[{}] {}'.format(i18n['core.config.tab.backup'].lower(), i18n['action.backup.substatus.delete']))
deleted, _ = handler.handle_simple(timeshift.delete_all_snapshots(root_pwd))
if not deleted and not self.request_confirmation(title=i18n['core.config.tab.backup'],
body='{}. {}'.format(i18n['action.backup.error.delete'], i18n['action.backup.error.proceed']),
confirmation_label=i18n['yes'].capitalize(),
deny_label=i18n['no'].capitalize()):
return False
self.change_substatus('[{}] {}'.format(i18n['core.config.tab.backup'].lower(), i18n['action.backup.substatus.create']))
created, _ = handler.handle_simple(timeshift.create_snapshot(root_pwd, app_config['backup']['type']))
if not created and not self.request_confirmation(title=i18n['core.config.tab.backup'],
body='{}. {}'.format(i18n['action.backup.error.create'],
if not deleted and not self.request_confirmation(title=i18n['core.config.tab.backup'],
body='{}. {}'.format(i18n['action.backup.error.delete'],
i18n['action.backup.error.proceed']),
confirmation_label=i18n['yes'].capitalize(),
deny_label=i18n['no'].capitalize()):
self.change_substatus('')
return False
self.change_substatus('[{}] {}'.format(i18n['core.config.tab.backup'].lower(), i18n['action.backup.substatus.create']))
created, _ = handler.handle_simple(timeshift.create_snapshot(root_password, app_config['backup']['type']))
if not created and not self.request_confirmation(title=i18n['core.config.tab.backup'],
body='{}. {}'.format(i18n['action.backup.error.create'],
i18n['action.backup.error.proceed']),
confirmation_label=i18n['yes'].capitalize(),
deny_label=i18n['no'].capitalize()):
self.change_substatus('')
return False
self.change_substatus('')
return True
def request_backup(self, app_config: dict, key: str, i18n: I18n, root_password: str = None) -> bool:
if bool(app_config['backup']['enabled']) and timeshift.is_available():
val = app_config['backup'][key] if key else None
def _check_backup_requirements(self, app_config: dict, pkg: Optional[PackageView], action_key: Optional[str]) -> bool:
if pkg and not pkg.model.supports_backup():
return False
if val is None: # ask mode
if self.request_confirmation(title=i18n['core.config.tab.backup'],
if action_key and app_config['backup'][action_key] is False:
return False
return bool(app_config['backup']['enabled']) and timeshift.is_available()
def _should_backup(self, action_key: str, app_config: dict, i18n: I18n) -> bool:
# backup -> true: do not ask, only execute | false: do not ask or execute | None: ask
backup = app_config['backup'][action_key] if action_key else None
if backup is None:
return self.request_confirmation(title=i18n['core.config.tab.backup'],
body=i18n['action.backup.msg'],
confirmation_label=i18n['yes'].capitalize(),
deny_label=i18n['no'].capitalize()):
res = self._generate_backup(app_config, i18n, root_password)
self.change_substatus('')
return res
deny_label=i18n['no'].capitalize())
else:
return backup
elif val is True: # direct mode
res = self._generate_backup(app_config, i18n, root_password)
self.change_substatus('')
return res
def request_backup(self, action_key: Optional[str], pkg: Optional[PackageView], i18n: I18n, app_config: dict, root_password: Optional[str], backup_only: bool = False) -> Tuple[bool, Optional[str]]:
if not backup_only:
if not self._check_backup_requirements(app_config=app_config, pkg=pkg, action_key=action_key):
return True, root_password
return True
if not self._should_backup(action_key=action_key, app_config=app_config, i18n=i18n):
return True, root_password
pwd = root_password
if not user.is_root() and pwd is None:
valid_password, pwd = self.request_root_password()
if not valid_password:
return False, None
return self._generate_backup(app_config=app_config, i18n=i18n, root_password=pwd), pwd
class UpgradeSelected(AsyncAction):
@@ -373,15 +387,29 @@ class UpgradeSelected(AsyncAction):
traceback.print_exc()
def run(self):
valid_password, root_password = self._request_password()
root_user = user.is_root()
to_update, upgrade_requires_root, bkp_supported = [], False, False
for pkg in self.pkgs:
if pkg.model.update and not pkg.model.is_update_ignored() and pkg.update_checked:
to_update.append(pkg)
if not bkp_supported and pkg.model.supports_backup():
bkp_supported = True
if not root_user and not upgrade_requires_root and self.manager.requires_root(SoftwareAction.UPGRADE, pkg.model):
upgrade_requires_root = True
if upgrade_requires_root:
valid_password, root_password = self._request_password()
else:
valid_password, root_password = True, None
if not valid_password:
self.notify_finished({'success': False, 'updated': 0, 'types': set(), 'id': None})
self.pkgs = None
return
to_update = [pkg for pkg in self.pkgs if pkg.model.update and not pkg.model.is_update_ignored() and pkg.update_checked]
if len(to_update) > 1:
self.disable_progress_controll()
else:
@@ -440,26 +468,37 @@ class UpgradeSelected(AsyncAction):
app_config = read_config()
# trim dialog
# backup dialog ( if enabled, supported and accepted )
should_backup = bkp_supported
should_backup = should_backup and self._check_backup_requirements(app_config=app_config, pkg=None, action_key='upgrade')
should_backup = should_backup and self._should_backup(action_key='upgrade', app_config=app_config, i18n=self.i18n)
# trim dialog ( if enabled and accepted )
if app_config['disk']['trim']['after_upgrade'] is not False:
should_trim = app_config['disk']['trim']['after_upgrade'] or self._ask_for_trim()
else:
should_trim = False
# backup process ( if enabled, supported and accepted )
if bool(app_config['backup']['enabled']) and app_config['backup']['upgrade'] in (True, None) and timeshift.is_available():
any_requires_bkp = False
if should_trim and not root_user and root_password is None: # trim requires root and the password might have not being asked yet
valid_password, root_password = self._request_password()
for dep in requirements.to_upgrade:
if dep.pkg.supports_backup():
any_requires_bkp = True
break
if not valid_password:
self.notify_finished({'success': False, 'updated': 0, 'types': set(), 'id': None})
self.pkgs = None
return
if any_requires_bkp:
if not self.request_backup(app_config, 'upgrade', self.i18n, root_password):
self.notify_finished({'success': success, 'updated': updated, 'types': updated_types, 'id': None})
self.pkgs = None
return
# performing backup
if should_backup:
proceed, root_password = self.request_backup(action_key='upgrade',
app_config=app_config,
i18n=self.i18n,
root_password=root_password,
pkg=None,
backup_only=True)
if not proceed:
self.notify_finished({'success': False, 'updated': 0, 'types': set(), 'id': None})
self.pkgs = None
return
self.change_substatus('')
@@ -497,7 +536,8 @@ class RefreshApps(AsyncAction):
def run(self):
try:
res = self.manager.read_installed(pkg_types=self.pkg_types)
res = self.manager.read_installed(pkg_types=self.pkg_types, disk_loader=None, only_apps=False,
internet_available=True, limit=-1)
refreshed_types = set()
if res:
@@ -528,12 +568,16 @@ class UninstallPackage(AsyncAction):
def run(self):
if self.pkg:
if self.pkg.model.supports_backup():
if not self.request_backup(read_config(), 'uninstall', self.i18n, self.root_pwd):
self.notify_finished(False)
self.pkg = None
self.root_pwd = None
return
proceed, _ = self.request_backup(action_key='uninstall',
pkg=self.pkg,
i18n=self.i18n,
root_password=self.root_pwd,
app_config=read_config())
if not proceed:
self.notify_finished({'success': False, 'removed': None, 'pkg': self.pkg})
self.pkg = None
self.root_pwd = None
return
try:
res = self.manager.uninstall(self.pkg.model, self.root_pwd, self, None)
@@ -567,12 +611,17 @@ class DowngradePackage(AsyncAction):
if self.pkg:
success = False
if self.pkg.model.supports_backup():
if not self.request_backup(read_config(), 'downgrade', self.i18n, self.root_pwd):
self.notify_finished({'app': self.pkg, 'success': success})
self.pkg = None
self.root_pwd = None
return
proceed, _ = self.request_backup(action_key='downgrade',
pkg=self.pkg,
i18n=self.i18n,
root_password=self.root_pwd,
app_config=read_config())
if not proceed:
self.notify_finished({'app': self.pkg, 'success': success})
self.pkg = None
self.root_pwd = None
return
try:
success = self.manager.downgrade(self.pkg.model, self.root_pwd, self)
@@ -630,7 +679,7 @@ class SearchPackages(AsyncAction):
if self.word:
try:
res = self.manager.search(self.word)
res = self.manager.search(words=self.word, disk_loader=None, limit=-1, is_url=False)
search_res['pkgs_found'].extend(res.installed)
search_res['pkgs_found'].extend(res.new)
except NoInternetException:
@@ -654,10 +703,17 @@ class InstallPackage(AsyncAction):
if self.pkg:
res = {'success': False, 'installed': None, 'removed': None, 'pkg': self.pkg}
if self.pkg.model.supports_backup():
if not self.request_backup(read_config(), 'install', self.i18n, self.root_pwd):
self.signal_finished.emit(res)
return
proceed, _ = self.request_backup(action_key='install',
pkg=self.pkg,
i18n=self.i18n,
root_password=self.root_pwd,
app_config=read_config())
if not proceed:
self.signal_finished.emit(res)
self.pkg = None
self.root_pwd = None
return
try:
transaction_res = self.manager.install(self.pkg.model, self.root_pwd, None, self)
@@ -834,7 +890,7 @@ class ListWarnings(QThread):
self.man = man
def run(self):
warnings = self.man.list_warnings()
warnings = self.man.list_warnings(internet_available=True)
if warnings:
self.signal_warnings.emit(warnings)
@@ -949,8 +1005,12 @@ class CustomAction(AsyncAction):
def run(self):
if self.custom_action.backup:
app_config = read_config()
if not self.request_backup(app_config, None, self.i18n, self.root_pwd):
proceed, _ = self.request_backup(app_config=read_config(),
action_key=None,
i18n=self.i18n,
root_password=self.root_pwd,
pkg=self.pkg)
if not proceed:
self.notify_finished({'success': False, 'pkg': self.pkg, 'action': self.custom_action})
self.pkg = None
self.custom_action = None

View File

@@ -14,7 +14,7 @@ from PyQt5.QtWidgets import QWidget, QVBoxLayout, QCheckBox, QHeaderView, QToolB
from bauh import LOGS_PATH
from bauh.api.abstract.cache import MemoryCache
from bauh.api.abstract.context import ApplicationContext
from bauh.api.abstract.controller import SoftwareManager
from bauh.api.abstract.controller import SoftwareManager, SoftwareAction
from bauh.api.abstract.model import SoftwarePackage
from bauh.api.abstract.view import MessageType
from bauh.api.http import HttpClient
@@ -712,7 +712,7 @@ class ManageWindow(QWidget):
self._finish_search(res)
def begin_uninstall(self, pkg: PackageView):
pwd, proceed = self._ask_root_password('uninstall', pkg)
pwd, proceed = self._ask_root_password(SoftwareAction.UNINSTALL, pkg)
if not proceed:
return
@@ -1160,7 +1160,7 @@ class ManageWindow(QWidget):
self.progress_controll_enabled = True
def begin_downgrade(self, pkg: PackageView):
pwd, proceed = self._ask_root_password('downgrade', pkg)
pwd, proceed = self._ask_root_password(SoftwareAction.DOWNGRADE, pkg)
if not proceed:
return
@@ -1281,7 +1281,7 @@ class ManageWindow(QWidget):
self.comp_manager.restore_state(ACTION_SEARCH)
dialog.show_message(title=self.i18n['warning'].capitalize(), body=self.i18n[res['error']], type_=MessageType.WARNING)
def _ask_root_password(self, action: str, pkg: PackageView) -> Tuple[Optional[str], bool]:
def _ask_root_password(self, action: SoftwareAction, pkg: PackageView) -> Tuple[Optional[str], bool]:
pwd = None
requires_root = self.manager.requires_root(action, pkg.model)
@@ -1293,7 +1293,7 @@ class ManageWindow(QWidget):
return pwd, True
def install(self, pkg: PackageView):
pwd, proceed = self._ask_root_password('install', pkg)
pwd, proceed = self._ask_root_password(SoftwareAction.INSTALL, pkg)
if not proceed:
return

View File

@@ -58,7 +58,7 @@ action.cancelled=lusuari ha cancel·lat loperació
action.disk_trim=Optimizing disc (TRIM)
action.disk_trim.error=There was a problem while optimizing the disk
action.failed=L'acció ha fallat
action.history.no_history.body=There is no available history for {}
action.history.no_history.body=There is no history available for {}
action.history.no_history.title=No history
action.ignore_updates.fail=It was not possible to ignore updates from {}
action.ignore_updates.success=Updates from {} will be ignored from now on

View File

@@ -58,7 +58,7 @@ action.cancelled=Aktion vom Nutzer abgebrochen
action.disk_trim=Optimizing disc (TRIM)
action.disk_trim.error=There was a problem while optimizing the disk
action.failed=Aktion fehlgeschlagen
action.history.no_history.body=There is no available history for {}
action.history.no_history.body=There is no history available for {}
action.history.no_history.title=No history
action.ignore_updates.fail=It was not possible to ignore updates from {}
action.ignore_updates.success=Updates from {} will be ignored from now on

View File

@@ -58,7 +58,7 @@ action.cancelled=operation cancelled by the user
action.disk_trim.error=There was a problem while optimizing the disk
action.disk_trim=Optimizing disc (TRIM)
action.failed=Action failed
action.history.no_history.body=There is no available history for {}
action.history.no_history.body=There is no history available for {}
action.history.no_history.title=No history
action.ignore_updates.fail=It was not possible to ignore updates from {}
action.ignore_updates.success=Updates from {} will be ignored from now on

View File

@@ -58,7 +58,7 @@ action.cancelled=operación cancelada por el usuario
action.disk_trim=Optimizando el disco (TRIM)
action.disk_trim.error=Hubo un problema al optimizar el disco
action.failed=Acción fallida
action.history.no_history.body=There is no available history for {}
action.history.no_history.body=No hay historial disponible para {}
action.history.no_history.title=No history
action.ignore_updates.fail=No fue posible ignorar las actualizaciones de {}
action.ignore_updates.success=Las actualizaciones de {} serán ignoradas de ahora en adelante

View File

@@ -58,7 +58,7 @@ action.cancelled=operazione annullata dall'utente
action.disk_trim=Optimizing disc (TRIM)
action.disk_trim.error=There was a problem while optimizing the disk
action.failed=Azione non riuscita
action.history.no_history.body=There is no available history for {}
action.history.no_history.body=There is no history available for {}
action.history.no_history.title=No history
action.ignore_updates.fail=It was not possible to ignore updates from {}
action.ignore_updates.success=Updates from {} will be ignored from now on

View File

@@ -58,7 +58,7 @@ action.cancelled=Операция отменена пользователем
action.disk_trim=Optimizing disc (TRIM)
action.disk_trim.error=There was a problem while optimizing the disk
action.failed=Действие не удалось
action.history.no_history.body=There is no available history for {}
action.history.no_history.body=There is no history available for {}
action.history.no_history.title=No history
action.ignore_updates.fail=It was not possible to ignore updates from {}
action.ignore_updates.success=Updates from {} will be ignored from now on