refactor(update): move update logic to unified UpdateManager and extend system support
Some checks failed
CI / test-unit (push) Has been cancelled
CI / test-integration (push) Has been cancelled
CI / test-env-virtual (push) Has been cancelled
CI / test-env-nix (push) Has been cancelled
CI / test-e2e (push) Has been cancelled
CI / test-virgin-user (push) Has been cancelled
CI / test-virgin-root (push) Has been cancelled
CI / codesniffer-shellcheck (push) Has been cancelled
CI / codesniffer-ruff (push) Has been cancelled

- Move update orchestration from repository scope to actions/update
- Introduce UpdateManager and SystemUpdater with distro detection
- Add Arch, Debian/Ubuntu, and Fedora/RHEL system update handling
- Rename CLI flag from --system-update to --system
- Route update as a top-level command in CLI dispatch
- Remove legacy update_repos implementation
- Add E2E tests for:
  - update all without system updates
  - update single repo (pkgmgr) with system updates

https://chatgpt.com/share/693e76ec-5ee4-800f-9623-3983f56d5430
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-14 09:35:52 +01:00
parent 2a4ec18532
commit 55f4a1e941
10 changed files with 390 additions and 92 deletions

View File

@@ -1,58 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import shutil
from pkgmgr.actions.install import install_repos
from pkgmgr.actions.repository.pull import pull_with_verification
def update_repos(
selected_repos,
repositories_base_dir,
bin_dir,
all_repos,
no_verification,
system_update,
preview: bool,
quiet: bool,
update_dependencies: bool,
clone_mode: str,
force_update: bool = True,
) -> None:
"""
Update repositories by pulling latest changes and installing them.
"""
pull_with_verification(
selected_repos,
repositories_base_dir,
all_repos,
[],
no_verification,
preview,
)
install_repos(
selected_repos,
repositories_base_dir,
bin_dir,
all_repos,
no_verification,
preview,
quiet,
clone_mode,
update_dependencies,
force_update=force_update,
)
if system_update:
from pkgmgr.core.command.run import run_command
if shutil.which("nix") is not None:
try:
run_command("nix profile upgrade '.*'", preview=preview)
except SystemExit as e:
print(f"[Warning] 'nix profile upgrade' failed: {e}")
run_command("sudo -u aur_builder yay -Syu --noconfirm", preview=preview)
run_command("sudo pacman -Syyu --noconfirm", preview=preview)

View File

@@ -0,0 +1,10 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
from pkgmgr.actions.update.manager import UpdateManager
__all__ = [
"UpdateManager",
]

View File

@@ -0,0 +1,61 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
from typing import Any, Iterable
from pkgmgr.actions.update.system_updater import SystemUpdater
class UpdateManager:
"""
Orchestrates:
- repository pull + installation
- optional system update
"""
def __init__(self) -> None:
self._system_updater = SystemUpdater()
def run(
self,
selected_repos: Iterable[Any],
repositories_base_dir: str,
bin_dir: str,
all_repos: Any,
no_verification: bool,
system_update: bool,
preview: bool,
quiet: bool,
update_dependencies: bool,
clone_mode: str,
force_update: bool = True,
) -> None:
from pkgmgr.actions.install import install_repos
from pkgmgr.actions.repository.pull import pull_with_verification
pull_with_verification(
selected_repos,
repositories_base_dir,
all_repos,
[],
no_verification,
preview,
)
install_repos(
selected_repos,
repositories_base_dir,
bin_dir,
all_repos,
no_verification,
preview,
quiet,
clone_mode,
update_dependencies,
force_update=force_update,
)
if system_update:
self._system_updater.run(preview=preview)

View File

@@ -0,0 +1,66 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Dict
def read_os_release(path: str = "/etc/os-release") -> Dict[str, str]:
"""
Parse /etc/os-release into a dict. Returns empty dict if missing.
"""
if not os.path.exists(path):
return {}
result: Dict[str, str] = {}
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
result[key.strip()] = value.strip().strip('"')
return result
@dataclass(frozen=True)
class OSReleaseInfo:
"""
Minimal /etc/os-release representation for distro detection.
"""
id: str = ""
id_like: str = ""
pretty_name: str = ""
@staticmethod
def load() -> "OSReleaseInfo":
data = read_os_release()
return OSReleaseInfo(
id=(data.get("ID") or "").lower(),
id_like=(data.get("ID_LIKE") or "").lower(),
pretty_name=(data.get("PRETTY_NAME") or ""),
)
def ids(self) -> set[str]:
ids: set[str] = set()
if self.id:
ids.add(self.id)
if self.id_like:
for part in self.id_like.split():
ids.add(part.strip())
return ids
def is_arch_family(self) -> bool:
ids = self.ids()
return ("arch" in ids) or ("archlinux" in ids)
def is_debian_family(self) -> bool:
ids = self.ids()
return bool(ids.intersection({"debian", "ubuntu"}))
def is_fedora_family(self) -> bool:
ids = self.ids()
return bool(ids.intersection({"fedora", "rhel", "centos", "rocky", "almalinux"}))

