Coverage for apio/managers/packages.py: 69%
204 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 01:55 +0000
1# -*- coding: utf-8 -*-
2# -- This file is part of the Apio project
3# -- (C) 2016-2021 FPGAwars
4# -- Author Jesús Arroyo
5# -- License GPLv2
6"""Package install/uninstall functionality.
7Used by the 'apio packages' command.
8"""
10import sys
11import os
12from dataclasses import dataclass
13from typing import Dict, List
14from pathlib import Path
15import shutil
16from apio.common.apio_console import cout, cerror, cstyle
17from apio.common.apio_styles import WARNING, ERROR, SUCCESS, EMPH3
18from apio.managers.downloader import FileDownloader
19from apio.managers.unpacker import FileUnpacker
20from apio.utils import util
21from apio.utils.apio_platforms import ApioPlatform
22from apio.profile import Profile, PackageRemoteConfig
25@dataclass(frozen=True)
26class PackagesContext:
27 """Context for package managements operations.
28 This class provides the information needed for package management
29 operations. This is a subset of the information contained by ApioContext
30 and we use it, instead of passing the ApioContext, because we need to
31 perform package management operations (e.g. updating packages) before
32 the ApioContext object is fully initialized.
33 """
35 # -- Same as ApioContext.profile
36 profile: Profile
37 # -- Same as ApioContext.required_packages
38 required_packages: Dict
39 # -- Same as ApioContext.platform
40 platform: ApioPlatform
41 # -- Same as ApioContext.packages_dir
42 packages_dir: Path
44 def __post_init__(self):
45 """Assert that all fields initialized to actual values."""
46 assert self.profile
47 assert self.required_packages
48 assert self.platform
49 assert self.packages_dir
52def _construct_package_download_url(
53 packages_ctx: PackagesContext,
54 package_remote_config: PackageRemoteConfig,
55) -> str:
56 """Construct the download URL for the given package name and version."""
58 # -- Create vars mapping.
59 url_vars = {
60 "${PLATFORM}": packages_ctx.platform.id,
61 "${YYYYMMDD}": package_remote_config.release_tag.replace("-", ""),
62 }
63 if util.is_debug(1): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 cout(f"Package URL vars: {url_vars}")
66 # -- Define the url parts.
67 url_parts = [
68 "https://github.com/",
69 package_remote_config.repo_organization,
70 "/",
71 package_remote_config.repo_name,
72 "/releases/download/",
73 package_remote_config.release_tag,
74 "/",
75 package_remote_config.release_file,
76 ]
78 if util.is_debug(1): 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 cout(f"package url parts = {url_parts}")
81 # -- Concatenate the URL parts.
82 url = "".join(url_parts)
84 if util.is_debug(1): 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 cout(f"Combined package url: {url}")
87 # -- Replace placeholders with values.
88 for name, val in url_vars.items():
89 url = url.replace(name, val)
91 if util.is_debug(1): 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 cout(f"Resolved package url: {url}")
94 # -- All done.
95 return url
98def _download_package_file(url: str, dir_path: Path) -> Path:
99 """Download the given file (url). Return the path of local destination
100 file. Exits with a user message and error code if any error.
102 * INPUTS:
103 * url: File to download
104 * OUTPUTS:
105 * The path of the destination file
106 """
108 filepath: Path | None = None
110 try:
111 # -- Object for downloading the file
112 downloader = FileDownloader(url, dir_path)
114 # -- Get the destination path
115 filepath = downloader.destination
117 downloader.start()
119 # -- If the user press Ctrl-C (Abort)
120 except KeyboardInterrupt:
122 # -- Remove the file
123 if filepath and filepath.is_file():
124 filepath.unlink()
126 # -- Inform the user
127 cout("User aborted download", style=ERROR)
128 sys.exit(1)
130 except IOError as exc:
131 cout("I/O error while downloading", style=ERROR)
132 cout(str(exc), style=ERROR)
133 sys.exit(1)
135 except util.ApioException:
136 cerror("Package not found")
137 sys.exit(1)
139 # -- Return the destination path
140 return filepath
143def _unpack_package_file(package_file: Path, package_dir: Path) -> None:
144 """Unpack the package_file in the package_dir directory.
145 Exit with an error message and error status if any error."""
147 # -- Create the unpacker.
148 operation = FileUnpacker(package_file, package_dir)
150 # -- Perform the operation.
151 ok = operation.start()
153 # -- Exit if error.
154 if not ok: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 cerror(f"Failed to unpack package file {package_file}")
156 sys.exit(1)
159def _delete_package_dir(
160 packages_ctx: PackagesContext, package_name: str, verbose: bool
161) -> bool:
162 """Delete the directory of the package with given name. Returns
163 True if the packages existed. Exits with an error message on error."""
164 package_dir = packages_ctx.packages_dir / package_name
166 dir_found = package_dir.is_dir()
167 if dir_found:
168 if verbose: 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true
169 cout(f"Deleting {str(package_dir)}")
171 # -- Sanity check the path and delete.
172 assert "packages" in str(package_dir).lower(), package_dir
173 shutil.rmtree(package_dir)
175 if package_dir.exists(): 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true
176 cerror(f"Directory deletion failed: {str(package_dir.absolute())}")
177 sys.exit(1)
179 return dir_found
182def scan_and_fix_packages(packages_ctx: PackagesContext) -> bool:
183 """Scan the packages and fix if there are errors. Returns true
184 if the packages are installed ok."""
186 # -- Scan the packages.
187 scan = scan_packages(packages_ctx)
189 # -- If there are fixable errors, fix them.
190 if scan.num_errors_to_fix() > 0:
191 _fix_packages(packages_ctx, scan)
193 # -- Return a flag that indicates if all packages are installed ok. We
194 # -- use a scan from before the fixing but the fixing does not touch
195 # -- installed ok packages.
196 return scan.packages_installed_ok()
199def install_missing_packages_on_the_fly(
200 packages_ctx: PackagesContext, verbose=False
201) -> None:
202 """Install on the fly any missing packages. Does not print a thing if
203 all packages are already ok. This function is intended for on demand
204 package fetching by commands such as apio build, and thus is allowed
205 to use fetched remote config instead of fetching a fresh one."""
207 # -- Scan and fix broken package.
208 # -- Since this is a on-the-fly operation, we don't require a fresh
209 # -- remote config file for required packages versions.
210 installed_ok = scan_and_fix_packages(packages_ctx)
212 # -- If all the packages are installed, we are done.
213 if installed_ok: 213 ↛ 221line 213 didn't jump to line 221 because the condition on line 213 was always true
214 return
216 # -- Here when we need to install some packages. Since we just fixed
217 # -- we can't have broken or packages with version mismatch, just
218 # -- installed ok, and not installed.
219 # --
220 # -- Get lists of installed and required packages.
221 installed_packages = packages_ctx.profile.installed_packages
222 required_packages_names = packages_ctx.required_packages.keys()
224 # -- Install any required package that is not installed.
225 for package_name in required_packages_names:
226 if package_name not in installed_packages:
227 install_package(
228 packages_ctx,
229 package_name=package_name,
230 force_reinstall=False,
231 verbose=verbose,
232 )
234 # -- Here all packages should be ok but we check again just in case.
235 scan_results = scan_packages(packages_ctx)
236 if not scan_results.is_all_ok():
237 cout(
238 "Warning: packages issues detected. Use "
239 "'apio packages list' to investigate.",
240 style=WARNING,
241 )
244def install_package(
245 packages_ctx: PackagesContext,
246 *,
247 package_name: str,
248 force_reinstall: bool,
249 verbose: bool,
250) -> None:
251 """Install a given package.
253 'packages_ctx' is the context object of this apio invocation.
254 'package_name' is the package name, e.g. 'examples' or 'oss-cad-suite'.
255 'force' indicates if to perform the installation even if a matching
256 package is already installed.
257 'explicit' indicates that the user specified the package name(s) explicitly
258 and thus expect more feedback in case of a 'no change'
259 'verbose' indicates if to print extra information.
261 Returns normally if no error, exits the program with an error status
262 and a user message if an error is detected.
263 """
265 # -- Caller is responsible to check check that package name is valid
266 # -- on this platform.
267 assert package_name in packages_ctx.required_packages, package_name
269 # -- Set up installation announcement
270 pending_announcement = cstyle(
271 f"Installing apio package '{package_name}'", style=EMPH3
272 )
274 # -- If in chatty mode, announce now and clear. Otherwise we will
275 # -- announce later only if actually installing.
276 if verbose: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 cout(pending_announcement)
278 pending_announcement = None
280 # -- Get package remote config from the cache. Caller can refresh the
281 # -- cache with the latest remote config if desired.
282 package_config: PackageRemoteConfig = (
283 packages_ctx.profile.get_package_config(package_name)
284 )
286 # -- Get the version we should have.
287 target_version = package_config.release_version
289 # -- If not forcing and the target version already installed then
290 # -- nothing to do and we leave quietly.
291 if not force_reinstall:
292 # -- Get the version of the installed package, None if not installed.
293 installed_version, package_platform_id = (
294 packages_ctx.profile.get_installed_package_info(package_name)
295 )
297 if verbose: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 cout(
299 f"Installed version: {installed_version} "
300 f"({package_platform_id})"
301 )
303 # -- If the installed and the target versions are the same then
304 # -- nothing to do.
305 if (
306 target_version == installed_version
307 and package_platform_id == packages_ctx.platform.id
308 ):
309 if verbose: 309 ↛ 310line 309 didn't jump to line 310 because the condition on line 309 was never true
310 cout(
311 f"Version {target_version} ({package_platform_id}) "
312 "already installed",
313 style=SUCCESS,
314 )
315 return
317 # -- Here we need to fetch and install so can be more chatty.
319 # -- Here we actually do the work. Announce if we haven't done it yet.
320 if pending_announcement: 320 ↛ 324line 320 didn't jump to line 324 because the condition on line 320 was always true
321 cout(pending_announcement)
322 pending_announcement = True
324 cout(f"Fetching version {target_version} ({packages_ctx.platform.id})")
326 # -- Construct the download URL.
327 download_url = _construct_package_download_url(
328 packages_ctx, package_config
329 )
330 if verbose: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 cout(f"Download URL: {download_url}")
333 # -- Prepare the packages directory.
334 packages_ctx.packages_dir.mkdir(exist_ok=True)
336 # -- Prepare the package directory.
337 # package_dir = packages_ctx.get_package_dir(package_name)
338 package_dir = packages_ctx.packages_dir / package_name
339 cout(f"Package dir: {package_dir}")
341 # -- Download the package file from the remote server.
342 local_package_file = _download_package_file(
343 download_url, packages_ctx.packages_dir
344 )
345 if verbose: 345 ↛ 346line 345 didn't jump to line 346 because the condition on line 345 was never true
346 cout(f"Local package file: {local_package_file}")
348 # -- Delete the old package dir, if exists, to avoid name conflicts and
349 # -- left over files.
350 _delete_package_dir(packages_ctx, package_name, verbose)
352 # -- Unpack the package. This creates a new package dir.
353 _unpack_package_file(local_package_file, package_dir)
355 # -- Remove the package file. We don't need it anymore.
356 if verbose: 356 ↛ 357line 356 didn't jump to line 357 because the condition on line 356 was never true
357 cout(f"Deleting package file {local_package_file}")
358 local_package_file.unlink()
360 # -- Add package to profile and save.
361 packages_ctx.profile.add_package(
362 package_name, target_version, packages_ctx.platform.id, download_url
363 )
364 # packages_ctx.profile.save()
366 # -- Inform the user!
367 cout(f"Package '{package_name}' installed successfully", style=SUCCESS)
370def _fix_packages(
371 packages_ctx: PackagesContext, scan: "PackageScanResults"
372) -> None:
373 """If the package scan result contains errors, fix them."""
375 for package_name in scan.bad_version_package_names: 375 ↛ 376line 375 didn't jump to line 376 because the loop on line 375 never started
376 cout(f"Uninstalling incompatible version of '{package_name}'")
377 _delete_package_dir(packages_ctx, package_name, verbose=False)
378 packages_ctx.profile.remove_package(package_name)
380 for package_name in scan.broken_package_names:
381 cout(f"Uninstalling broken package '{package_name}'")
382 _delete_package_dir(packages_ctx, package_name, verbose=False)
383 packages_ctx.profile.remove_package(package_name)
385 for package_name in scan.orphan_package_names: 385 ↛ 386line 385 didn't jump to line 386 because the loop on line 385 never started
386 cout(f"Uninstalling unknown package '{package_name}'")
387 packages_ctx.profile.remove_package(package_name)
389 for dir_name in scan.orphan_dir_names:
390 cout(f"Deleting unknown package dir '{dir_name}'")
391 # -- Sanity check. Since packages_ctx.packages_dir is guaranteed to
392 # -- include the word packages, this can fail only due to programming
393 # -- error.
394 dir_path = packages_ctx.packages_dir / dir_name
395 assert "packages" in str(dir_path).lower(), dir_path
396 # -- Delete.
397 shutil.rmtree(dir_path)
399 for file_name in scan.orphan_file_names: 399 ↛ 400line 399 didn't jump to line 400 because the loop on line 399 never started
400 cout(f"Deleting unknown package file '{file_name}'")
401 # -- Sanity check. Since packages_ctx.packages_dir is guaranteed to
402 # -- include the word packages, this can fail only due to programming
403 # -- error.
404 file_path = packages_ctx.packages_dir / file_name
405 assert "packages" in str(file_path).lower(), dir_path
406 # -- Delete.
407 file_path.unlink()
410@dataclass
411class PackageScanResults:
412 """Represents results of packages scan."""
414 # -- Normal and Error. Packages in required_packages that are installed
415 # -- regardless if the version matches or not.
416 installed_ok_package_names: List[str]
417 # -- Error. Packages in required_packages that are installed but with
418 # -- version mismatch.
419 bad_version_package_names: List[str]
420 # -- Normal. Packages in required_packages that are uninstalled properly.
421 uninstalled_package_names: List[str]
422 # -- Error. Packages in required_packages with broken installation. E.g,
423 # -- registered in profile but package directory is missing.
424 broken_package_names: List[str]
425 # -- Error. Packages that are marked in profile as registered but are not
426 # -- in required_packages.
427 orphan_package_names: List[str]
428 # -- Error. Basenames of directories in packages dir that don't match
429 # -- folder_name of packages in required_packages.
430 orphan_dir_names: List[str]
431 # -- Error. Basenames of all files in packages directory. That directory is
432 # -- expected to contain only directories for packages.a
433 orphan_file_names: List[str]
435 def packages_installed_ok(self) -> bool:
436 """Returns true if all packages are installed ok, regardless of
437 other fixable errors."""
438 return (
439 len(self.bad_version_package_names) == 0
440 and len(self.uninstalled_package_names) == 0
441 and len(self.broken_package_names) == 0
442 )
444 def num_errors_to_fix(self) -> int:
445 """Returns the number of errors that required , having a non installed
446 packages is not considered an error that need to be fix."""
447 return (
448 len(self.bad_version_package_names)
449 + len(self.broken_package_names)
450 + len(self.orphan_package_names)
451 + len(self.orphan_dir_names)
452 + len(self.orphan_file_names)
453 )
455 def is_all_ok(self) -> bool:
456 """Return True if all packages are installed properly with no
457 issues."""
458 return (
459 not self.num_errors_to_fix() and not self.uninstalled_package_names
460 )
462 def dump(self):
463 """Dump the content of this object. For debugging."""
464 cout()
465 cout("Package scan results:")
466 cout(f" Installed {self.installed_ok_package_names}")
467 cout(f" bad version {self.bad_version_package_names}")
468 cout(f" Uninstalled {self.uninstalled_package_names}")
469 cout(f" Broken {self.broken_package_names}")
470 cout(f" Orphan ids {self.orphan_package_names}")
471 cout(f" Orphan dirs {self.orphan_dir_names}")
472 cout(f" Orphan files {self.orphan_file_names}")
475def package_version_ok(
476 packages_ctx: PackagesContext,
477 package_name: str,
478) -> bool:
479 """Return true if the package is both in profile and required packages
480 and its version in the profile meet the requirements in the
481 config.jsonc file. Otherwise return false."""
483 # If this package is not applicable to this platform, return False.
484 if package_name not in packages_ctx.required_packages: 484 ↛ 485line 484 didn't jump to line 485 because the condition on line 484 was never true
485 return False
487 # -- If the current version is not available, the package is not installed.
488 current_ver, package_platform_id = (
489 packages_ctx.profile.get_installed_package_info(package_name)
490 )
491 if not current_ver or package_platform_id != packages_ctx.platform.id:
492 return False
494 # -- Get the package remote config.
495 package_config: PackageRemoteConfig = (
496 packages_ctx.profile.get_package_config(package_name)
497 )
499 # -- Compare to the required version. We expect the two version to be
500 # -- normalized and ths a string comparison is sufficient.
501 return current_ver == package_config.release_version
504def scan_packages(packages_ctx: PackagesContext) -> PackageScanResults:
505 """Scans the available and installed packages and returns
506 the findings as a PackageScanResults object."""
508 # pylint: disable=too-many-branches
510 assert isinstance(packages_ctx, PackagesContext)
512 # Initialize the result with empty data.
513 result = PackageScanResults([], [], [], [], [], [], [])
515 # -- A helper set that we populate with the 'folder_name' values of the
516 # -- all the packages for this platform.
517 platform_folder_names = set()
519 # -- Scan packages ids in required_packages and populate
520 # -- the installed/uninstall/broken packages lists.
521 for package_name in packages_ctx.required_packages.keys():
522 # -- Collect package's folder names in a set. For a later use.
523 platform_folder_names.add(package_name)
525 # -- Classify the package as one of four cases.
526 in_profile = package_name in packages_ctx.profile.installed_packages
527 # has_dir = packages_ctx.get_package_dir(package_name).is_dir()
528 package_dir = packages_ctx.packages_dir / package_name
529 has_dir = package_dir.is_dir()
530 version_ok = package_version_ok(packages_ctx, package_name)
531 if in_profile and has_dir:
532 if version_ok: 532 ↛ 537line 532 didn't jump to line 537 because the condition on line 532 was always true
533 # Case 1: Package installed ok.
534 result.installed_ok_package_names.append(package_name)
535 else:
536 # -- Case 2: Package installed but version mismatch.
537 result.bad_version_package_names.append(package_name)
538 elif not in_profile and not has_dir:
539 # -- Case 3: Package not installed.
540 result.uninstalled_package_names.append(package_name)
541 else:
542 # -- Case 4: Package is broken.
543 result.broken_package_names.append(package_name)
545 # -- Scan the packages ids that are registered in profile as installed
546 # -- the ones that are not required_packages as orphans.
547 for package_name in packages_ctx.profile.installed_packages:
548 if package_name not in packages_ctx.required_packages: 548 ↛ 549line 548 didn't jump to line 549 because the condition on line 548 was never true
549 result.orphan_package_names.append(package_name)
551 # -- Scan the packages directory and identify orphan dirs and files.
552 for path in packages_ctx.packages_dir.glob("*"):
553 base_name = os.path.basename(path)
554 if path.is_dir():
555 if base_name not in platform_folder_names:
556 result.orphan_dir_names.append(base_name)
557 else:
558 # -- Skip the packages installed file, so we don't consider it as
559 # -- an orphan file.
560 # TODO Make this a const.
561 if base_name == "installed_packages.json": 561 ↛ 563line 561 didn't jump to line 563 because the condition on line 561 was always true
562 continue
563 result.orphan_file_names.append(base_name)
565 # -- Return results
566 if util.is_debug(1): 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true
567 result.dump()
569 return result