Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
PackageDB.get_local_dict()[self.repo2_pkg_name].version,
self.repo2_old_version
)
self.assertEqual(
PackageDB.get_local_dict()[self.aur2_pkg_name].version,
self.aur2_old_version
)
# ignore one of repo pkgs and one of AUR pkgs
pikaur(
'-Su --noconfirm '
f'--ignore {self.repo_pkg_name} '
f'--ignore {self.aur_pkg_name}'
)
self.assertEqual(
PackageDB.get_local_dict()[self.repo_pkg_name].version,
self.repo_old_version
)
self.assertEqual(
PackageDB.get_local_dict()[self.aur_pkg_name].version,
self.aur_old_version
)
self.assertNotEqual(
PackageDB.get_local_dict()[self.repo2_pkg_name].version,
self.repo2_old_version
)
self.assertNotEqual(
PackageDB.get_local_dict()[self.aur2_pkg_name].version,
self.aur2_old_version
)
# ignore the only remaining AUR package
self.assertEqual(
PackageDB.get_local_dict()[self.aur_pkg_name].version,
self.aur_old_version
)
self.downgrade_repo1_pkg()
# ignore the only one remaining repo package
pikaur('-Su --noconfirm '
f'--ignore {self.repo_pkg_name}')
self.assertEqual(
PackageDB.get_local_dict()[self.repo_pkg_name].version,
self.repo_old_version
)
self.assertNotEqual(
PackageDB.get_local_dict()[self.aur_pkg_name].version,
self.aur_old_version
)
self.repo_old_version
)
self.assertEqual(
PackageDB.get_local_dict()[self.aur_pkg_name].version,
self.aur_old_version
)
self.downgrade_repo1_pkg()
pikaur('-Su --noconfirm --aur')
self.assertEqual(
PackageDB.get_local_dict()[self.repo_pkg_name].version,
self.repo_old_version
)
self.assertNotEqual(
PackageDB.get_local_dict()[self.aur_pkg_name].version,
self.aur_old_version
)
def get_all_local_pkgs_conflicts() -> Dict[str, List[str]]:
all_local_pkgs_info = PackageDB.get_local_dict()
all_local_pgks_conflicts_lists = {}
for local_pkg_info in all_local_pkgs_info.values():
conflicts: List[str] = []
if local_pkg_info.conflicts:
conflicts += local_pkg_info.conflicts
if local_pkg_info.replaces:
conflicts += local_pkg_info.replaces
if conflicts:
all_local_pgks_conflicts_lists[local_pkg_info.name] = list(set(conflicts))
return all_local_pgks_conflicts_lists
def find_repo_upgradeable() -> List[InstallInfo]:
all_local_pkgs = PackageDB.get_local_dict()
repo_packages_updates = []
for repo_pkg in find_upgradeable_packages():
local_pkg = all_local_pkgs[repo_pkg.name]
repo_packages_updates.append(
InstallInfo(
name=local_pkg.name,
new_version=repo_pkg.version,
current_version=local_pkg.version,
description=repo_pkg.desc,
repository=repo_pkg.db.name,
package=repo_pkg,
)
)
return repo_packages_updates
def install_built_deps(
deps_names_and_paths: Dict[str, str],
resolved_conflicts: Optional[List[List[str]]] = None
) -> None:
if not deps_names_and_paths:
return
local_packages = PackageDB.get_local_dict()
args = parse_args()
def _get_pacman_command() -> List[str]:
return get_pacman_command() + reconstruct_args(args, ignore_args=[
'upgrade',
'asdeps',
'sync',
'sysupgrade',
'refresh',
'ignore',
'downloadonly',
])
explicitly_installed_deps = []
for pkg_name, _path in deps_names_and_paths.items():
print(color_line(pkg_name, 14))
def find_aur_deps(package_names):
# @TODO: refactor: split to smaller routines
def _get_deps_and_version_matchers(result):
deps = {}
for dep in (result.Depends or []) + (result.MakeDepends or []):
name, version_matcher = get_package_name_and_version_matcher_from_depend_line(dep)
deps[name] = version_matcher
return deps
all_repo_pkgs_info = PackageDB.get_repo_dict()
all_local_pkgs_info = PackageDB.get_local_dict()
new_aur_deps = []
while package_names:
all_deps_for_aur_packages = {}
aur_pkgs_info, not_found_aur_pkgs = find_aur_packages(package_names)
if not_found_aur_pkgs:
raise PackagesNotFoundInAUR(packages=not_found_aur_pkgs)
for result in aur_pkgs_info:
aur_pkg_deps = _get_deps_and_version_matchers(result)
if aur_pkg_deps:
all_deps_for_aur_packages[result.Name] = aur_pkg_deps
not_found_local_pkgs = []
for aur_pkg_name, deps_for_aur_package in all_deps_for_aur_packages.items():
# @TODO: refactor: dedup this and next blocks
def package_search_thread_local() -> Dict[str, str]:
result = {
pkg_name: pkg.version
for pkg_name, pkg in PackageDB.get_local_dict(quiet=True).items()
}
if not parse_args().quiet:
sys.stderr.write('#')
return result
# latest sources for quite a long time)
with ThreadPool(processes=num_threads) as pool:
threads = []
for repo_status in all_package_builds:
threads.append(
pool.apply_async(getattr, (repo_status, 'version_already_installed'))
)
for thread in threads:
thread.get()
pool.close()
pool.join()
# handle if version is already installed
if not self.args.needed:
return
local_db = PackageDB.get_local_dict()
for repo_status in all_package_builds:
if not repo_status.reviewed:
continue
# pragma: no cover
repo_status.update_last_installed_file()
for package_name in repo_status.package_names:
if package_name not in local_db:
continue
if repo_status.version_already_installed:
print_package_uptodate(package_name, PackageSource.AUR)
self.discard_install_info(package_name)
elif (
self.args.sysupgrade > 1
) or (
is_devel_pkg(repo_status.package_base) and (self.args.devel > 1)
):