View File

@@ -0,0 +1,96 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import platform
import shutil
from pkgmgr.actions.update.os_release import OSReleaseInfo
class SystemUpdater:
"""
Executes distro-specific system update commands, plus Nix profile upgrades if available.
"""
def run(self, *, preview: bool) -> None:
from pkgmgr.core.command.run import run_command
# Distro-agnostic: Nix profile upgrades (if Nix is present).
if shutil.which("nix") is not None:
try:
run_command("nix profile upgrade '.*'", preview=preview)
except SystemExit as e:
print(f"[Warning] 'nix profile upgrade' failed: {e}")
osr = OSReleaseInfo.load()
if osr.is_arch_family():
self._update_arch(preview=preview)
return
if osr.is_debian_family():
self._update_debian(preview=preview)
return
if osr.is_fedora_family():
self._update_fedora(preview=preview)
return
distro = osr.pretty_name or platform.platform()
print(f"[Warning] Unsupported distribution for system update: {distro}")
def _update_arch(self, *, preview: bool) -> None:
from pkgmgr.core.command.run import run_command
yay = shutil.which("yay")
pacman = shutil.which("pacman")
sudo = shutil.which("sudo")
# Prefer yay if available (repo + AUR in one pass).
# Avoid running yay and pacman afterwards to prevent double update passes.
if yay and sudo:
run_command("sudo -u aur_builder yay -Syu --noconfirm", preview=preview)
return
if pacman and sudo:
run_command("sudo pacman -Syu --noconfirm", preview=preview)
return
print("[Warning] Cannot update Arch system: missing required tools (sudo/yay/pacman).")
def _update_debian(self, *, preview: bool) -> None:
from pkgmgr.core.command.run import run_command
sudo = shutil.which("sudo")
apt_get = shutil.which("apt-get")
if not (sudo and apt_get):
print("[Warning] Cannot update Debian/Ubuntu system: missing required tools (sudo/apt-get).")
return
env = "DEBIAN_FRONTEND=noninteractive"
run_command(f"sudo {env} apt-get update -y", preview=preview)
run_command(f"sudo {env} apt-get -y dist-upgrade", preview=preview)
def _update_fedora(self, *, preview: bool) -> None:
from pkgmgr.core.command.run import run_command
sudo = shutil.which("sudo")
dnf = shutil.which("dnf")
microdnf = shutil.which("microdnf")
if not sudo:
print("[Warning] Cannot update Fedora/RHEL-like system: missing sudo.")
return
if dnf:
run_command("sudo dnf -y upgrade", preview=preview)
return
if microdnf:
run_command("sudo microdnf -y upgrade", preview=preview)
return
print("[Warning] Cannot update Fedora/RHEL-like system: missing dnf/microdnf.")

View File

@@ -8,7 +8,6 @@ from typing import Any, Dict, List
from pkgmgr.cli.context import CLIContext from pkgmgr.cli.context import CLIContext
from pkgmgr.actions.install import install_repos from pkgmgr.actions.install import install_repos
from pkgmgr.actions.repository.update import update_repos
from pkgmgr.actions.repository.deinstall import deinstall_repos from pkgmgr.actions.repository.deinstall import deinstall_repos
from pkgmgr.actions.repository.delete import delete_repos from pkgmgr.actions.repository.delete import delete_repos
from pkgmgr.actions.repository.status import status_repos from pkgmgr.actions.repository.status import status_repos
@@ -72,25 +71,6 @@ def handle_repos_command(
) )
return return
# ------------------------------------------------------------
# update
# ------------------------------------------------------------
if args.command == "update":
update_repos(
selected,
ctx.repositories_base_dir,
ctx.binaries_dir,
ctx.all_repositories,
args.no_verification,
args.system_update,
args.preview,
args.quiet,
args.dependencies,
args.clone_mode,
force_update=True,
)
return
# ------------------------------------------------------------ # ------------------------------------------------------------
# deinstall # deinstall
# ------------------------------------------------------------ # ------------------------------------------------------------

