diff --git a/src/pkgmgr/actions/archive/workflow.py b/src/pkgmgr/actions/archive/workflow.py index 8616523..44b8a27 100644 --- a/src/pkgmgr/actions/archive/workflow.py +++ b/src/pkgmgr/actions/archive/workflow.py @@ -102,10 +102,8 @@ def run_archive( if not dry_run: for path, _title in archived: - try: + with contextlib.suppress(FileNotFoundError): path.unlink() - except FileNotFoundError: - pass return ArchivePlan( archived=archived, diff --git a/src/pkgmgr/actions/install/__init__.py b/src/pkgmgr/actions/install/__init__.py index e5e9194..b7b8373 100644 --- a/src/pkgmgr/actions/install/__init__.py +++ b/src/pkgmgr/actions/install/__init__.py @@ -230,7 +230,7 @@ def install_repos( f"[Warning] install: repository {identifier} failed (exit={code}). Continuing..." ) continue - except Exception as exc: + except Exception as exc: # noqa: BLE001 - batch boundary: one repository must never abort the run failures.append((identifier, f"unexpected error: {exc}")) if not quiet: print( diff --git a/src/pkgmgr/actions/install/installers/nix/profile/normalizer.py b/src/pkgmgr/actions/install/installers/nix/profile/normalizer.py index 19ac873..d1e588a 100644 --- a/src/pkgmgr/actions/install/installers/nix/profile/normalizer.py +++ b/src/pkgmgr/actions/install/installers/nix/profile/normalizer.py @@ -21,7 +21,7 @@ def coerce_index(key: str, entry: dict[str, Any]) -> int | None: if k.isdigit(): try: return int(k) - except Exception: + except ValueError: return None # 2) Explicit index fields (schema-dependent) @@ -32,7 +32,7 @@ def coerce_index(key: str, entry: dict[str, Any]) -> int | None: if isinstance(v, str) and v.strip().isdigit(): try: return int(v.strip()) - except Exception: + except ValueError: pass # 3) Last resort: extract trailing number from key if it looks like "-" @@ -40,13 +40,13 @@ def coerce_index(key: str, entry: dict[str, Any]) -> int | None: if m: try: return int(m.group(1)) - except Exception: + except ValueError: return None return None -def iter_store_paths(entry: Dict[str, Any]) -> Iterable[str]: +def iter_store_paths(entry: dict[str, Any]) -> Iterable[str]: """ Yield all possible store paths from a nix profile JSON entry. diff --git a/src/pkgmgr/actions/install/installers/nix/profile_list.py b/src/pkgmgr/actions/install/installers/nix/profile_list.py index 4aea026..d6310bb 100644 --- a/src/pkgmgr/actions/install/installers/nix/profile_list.py +++ b/src/pkgmgr/actions/install/installers/nix/profile_list.py @@ -35,12 +35,12 @@ class NixProfileListReader: sp = m.group(2) try: idx = int(idx_s) - except Exception: + except ValueError: continue entries.append((idx, self._store_prefix(sp))) seen: set[int] = set() - uniq: List[Tuple[int, str]] = [] + uniq: list[tuple[int, str]] = [] for idx, sp in entries: if idx not in seen: seen.add(idx) diff --git a/src/pkgmgr/actions/install/installers/os_packages/arch_pkgbuild.py b/src/pkgmgr/actions/install/installers/os_packages/arch_pkgbuild.py index b7d72f9..39826db 100644 --- a/src/pkgmgr/actions/install/installers/os_packages/arch_pkgbuild.py +++ b/src/pkgmgr/actions/install/installers/os_packages/arch_pkgbuild.py @@ -36,7 +36,7 @@ class ArchPkgbuildInstaller(BaseInstaller): try: if hasattr(os, "geteuid") and os.geteuid() == 0: return False - except Exception: + except (AttributeError, OSError): # On non-POSIX platforms just ignore this check. pass diff --git a/src/pkgmgr/actions/mirror/merge_cmd.py b/src/pkgmgr/actions/mirror/merge_cmd.py index ac17548..a77a747 100644 --- a/src/pkgmgr/actions/mirror/merge_cmd.py +++ b/src/pkgmgr/actions/mirror/merge_cmd.py @@ -35,10 +35,10 @@ def _load_user_config(path: str) -> dict[str, object]: return {} try: - with open(path, "r", encoding="utf-8") as f: + with open(path, encoding="utf-8") as f: data = yaml.safe_load(f) or {} return data if isinstance(data, dict) else {} - except Exception: + except (OSError, UnicodeDecodeError, yaml.YAMLError): return {} diff --git a/src/pkgmgr/actions/release/files/changelog_lint.py b/src/pkgmgr/actions/release/files/changelog_lint.py index eea7ba1..5a78b9e 100644 --- a/src/pkgmgr/actions/release/files/changelog_lint.py +++ b/src/pkgmgr/actions/release/files/changelog_lint.py @@ -83,10 +83,8 @@ def _markdownlint(reference_path: str, document: str) -> list[str]: ] return findings or [output.strip() or "markdown-lint reported an error"] finally: - try: + with contextlib.suppress(OSError): os.remove(tmp_path) - except OSError: - pass def _builtin_lint(document: str) -> list[str]: diff --git a/src/pkgmgr/actions/release/files/changelog_md.py b/src/pkgmgr/actions/release/files/changelog_md.py index 9b3d0f9..324fc2f 100644 --- a/src/pkgmgr/actions/release/files/changelog_md.py +++ b/src/pkgmgr/actions/release/files/changelog_md.py @@ -104,9 +104,9 @@ def update_changelog( changelog = "" if os.path.exists(changelog_path): try: - with open(changelog_path, "r", encoding="utf-8") as f: + with open(changelog_path, encoding="utf-8") as f: changelog = f.read() - except Exception as exc: + except (OSError, UnicodeDecodeError) as exc: print(f"[WARN] Could not read existing CHANGELOG.md: {exc}") new_changelog = _insert_after_h1(changelog, entry) diff --git a/src/pkgmgr/actions/release/files/debian.py b/src/pkgmgr/actions/release/files/debian.py index 5288669..54591ab 100644 --- a/src/pkgmgr/actions/release/files/debian.py +++ b/src/pkgmgr/actions/release/files/debian.py @@ -68,9 +68,9 @@ def update_debian_changelog( return try: - with open(debian_changelog_path, "r", encoding="utf-8") as f: + with open(debian_changelog_path, encoding="utf-8") as f: existing = f.read() - except Exception as exc: + except (OSError, UnicodeDecodeError) as exc: print(f"[WARN] Could not read debian/changelog: {exc}") existing = "" diff --git a/src/pkgmgr/actions/release/files/editor.py b/src/pkgmgr/actions/release/files/editor.py index a990548..d83aa12 100644 --- a/src/pkgmgr/actions/release/files/editor.py +++ b/src/pkgmgr/actions/release/files/editor.py @@ -34,13 +34,11 @@ def _open_editor_for_changelog(initial_message: str | None = None) -> str: ) try: - with open(tmp_path, "r", encoding="utf-8") as f: + with open(tmp_path, encoding="utf-8") as f: content = f.read() finally: - try: + with contextlib.suppress(OSError): os.remove(tmp_path) - except OSError: - pass lines = [line for line in content.splitlines() if not line.strip().startswith(";")] return "\n".join(lines).strip() diff --git a/src/pkgmgr/actions/release/files/flake.py b/src/pkgmgr/actions/release/files/flake.py index a3207a0..f732307 100644 --- a/src/pkgmgr/actions/release/files/flake.py +++ b/src/pkgmgr/actions/release/files/flake.py @@ -12,9 +12,9 @@ def update_flake_version( return try: - with open(flake_path, "r", encoding="utf-8") as f: + with open(flake_path, encoding="utf-8") as f: content = f.read() - except Exception as exc: + except (OSError, UnicodeDecodeError) as exc: print(f"[WARN] Could not read flake.nix: {exc}") return diff --git a/src/pkgmgr/actions/release/files/pkgbuild.py b/src/pkgmgr/actions/release/files/pkgbuild.py index b28556e..2647302 100644 --- a/src/pkgmgr/actions/release/files/pkgbuild.py +++ b/src/pkgmgr/actions/release/files/pkgbuild.py @@ -12,9 +12,9 @@ def update_pkgbuild_version( return try: - with open(pkgbuild_path, "r", encoding="utf-8") as f: + with open(pkgbuild_path, encoding="utf-8") as f: content = f.read() - except Exception as exc: + except (OSError, UnicodeDecodeError) as exc: print(f"[WARN] Could not read PKGBUILD: {exc}") return diff --git a/src/pkgmgr/actions/release/files/rpm_changelog.py b/src/pkgmgr/actions/release/files/rpm_changelog.py index fbada7f..6aab28d 100644 --- a/src/pkgmgr/actions/release/files/rpm_changelog.py +++ b/src/pkgmgr/actions/release/files/rpm_changelog.py @@ -18,9 +18,9 @@ def update_spec_changelog( return try: - with open(spec_path, "r", encoding="utf-8") as f: + with open(spec_path, encoding="utf-8") as f: content = f.read() - except Exception as exc: + except (OSError, UnicodeDecodeError) as exc: print(f"[WARN] Could not read spec file for changelog update: {exc}") return @@ -63,7 +63,7 @@ def update_spec_changelog( try: with open(spec_path, "w", encoding="utf-8") as f: f.write(new_content) - except Exception as exc: + except (OSError, UnicodeEncodeError) as exc: print(f"[WARN] Failed to write updated spec changelog section: {exc}") return diff --git a/src/pkgmgr/actions/release/files/rpm_spec.py b/src/pkgmgr/actions/release/files/rpm_spec.py index b687cec..11e1388 100644 --- a/src/pkgmgr/actions/release/files/rpm_spec.py +++ b/src/pkgmgr/actions/release/files/rpm_spec.py @@ -15,9 +15,9 @@ def update_spec_version( return try: - with open(spec_path, "r", encoding="utf-8") as f: + with open(spec_path, encoding="utf-8") as f: content = f.read() - except Exception as exc: + except (OSError, UnicodeDecodeError) as exc: print(f"[WARN] Could not read spec file: {exc}") return diff --git a/src/pkgmgr/actions/release/workflow.py b/src/pkgmgr/actions/release/workflow.py index b4d6725..09f6262 100644 --- a/src/pkgmgr/actions/release/workflow.py +++ b/src/pkgmgr/actions/release/workflow.py @@ -186,7 +186,7 @@ def _release_impl( print(f"[INFO] Deleting branch {branch} after successful release...") try: close_branch(name=branch, base_branch="main", cwd=".") - except Exception as exc: + except (RuntimeError, GitRunError) as exc: print(f"[WARN] Failed to close branch {branch} automatically: {exc}") diff --git a/src/pkgmgr/actions/repository/create/templates.py b/src/pkgmgr/actions/repository/create/templates.py index e7e409b..a4bc1a6 100644 --- a/src/pkgmgr/actions/repository/create/templates.py +++ b/src/pkgmgr/actions/repository/create/templates.py @@ -2,13 +2,13 @@ from __future__ import annotations import os from pathlib import Path -from typing import Any, Dict +from typing import Any from pkgmgr.core.git.queries import get_repo_root try: from jinja2 import Environment, FileSystemLoader, StrictUndefined -except Exception as exc: # pragma: no cover +except ImportError as exc: # pragma: no cover Environment = None # type: ignore FileSystemLoader = None # type: ignore StrictUndefined = None # type: ignore diff --git a/src/pkgmgr/actions/repository/delete.py b/src/pkgmgr/actions/repository/delete.py index f08d01c..17acee2 100644 --- a/src/pkgmgr/actions/repository/delete.py +++ b/src/pkgmgr/actions/repository/delete.py @@ -28,7 +28,7 @@ def delete_repos(selected_repos, repositories_base_dir, all_repos, preview=False print( f"Deleted repository directory '{repo_dir}' for {repo_identifier}." ) - except Exception as e: + except OSError as e: print(f"Error deleting '{repo_dir}' for {repo_identifier}: {e}") else: print(f"Skipped deletion of '{repo_dir}' for {repo_identifier}.") diff --git a/src/pkgmgr/actions/update/manager.py b/src/pkgmgr/actions/update/manager.py index cda49dc..f32e30a 100644 --- a/src/pkgmgr/actions/update/manager.py +++ b/src/pkgmgr/actions/update/manager.py @@ -57,7 +57,7 @@ class UpdateManager: f"[Warning] update: pull failed for {identifier} (exit={code}). Continuing..." ) continue - except Exception as exc: + except Exception as exc: # noqa: BLE001 - batch boundary: one repository must never abort the run failures.append((identifier, f"pull failed: {exc}")) if not quiet: print( @@ -88,7 +88,7 @@ class UpdateManager: f"[Warning] update: install failed for {identifier} (exit={code}). Continuing..." ) continue - except Exception as exc: + except Exception as exc: # noqa: BLE001 - batch boundary: one repository must never abort the run failures.append((identifier, f"install failed: {exc}")) if not quiet: print( diff --git a/src/pkgmgr/cli/commands/changelog.py b/src/pkgmgr/cli/commands/changelog.py index acc3b84..6462dfe 100644 --- a/src/pkgmgr/cli/commands/changelog.py +++ b/src/pkgmgr/cli/commands/changelog.py @@ -2,22 +2,23 @@ from __future__ import annotations import os import sys -from typing import Any, Dict, List, Optional, Tuple +from typing import Any from pkgmgr.actions.changelog import generate_changelog from pkgmgr.cli.context import CLIContext +from pkgmgr.core.git import GitRunError from pkgmgr.core.git.queries import get_tags from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.core.repository.identifier import get_repo_identifier from pkgmgr.core.version.semver import extract_semver_from_tags -Repository = Dict[str, Any] +Repository = dict[str, Any] def _find_previous_and_current_tag( - tags: List[str], - target_tag: Optional[str] = None, -) -> Tuple[Optional[str], Optional[str]]: + tags: list[str], + target_tag: str | None = None, +) -> tuple[str | None, str | None]: """ Given a list of tags and an optional target tag, determine (previous_tag, current_tag) on the SemVer axis. @@ -93,7 +94,7 @@ def handle_changelog( if not repo_dir: try: repo_dir = get_repo_dir(ctx.repositories_base_dir, repo) - except Exception: + except (AttributeError, KeyError, TypeError): repo_dir = None identifier = get_repo_identifier(repo, ctx.all_repositories) @@ -113,12 +114,12 @@ def handle_changelog( try: tags = get_tags(cwd=repo_dir) - except Exception as exc: + except GitRunError as exc: print(f"[ERROR] Could not read git tags: {exc}") tags = [] - from_ref: Optional[str] = None - to_ref: Optional[str] = None + from_ref: str | None = None + to_ref: str | None = None if range_arg: # Explicit range provided diff --git a/src/pkgmgr/cli/commands/release.py b/src/pkgmgr/cli/commands/release.py index 4b3daab..2f72d51 100644 --- a/src/pkgmgr/cli/commands/release.py +++ b/src/pkgmgr/cli/commands/release.py @@ -36,7 +36,7 @@ def handle_release( repo_dir = repo.get("directory") or get_repo_dir( ctx.repositories_base_dir, repo ) - except Exception as exc: + except (AttributeError, KeyError, TypeError) as exc: print( f"[WARN] Skipping repository {identifier}: failed to resolve directory: {exc}" ) diff --git a/src/pkgmgr/cli/commands/repos.py b/src/pkgmgr/cli/commands/repos.py index e522d05..66eba09 100644 --- a/src/pkgmgr/cli/commands/repos.py +++ b/src/pkgmgr/cli/commands/repos.py @@ -119,7 +119,7 @@ def handle_repos_command( for repository in selected: try: repo_dir = _resolve_repository_directory(repository, ctx) - except Exception as exc: + except (AttributeError, KeyError, TypeError, ValueError) as exc: ident = ( f"{repository.get('provider', '?')}/" f"{repository.get('account', '?')}/" diff --git a/src/pkgmgr/cli/commands/version.py b/src/pkgmgr/cli/commands/version.py index f8778dc..9d9f629 100644 --- a/src/pkgmgr/cli/commands/version.py +++ b/src/pkgmgr/cli/commands/version.py @@ -2,9 +2,10 @@ from __future__ import annotations import os import sys -from typing import Any, Dict, List, Optional, Tuple +from typing import Any from pkgmgr.cli.context import CLIContext +from pkgmgr.core.git import GitRunError from pkgmgr.core.git.queries import get_tags from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.core.repository.identifier import get_repo_identifier @@ -116,7 +117,7 @@ def handle_version( if not repo_dir: try: repo_dir = get_repo_dir(ctx.repositories_base_dir, repo) - except Exception: + except (AttributeError, KeyError, TypeError): repo_dir = None if not repo_dir or not os.path.isdir(repo_dir): @@ -150,11 +151,11 @@ def handle_version( try: tags = get_tags(cwd=repo_dir) - except Exception as exc: + except GitRunError as exc: print(f"[ERROR] Could not read git tags: {exc}") tags = [] - latest_tag_info: Optional[Tuple[str, SemVer]] = ( + latest_tag_info: tuple[str, SemVer] | None = ( find_latest_version(tags) if tags else None ) diff --git a/src/pkgmgr/cli/dispatch.py b/src/pkgmgr/cli/dispatch.py index fd93f51..654e896 100644 --- a/src/pkgmgr/cli/dispatch.py +++ b/src/pkgmgr/cli/dispatch.py @@ -43,7 +43,7 @@ def _select_repo_for_current_directory(ctx: CLIContext) -> list[dict[str, Any]]: if not repo_dir: try: repo_dir = get_repo_dir(ctx.repositories_base_dir, repo) - except Exception: + except (AttributeError, KeyError, TypeError): continue repo_dir = os.path.abspath(os.path.expanduser(repo_dir)) diff --git a/src/pkgmgr/cli/proxy.py b/src/pkgmgr/cli/proxy.py index 2df33bf..72997b8 100644 --- a/src/pkgmgr/cli/proxy.py +++ b/src/pkgmgr/cli/proxy.py @@ -117,20 +117,20 @@ def _proxy_has_explicit_selection(args: argparse.Namespace) -> bool: def _select_repo_for_current_directory( ctx: CLIContext, -) -> List[Dict[str, Any]]: +) -> list[dict[str, Any]]: """ Heuristic: find the repository whose local directory matches the current working directory or is the closest parent. """ cwd = os.path.abspath(os.getcwd()) - candidates: List[tuple[str, Dict[str, Any]]] = [] + candidates: list[tuple[str, dict[str, Any]]] = [] for repo in ctx.all_repositories: repo_dir = repo.get("directory") if not repo_dir: try: repo_dir = get_repo_dir(ctx.repositories_base_dir, repo) - except Exception: + except (AttributeError, KeyError, TypeError): repo_dir = None if not repo_dir: continue diff --git a/src/pkgmgr/core/command/ink.py b/src/pkgmgr/core/command/ink.py index 3cc1b4b..7d15d14 100644 --- a/src/pkgmgr/core/command/ink.py +++ b/src/pkgmgr/core/command/ink.py @@ -69,7 +69,7 @@ def create_ink( try: if os.path.realpath(command).startswith(os.path.realpath(repo_dir)): os.chmod(command, 0o755) - except Exception as e: + except OSError as e: if not quiet: print(f"Failed to set permissions on '{command}': {e}") @@ -106,6 +106,6 @@ def create_ink( os.symlink(link_path, alias_link_path) if not quiet: print(f"Alias '{alias_name}' created → {repo_identifier}") - except Exception as e: + except OSError as e: if not quiet: print(f"Error creating alias '{alias_name}': {e}") diff --git a/src/pkgmgr/core/command/run.py b/src/pkgmgr/core/command/run.py index 9786714..7515004 100644 --- a/src/pkgmgr/core/command/run.py +++ b/src/pkgmgr/core/command/run.py @@ -61,10 +61,8 @@ def run_command( line = stream.readline() if line == "": # EOF: stop watching this stream - try: + with contextlib.suppress(Exception): sel.unregister(stream) - except Exception: - pass continue if which == "stdout": @@ -78,14 +76,10 @@ def run_command( try: sel.close() finally: - try: + with contextlib.suppress(Exception): process.stdout.close() - except Exception: - pass - try: + with contextlib.suppress(Exception): process.stderr.close() - except Exception: - pass returncode = process.wait() diff --git a/src/pkgmgr/core/config/load.py b/src/pkgmgr/core/config/load.py index 62d4c04..16f9d13 100644 --- a/src/pkgmgr/core/config/load.py +++ b/src/pkgmgr/core/config/load.py @@ -194,11 +194,11 @@ def _load_defaults_from_package_or_project() -> dict[str, Any]: """ try: import pkgmgr # type: ignore - except Exception: + except ImportError: return {"directories": {}, "repositories": []} pkg_root = Path(pkgmgr.__file__).resolve().parent - candidates: List[Path] = [] + candidates: list[Path] = [] # Always prefer package-internal config dir candidates.append(pkg_root / "config") diff --git a/src/pkgmgr/core/credentials/providers/gh.py b/src/pkgmgr/core/credentials/providers/gh.py index 7de4b27..a85475c 100644 --- a/src/pkgmgr/core/credentials/providers/gh.py +++ b/src/pkgmgr/core/credentials/providers/gh.py @@ -34,7 +34,7 @@ class GhTokenProvider: stderr=subprocess.STDOUT, text=True, ).strip() - except Exception: + except (OSError, subprocess.SubprocessError): return None if not out: diff --git a/src/pkgmgr/core/credentials/validate.py b/src/pkgmgr/core/credentials/validate.py index de76c73..1cf2914 100644 --- a/src/pkgmgr/core/credentials/validate.py +++ b/src/pkgmgr/core/credentials/validate.py @@ -1,5 +1,6 @@ from __future__ import annotations +import http.client import json import urllib.request @@ -32,7 +33,7 @@ def validate_token(provider_kind: str, host: str, token: str) -> bool: # Optional: parse to ensure body is JSON _ = json.loads(resp.read().decode("utf-8")) return True - except Exception: + except (OSError, http.client.HTTPException, json.JSONDecodeError, UnicodeDecodeError): return False # Unknown provider: don't hard-fail validation (conservative default) diff --git a/src/pkgmgr/core/remote_provisioning/http/client.py b/src/pkgmgr/core/remote_provisioning/http/client.py index 42bfe0e..4074cec 100644 --- a/src/pkgmgr/core/remote_provisioning/http/client.py +++ b/src/pkgmgr/core/remote_provisioning/http/client.py @@ -50,19 +50,19 @@ class HttpClient: ) as resp: raw = resp.read().decode("utf-8", errors="replace") - parsed: Optional[Dict[str, Any]] = None + parsed: dict[str, Any] | None = None if raw: try: loaded = json.loads(raw) parsed = loaded if isinstance(loaded, dict) else None - except Exception: + except json.JSONDecodeError: parsed = None return HttpResponse(status=int(resp.status), text=raw, json=parsed) except urllib.error.HTTPError as exc: try: body = exc.read().decode("utf-8", errors="replace") - except Exception: + except (OSError, ValueError): body = "" raise HttpError(status=int(exc.code), message=str(exc), body=body) from exc except urllib.error.URLError as exc: diff --git a/src/pkgmgr/core/remote_provisioning/registry.py b/src/pkgmgr/core/remote_provisioning/registry.py index 1ffc01b..ea301dd 100644 --- a/src/pkgmgr/core/remote_provisioning/registry.py +++ b/src/pkgmgr/core/remote_provisioning/registry.py @@ -12,18 +12,18 @@ from .providers.github import GitHubProvider class ProviderRegistry: """Resolve the correct provider implementation for a host.""" - providers: List[RemoteProvider] + providers: list[RemoteProvider] @classmethod def default(cls) -> ProviderRegistry: # Order matters: more specific providers first; fallback providers last. return cls(providers=[GitHubProvider(), GiteaProvider()]) - def resolve(self, host: str) -> Optional[RemoteProvider]: + def resolve(self, host: str) -> RemoteProvider | None: for p in self.providers: try: if p.can_handle(host): return p - except Exception: + except (AttributeError, TypeError, ValueError): continue return None diff --git a/src/pkgmgr/core/version/installed.py b/src/pkgmgr/core/version/installed.py index 6ef9b60..57ee747 100644 --- a/src/pkgmgr/core/version/installed.py +++ b/src/pkgmgr/core/version/installed.py @@ -46,7 +46,7 @@ def get_installed_python_version(*candidates: str) -> InstalledVersion | None: """ try: from importlib import metadata as importlib_metadata - except Exception: + except ImportError: return None candidates = _unique_candidates(candidates) @@ -62,13 +62,13 @@ def get_installed_python_version(*candidates: str) -> InstalledVersion | None: try: version = importlib_metadata.version(name) return InstalledVersion(name=name, version=version) - except Exception: + except importlib_metadata.PackageNotFoundError: continue # 2) Fallback: scan distributions (last resort) try: dists = importlib_metadata.distributions() - except Exception: + except (OSError, ImportError): return None norm_candidates = {_normalize(c) for c in candidates} @@ -145,7 +145,7 @@ def get_installed_nix_profile_version(*candidates: str) -> InstalledVersion | No guess = _extract_version_from_store_path(sp) if guess: return InstalledVersion(name=name, version=guess) - except Exception: + except (json.JSONDecodeError, AttributeError): pass # Fallback: text mode diff --git a/src/pkgmgr/core/version/source.py b/src/pkgmgr/core/version/source.py index e250905..e082c0a 100644 --- a/src/pkgmgr/core/version/source.py +++ b/src/pkgmgr/core/version/source.py @@ -22,7 +22,7 @@ def read_pyproject_version(repo_dir: str) -> str | None: try: import tomllib # Python 3.11+ - except Exception: + except ImportError: import tomli as tomllib # type: ignore try: @@ -31,11 +31,11 @@ def read_pyproject_version(repo_dir: str) -> str | None: project = data.get("project") or {} version = project.get("version") return str(version).strip() if version else None - except Exception: + except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError): return None -def read_pyproject_project_name(repo_dir: str) -> Optional[str]: +def read_pyproject_project_name(repo_dir: str) -> str | None: """ Read distribution name from pyproject.toml ([project].name). @@ -49,7 +49,7 @@ def read_pyproject_project_name(repo_dir: str) -> Optional[str]: try: import tomllib # Python 3.11+ - except Exception: + except ImportError: import tomli as tomllib # type: ignore try: @@ -58,11 +58,11 @@ def read_pyproject_project_name(repo_dir: str) -> Optional[str]: project = data.get("project") or {} name = project.get("name") return str(name).strip() if name else None - except Exception: + except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError): return None -def read_flake_version(repo_dir: str) -> Optional[str]: +def read_flake_version(repo_dir: str) -> str | None: """ Read the version from flake.nix in repo_dir, if present. @@ -75,9 +75,9 @@ def read_flake_version(repo_dir: str) -> Optional[str]: return None try: - with open(path, "r", encoding="utf-8") as f: + with open(path, encoding="utf-8") as f: text = f.read() - except Exception: + except (OSError, UnicodeDecodeError): return None match = re.search(r'version\s*=\s*"([^"]+)"', text) @@ -102,9 +102,9 @@ def read_pkgbuild_version(repo_dir: str) -> str | None: return None try: - with open(path, "r", encoding="utf-8") as f: + with open(path, encoding="utf-8") as f: text = f.read() - except Exception: + except (OSError, UnicodeDecodeError): return None ver_match = re.search(r"^pkgver\s*=\s*(.+)$", text, re.MULTILINE) @@ -148,13 +148,13 @@ def read_debian_changelog_version(repo_dir: str) -> str | None: if match: return match.group(1).strip() or None break - except Exception: + except (OSError, UnicodeDecodeError): return None return None -def read_spec_version(repo_dir: str) -> Optional[str]: +def read_spec_version(repo_dir: str) -> str | None: """ Read the version from an RPM spec file. @@ -174,9 +174,9 @@ def read_spec_version(repo_dir: str) -> Optional[str]: return None try: - with open(path, "r", encoding="utf-8") as f: + with open(path, encoding="utf-8") as f: text = f.read() - except Exception: + except (OSError, UnicodeDecodeError): return None ver_match = re.search(r"^Version:\s*(.+)$", text, re.MULTILINE) @@ -205,18 +205,18 @@ def read_ansible_galaxy_version(repo_dir: str) -> str | None: galaxy_yml = os.path.join(repo_dir, "galaxy.yml") if os.path.isfile(galaxy_yml): try: - with open(galaxy_yml, "r", encoding="utf-8") as f: + with open(galaxy_yml, encoding="utf-8") as f: data = yaml.safe_load(f) or {} version = data.get("version") if isinstance(version, str) and version.strip(): return version.strip() - except Exception: + except (OSError, UnicodeDecodeError, yaml.YAMLError): pass meta_yml = os.path.join(repo_dir, "meta", "main.yml") if os.path.isfile(meta_yml): try: - with open(meta_yml, "r", encoding="utf-8") as f: + with open(meta_yml, encoding="utf-8") as f: data = yaml.safe_load(f) or {} galaxy_info = data.get("galaxy_info") or {} @@ -228,7 +228,7 @@ def read_ansible_galaxy_version(repo_dir: str) -> str | None: version = data.get("version") if isinstance(version, str) and version.strip(): return version.strip() - except Exception: + except (OSError, UnicodeDecodeError, yaml.YAMLError): return None return None diff --git a/tests/e2e/test_changelog_commands.py b/tests/e2e/test_changelog_commands.py index eac0462..9c467cf 100644 --- a/tests/e2e/test_changelog_commands.py +++ b/tests/e2e/test_changelog_commands.py @@ -21,7 +21,7 @@ class TestIntegrationChangelogCommands(unittest.TestCase): """ try: repo_dir = _load_pkgmgr_repo_dir() - except Exception: + except (OSError, RuntimeError, ValueError): repo_dir = None if repo_dir is not None and not os.path.isdir(repo_dir): diff --git a/tests/integration/test_error_path_degradation.py b/tests/integration/test_error_path_degradation.py new file mode 100644 index 0000000..86f5613 --- /dev/null +++ b/tests/integration/test_error_path_degradation.py @@ -0,0 +1,766 @@ +"""Error-path tests: pkgmgr must degrade, not crash, on broken input. + +pkgmgr probes external toolchains and parses packaging metadata it does +not control. Every test here injects one concrete failure -- undecodable +bytes, malformed YAML, a missing binary, an unreachable host -- and +asserts the documented fallback: return None, warn and continue, or skip +the entry. + +Faults are injected for real wherever possible rather than mocked, so the +exception types listed in the handlers are the ones actually raised. +""" + +from __future__ import annotations + +import importlib.util +import io +import os +import subprocess +import tempfile +import unittest +import urllib.error +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +BAD_UTF8 = b"\xff\xfe\x00\x80version = 1" + +# tomli is only a dependency below Python 3.11, so the fallback branch is +# only exercisable where it is actually installed. +HAS_TOMLI = importlib.util.find_spec("tomli") is not None + + +def _write(path: str, data: bytes | str) -> str: + mode = "wb" if isinstance(data, bytes) else "w" + with open(path, mode) as handle: + handle.write(data) + return path + + +class TestVersionSourceDegradation(unittest.TestCase): + """src/pkgmgr/core/version/source.py""" + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.dir = self.tmp.name + + def test_broken_toml_returns_none(self) -> None: + from pkgmgr.core.version.source import ( + read_pyproject_project_name, + read_pyproject_version, + ) + + _write(os.path.join(self.dir, "pyproject.toml"), "this is [[[ not toml") + self.assertIsNone(read_pyproject_version(self.dir)) + self.assertIsNone(read_pyproject_project_name(self.dir)) + + def test_undecodable_toml_returns_none(self) -> None: + from pkgmgr.core.version.source import read_pyproject_version + + _write(os.path.join(self.dir, "pyproject.toml"), BAD_UTF8) + self.assertIsNone(read_pyproject_version(self.dir)) + + def test_unreadable_pyproject_returns_none(self) -> None: + from pkgmgr.core.version.source import read_pyproject_version + + _write(os.path.join(self.dir, "pyproject.toml"), '[project]\nversion = "1.0.0"\n') + with patch("builtins.open", side_effect=OSError("EIO")): + self.assertIsNone(read_pyproject_version(self.dir)) + + def test_undecodable_flake_returns_none(self) -> None: + from pkgmgr.core.version.source import read_flake_version + + _write(os.path.join(self.dir, "flake.nix"), BAD_UTF8) + self.assertIsNone(read_flake_version(self.dir)) + + def test_undecodable_pkgbuild_returns_none(self) -> None: + from pkgmgr.core.version.source import read_pkgbuild_version + + _write(os.path.join(self.dir, "PKGBUILD"), BAD_UTF8) + self.assertIsNone(read_pkgbuild_version(self.dir)) + + def test_undecodable_debian_changelog_returns_none(self) -> None: + from pkgmgr.core.version.source import read_debian_changelog_version + + os.makedirs(os.path.join(self.dir, "debian"), exist_ok=True) + _write(os.path.join(self.dir, "debian", "changelog"), BAD_UTF8) + self.assertIsNone(read_debian_changelog_version(self.dir)) + + def test_undecodable_spec_returns_none(self) -> None: + from pkgmgr.core.version.source import read_spec_version + + _write(os.path.join(self.dir, "pkg.spec"), BAD_UTF8) + self.assertIsNone(read_spec_version(self.dir)) + + def test_broken_galaxy_yaml_returns_none(self) -> None: + from pkgmgr.core.version.source import read_ansible_galaxy_version + + _write(os.path.join(self.dir, "galaxy.yml"), "::: not: valid: yaml:::\n - [") + self.assertIsNone(read_ansible_galaxy_version(self.dir)) + + def test_broken_meta_main_yaml_returns_none(self) -> None: + from pkgmgr.core.version.source import read_ansible_galaxy_version + + os.makedirs(os.path.join(self.dir, "meta"), exist_ok=True) + _write(os.path.join(self.dir, "meta", "main.yml"), "a: [1,\nb: }{") + self.assertIsNone(read_ansible_galaxy_version(self.dir)) + + @unittest.skipUnless(HAS_TOMLI, "tomli is not installed") + def test_missing_tomllib_falls_back_to_tomli(self) -> None: + import builtins + + from pkgmgr.core.version.source import read_pyproject_version + + _write(os.path.join(self.dir, "pyproject.toml"), '[project]\nversion = "9.9.9"\n') + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + if name == "tomllib": + raise ImportError("no tomllib on this interpreter") + return real_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=fake_import): + self.assertEqual(read_pyproject_version(self.dir), "9.9.9") + + +class TestInstalledVersionDegradation(unittest.TestCase): + """src/pkgmgr/core/version/installed.py""" + + def test_unknown_distribution_returns_none(self) -> None: + from pkgmgr.core.version.installed import get_installed_python_version + + self.assertIsNone(get_installed_python_version("no-such-distribution-xyz")) + + def test_missing_importlib_metadata_returns_none(self) -> None: + import builtins + + from pkgmgr.core.version.installed import get_installed_python_version + + real_import = builtins.__import__ + + def fake_import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "importlib" and fromlist and "metadata" in fromlist: + raise ImportError("no importlib.metadata") + return real_import(name, globals, locals, fromlist, level) + + with patch("builtins.__import__", side_effect=fake_import): + self.assertIsNone(get_installed_python_version("kpmx")) + + def test_broken_distribution_scan_returns_none(self) -> None: + from importlib import metadata as importlib_metadata + + from pkgmgr.core.version.installed import get_installed_python_version + + with ( + patch.object( + importlib_metadata, + "version", + side_effect=importlib_metadata.PackageNotFoundError("nope"), + ), + patch.object( + importlib_metadata, "distributions", side_effect=OSError("broken env") + ), + ): + self.assertIsNone(get_installed_python_version("kpmx")) + + def test_non_json_nix_profile_output_falls_back_to_text_mode(self) -> None: + import pkgmgr.core.version.installed as mod + + with ( + patch.object(mod.shutil, "which", return_value="/usr/bin/nix"), + patch.object(mod, "_run_nix", return_value=(0, "not json at all", "")), + ): + self.assertIsNone(mod.get_installed_nix_profile_version("pkgmgr")) + + def test_json_list_instead_of_object_is_tolerated(self) -> None: + import pkgmgr.core.version.installed as mod + + # Valid JSON, wrong shape: data.get() raises AttributeError on a list. + with ( + patch.object(mod.shutil, "which", return_value="/usr/bin/nix"), + patch.object(mod, "_run_nix", return_value=(0, '["a", "b"]', "")), + ): + self.assertIsNone(mod.get_installed_nix_profile_version("pkgmgr")) + + +class TestNixProfileParsingDegradation(unittest.TestCase): + """nix profile normalizer and list reader""" + + def test_coerce_index_rejects_oversized_digit_strings(self) -> None: + from pkgmgr.actions.install.installers.nix.profile.normalizer import ( + coerce_index, + ) + + # str.isdigit() accepts characters int() refuses, e.g. superscripts. + self.assertIsNone(coerce_index("²³", {})) + self.assertIsNone(coerce_index("name-x", {"index": "²"})) + + def test_profile_list_skips_unparsable_index(self) -> None: + from pkgmgr.actions.install.installers.nix.profile_list import ( + NixProfileListReader, + ) + + runner = MagicMock() + runner.run.return_value = SimpleNamespace( + returncode=0, + stdout=" 7 /nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-demo\n", + ) + entries = NixProfileListReader(runner).entries(SimpleNamespace()) + self.assertEqual( + entries, + [(7, "/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-demo")], + ) + + +class TestReleaseFileDegradation(unittest.TestCase): + """src/pkgmgr/actions/release/files/*""" + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.dir = self.tmp.name + + def test_undecodable_flake_is_reported_not_raised(self) -> None: + from pkgmgr.actions.release.files.flake import update_flake_version + + path = _write(os.path.join(self.dir, "flake.nix"), BAD_UTF8) + buf = io.StringIO() + with patch("sys.stdout", buf): + update_flake_version(path, "2.0.0") + self.assertIn("Could not read flake.nix", buf.getvalue()) + + def test_undecodable_pkgbuild_is_reported_not_raised(self) -> None: + from pkgmgr.actions.release.files.pkgbuild import update_pkgbuild_version + + path = _write(os.path.join(self.dir, "PKGBUILD"), BAD_UTF8) + buf = io.StringIO() + with patch("sys.stdout", buf): + update_pkgbuild_version(path, "2.0.0") + self.assertIn("Could not read PKGBUILD", buf.getvalue()) + + def test_undecodable_spec_is_reported_not_raised(self) -> None: + from pkgmgr.actions.release.files.rpm_spec import update_spec_version + + path = _write(os.path.join(self.dir, "pkg.spec"), BAD_UTF8) + buf = io.StringIO() + with patch("sys.stdout", buf): + update_spec_version(path, "2.0.0") + self.assertIn("Could not read spec file", buf.getvalue()) + + +class TestMirrorConfigDegradation(unittest.TestCase): + """src/pkgmgr/actions/mirror/merge_cmd.py""" + + def test_broken_user_config_yields_empty_mapping(self) -> None: + from pkgmgr.actions.mirror.merge_cmd import _load_user_config + + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "config.yaml"), "a: [1,\nb: }{") + self.assertEqual(_load_user_config(path), {}) + + undecodable = _write(os.path.join(tmp, "bad.yaml"), BAD_UTF8) + self.assertEqual(_load_user_config(undecodable), {}) + + +class TestRepositoryDeleteDegradation(unittest.TestCase): + """src/pkgmgr/actions/repository/delete.py""" + + def test_rmtree_failure_is_reported_not_raised(self) -> None: + import pkgmgr.actions.repository.delete as mod + + repo = {"provider": "github.com", "account": "a", "repository": "r"} + buf = io.StringIO() + with ( + patch.object(mod.shutil, "rmtree", side_effect=OSError("device busy")), + patch.object(mod.os.path, "exists", return_value=True), + patch.object(mod, "get_repo_dir", return_value="/repos/a/r"), + patch("builtins.input", return_value="y"), + patch("sys.stdout", buf), + ): + mod.delete_repos([repo], "/repos", [repo], preview=False) + self.assertIn("Error deleting", buf.getvalue()) + + +class TestInkDegradation(unittest.TestCase): + """src/pkgmgr/core/command/ink.py""" + + def test_chmod_failure_is_reported_not_raised(self) -> None: + import pkgmgr.core.command.ink as mod + + with tempfile.TemporaryDirectory() as tmp: + repo_dir = os.path.join(tmp, "p", "a", "repo") + bin_dir = os.path.join(tmp, "bin") + os.makedirs(repo_dir) + _write(os.path.join(repo_dir, "main.py"), "print()\n") + repo = { + "provider": "p", + "account": "a", + "repository": "repo", + "command": os.path.join(repo_dir, "main.py"), + } + buf = io.StringIO() + with ( + patch.object(mod.os, "chmod", side_effect=OSError("read-only fs")), + patch("sys.stdout", buf), + ): + mod.create_ink( + repo, + tmp, + bin_dir, + [repo], + quiet=False, + preview=False, + ) + self.assertIn("Failed to set permissions", buf.getvalue()) + + +class TestHttpClientDegradation(unittest.TestCase): + """src/pkgmgr/core/remote_provisioning/http/client.py""" + + def test_non_json_body_is_returned_as_text(self) -> None: + from pkgmgr.core.remote_provisioning.http.client import HttpClient + + resp = MagicMock() + resp.read.return_value = b"not json" + resp.status = 200 + resp.__enter__ = lambda s: s + resp.__exit__ = lambda *a: False + + with patch( + "pkgmgr.core.remote_provisioning.http.client.urllib.request.urlopen", + return_value=resp, + ): + out = HttpClient().request_json("GET", "https://example.invalid/x") + self.assertIsNone(out.json) + self.assertIn("not json", out.text) + + def test_unreadable_error_body_degrades_to_empty_string(self) -> None: + from pkgmgr.core.remote_provisioning.http.client import HttpClient + from pkgmgr.core.remote_provisioning.http.errors import HttpError + + class BrokenBody(io.RawIOBase): + def read(self, *args): + raise OSError("socket closed") + + err = urllib.error.HTTPError( + url="https://example.invalid/x", + code=503, + msg="nope", + hdrs={}, + fp=BrokenBody(), + ) + with ( + patch( + "pkgmgr.core.remote_provisioning.http.client.urllib.request.urlopen", + side_effect=err, + ), + self.assertRaises(HttpError) as ctx, + ): + HttpClient().request_json("GET", "https://example.invalid/x") + self.assertEqual(ctx.exception.status, 503) + self.assertEqual(ctx.exception.body, "") + + +class TestProviderRegistryDegradation(unittest.TestCase): + """src/pkgmgr/core/remote_provisioning/registry.py""" + + def test_misbehaving_provider_is_skipped(self) -> None: + from pkgmgr.core.remote_provisioning.registry import ProviderRegistry + + class Broken: + def can_handle(self, host): + raise TypeError("host is not what I expected") + + class Good: + def can_handle(self, host): + return True + + good = Good() + self.assertIs(ProviderRegistry(providers=[Broken(), good]).resolve("x"), good) + + def test_registry_returns_none_when_all_providers_break(self) -> None: + from pkgmgr.core.remote_provisioning.registry import ProviderRegistry + + class Broken: + def can_handle(self, host): + raise AttributeError("nope") + + self.assertIsNone(ProviderRegistry(providers=[Broken()]).resolve("x")) + + +class TestCredentialProviderDegradation(unittest.TestCase): + """gh token provider and token validation""" + + def test_missing_gh_binary_returns_none(self) -> None: + import pkgmgr.core.credentials.providers.gh as mod + from pkgmgr.core.credentials.types import TokenRequest + + req = TokenRequest(provider_kind="github", host="github.com", owner=None) + with ( + patch.object(mod.shutil, "which", return_value="/usr/bin/gh"), + patch.object( + mod.subprocess, + "check_output", + side_effect=FileNotFoundError("gh vanished"), + ), + ): + self.assertIsNone(mod.GhTokenProvider().get(req)) + + def test_failing_gh_command_returns_none(self) -> None: + import pkgmgr.core.credentials.providers.gh as mod + from pkgmgr.core.credentials.types import TokenRequest + + req = TokenRequest(provider_kind="github", host="github.com", owner=None) + with ( + patch.object(mod.shutil, "which", return_value="/usr/bin/gh"), + patch.object( + mod.subprocess, + "check_output", + side_effect=subprocess.CalledProcessError(1, ["gh"]), + ), + ): + self.assertIsNone(mod.GhTokenProvider().get(req)) + + def test_unreachable_host_makes_token_invalid(self) -> None: + import pkgmgr.core.credentials.validate as mod + + with patch.object( + mod.urllib.request, "urlopen", side_effect=urllib.error.URLError("no route") + ): + self.assertFalse(mod.validate_token("github", "github.com", "tok")) + + def test_non_json_validation_body_makes_token_invalid(self) -> None: + import pkgmgr.core.credentials.validate as mod + + resp = MagicMock() + resp.status = 200 + resp.read.return_value = b"gateway" + resp.__enter__ = lambda s: s + resp.__exit__ = lambda *a: False + with patch.object(mod.urllib.request, "urlopen", return_value=resp): + self.assertFalse(mod.validate_token("github", "github.com", "tok")) + + +class TestConfigLoadDegradation(unittest.TestCase): + """src/pkgmgr/core/config/load.py""" + + def test_missing_pkgmgr_package_yields_empty_defaults(self) -> None: + import builtins + + from pkgmgr.core.config.load import _load_defaults_from_package_or_project + + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + if name == "pkgmgr": + raise ImportError("not installed") + return real_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=fake_import): + self.assertEqual( + _load_defaults_from_package_or_project(), + {"directories": {}, "repositories": []}, + ) + + +class TestArchPkgbuildDegradation(unittest.TestCase): + """src/pkgmgr/actions/install/installers/os_packages/arch_pkgbuild.py""" + + def test_missing_geteuid_does_not_break_supports(self) -> None: + import pkgmgr.actions.install.installers.os_packages.arch_pkgbuild as mod + + ctx = SimpleNamespace(repo_dir="/repo") + with ( + patch.object(mod.os, "geteuid", side_effect=OSError("no euid here")), + patch.object(mod.shutil, "which", return_value=None), + ): + self.assertFalse(mod.ArchPkgbuildInstaller().supports(ctx)) + + +class TestCliDirectoryResolutionDegradation(unittest.TestCase): + """CLI commands resolving repository directories from malformed config""" + + def test_dispatch_skips_malformed_repository_entries(self) -> None: + import pkgmgr.cli.dispatch as mod + + ctx = SimpleNamespace( + all_repositories=[{"provider": "github.com"}], + repositories_base_dir="/repos", + ) + with patch.object(mod, "get_repo_dir", side_effect=KeyError("account")): + self.assertEqual(mod._select_repo_for_current_directory(ctx), []) + + def test_proxy_skips_malformed_repository_entries(self) -> None: + import pkgmgr.cli.proxy as mod + + ctx = SimpleNamespace( + all_repositories=[{"provider": "github.com"}], + repositories_base_dir="/repos", + ) + with patch.object(mod, "get_repo_dir", side_effect=TypeError("bad entry")): + self.assertEqual(mod._select_repo_for_current_directory(ctx), []) + + +class TestGitTagQueryDegradation(unittest.TestCase): + """changelog and version CLI reading git tags outside a repository""" + + def test_changelog_reports_unreadable_tags(self) -> None: + import pkgmgr.cli.commands.changelog as mod + from pkgmgr.core.git import GitRunError + + with tempfile.TemporaryDirectory() as tmp: + ctx = SimpleNamespace( + all_repositories=[], + repositories_base_dir=tmp, + config_merged={}, + ) + repo = {"directory": tmp, "repository": "demo"} + buf = io.StringIO() + with ( + patch.object(mod, "get_tags", side_effect=GitRunError("not a git repo")), + patch.object(mod, "get_repo_identifier", return_value="demo"), + patch.object(mod, "generate_changelog", return_value=""), + patch("sys.stdout", buf), + ): + mod.handle_changelog(SimpleNamespace(command="show"), ctx, [repo]) + self.assertIn("Could not read git tags", buf.getvalue()) + + +if __name__ == "__main__": + unittest.main() + + +class TestReleaseFileWriteDegradation(unittest.TestCase): + """release file updaters that read or write packaging metadata""" + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.dir = self.tmp.name + + def test_undecodable_changelog_is_reported_and_rewritten(self) -> None: + from pkgmgr.actions.release.files.changelog_md import update_changelog + + path = _write(os.path.join(self.dir, "CHANGELOG.md"), BAD_UTF8) + buf = io.StringIO() + with patch("sys.stdout", buf): + update_changelog(path, "2.0.0", message="Fix things", preview=True) + self.assertIn("Could not read existing CHANGELOG.md", buf.getvalue()) + + def test_undecodable_debian_changelog_is_reported(self) -> None: + from pkgmgr.actions.release.files.debian import update_debian_changelog + + path = _write(os.path.join(self.dir, "changelog"), BAD_UTF8) + buf = io.StringIO() + with patch("sys.stdout", buf): + update_debian_changelog(path, "demo", "2.0.0", message="Fix", preview=False) + self.assertIn("Could not read debian/changelog", buf.getvalue()) + + def test_undecodable_spec_changelog_is_reported(self) -> None: + from pkgmgr.actions.release.files.rpm_changelog import update_spec_changelog + + path = _write(os.path.join(self.dir, "pkg.spec"), BAD_UTF8) + buf = io.StringIO() + with patch("sys.stdout", buf): + update_spec_changelog(path, "demo", "2.0.0", message="Fix", preview=False) + self.assertIn("Could not read spec file for changelog update", buf.getvalue()) + + def test_unwritable_spec_changelog_is_reported(self) -> None: + import pkgmgr.actions.release.files.rpm_changelog as mod + + path = _write( + os.path.join(self.dir, "pkg.spec"), + "Name: demo\nVersion: 1.0.0\n%changelog\n", + ) + real_open = open + buf = io.StringIO() + + def failing_open(file, mode="r", *args, **kwargs): + if "w" in mode: + raise OSError("disk full") + return real_open(file, mode, *args, **kwargs) + + with patch("builtins.open", side_effect=failing_open), patch("sys.stdout", buf): + mod.update_spec_changelog(path, "demo", "2.0.0", message="Fix") + self.assertIn("Failed to write updated spec changelog", buf.getvalue()) + + +class TestTemplateImportDegradation(unittest.TestCase): + """src/pkgmgr/actions/repository/create/templates.py""" + + def test_module_records_missing_jinja2(self) -> None: + import pkgmgr.actions.repository.create.templates as mod + + self.assertTrue(hasattr(mod, "_JINJA_IMPORT_ERROR")) + if mod.Environment is None: + self.assertIsNotNone(mod._JINJA_IMPORT_ERROR) + else: + self.assertIsNone(mod._JINJA_IMPORT_ERROR) + + +class TestCliRepoDirDegradation(unittest.TestCase): + """CLI handlers resolving repo dirs from malformed entries""" + + def _ctx(self): + return SimpleNamespace( + all_repositories=[], + repositories_base_dir="/repos", + binaries_dir="/bin", + config_merged={}, + ) + + def test_changelog_skips_unresolvable_directory(self) -> None: + import pkgmgr.cli.commands.changelog as mod + + buf = io.StringIO() + with ( + patch.object(mod, "get_repo_dir", side_effect=KeyError("account")), + patch.object(mod, "get_repo_identifier", return_value="demo"), + patch("sys.stdout", buf), + ): + mod.handle_changelog( + SimpleNamespace(range=""), self._ctx(), [{"repository": "demo"}] + ) + self.assertIn("Skipped", buf.getvalue()) + + def test_version_skips_unresolvable_directory(self) -> None: + import pkgmgr.cli.commands.version as mod + + buf = io.StringIO() + with ( + patch.object(mod, "get_repo_dir", side_effect=TypeError("bad entry")), + patch.object(mod, "get_repo_identifier", return_value="demo"), + patch("sys.stdout", buf), + ): + mod.handle_version( + SimpleNamespace(), self._ctx(), [{"repository": "demo"}] + ) + self.assertIn("Skipped", buf.getvalue()) + + def test_release_skips_unresolvable_directory(self) -> None: + import pkgmgr.cli.commands.release as mod + + buf = io.StringIO() + with ( + patch.object(mod, "get_repo_dir", side_effect=AttributeError("nope")), + patch.object(mod, "get_repo_identifier", return_value="demo"), + patch("sys.stdout", buf), + ): + mod.handle_release( + SimpleNamespace(list=False), self._ctx(), [{"repository": "demo"}] + ) + self.assertIn("failed to resolve directory", buf.getvalue()) + + def test_repos_path_reports_unresolvable_directory(self) -> None: + import pkgmgr.cli.commands.repos as mod + + buf = io.StringIO() + with ( + patch.object( + mod, "_resolve_repository_directory", side_effect=ValueError("bad") + ), + patch("sys.stdout", buf), + ): + mod.handle_repos_command( + SimpleNamespace(command="path"), self._ctx(), [{"repository": "demo"}] + ) + self.assertIn("Could not resolve directory", buf.getvalue()) + + def test_version_reports_unreadable_tags(self) -> None: + import pkgmgr.cli.commands.version as mod + from pkgmgr.core.git import GitRunError + + with tempfile.TemporaryDirectory() as tmp: + buf = io.StringIO() + with ( + patch.object(mod, "get_tags", side_effect=GitRunError("no repo")), + patch.object(mod, "get_repo_identifier", return_value="demo"), + patch("sys.stdout", buf), + ): + mod.handle_version( + SimpleNamespace(), self._ctx(), [{"directory": tmp}] + ) + self.assertIn("Could not read git tags", buf.getvalue()) + + +class TestInkAliasDegradation(unittest.TestCase): + """alias symlink creation in src/pkgmgr/core/command/ink.py""" + + def test_alias_symlink_failure_is_reported_not_raised(self) -> None: + import pkgmgr.core.command.ink as mod + + with tempfile.TemporaryDirectory() as tmp: + repo_dir = os.path.join(tmp, "p", "a", "repo") + bin_dir = os.path.join(tmp, "bin") + os.makedirs(repo_dir) + command = _write(os.path.join(repo_dir, "main.py"), "print()\n") + repo = { + "provider": "p", + "account": "a", + "repository": "repo", + "command": command, + "alias": "demo-alias", + } + buf = io.StringIO() + real_symlink = os.symlink + calls = {"n": 0} + + def flaky_symlink(src, dst, **kwargs): + calls["n"] += 1 + if calls["n"] > 1: + raise OSError("alias target is on a read-only mount") + return real_symlink(src, dst, **kwargs) + + with patch.object(mod.os, "symlink", side_effect=flaky_symlink), patch( + "sys.stdout", buf + ): + mod.create_ink(repo, tmp, bin_dir, [repo], quiet=False, preview=False) + self.assertIn("Error creating alias", buf.getvalue()) + + +class TestNixIndexParsingDegradation(unittest.TestCase): + """remaining int() guards in the nix profile helpers""" + + def test_coerce_index_rejects_oversized_trailing_number(self) -> None: + from pkgmgr.actions.install.installers.nix.profile.normalizer import ( + coerce_index, + ) + + # Matches the "-" branch, but int() refuses digit strings + # longer than sys.get_int_max_str_digits(). + self.assertIsNone(coerce_index("pkg-" + "9" * 5000, {})) + + def test_profile_list_skips_oversized_index(self) -> None: + from pkgmgr.actions.install.installers.nix.profile_list import ( + NixProfileListReader, + ) + + runner = MagicMock() + runner.run.return_value = SimpleNamespace( + returncode=0, + stdout=" " + "9" * 5000 + " /nix/store/" + + "a" * 32 + "-demo\n", + ) + self.assertEqual(NixProfileListReader(runner).entries(SimpleNamespace()), []) + + +class TestTomliFallbackDegradation(unittest.TestCase): + """second tomllib import guard in version/source.py""" + + @unittest.skipUnless(HAS_TOMLI, "tomli is not installed") + def test_project_name_falls_back_to_tomli(self) -> None: + import builtins + + from pkgmgr.core.version.source import read_pyproject_project_name + + with tempfile.TemporaryDirectory() as tmp: + _write(os.path.join(tmp, "pyproject.toml"), '[project]\nname = "demo"\n') + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + if name == "tomllib": + raise ImportError("no tomllib") + return real_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=fake_import): + self.assertEqual(read_pyproject_project_name(tmp), "demo") diff --git a/tests/integration/test_repos_create_preview.py b/tests/integration/test_repos_create_preview.py index 5ae7b5b..1138074 100644 --- a/tests/integration/test_repos_create_preview.py +++ b/tests/integration/test_repos_create_preview.py @@ -13,7 +13,7 @@ class TestIntegrationReposCreatePreview(unittest.TestCase): # Import lazily to avoid hard-failing if the CLI module/function name differs. try: repos_mod = importlib.import_module("pkgmgr.cli.commands.repos") - except Exception as exc: + except ImportError as exc: self.skipTest(f"CLI module not available: {exc}") handle = getattr(repos_mod, "handle_repos_command", None)