diff --git a/CHANGELOG.md b/CHANGELOG.md
index 733d947a..832e524b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -25,6 +25,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
- some unneeded cursor icons removed from the apps table
- some icons replaced
+### Fixes
+- General
+ - not accepting blank root passwords [#235](https://github.com/vinifmor/bauh/issues/235)
+
+
## [0.9.28] 2022-02-14
### Fixes
diff --git a/bauh/api/abstract/controller.py b/bauh/api/abstract/controller.py
index 6b68f4e0..e80f12b4 100644
--- a/bauh/api/abstract/controller.py
+++ b/bauh/api/abstract/controller.py
@@ -143,7 +143,7 @@ class SoftwareManager(ABC):
pass
@abstractmethod
- def downgrade(self, pkg: SoftwarePackage, root_password: str, handler: ProcessWatcher) -> bool:
+ def downgrade(self, pkg: SoftwarePackage, root_password: Optional[str], handler: ProcessWatcher) -> bool:
"""
downgrades a package version
:param pkg:
@@ -162,7 +162,7 @@ class SoftwareManager(ABC):
if pkg.supports_disk_cache() and os.path.exists(pkg.get_disk_cache_path()):
shutil.rmtree(pkg.get_disk_cache_path())
- def get_upgrade_requirements(self, pkgs: List[SoftwarePackage], root_password: str, watcher: ProcessWatcher) -> UpgradeRequirements:
+ def get_upgrade_requirements(self, pkgs: List[SoftwarePackage], root_password: Optional[str], watcher: ProcessWatcher) -> UpgradeRequirements:
"""
return additional required software that needs to be installed / removed / updated before updating a list of packages
:param pkgs:
@@ -172,7 +172,7 @@ class SoftwareManager(ABC):
return UpgradeRequirements(None, None, [UpgradeRequirement(p) for p in pkgs], None)
@abstractmethod
- def upgrade(self, requirements: UpgradeRequirements, root_password: str, watcher: ProcessWatcher) -> bool:
+ def upgrade(self, requirements: UpgradeRequirements, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
"""
:param requirements:
:param root_password: the root user password (if required)
@@ -182,7 +182,7 @@ class SoftwareManager(ABC):
pass
@abstractmethod
- def uninstall(self, pkg: SoftwarePackage, root_password: str, watcher: ProcessWatcher, disk_loader: Optional[DiskCacheLoader]) -> TransactionResult:
+ def uninstall(self, pkg: SoftwarePackage, root_password: Optional[str], watcher: ProcessWatcher, disk_loader: Optional[DiskCacheLoader]) -> TransactionResult:
"""
:param pkg:
:param root_password: the root user password (if required)
@@ -217,7 +217,7 @@ class SoftwareManager(ABC):
pass
@abstractmethod
- def install(self, pkg: SoftwarePackage, root_password: str, disk_loader: Optional[DiskCacheLoader], watcher: ProcessWatcher) -> TransactionResult:
+ def install(self, pkg: SoftwarePackage, root_password: Optional[str], disk_loader: Optional[DiskCacheLoader], watcher: ProcessWatcher) -> TransactionResult:
"""
:param pkg:
:param root_password: the root user password (if required)
@@ -334,7 +334,7 @@ class SoftwareManager(ABC):
"""
pass
- def execute_custom_action(self, action: CustomSoftwareAction, pkg: SoftwarePackage, root_password: str, watcher: ProcessWatcher) -> bool:
+ def execute_custom_action(self, action: CustomSoftwareAction, pkg: SoftwarePackage, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
"""
At the moment the GUI implements this action. No need to implement it yourself.
:param action:
diff --git a/bauh/api/abstract/download.py b/bauh/api/abstract/download.py
index 8c4eb970..eead54d7 100644
--- a/bauh/api/abstract/download.py
+++ b/bauh/api/abstract/download.py
@@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
-from typing import Iterable, List, Optional, Tuple
+from typing import Iterable, List, Optional
from bauh.api.abstract.handler import ProcessWatcher
@@ -7,7 +7,7 @@ from bauh.api.abstract.handler import ProcessWatcher
class FileDownloader(ABC):
@abstractmethod
- def download(self, file_url: str, watcher: Optional[ProcessWatcher], output_path: str, cwd: str, root_password: str = None, substatus_prefix: str = None, display_file_size: bool = True, max_threads: int = None, known_size: int = None) -> bool:
+ def download(self, file_url: str, watcher: Optional[ProcessWatcher], output_path: str, cwd: str, root_password: Optional[str] = None, substatus_prefix: str = None, display_file_size: bool = True, max_threads: int = None, known_size: int = None) -> bool:
"""
:param file_url:
:param watcher:
diff --git a/bauh/api/abstract/model.py b/bauh/api/abstract/model.py
index d33bfc06..06c352ff 100644
--- a/bauh/api/abstract/model.py
+++ b/bauh/api/abstract/model.py
@@ -18,7 +18,7 @@ class CustomSoftwareAction:
:param i18n_label_key: the i18n key that will be used to display the action name
:param i18n_status_key: the i18n key that will be used to display the action name being executed
:param icon_path: the action icon path. Use None for no icon
- :param manager_method: the SoftwareManager method name that should be called. The method must has the following parameters: (pkg: SoftwarePackage, root_password: str, watcher: ProcessWatcher)
+ :param manager_method: the SoftwareManager method name that should be called. The method must has the following parameters: (pkg: SoftwarePackage, root_password: Optional[str], watcher: ProcessWatcher)
:param manager: the instance that will execute the action ( optional )
:param backup: if a system backup should be performed before executing the action
:param requires_root:
diff --git a/bauh/commons/system.py b/bauh/commons/system.py
index c18dd383..0238029a 100644
--- a/bauh/commons/system.py
+++ b/bauh/commons/system.py
@@ -65,7 +65,7 @@ class SystemProcess:
class SimpleProcess:
def __init__(self, cmd: List[str], cwd: str = '.', expected_code: int = 0,
- global_interpreter: bool = USE_GLOBAL_INTERPRETER, lang: str = DEFAULT_LANG, root_password: str = None,
+ global_interpreter: bool = USE_GLOBAL_INTERPRETER, lang: str = DEFAULT_LANG, root_password: Optional[str] = None,
extra_paths: Set[str] = None, error_phrases: Set[str] = None, wrong_error_phrases: Set[str] = None,
shell: bool = False, success_phrases: Set[str] = None, extra_env: Optional[Dict[str, str]] = None,
custom_user: Optional[str] = None):
@@ -75,7 +75,7 @@ class SimpleProcess:
if custom_user:
final_cmd.extend(['runuser', '-u', custom_user, '--'])
- elif root_password is not None:
+ elif isinstance(root_password, str):
final_cmd.extend(['sudo', '-S'])
pwdin = self._new(['echo', root_password], cwd, global_interpreter, lang).stdout
@@ -279,12 +279,12 @@ def new_subprocess(cmd: List[str], cwd: str = '.', shell: bool = False, stdin =
return subprocess.Popen(final_cmd, **args)
-def new_root_subprocess(cmd: List[str], root_password: str, cwd: str = '.',
+def new_root_subprocess(cmd: List[str], root_password: Optional[str], cwd: str = '.',
global_interpreter: bool = USE_GLOBAL_INTERPRETER, lang: str = DEFAULT_LANG,
extra_paths: Set[str] = None) -> subprocess.Popen:
pwdin, final_cmd = None, []
- if root_password is not None:
+ if isinstance(root_password, str):
final_cmd.extend(['sudo', '-S'])
pwdin = new_subprocess(['echo', root_password], global_interpreter=global_interpreter, lang=lang).stdout
diff --git a/bauh/gems/appimage/controller.py b/bauh/gems/appimage/controller.py
index f3016908..0c431837 100644
--- a/bauh/gems/appimage/controller.py
+++ b/bauh/gems/appimage/controller.py
@@ -80,7 +80,7 @@ class AppImageManager(SoftwareManager):
self.configman = AppImageConfigManager()
self._custom_actions: Optional[Iterable[CustomSoftwareAction]] = None
- def install_file(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def install_file(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
file_chooser = FileChooserComponent(label=self.i18n['file'].capitalize(),
allowed_extensions={'AppImage', '*'},
search_path=get_default_manual_installation_file_dir())
@@ -134,7 +134,7 @@ class AppImageManager(SoftwareManager):
return res
- def update_file(self, pkg: AppImage, root_password: str, watcher: ProcessWatcher):
+ def update_file(self, pkg: AppImage, root_password: Optional[str], watcher: ProcessWatcher):
file_chooser = FileChooserComponent(label=self.i18n['file'].capitalize(),
allowed_extensions={'AppImage', '*'},
search_path=get_default_manual_installation_file_dir())
@@ -298,7 +298,7 @@ class AppImageManager(SoftwareManager):
res.total = len(res.installed)
return res
- def downgrade(self, pkg: AppImage, root_password: str, watcher: ProcessWatcher) -> bool:
+ def downgrade(self, pkg: AppImage, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
versions = self.get_history(pkg)
if len(versions.history) == 1:
@@ -342,7 +342,7 @@ class AppImageManager(SoftwareManager):
return False
- def upgrade(self, requirements: UpgradeRequirements, root_password: str, watcher: ProcessWatcher) -> bool:
+ def upgrade(self, requirements: UpgradeRequirements, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
not_upgraded = []
for req in requirements.to_upgrade:
@@ -381,7 +381,7 @@ class AppImageManager(SoftwareManager):
watcher.change_substatus('')
return not all_failed
- def uninstall(self, pkg: AppImage, root_password: str, watcher: ProcessWatcher, disk_loader: DiskCacheLoader = None) -> TransactionResult:
+ def uninstall(self, pkg: AppImage, root_password: Optional[str], watcher: ProcessWatcher, disk_loader: DiskCacheLoader = None) -> TransactionResult:
if os.path.exists(pkg.get_disk_cache_path()):
handler = ProcessHandler(watcher)
@@ -530,7 +530,7 @@ class AppImageManager(SoftwareManager):
return file_name, file_path
- def install(self, pkg: AppImage, root_password: str, disk_loader: Optional[DiskCacheLoader], watcher: ProcessWatcher) -> TransactionResult:
+ def install(self, pkg: AppImage, root_password: Optional[str], disk_loader: Optional[DiskCacheLoader], watcher: ProcessWatcher) -> TransactionResult:
return self._install(pkg=pkg, watcher=watcher)
def _install(self, pkg: AppImage, watcher: ProcessWatcher, pre_downloaded_file: Optional[Tuple[str, str]] = None):
@@ -690,7 +690,7 @@ class AppImageManager(SoftwareManager):
def requires_root(self, action: SoftwareAction, pkg: AppImage) -> bool:
return False
- def prepare(self, task_manager: TaskManager, root_password: str, internet_available: bool):
+ def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
create_config = CreateConfigFile(taskman=task_manager, configman=self.configman, i18n=self.i18n,
task_icon_path=get_icon_path(), logger=self.logger)
create_config.start()
@@ -863,7 +863,7 @@ class AppImageManager(SoftwareManager):
requires_internet=True))
yield from self._custom_actions
- def get_upgrade_requirements(self, pkgs: List[AppImage], root_password: str, watcher: ProcessWatcher) -> UpgradeRequirements:
+ def get_upgrade_requirements(self, pkgs: List[AppImage], root_password: Optional[str], watcher: ProcessWatcher) -> UpgradeRequirements:
to_update = []
for pkg in pkgs:
@@ -925,7 +925,7 @@ class AppImageManager(SoftwareManager):
pkg.updates_ignored = False
- def update_database(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def update_database(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
db_updater = DatabaseUpdater(i18n=self.i18n, http_client=self.context.http_client,
logger=self.context.logger, watcher=watcher, taskman=TaskManager())
diff --git a/bauh/gems/arch/controller.py b/bauh/gems/arch/controller.py
index af8b2aac..06939bc0 100644
--- a/bauh/gems/arch/controller.py
+++ b/bauh/gems/arch/controller.py
@@ -70,7 +70,7 @@ RE_PKG_ENDS_WITH_BIN = re.compile(r'.+[\-_]bin$')
class TransactionContext:
def __init__(self, aur_supported: bool, name: str = None, base: str = None, maintainer: str = None, watcher: ProcessWatcher = None,
- handler: ProcessHandler = None, dependency: bool = None, skip_opt_deps: bool = False, root_password: str = None,
+ handler: ProcessHandler = None, dependency: bool = None, skip_opt_deps: bool = False, root_password: Optional[str] = None,
build_dir: str = None, project_dir: str = None, change_progress: bool = False, arch_config: dict = None,
install_files: Set[str] = None, repository: str = None, pkg: ArchPackage = None,
remote_repo_map: Dict[str, str] = None, provided_map: Dict[str, Set[str]] = None,
@@ -115,7 +115,7 @@ class TransactionContext:
self.update_aur_index = update_aur_index
@classmethod
- def gen_context_from(cls, pkg: ArchPackage, arch_config: dict, root_password: str, handler: ProcessHandler, aur_supported: Optional[bool] = None) -> "TransactionContext":
+ def gen_context_from(cls, pkg: ArchPackage, arch_config: dict, root_password: Optional[str], handler: ProcessHandler, aur_supported: Optional[bool] = None) -> "TransactionContext":
return cls(name=pkg.name, base=pkg.get_base_name(), maintainer=pkg.maintainer, repository=pkg.repository,
arch_config=arch_config, watcher=handler.watcher, handler=handler, skip_opt_deps=True,
change_progress=True, root_password=root_password, dependency=False,
@@ -216,7 +216,7 @@ class ArchManager(SoftwareManager):
self.disk_cache_updater = disk_cache_updater
self.pkgbuilder_user: Optional[str] = f'{__app_name__}-aur' if context.root_user else None
- def refresh_mirrors(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def refresh_mirrors(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
handler = ProcessHandler(watcher)
if self._is_database_locked(handler, root_password):
@@ -285,7 +285,7 @@ class ArchManager(SoftwareManager):
watcher.change_substatus(self.i18n['arch.sync_databases.substatus'])
return self.sync_databases(root_password=root_password, watcher=watcher)
- def sync_databases(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def sync_databases(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
handler = ProcessHandler(watcher)
if self._is_database_locked(handler, root_password):
@@ -797,7 +797,7 @@ class ArchManager(SoftwareManager):
return self._install(context)
- def downgrade(self, pkg: ArchPackage, root_password: str, watcher: ProcessWatcher) -> bool:
+ def downgrade(self, pkg: ArchPackage, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
if not self.check_action_allowed(pkg, watcher):
return False
@@ -831,7 +831,7 @@ class ArchManager(SoftwareManager):
if os.path.exists(pkg.get_disk_cache_path()):
shutil.rmtree(pkg.get_disk_cache_path())
- def _is_database_locked(self, handler: ProcessHandler, root_password: str) -> bool:
+ def _is_database_locked(self, handler: ProcessHandler, root_password: Optional[str]) -> bool:
if os.path.exists('/var/lib/pacman/db.lck'):
handler.watcher.print('pacman database is locked')
msg = '
{}
{}
'.format(self.i18n['arch.action.db_locked.body.l1'],
@@ -928,7 +928,7 @@ class ArchManager(SoftwareManager):
return related
- def _upgrade_repo_pkgs(self, to_upgrade: List[str], to_remove: Optional[Set[str]], handler: ProcessHandler, root_password: str,
+ def _upgrade_repo_pkgs(self, to_upgrade: List[str], to_remove: Optional[Set[str]], handler: ProcessHandler, root_password: Optional[str],
multithread_download: bool, pkgs_data: Dict[str, dict], overwrite_files: bool = False,
status_handler: TransactionStatusHandler = None, sizes: Dict[str, int] = None, download: bool = True,
check_syncfirst: bool = True, skip_dependency_checks: bool = False) -> bool:
@@ -1069,7 +1069,7 @@ class ArchManager(SoftwareManager):
traceback.print_exc()
return False
- def _remove_transaction_packages(self, to_remove: Set[str], handler: ProcessHandler, root_password: str) -> bool:
+ def _remove_transaction_packages(self, to_remove: Set[str], handler: ProcessHandler, root_password: Optional[str]) -> bool:
output_handler = TransactionStatusHandler(watcher=handler.watcher,
i18n=self.i18n,
names=set(),
@@ -1101,7 +1101,7 @@ class ArchManager(SoftwareManager):
body=self.i18n['arch.upgrade.mthreaddownload.fail'],
type_=MessageType.ERROR)
- def upgrade(self, requirements: UpgradeRequirements, root_password: str, watcher: ProcessWatcher) -> bool:
+ def upgrade(self, requirements: UpgradeRequirements, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
self.aur_client.clean_caches()
watcher.change_status("{}...".format(self.i18n['manage_window.status.upgrading']))
@@ -1193,7 +1193,7 @@ class ArchManager(SoftwareManager):
watcher.change_substatus('')
return True
- def _uninstall_pkgs(self, pkgs: Iterable[str], root_password: str, handler: ProcessHandler, ignore_dependencies: bool = False) -> bool:
+ def _uninstall_pkgs(self, pkgs: Iterable[str], root_password: Optional[str], handler: ProcessHandler, ignore_dependencies: bool = False) -> bool:
status_handler = TransactionStatusHandler(watcher=handler.watcher,
i18n=self.i18n,
names={*pkgs},
@@ -1419,7 +1419,7 @@ class ArchManager(SoftwareManager):
if not exp_provided:
del context.provided_map[exp]
- def uninstall(self, pkg: ArchPackage, root_password: str, watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
+ def uninstall(self, pkg: ArchPackage, root_password: Optional[str], watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
self.aur_client.clean_caches()
handler = ProcessHandler(watcher)
@@ -2282,7 +2282,7 @@ class ArchManager(SoftwareManager):
and self.context.file_downloader.is_multithreaded() \
and pacman.is_mirrors_available()
- def _download_packages(self, pkgnames: List[str], handler: ProcessHandler, root_password: str, sizes: Dict[str, int] = None, multithreaded: bool = True) -> int:
+ def _download_packages(self, pkgnames: List[str], handler: ProcessHandler, root_password: Optional[str], sizes: Dict[str, int] = None, multithreaded: bool = True) -> int:
if multithreaded:
download_service = MultithreadedDownloadService(file_downloader=self.context.file_downloader,
http_client=self.http_client,
@@ -2493,7 +2493,7 @@ class ArchManager(SoftwareManager):
if context.change_progress:
context.watcher.change_progress(val)
- def _import_pgp_keys(self, pkgname: str, root_password: str, handler: ProcessHandler):
+ def _import_pgp_keys(self, pkgname: str, root_password: Optional[str], handler: ProcessHandler):
srcinfo = self.aur_client.get_src_info(pkgname)
if srcinfo.get('validpgpkeys'):
@@ -2560,7 +2560,7 @@ class ArchManager(SoftwareManager):
return False
- def _sync_databases(self, arch_config: dict, aur_supported: bool, root_password: str, handler: ProcessHandler, change_substatus: bool = True):
+ def _sync_databases(self, arch_config: dict, aur_supported: bool, root_password: Optional[str], handler: ProcessHandler, change_substatus: bool = True):
if bool(arch_config['sync_databases']) and database.should_sync(arch_config, aur_supported, handler, self.logger):
if change_substatus:
handler.watcher.change_substatus(self.i18n['arch.sync_databases.substatus'])
@@ -2577,7 +2577,7 @@ class ArchManager(SoftwareManager):
watcher.change_substatus(self.i18n['arch.makepkg.optimizing'])
ArchCompilationOptimizer(i18n=self.i18n, logger=self.context.logger, taskman=TaskManager()).optimize()
- def install(self, pkg: ArchPackage, root_password: str, disk_loader: Optional[DiskCacheLoader], watcher: ProcessWatcher, context: TransactionContext = None) -> TransactionResult:
+ def install(self, pkg: ArchPackage, root_password: Optional[str], disk_loader: Optional[DiskCacheLoader], watcher: ProcessWatcher, context: TransactionContext = None) -> TransactionResult:
if not self.check_action_allowed(pkg, watcher):
return TransactionResult.fail()
@@ -2701,7 +2701,7 @@ class ArchManager(SoftwareManager):
taskman.update_progress('arch_aur_cats', 100, None)
taskman.finish_task('arch_aur_cats')
- def prepare(self, task_manager: TaskManager, root_password: str, internet_available: bool):
+ def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
create_config = CreateConfigFile(taskman=task_manager, configman=self.configman, i18n=self.i18n,
task_icon_path=get_icon_path(), logger=self.logger)
create_config.start()
@@ -3012,7 +3012,7 @@ class ArchManager(SoftwareManager):
except:
return False, [traceback.format_exc()]
- def get_upgrade_requirements(self, pkgs: List[ArchPackage], root_password: str, watcher: ProcessWatcher) -> UpgradeRequirements:
+ def get_upgrade_requirements(self, pkgs: List[ArchPackage], root_password: Optional[str], watcher: ProcessWatcher) -> UpgradeRequirements:
self.aur_client.clean_caches()
arch_config = self.configman.get_config()
@@ -3123,7 +3123,7 @@ class ArchManager(SoftwareManager):
elif new_size is not None:
p.size = new_size - p.size
- def upgrade_system(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def upgrade_system(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
# repo_map = pacman.map_repositories()
net_available = self.context.internet_checker.is_available()
installed = self.read_installed(limit=-1, only_apps=False, pkg_types=None, internet_available=net_available, disk_loader=None).installed
@@ -3198,7 +3198,7 @@ class ArchManager(SoftwareManager):
watcher.request_reboot(msg)
return True
- def clean_cache(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def clean_cache(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
cache_dir = pacman.get_cache_dir()
@@ -3331,15 +3331,15 @@ class ArchManager(SoftwareManager):
return set()
- def enable_pkgbuild_edition(self, pkg: ArchPackage, root_password: str, watcher: ProcessWatcher):
+ def enable_pkgbuild_edition(self, pkg: ArchPackage, root_password: Optional[str], watcher: ProcessWatcher):
if self._add_as_editable_pkgbuild(pkg.name):
pkg.pkgbuild_editable = True
- def disable_pkgbuild_edition(self, pkg: ArchPackage, root_password: str, watcher: ProcessWatcher):
+ def disable_pkgbuild_edition(self, pkg: ArchPackage, root_password: Optional[str], watcher: ProcessWatcher):
if self._remove_from_editable_pkgbuilds(pkg.name):
pkg.pkgbuild_editable = False
- def setup_snapd(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def setup_snapd(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
# checking services
missing_items = []
for serv, active in system.check_enabled_services('snapd.service', 'snapd.socket').items():
@@ -3521,7 +3521,7 @@ class ArchManager(SoftwareManager):
return res
- def reinstall(self, pkg: ArchPackage, root_password: str, watcher: ProcessWatcher) -> bool: # only available for AUR packages
+ def reinstall(self, pkg: ArchPackage, root_password: Optional[str], watcher: ProcessWatcher) -> bool: # only available for AUR packages
if not self.context.internet_checker.is_available():
raise NoInternetException()
@@ -3550,7 +3550,7 @@ class ArchManager(SoftwareManager):
context=context,
disk_loader=self.context.disk_loader_factory.new()).success
- def set_rebuild_check(self, pkg: ArchPackage, root_password: str, watcher: ProcessWatcher) -> bool:
+ def set_rebuild_check(self, pkg: ArchPackage, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
if pkg.repository != 'aur':
return False
diff --git a/bauh/gems/arch/cpu_manager.py b/bauh/gems/arch/cpu_manager.py
index e10dd249..38b3c83f 100644
--- a/bauh/gems/arch/cpu_manager.py
+++ b/bauh/gems/arch/cpu_manager.py
@@ -24,7 +24,7 @@ def current_governors() -> Dict[str, Set[int]]:
return governors
-def set_governor(governor: str, root_password: str, cpu_idxs: Optional[Set[int]] = None):
+def set_governor(governor: str, root_password: Optional[str], cpu_idxs: Optional[Set[int]] = None):
new_gov_file = f'{TEMP_DIR}/bauh_scaling_governor'
with open(new_gov_file, 'w+') as f:
f.write(governor)
@@ -39,7 +39,7 @@ def set_governor(governor: str, root_password: str, cpu_idxs: Optional[Set[int]]
traceback.print_exc()
-def _change_governor(cpu_idx: int, new_gov_file_path: str, root_password: str):
+def _change_governor(cpu_idx: int, new_gov_file_path: str, root_password: Optional[str]):
try:
gov_file = '/sys/devices/system/cpu/cpu{}/cpufreq/scaling_governor'.format(cpu_idx)
replace = new_root_subprocess(['cp', new_gov_file_path, gov_file], root_password=root_password)
@@ -48,7 +48,7 @@ def _change_governor(cpu_idx: int, new_gov_file_path: str, root_password: str):
traceback.print_exc()
-def set_all_cpus_to(governor: str, root_password: str, logger: Optional[Logger] = None) -> Tuple[bool, Optional[Dict[str, Set[int]]]]:
+def set_all_cpus_to(governor: str, root_password: Optional[str], logger: Optional[Logger] = None) -> Tuple[bool, Optional[Dict[str, Set[int]]]]:
"""
"""
cpus_changed, cpu_governors = False, current_governors()
@@ -69,7 +69,7 @@ def set_all_cpus_to(governor: str, root_password: str, logger: Optional[Logger]
return cpus_changed, cpu_governors
-def set_cpus(governors: Dict[str, Set[int]], root_password: str, ignore_governors: Optional[Set[str]] = None, logger: Optional[Logger] = None):
+def set_cpus(governors: Dict[str, Set[int]], root_password: Optional[str], ignore_governors: Optional[Set[str]] = None, logger: Optional[Logger] = None):
for gov, cpus in governors.items():
if not ignore_governors or gov not in ignore_governors:
if logger:
diff --git a/bauh/gems/arch/download.py b/bauh/gems/arch/download.py
index a15f3a4a..acf10c5f 100644
--- a/bauh/gems/arch/download.py
+++ b/bauh/gems/arch/download.py
@@ -4,7 +4,7 @@ import os
import time
import traceback
from threading import Lock, Thread
-from typing import List, Iterable, Dict
+from typing import List, Iterable, Dict, Optional
from bauh.api.abstract.download import FileDownloader
from bauh.api.abstract.handler import ProcessWatcher
@@ -38,7 +38,7 @@ class MultiThreadedDownloader:
self.async_downloads = []
self.async_downloads_lock = Lock()
- def download_package_signature(self, pkg: dict, file_url: str, output_path: str, root_password: str, watcher: ProcessWatcher):
+ def download_package_signature(self, pkg: dict, file_url: str, output_path: str, root_password: Optional[str], watcher: ProcessWatcher):
try:
self.logger.info("Downloading package '{}' signature".format(pkg['n']))
@@ -60,7 +60,7 @@ class MultiThreadedDownloader:
self.logger.warning("An error occurred while download package '{}' signature".format(pkg['n']))
traceback.print_exc()
- def download_package(self, pkg: Dict[str, str], root_password: str, substatus_prefix: str, watcher: ProcessWatcher, size: int) -> bool:
+ def download_package(self, pkg: Dict[str, str], root_password: Optional[str], substatus_prefix: str, watcher: ProcessWatcher, size: int) -> bool:
if self.mirrors and self.branch:
pkgname = '{}-{}{}.pkg'.format(pkg['n'], pkg['v'], ('-{}'.format(pkg['a']) if pkg['a'] else ''))
@@ -117,7 +117,7 @@ class MultithreadedDownloadService:
self.logger = logger
self.i18n = i18n
- def download_packages(self, pkgs: List[str], handler: ProcessHandler, root_password: str, sizes: Dict[str, int] = None) -> int:
+ def download_packages(self, pkgs: List[str], handler: ProcessHandler, root_password: Optional[str], sizes: Dict[str, int] = None) -> int:
ti = time.time()
watcher = handler.watcher
mirrors = pacman.list_available_mirrors()
diff --git a/bauh/gems/arch/pacman.py b/bauh/gems/arch/pacman.py
index 7835a602..2fd00874 100644
--- a/bauh/gems/arch/pacman.py
+++ b/bauh/gems/arch/pacman.py
@@ -146,7 +146,7 @@ def map_installed(names: Iterable[str] = None) -> dict: # returns a dict with w
return pkgs
-def install_as_process(pkgpaths: Iterable[str], root_password: str, file: bool, pkgdir: str = '.',
+def install_as_process(pkgpaths: Iterable[str], root_password: Optional[str], file: bool, pkgdir: str = '.',
overwrite_conflicting_files: bool = False, simulate: bool = False, as_deps: bool = False) -> SimpleProcess:
cmd = ['pacman', '-U'] if file else ['pacman', '-S']
cmd.extend(pkgpaths)
@@ -211,11 +211,11 @@ def verify_pgp_key(key: str) -> bool:
return False
-def receive_key(key: str, root_password: str) -> SystemProcess:
+def receive_key(key: str, root_password: Optional[str]) -> SystemProcess:
return SystemProcess(new_root_subprocess(['pacman-key', '-r', key], root_password=root_password), check_error_output=False)
-def sign_key(key: str, root_password: str) -> SystemProcess:
+def sign_key(key: str, root_password: Optional[str]) -> SystemProcess:
return SystemProcess(new_root_subprocess(['pacman-key', '--lsign-key', key], root_password=root_password), check_error_output=False)
@@ -361,7 +361,7 @@ def read_dependencies(name: str) -> Set[str]:
return depends_on
-def sync_databases(root_password: str, force: bool = False) -> SimpleProcess:
+def sync_databases(root_password: Optional[str], force: bool = False) -> SimpleProcess:
return SimpleProcess(cmd=['pacman', '-Sy{}'.format('y' if force else '')],
root_password=root_password)
@@ -469,15 +469,15 @@ def can_refresh_mirrors() -> bool:
return is_mirrors_available()
-def refresh_mirrors(root_password: str) -> SimpleProcess:
+def refresh_mirrors(root_password: Optional[str]) -> SimpleProcess:
return SimpleProcess(cmd=['pacman-mirrors', '-g'], root_password=root_password)
-def update_mirrors(root_password: str, countries: List[str]) -> SimpleProcess:
+def update_mirrors(root_password: Optional[str], countries: List[str]) -> SimpleProcess:
return SimpleProcess(cmd=['pacman-mirrors', '-c', ','.join(countries)], root_password=root_password)
-def sort_fastest_mirrors(root_password: str, limit: int) -> SimpleProcess:
+def sort_fastest_mirrors(root_password: Optional[str], limit: int) -> SimpleProcess:
cmd = ['pacman-mirrors', '--fasttrack']
if limit > 0:
@@ -529,7 +529,7 @@ def get_installed_size(pkgs: List[str]) -> Dict[str, int]: # bytes
return {}
-def upgrade_system(root_password: str) -> SimpleProcess:
+def upgrade_system(root_password: Optional[str]) -> SimpleProcess:
return SimpleProcess(cmd=['pacman', '-Syyu', '--noconfirm'], root_password=root_password)
@@ -717,7 +717,7 @@ def map_updates_data(pkgs: Iterable[str], files: bool = False) -> dict:
return res
-def upgrade_several(pkgnames: Iterable[str], root_password: str, overwrite_conflicting_files: bool = False, skip_dependency_checks: bool = False) -> SimpleProcess:
+def upgrade_several(pkgnames: Iterable[str], root_password: Optional[str], overwrite_conflicting_files: bool = False, skip_dependency_checks: bool = False) -> SimpleProcess:
cmd = ['pacman', '-S', *pkgnames, '--noconfirm']
if overwrite_conflicting_files:
@@ -732,14 +732,14 @@ def upgrade_several(pkgnames: Iterable[str], root_password: str, overwrite_confl
shell=True)
-def download(root_password: str, *pkgnames: str) -> SimpleProcess:
+def download(root_password: Optional[str], *pkgnames: str) -> SimpleProcess:
return SimpleProcess(cmd=['pacman', '-Swdd', *pkgnames, '--noconfirm', '--noprogressbar'],
root_password=root_password,
error_phrases={'error: failed to prepare transaction', 'error: failed to commit transaction', 'error: target not found'},
shell=True)
-def remove_several(pkgnames: Iterable[str], root_password: str, skip_checks: bool = False) -> SimpleProcess:
+def remove_several(pkgnames: Iterable[str], root_password: Optional[str], skip_checks: bool = False) -> SimpleProcess:
cmd = ['pacman', '-R', *pkgnames, '--noconfirm']
if skip_checks:
diff --git a/bauh/gems/arch/updates.py b/bauh/gems/arch/updates.py
index 2f5113ce..888e3833 100644
--- a/bauh/gems/arch/updates.py
+++ b/bauh/gems/arch/updates.py
@@ -24,7 +24,7 @@ class UpdateRequirementsContext:
pkgs_data: Dict[str, dict], cannot_upgrade: Dict[str, UpgradeRequirement],
to_remove: Dict[str, UpgradeRequirement], installed_names: Set[str], provided_map: Dict[str, Set[str]],
aur_index: Set[str], arch_config: dict, remote_provided_map: Dict[str, Set[str]], remote_repo_map: Dict[str, str],
- root_password: str, aur_supported: bool):
+ root_password: Optional[str], aur_supported: bool):
self.to_update = to_update
self.repo_to_update = repo_to_update
self.aur_to_update = aur_to_update
@@ -367,7 +367,7 @@ class UpdatesSummarizer:
return requirement
- def summarize(self, pkgs: List[ArchPackage], root_password: str, arch_config: dict) -> Optional[UpgradeRequirements]:
+ def summarize(self, pkgs: List[ArchPackage], root_password: Optional[str], arch_config: dict) -> Optional[UpgradeRequirements]:
res = UpgradeRequirements([], [], [], [])
remote_provided_map = pacman.map_provided(remote=True)
diff --git a/bauh/gems/arch/worker.py b/bauh/gems/arch/worker.py
index c33cd28b..bd1bdae4 100644
--- a/bauh/gems/arch/worker.py
+++ b/bauh/gems/arch/worker.py
@@ -409,7 +409,7 @@ class ArchCompilationOptimizer(Thread):
class RefreshMirrors(Thread):
- def __init__(self, taskman: TaskManager, root_password: str, i18n: I18n, logger: logging.Logger,
+ def __init__(self, taskman: TaskManager, root_password: Optional[str], i18n: I18n, logger: logging.Logger,
create_config: CreateConfigFile):
super(RefreshMirrors, self).__init__(daemon=True)
self.taskman = taskman
@@ -488,7 +488,7 @@ class RefreshMirrors(Thread):
class SyncDatabases(Thread):
- def __init__(self, taskman: TaskManager, root_password: str, i18n: I18n, logger: logging.Logger,
+ def __init__(self, taskman: TaskManager, root_password: Optional[str], i18n: I18n, logger: logging.Logger,
refresh_mirrors: RefreshMirrors, create_config: CreateConfigFile):
super(SyncDatabases, self).__init__(daemon=True)
self.task_man = taskman
diff --git a/bauh/gems/flatpak/controller.py b/bauh/gems/flatpak/controller.py
index e13213fa..94cb8a8b 100644
--- a/bauh/gems/flatpak/controller.py
+++ b/bauh/gems/flatpak/controller.py
@@ -190,7 +190,7 @@ class FlatpakManager(SoftwareManager):
return SearchResult([*models.values()], None, len(models))
- def downgrade(self, pkg: FlatpakApplication, root_password: str, watcher: ProcessWatcher) -> bool:
+ def downgrade(self, pkg: FlatpakApplication, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
if not self._make_exports_dir(watcher):
return False
@@ -220,7 +220,7 @@ class FlatpakManager(SoftwareManager):
super(FlatpakManager, self).clean_cache_for(pkg)
self.api_cache.delete(pkg.id)
- def upgrade(self, requirements: UpgradeRequirements, root_password: str, watcher: ProcessWatcher) -> bool:
+ def upgrade(self, requirements: UpgradeRequirements, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
flatpak_version = flatpak.get_version()
if not self._make_exports_dir(watcher):
@@ -260,7 +260,7 @@ class FlatpakManager(SoftwareManager):
watcher.change_substatus('')
return True
- def uninstall(self, pkg: FlatpakApplication, root_password: str, watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
+ def uninstall(self, pkg: FlatpakApplication, root_password: Optional[str], watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
if not self._make_exports_dir(watcher):
return TransactionResult.fail()
@@ -375,7 +375,7 @@ class FlatpakManager(SoftwareManager):
return True
- def install(self, pkg: FlatpakApplication, root_password: str, disk_loader: DiskCacheLoader, watcher: ProcessWatcher) -> TransactionResult:
+ def install(self, pkg: FlatpakApplication, root_password: Optional[str], disk_loader: DiskCacheLoader, watcher: ProcessWatcher) -> TransactionResult:
if not self.context.root_user:
flatpak_config = self.configman.get_config()
install_level = flatpak_config['installation_level']
@@ -493,7 +493,7 @@ class FlatpakManager(SoftwareManager):
def requires_root(self, action: SoftwareAction, pkg: FlatpakApplication) -> bool:
return action == SoftwareAction.DOWNGRADE and pkg.installation == 'system'
- def prepare(self, task_manager: TaskManager, root_password: str, internet_available: bool):
+ def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
CreateConfigFile(taskman=task_manager, configman=self.configman, i18n=self.i18n,
task_icon_path=get_icon_path(), logger=self.logger).start()
@@ -626,7 +626,7 @@ class FlatpakManager(SoftwareManager):
except:
return False, [traceback.format_exc()]
- def get_upgrade_requirements(self, pkgs: List[FlatpakApplication], root_password: str, watcher: ProcessWatcher) -> UpgradeRequirements:
+ def get_upgrade_requirements(self, pkgs: List[FlatpakApplication], root_password: Optional[str], watcher: ProcessWatcher) -> UpgradeRequirements:
flatpak_version = flatpak.get_version()
user_pkgs, system_pkgs = [], []
diff --git a/bauh/gems/flatpak/flatpak.py b/bauh/gems/flatpak/flatpak.py
index 8586af4c..1882e7f9 100755
--- a/bauh/gems/flatpak/flatpak.py
+++ b/bauh/gems/flatpak/flatpak.py
@@ -229,7 +229,7 @@ def read_updates(version: Version, installation: str) -> Dict[str, set]:
return res
-def downgrade(app_ref: str, commit: str, installation: str, root_password: str) -> SimpleProcess:
+def downgrade(app_ref: str, commit: str, installation: str, root_password: Optional[str]) -> SimpleProcess:
cmd = ['flatpak', 'update', '--no-related', '--no-deps', '--commit={}'.format(commit), app_ref, '-y', '--{}'.format(installation)]
return SimpleProcess(cmd=cmd,
@@ -366,7 +366,7 @@ def install(app_id: str, origin: str, installation: str) -> SimpleProcess:
shell=True)
-def set_default_remotes(installation: str, root_password: str = None) -> SimpleProcess:
+def set_default_remotes(installation: str, root_password: Optional[str] = None) -> SimpleProcess:
cmd = ['flatpak', 'remote-add', '--if-not-exists', 'flathub', 'https://flathub.org/repo/flathub.flatpakrepo', '--{}'.format(installation)]
return SimpleProcess(cmd, root_password=root_password)
diff --git a/bauh/gems/snap/controller.py b/bauh/gems/snap/controller.py
index 48d39be5..4cd443c4 100644
--- a/bauh/gems/snap/controller.py
+++ b/bauh/gems/snap/controller.py
@@ -98,7 +98,7 @@ class SnapManager(SoftwareManager):
else:
return SearchResult([], None, 0)
- def downgrade(self, pkg: SnapApplication, root_password: str, watcher: ProcessWatcher) -> bool:
+ def downgrade(self, pkg: SnapApplication, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
if not snap.is_installed():
watcher.print("'snap' seems not to be installed")
return False
@@ -108,10 +108,10 @@ class SnapManager(SoftwareManager):
return ProcessHandler(watcher).handle_simple(snap.downgrade_and_stream(pkg.name, root_password))[0]
- def upgrade(self, requirements: UpgradeRequirements, root_password: str, watcher: ProcessWatcher) -> SystemProcess:
+ def upgrade(self, requirements: UpgradeRequirements, root_password: Optional[str], watcher: ProcessWatcher) -> SystemProcess:
raise Exception(f"'upgrade' is not supported by {SnapManager.__class__.__name__}")
- def uninstall(self, pkg: SnapApplication, root_password: str, watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
+ def uninstall(self, pkg: SnapApplication, root_password: Optional[str], watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
if snap.is_installed() and snapd.is_running():
uninstalled = ProcessHandler(watcher).handle_simple(snap.uninstall_and_stream(pkg.name, root_password))[0]
@@ -160,7 +160,7 @@ class SnapManager(SoftwareManager):
def get_history(self, pkg: SnapApplication) -> PackageHistory:
raise Exception(f"'get_history' is not supported by {pkg.__class__.__name__}")
- def install(self, pkg: SnapApplication, root_password: str, disk_loader: DiskCacheLoader, watcher: ProcessWatcher) -> TransactionResult:
+ def install(self, pkg: SnapApplication, root_password: Optional[str], disk_loader: DiskCacheLoader, watcher: ProcessWatcher) -> TransactionResult:
# retrieving all installed so it will be possible to know the additional installed runtimes after the operation succeeds
if not snap.is_installed():
watcher.print("'snap' seems not to be installed")
@@ -244,10 +244,10 @@ class SnapManager(SoftwareManager):
def requires_root(self, action: SoftwareAction, pkg: SnapApplication) -> bool:
return action not in (SoftwareAction.PREPARE, SoftwareAction.SEARCH)
- def refresh(self, pkg: SnapApplication, root_password: str, watcher: ProcessWatcher) -> bool:
+ def refresh(self, pkg: SnapApplication, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
return ProcessHandler(watcher).handle_simple(snap.refresh_and_stream(pkg.name, root_password))[0]
- def change_channel(self, pkg: SnapApplication, root_password: str, watcher: ProcessWatcher) -> bool:
+ def change_channel(self, pkg: SnapApplication, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
if not self.context.internet_checker.is_available():
raise NoInternetException()
@@ -283,7 +283,7 @@ class SnapManager(SoftwareManager):
taskman.update_progress('snap_cats', 100, None)
taskman.finish_task('snap_cats')
- def prepare(self, task_manager: TaskManager, root_password: str, internet_available: bool):
+ def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
create_config = CreateConfigFile(taskman=task_manager, configman=self.configman, i18n=self.i18n,
task_icon_path=get_icon_path(), logger=self.logger)
create_config.start()
diff --git a/bauh/gems/snap/snap.py b/bauh/gems/snap/snap.py
index 82cba496..002383db 100644
--- a/bauh/gems/snap/snap.py
+++ b/bauh/gems/snap/snap.py
@@ -13,13 +13,13 @@ def is_installed() -> bool:
return bool(shutil.which(BASE_CMD))
-def uninstall_and_stream(app_name: str, root_password: str) -> SimpleProcess:
+def uninstall_and_stream(app_name: str, root_password: Optional[str]) -> SimpleProcess:
return SimpleProcess(cmd=[BASE_CMD, 'remove', app_name],
root_password=root_password,
shell=True)
-def install_and_stream(app_name: str, confinement: str, root_password: str, channel: Optional[str] = None) -> SimpleProcess:
+def install_and_stream(app_name: str, confinement: str, root_password: Optional[str], channel: Optional[str] = None) -> SimpleProcess:
install_cmd = [BASE_CMD, 'install', app_name] # default
@@ -32,13 +32,13 @@ def install_and_stream(app_name: str, confinement: str, root_password: str, chan
return SimpleProcess(install_cmd, root_password=root_password, shell=True)
-def downgrade_and_stream(app_name: str, root_password: str) -> SimpleProcess:
+def downgrade_and_stream(app_name: str, root_password: Optional[str]) -> SimpleProcess:
return SimpleProcess(cmd=[BASE_CMD, 'revert', app_name],
root_password=root_password,
shell=True)
-def refresh_and_stream(app_name: str, root_password: str, channel: Optional[str] = None) -> SimpleProcess:
+def refresh_and_stream(app_name: str, root_password: Optional[str], channel: Optional[str] = None) -> SimpleProcess:
cmd = [BASE_CMD, 'refresh', app_name]
if channel:
diff --git a/bauh/gems/web/controller.py b/bauh/gems/web/controller.py
index 5d6093f2..9d965c81 100644
--- a/bauh/gems/web/controller.py
+++ b/bauh/gems/web/controller.py
@@ -83,7 +83,7 @@ class WebApplicationManager(SoftwareManager):
except:
return 'en_US'
- def clean_environment(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def clean_environment(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
handler = ProcessHandler(watcher)
success = True
@@ -368,13 +368,13 @@ class WebApplicationManager(SoftwareManager):
return res
- def downgrade(self, pkg: SoftwarePackage, root_password: str, handler: ProcessWatcher) -> bool:
+ def downgrade(self, pkg: SoftwarePackage, root_password: Optional[str], handler: ProcessWatcher) -> bool:
pass
- def upgrade(self, requirements: UpgradeRequirements, root_password: str, watcher: ProcessWatcher) -> bool:
+ def upgrade(self, requirements: UpgradeRequirements, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
pass
- def uninstall(self, pkg: WebApplication, root_password: str, watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
+ def uninstall(self, pkg: WebApplication, root_password: Optional[str], watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult:
self.logger.info("Checking if {} installation directory {} exists".format(pkg.name, pkg.installation_dir))
if not os.path.exists(pkg.installation_dir):
@@ -842,7 +842,7 @@ class WebApplicationManager(SoftwareManager):
return TransactionResult(success=True, installed=[pkg], removed=[])
- def install(self, pkg: WebApplication, root_password: str, disk_loader: DiskCacheLoader, watcher: ProcessWatcher) -> TransactionResult:
+ def install(self, pkg: WebApplication, root_password: Optional[str], disk_loader: DiskCacheLoader, watcher: ProcessWatcher) -> TransactionResult:
continue_install, install_options = self._ask_install_options(pkg, watcher, pre_validated=True)
if not continue_install:
@@ -850,7 +850,7 @@ class WebApplicationManager(SoftwareManager):
return self._install(pkg, install_options, watcher)
- def install_app(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def install_app(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
pkg = WebApplication()
continue_install, install_options = self._ask_install_options(pkg, watcher, pre_validated=False)
@@ -908,7 +908,7 @@ class WebApplicationManager(SoftwareManager):
def _assign_suggestions(self, suggestions: dict):
self.suggestions = suggestions
- def prepare(self, task_manager: TaskManager, root_password: str, internet_available: bool):
+ def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
create_config = CreateConfigFile(taskman=task_manager, configman=self.configman, i18n=self.i18n,
task_icon_path=get_icon_path(), logger=self.logger)
create_config.start()
@@ -1063,7 +1063,7 @@ class WebApplicationManager(SoftwareManager):
return res
- def execute_custom_action(self, action: CustomSoftwareAction, pkg: SoftwarePackage, root_password: str, watcher: ProcessWatcher) -> bool:
+ def execute_custom_action(self, action: CustomSoftwareAction, pkg: SoftwarePackage, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
pass
def is_default_enabled(self) -> bool:
diff --git a/bauh/view/core/controller.py b/bauh/view/core/controller.py
index 04be6374..f9835e01 100755
--- a/bauh/view/core/controller.py
+++ b/bauh/view/core/controller.py
@@ -98,7 +98,7 @@ class GenericSoftwareManager(SoftwareManager):
self._available_cache = {}
self.working_managers.clear()
- def launch_timeshift(self, root_password: str, watcher: ProcessWatcher):
+ def launch_timeshift(self, root_password: Optional[str], watcher: ProcessWatcher):
if self._is_timeshift_launcher_available():
try:
Popen(['timeshift-launcher'], stderr=STDOUT)
@@ -281,7 +281,7 @@ class GenericSoftwareManager(SoftwareManager):
self.logger.info(f'Took {tf - ti:.2f} seconds')
return res
- def downgrade(self, app: SoftwarePackage, root_password: str, handler: ProcessWatcher) -> bool:
+ def downgrade(self, app: SoftwarePackage, root_password: Optional[str], handler: ProcessWatcher) -> bool:
man = self._get_manager_for(app)
if man and app.can_be_downgraded():
@@ -299,7 +299,7 @@ class GenericSoftwareManager(SoftwareManager):
if man:
return man.clean_cache_for(app)
- def upgrade(self, requirements: GenericUpgradeRequirements, root_password: str, handler: ProcessWatcher) -> bool:
+ def upgrade(self, requirements: GenericUpgradeRequirements, root_password: Optional[str], handler: ProcessWatcher) -> bool:
for man, man_reqs in requirements.sub_requirements.items():
res = man.upgrade(man_reqs, root_password, handler)
@@ -324,7 +324,7 @@ class GenericSoftwareManager(SoftwareManager):
for p in res.removed:
self._fill_post_transaction_status(p, False)
- def uninstall(self, pkg: SoftwarePackage, root_password: str, handler: ProcessWatcher, disk_loader: DiskCacheLoader = None) -> TransactionResult:
+ def uninstall(self, pkg: SoftwarePackage, root_password: Optional[str], handler: ProcessWatcher, disk_loader: DiskCacheLoader = None) -> TransactionResult:
man = self._get_manager_for(pkg)
if man:
@@ -345,7 +345,7 @@ class GenericSoftwareManager(SoftwareManager):
tf = time.time()
self.logger.info(f'Uninstallation of {pkg} took {(tf - ti) / 60:.2f} minutes')
- def install(self, app: SoftwarePackage, root_password: str, disk_loader: DiskCacheLoader, handler: ProcessWatcher) -> TransactionResult:
+ def install(self, app: SoftwarePackage, root_password: Optional[str], disk_loader: DiskCacheLoader, handler: ProcessWatcher) -> TransactionResult:
man = self._get_manager_for(app)
if man:
@@ -418,7 +418,7 @@ class GenericSoftwareManager(SoftwareManager):
if man:
return man.requires_root(action, app)
- def prepare(self, task_manager: TaskManager, root_password: str, internet_available: bool):
+ def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
ti = time.time()
self.logger.info("Initializing")
taskman = task_manager if task_manager else TaskManager() # empty task manager to prevent null pointers
@@ -516,7 +516,7 @@ class GenericSoftwareManager(SoftwareManager):
return suggestions
return []
- def execute_custom_action(self, action: CustomSoftwareAction, pkg: SoftwarePackage, root_password: str, watcher: ProcessWatcher):
+ def execute_custom_action(self, action: CustomSoftwareAction, pkg: SoftwarePackage, root_password: Optional[str], watcher: ProcessWatcher):
if action.requires_internet and not self.context.is_internet_available():
raise NoInternetException()
@@ -582,7 +582,7 @@ class GenericSoftwareManager(SoftwareManager):
return by_manager
- def get_upgrade_requirements(self, pkgs: List[SoftwarePackage], root_password: str, watcher: ProcessWatcher) -> UpgradeRequirements:
+ def get_upgrade_requirements(self, pkgs: List[SoftwarePackage], root_password: Optional[str], watcher: ProcessWatcher) -> UpgradeRequirements:
by_manager = self._map_pkgs_by_manager(pkgs)
res = GenericUpgradeRequirements([], [], [], [], {})
@@ -612,7 +612,7 @@ class GenericSoftwareManager(SoftwareManager):
return res
- def reset(self, root_password: str, watcher: ProcessWatcher) -> bool:
+ def reset(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
body = f"{self.i18n['action.reset.body_1'].format(bold(self.context.app_name))}
" \
f"{self.i18n['action.reset.body_2']}
"
diff --git a/bauh/view/core/downloader.py b/bauh/view/core/downloader.py
index 5c17d92c..993b4b14 100644
--- a/bauh/view/core/downloader.py
+++ b/bauh/view/core/downloader.py
@@ -8,7 +8,7 @@ from io import StringIO
from math import floor
from pathlib import Path
from threading import Thread
-from typing import Iterable, List
+from typing import Iterable, List, Optional
from bauh.api.abstract.download import FileDownloader
from bauh.api.abstract.handler import ProcessWatcher
@@ -42,7 +42,7 @@ class AdaptableFileDownloader(FileDownloader):
def is_wget_available() -> bool:
return bool(shutil.which('wget'))
- def _get_aria2c_process(self, url: str, output_path: str, cwd: str, root_password: str, threads: int) -> SimpleProcess:
+ def _get_aria2c_process(self, url: str, output_path: str, cwd: str, root_password: Optional[str], threads: int) -> SimpleProcess:
cmd = ['aria2c', url,
'--no-conf',
'-x', '16',
@@ -71,7 +71,7 @@ class AdaptableFileDownloader(FileDownloader):
return SimpleProcess(cmd=cmd, root_password=root_password, cwd=cwd)
- def _get_axel_process(self, url: str, output_path: str, cwd: str, root_password: str, threads: int) -> SimpleProcess:
+ def _get_axel_process(self, url: str, output_path: str, cwd: str, root_password: Optional[str], threads: int) -> SimpleProcess:
cmd = ['axel', url, '-n', str(threads), '-4', '-c', '-T', '5']
if output_path:
@@ -79,7 +79,7 @@ class AdaptableFileDownloader(FileDownloader):
return SimpleProcess(cmd=cmd, cwd=cwd, root_password=root_password)
- def _get_wget_process(self, url: str, output_path: str, cwd: str, root_password: str) -> SimpleProcess:
+ def _get_wget_process(self, url: str, output_path: str, cwd: str, root_password: Optional[str]) -> SimpleProcess:
cmd = ['wget', url, '-c', '--retry-connrefused', '-t', '10', '--no-config', '-nc']
if output_path:
@@ -88,7 +88,7 @@ class AdaptableFileDownloader(FileDownloader):
return SimpleProcess(cmd=cmd, cwd=cwd, root_password=root_password)
- def _rm_bad_file(self, file_name: str, output_path: str, cwd, handler: ProcessHandler, root_password: str):
+ def _rm_bad_file(self, file_name: str, output_path: str, cwd, handler: ProcessHandler, root_password: Optional[str]):
to_delete = output_path if output_path else f'{cwd}/{file_name}'
if to_delete and os.path.exists(to_delete):
@@ -121,7 +121,7 @@ class AdaptableFileDownloader(FileDownloader):
return threads
- def download(self, file_url: str, watcher: ProcessWatcher, output_path: str = None, cwd: str = None, root_password: str = None, substatus_prefix: str = None, display_file_size: bool = True, max_threads: int = None, known_size: int = None) -> bool:
+ def download(self, file_url: str, watcher: ProcessWatcher, output_path: str = None, cwd: str = None, root_password: Optional[str] = None, substatus_prefix: str = None, display_file_size: bool = True, max_threads: int = None, known_size: int = None) -> bool:
self.logger.info(f'Downloading {file_url}')
handler = ProcessHandler(watcher)
file_name = file_url.split('/')[-1]
diff --git a/bauh/view/core/timeshift.py b/bauh/view/core/timeshift.py
index 2a808c5a..e2d4ad73 100644
--- a/bauh/view/core/timeshift.py
+++ b/bauh/view/core/timeshift.py
@@ -1,4 +1,5 @@
import shutil
+from typing import Optional
from bauh.commons.system import SimpleProcess
@@ -7,9 +8,9 @@ def is_available() -> bool:
return bool(shutil.which('timeshift'))
-def delete_all_snapshots(root_password: str) -> SimpleProcess:
+def delete_all_snapshots(root_password: Optional[str]) -> SimpleProcess:
return SimpleProcess(['timeshift', '--delete-all', '--scripted'], root_password=root_password)
-def create_snapshot(root_password: str, mode: str) -> SimpleProcess:
+def create_snapshot(root_password: Optional[str], mode: str) -> SimpleProcess:
return SimpleProcess(['timeshift', '--create', '--scripted', '--{}'.format(mode)], root_password=root_password)
diff --git a/bauh/view/qt/root.py b/bauh/view/qt/root.py
index fd541d32..e0ac654d 100644
--- a/bauh/view/qt/root.py
+++ b/bauh/view/qt/root.py
@@ -26,7 +26,7 @@ class ValidatePassword(QThread):
self.password = password
def run(self):
- if self.password is not None:
+ if isinstance(self.password, str):
try:
valid = validate_password(self.password)
except:
@@ -91,9 +91,9 @@ class RootDialog(QDialog):
self.validate_password.signal_valid.connect(self._handle_password_validated)
def _validate_password(self):
- password = self.input_password.text().strip()
+ password = self.input_password.text()
- if password:
+ if isinstance(password, str):
self.password = password
self.tries += 1
self.bt_ok.setEnabled(False)
@@ -135,7 +135,7 @@ class RootDialog(QDialog):
store_password = bool(current_config['store_root_password'])
- if store_password and context.root_password and validate_password(context.root_password):
+ if store_password and isinstance(context.root_password, str) and validate_password(context.root_password):
return True, context.root_password
if comp_manager:
@@ -149,7 +149,7 @@ class RootDialog(QDialog):
if comp_manager:
comp_manager.restore_state(ACTION_ASK_ROOT)
- if password is not None and store_password:
+ if isinstance(password, str) and store_password:
context.root_password = password
return (True, password) if diag.valid else (False, None)
diff --git a/bauh/view/qt/thread.py b/bauh/view/qt/thread.py
index 55983bae..72536118 100644
--- a/bauh/view/qt/thread.py
+++ b/bauh/view/qt/thread.py
@@ -320,7 +320,7 @@ class UpgradeSelected(AsyncAction):
def _ask_for_trim(self) -> bool:
return self.request_confirmation(title=self.i18n['confirmation'].capitalize(), body=self.i18n['action.trim_disk.ask'])
- def _trim_disk(self, root_password: str):
+ def _trim_disk(self, root_password: Optional[str]):
self.change_status('{}...'.format(self.i18n['action.disk_trim'].capitalize()))
self.change_substatus('')
@@ -1016,7 +1016,7 @@ class ApplyFilters(AsyncAction):
class CustomAction(AsyncAction):
- def __init__(self, manager: SoftwareManager, i18n: I18n, custom_action: CustomSoftwareAction = None, pkg: PackageView = None, root_password: str = None):
+ def __init__(self, manager: SoftwareManager, i18n: I18n, custom_action: CustomSoftwareAction = None, pkg: PackageView = None, root_password: Optional[str] = None):
super(CustomAction, self).__init__()
self.manager = manager
self.pkg = pkg