View File

@@ -141,6 +141,27 @@ def dispatch_command(args, ctx: CLIContext) -> None:
handle_repos_command(args, ctx, selected) handle_repos_command(args, ctx, selected)
return return
# ------------------------------------------------------------
# update
# ------------------------------------------------------------
if args.command == "update":
from pkgmgr.actions.update import UpdateManager
UpdateManager().run(
selected_repos=selected,
repositories_base_dir=ctx.repositories_base_dir,
bin_dir=ctx.binaries_dir,
all_repos=ctx.all_repositories,
no_verification=args.no_verification,
system_update=args.system,
preview=args.preview,
quiet=args.quiet,
update_dependencies=args.dependencies,
clone_mode=args.clone_mode,
force_update=True,
)
return
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
# Tools (explore / terminal / code) # Tools (explore / terminal / code)
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #

View File

@@ -33,8 +33,8 @@ def add_install_update_subparsers(
) )
add_install_update_arguments(update_parser) add_install_update_arguments(update_parser)
update_parser.add_argument( update_parser.add_argument(
"--system-update", "--system",
dest="system_update", dest="system",
action="store_true", action="store_true",
help="Include system update commands", help="Include system update commands",
) )

View File

@@ -1,6 +1,6 @@
""" """
Integration test: update all configured repositories using Integration test: update all configured repositories using
--clone-mode https and --no-verification. --clone-mode shallow and --no-verification, WITHOUT system updates.
This test is intended to be run inside the Docker container where: This test is intended to be run inside the Docker container where:
- network access is available, - network access is available,
@@ -8,8 +8,8 @@ This test is intended to be run inside the Docker container where:
- and it is safe to perform real git operations. - and it is safe to perform real git operations.
It passes if BOTH commands complete successfully (in separate tests): It passes if BOTH commands complete successfully (in separate tests):
1) pkgmgr update --all --clone-mode https --no-verification --system-update 1) pkgmgr update --all --clone-mode shallow --no-verification
2) nix run .#pkgmgr -- update --all --clone-mode https --no-verification --system-update 2) nix run .#pkgmgr -- update --all --clone-mode shallow --no-verification
""" """
from __future__ import annotations from __future__ import annotations
@@ -38,7 +38,7 @@ def _make_temp_gitconfig_with_safe_dirs(home: Path) -> Path:
return gitconfig return gitconfig
class TestIntegrationUpdateAllHttps(unittest.TestCase): class TestIntegrationUpdateAllshallowNoSystem(unittest.TestCase):
def _common_env(self, home_dir: str) -> dict[str, str]: def _common_env(self, home_dir: str) -> dict[str, str]:
env = os.environ.copy() env = os.environ.copy()
env["HOME"] = home_dir env["HOME"] = home_dir
@@ -86,32 +86,30 @@ class TestIntegrationUpdateAllHttps(unittest.TestCase):
remove_pkgmgr_from_nix_profile() remove_pkgmgr_from_nix_profile()
nix_profile_list_debug("AFTER CLEANUP") nix_profile_list_debug("AFTER CLEANUP")
def test_update_all_repositories_https_pkgmgr(self) -> None: def test_update_all_repositories_shallow_pkgmgr_no_system(self) -> None:
self._common_setup() self._common_setup()
with tempfile.TemporaryDirectory(prefix="pkgmgr-updateall-") as tmp: with tempfile.TemporaryDirectory(prefix="pkgmgr-updateall-nosys-") as tmp:
env = self._common_env(tmp) env = self._common_env(tmp)
args = [ args = [
"update", "update",
"--all", "--all",
"--clone-mode", "--clone-mode",
"https", "shallow",
"--no-verification", "--no-verification",
"--system-update",
] ]
self._run_cmd(["pkgmgr", *args], label="pkgmgr", env=env) self._run_cmd(["pkgmgr", *args], label="pkgmgr", env=env)
pkgmgr_help_debug() pkgmgr_help_debug()
def test_update_all_repositories_https_nix_pkgmgr(self) -> None: def test_update_all_repositories_shallow_nix_pkgmgr_no_system(self) -> None:
self._common_setup() self._common_setup()
with tempfile.TemporaryDirectory(prefix="pkgmgr-updateall-nix-") as tmp: with tempfile.TemporaryDirectory(prefix="pkgmgr-updateall-nosys-nix-") as tmp:
env = self._common_env(tmp) env = self._common_env(tmp)
args = [ args = [
"update", "update",
"--all", "--all",
"--clone-mode", "--clone-mode",
"https", "shallow",
"--no-verification", "--no-verification",
"--system-update",
] ]
self._run_cmd( self._run_cmd(
["nix", "run", ".#pkgmgr", "--", *args], ["nix", "run", ".#pkgmgr", "--", *args],

View File

@@ -0,0 +1,124 @@
"""
Integration test: update ONLY the 'pkgmgr' repository with system updates enabled.
This test is intended to be run inside the Docker container where:
- network access is available,
- the config/config.yaml is present,
- and it is safe to perform real git operations.
It passes if BOTH commands complete successfully (in separate tests):
1) pkgmgr update pkgmgr --clone-mode shallow --no-verification --system
2) nix run .#pkgmgr -- update pkgmgr --clone-mode shallow --no-verification --system
"""
from __future__ import annotations
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
from test_install_pkgmgr_shallow import (
nix_profile_list_debug,
remove_pkgmgr_from_nix_profile,
pkgmgr_help_debug,
)
def _make_temp_gitconfig_with_safe_dirs(home: Path) -> Path:
gitconfig = home / ".gitconfig"
gitconfig.write_text(
"[safe]\n"
"\tdirectory = /src\n"
"\tdirectory = /src/.git\n"
"\tdirectory = *\n"
)
return gitconfig
class TestIntegrationUpdatePkgmgrWithSystem(unittest.TestCase):
def _common_env(self, home_dir: str) -> dict[str, str]:
env = os.environ.copy()
env["HOME"] = home_dir
home = Path(home_dir)
home.mkdir(parents=True, exist_ok=True)
env["GIT_CONFIG_GLOBAL"] = str(_make_temp_gitconfig_with_safe_dirs(home))
# Ensure nix is discoverable if the container has it
env["PATH"] = "/nix/var/nix/profiles/default/bin:" + env.get("PATH", "")
return env
def _run_cmd(self, cmd: list[str], label: str, env: dict[str, str]) -> None:
cmd_repr = " ".join(cmd)
print(f"\n[TEST] Running ({label}): {cmd_repr}")
proc = subprocess.run(
cmd,
check=False,
cwd=os.getcwd(),
env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
print(proc.stdout.rstrip())
if proc.returncode != 0:
print(f"\n[TEST] Command failed ({label})")
print(f"[TEST] Command : {cmd_repr}")
print(f"[TEST] Exit code: {proc.returncode}")
nix_profile_list_debug(f"ON FAILURE ({label})")
raise AssertionError(
f"({label}) {cmd_repr!r} failed with exit code {proc.returncode}.\n\n"
f"--- output ---\n{proc.stdout}\n"
)
def _common_setup(self) -> None:
nix_profile_list_debug("BEFORE CLEANUP")
remove_pkgmgr_from_nix_profile()
nix_profile_list_debug("AFTER CLEANUP")
def test_update_pkgmgr_shallow_pkgmgr_with_system(self) -> None:
self._common_setup()
with tempfile.TemporaryDirectory(prefix="pkgmgr-update-pkgmgr-sys-") as tmp:
env = self._common_env(tmp)
args = [
"update",
"pkgmgr",
"--clone-mode",
"shallow",
"--no-verification",
"--system",
]
self._run_cmd(["pkgmgr", *args], label="pkgmgr", env=env)
pkgmgr_help_debug()
def test_update_pkgmgr_shallow_nix_pkgmgr_with_system(self) -> None:
self._common_setup()
with tempfile.TemporaryDirectory(prefix="pkgmgr-update-pkgmgr-sys-nix-") as tmp:
env = self._common_env(tmp)
args = [
"update",
"pkgmgr",
"--clone-mode",
"shallow",
"--no-verification",
"--system",
]
self._run_cmd(
["nix", "run", ".#pkgmgr", "--", *args],
label="nix run .#pkgmgr",
env=env,
)
pkgmgr_help_debug()
if __name__ == "__main__":
unittest.main()