spalencsar/bearhub
1
0
Fork 0
mirror of https://github.com/spalencsar/bearhub.git synced 2026-07-11 22:34:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix: not accepting blank root passwords

Browse Source
This commit is contained in:
Vinicius Moreira
2022-02-28 11:25:41 -03:00
parent 8a0609de06
commit ff9d3be65b
22 changed files with 126 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -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

12
bauh/api/abstract/controller.py
View File

@@ -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:

4
bauh/api/abstract/download.py
View File

@@ -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:

2
bauh/api/abstract/model.py
View File

@@ -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:

8
bauh/commons/system.py
View File

@@ -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

18
bauh/gems/appimage/controller.py
View File

@@ -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())

48
bauh/gems/arch/controller.py
View File

@@ -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 = '<p>{}</p><p>{}</p><br/>'.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

8
bauh/gems/arch/cpu_manager.py
View File

@@ -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:

8
bauh/gems/arch/download.py
View File

@@ -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()

22
bauh/gems/arch/pacman.py
View File

@@ -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:

4
bauh/gems/arch/updates.py
View File

@@ -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)

4
bauh/gems/arch/worker.py
View File

@@ -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

12
bauh/gems/flatpak/controller.py
View File

@@ -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 = [], []

4
bauh/gems/flatpak/flatpak.py
View File

@@ -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)

14
bauh/gems/snap/controller.py
View File

@@ -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()

8
bauh/gems/snap/snap.py
View File

@@ -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:

16
bauh/gems/web/controller.py
View File

@@ -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:

18
bauh/view/core/controller.py
View File

@@ -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"<p>{self.i18n['action.reset.body_1'].format(bold(self.context.app_name))}</p>" \
f"<p>{self.i18n['action.reset.body_2']}</p>"

12
bauh/view/core/downloader.py
View File

@@ -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]

5
bauh/view/core/timeshift.py
View File

@@ -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)

10
bauh/view/qt/root.py
View File

@@ -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)

4
bauh/view/qt/thread.py
View File

@@ -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