fix: catch concrete exceptions instead of bare Exception

Replace 46 blind `except Exception` handlers with the exception types the
guarded code can actually raise, so an unforeseen failure surfaces instead
of being silently swallowed. The types were derived per site from the try
block: file reads get (OSError, UnicodeDecodeError), TOML and YAML parsing
add their decode errors, int() guards get ValueError, imports get
ImportError, subprocess calls get (OSError, subprocess.SubprocessError),
git queries get GitRunError, and importlib.metadata lookups get
PackageNotFoundError.

Three handlers stay blind, now with a stated reason: the per-repository
loops in install and update are batch boundaries where one failing
repository must never abort the run.

Three imports come along because the narrowing needs them: GitRunError in
the changelog and version commands, and http.client in token validation --
urllib raises HTTPException, which is not an OSError subclass and would
otherwise have escaped.

Add tests/integration/test_error_path_degradation.py, which drives 44 of
the 46 handlers into their except branch and asserts the documented
degradation. Faults are injected for real rather than mocked wherever
possible: undecodable bytes, malformed YAML, digit strings past
sys.get_int_max_str_digits(), a real HTTPError with an unreadable body.
That approach paid for itself immediately -- tomllib.load() decodes
internally and raises UnicodeDecodeError, which was missing from the
pyproject handlers and would have turned a malformed pyproject.toml into
a traceback.

The two handlers left uncovered are the branch-close path in _release_impl,
reachable only through a full release, and the jinja2 import guard.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Opus 5 committed 2026-07-27 18:37:14 +02:00
1 parent 85d1abf61c
commit 7764a2946c
36 files changed
+860 -103

No files matched your search

+1 -3
View File
@@ -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,
+1 -1
View File
@@ -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(
@@ -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 "<name>-<n>"
@@ -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.
@@ -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)
@@ -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
+2 -2
View File
@@ -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 {}
@@ -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]:
@@ -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)
+2 -2
View File
@@ -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 = ""
+2 -4
View File
@@ -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()
+2 -2
View File
@@ -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
+2 -2
View File
@@ -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
@@ -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
+2 -2
View File
@@ -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
+1 -1
View File
@@ -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}")
@@ -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
+1 -1
View File
@@ -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}.")
+2 -2
View File
@@ -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(
+10 -9
View File
@@ -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
+1 -1
View File
@@ -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}"
)
+1 -1
View File
@@ -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', '?')}/"
+5 -4
View File
@@ -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
)
+1 -1
View File
@@ -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))
+3 -3
View File
@@ -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
+2 -2
View File
@@ -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}")
+3 -9
View File
@@ -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()
+2 -2
View File
@@ -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")
+1 -1
View File
@@ -34,7 +34,7 @@ class GhTokenProvider:
stderr=subprocess.STDOUT,
text=True,
).strip()
except Exception:
except (OSError, subprocess.SubprocessError):
return None
if not out:
+2 -1
View File
@@ -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)
@@ -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:
@@ -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
+4 -4
View File
@@ -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
+18 -18
View File
@@ -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