[Piuparts-devel] [Git][debian/piuparts][helmutg/pathlib] differentiate path types using pathlib
Helmut Grohne (@helmutg)
gitlab at salsa.debian.org
Sun Jul 5 05:17:22 BST 2026
Helmut Grohne pushed to branch helmutg/pathlib at Debian / piuparts
Commits:
71ae753c by Helmut Grohne at 2026-07-05T06:16:56+02:00
differentiate path types using pathlib
Change most of filename variables to use pathlib's Path or PurePath. The
benefit of this is not so much in using those pathlib objects. It is
knowing what we talk about. The Path type is used for filenames that
actually may exist when looked at from the host system. When a variable
of type Path refers to a file inside the chroot, it includes the chroot
prefix (Chroot.name). In contrast, filenames that lack this prefix (and
thus cannot be operated on directly) become PurePath. In particular, the
Chroot.relative method that turns a path inside the chroot into a host
path now consumes a PurePath and returns a Path. The type checker now
also tells when mixing paths and str. Yes, this adds a bit verbosity all
over the place due to having to convert between these types explicitly.
In particular, arguments to run need to be converted using os.fspath.
Hopefully, this explicit conversion makes the code easier to reason
about.
- - - - -
2 changed files:
- piuparts.py
- tests/test_piuparts.py
Changes:
=====================================
piuparts.py
=====================================
@@ -34,10 +34,10 @@ Lars Wirzenius <liw at iki.fi>
from __future__ import print_function
import errno
+import collections.abc
import logging
import optparse
import os
-import pathlib
import pickle
import re
import shlex
@@ -52,6 +52,7 @@ import typing
import uuid
from collections import defaultdict, namedtuple
from contextlib import ExitStack
+from pathlib import Path, PurePath
from signal import SIGALRM, SIGKILL, SIGTERM, alarm, signal
import apt_pkg
@@ -90,7 +91,7 @@ class Defaults:
"""Return default distribution."""
raise NotImplementedError
- def get_keyring(self) -> str:
+ def get_keyring(self) -> Path:
"""Return default keyring."""
raise NotImplementedError
@@ -105,8 +106,8 @@ class DebianDefaults(Defaults):
def get_distribution(self) -> list[str]:
return [distro_info.DebianDistroInfo().devel()]
- def get_keyring(self) -> str:
- return "/usr/share/keyrings/debian-archive-keyring.gpg"
+ def get_keyring(self) -> Path:
+ return Path("/usr/share/keyrings/debian-archive-keyring.gpg")
class UbuntuDefaults(Defaults):
@@ -119,8 +120,8 @@ class UbuntuDefaults(Defaults):
def get_distribution(self) -> list[str]:
return [distro_info.UbuntuDistroInfo().devel()]
- def get_keyring(self) -> str:
- return "/usr/share/keyrings/ubuntu-archive-keyring.gpg"
+ def get_keyring(self) -> Path:
+ return Path("/usr/share/keyrings/ubuntu-archive-keyring.gpg")
class DefaultsFactory:
@@ -147,7 +148,7 @@ class Settings:
def __init__(self) -> None:
self.defaults: str | None = None
- self.tmpdir: str | None = None
+ self.tmpdir: Path | None = None
self.keep_env = False
self.shell_on_error = False
self.max_command_output_size = (
@@ -165,7 +166,7 @@ class Settings:
self.debian_distros: list[str] = []
self.bootstrapcmd: list[str] = []
self.keep_sources_list = False
- self.keyring: str | None = None
+ self.keyring: Path | None = None
self.do_not_verify_signatures = False
self.no_check_valid_until = False
self.install_recommends = False
@@ -173,21 +174,21 @@ class Settings:
self.eatmydata = True
self.dpkg_force_unsafe_io = True
self.dpkg_force_confdef = False
- self.scriptsdirs: list[str] = []
+ self.scriptsdirs: list[Path] = []
self.bindmounts: list[str] = []
self.allow_database = False
self.update_retries = None
# chroot setup
self.arch = None
- self.basetgz = None
- self.savetgz = None
+ self.basetgz: Path | None = None
+ self.savetgz: Path | None = None
self.lvm_volume = None
self.lvm_snapshot_size = "1G"
- self.existing_chroot = None
+ self.existing_chroot: Path | None = None
self.hard_link = False
self.schroot = None
- self.end_meta = None
- self.save_end_meta = None
+ self.end_meta: Path | None = None
+ self.save_end_meta: Path | None = None
self.skip_minimize = True
self.minimize = False
self.debfoster_options: list[str] | None = None
@@ -620,40 +621,39 @@ def run(command: list[str], ignore_errors: bool = False, timeout: int = 0) -> tu
return p.returncode, output
-def create_temp_file() -> tuple[int, str]:
+def create_temp_file() -> tuple[int, Path]:
"""Create a temporary file and return its full path."""
fd, path = tempfile.mkstemp(dir=settings.tmpdir)
logging.debug("Created temporary file %s" % path)
- return (fd, path)
+ return (fd, Path(path))
-def create_file(filename: str, contents: str) -> None:
+def create_file(filename: Path, contents: str) -> None:
"""Create a new file with the desired name and contents."""
try:
- with open(filename, "w") as f:
- f.write(contents)
+ filename.write_text(contents)
except IOError as detail:
logging.error("Couldn't create file %s: %s" % (filename, detail))
panic()
-def readlines_file(filename: str) -> list[str]:
- with open(filename, "r") as f:
+def readlines_file(filename: Path) -> list[str]:
+ with filename.open("r") as f:
return f.readlines()
-def remove_files(filenames: list[str]) -> None:
+def remove_files(filenames: collections.abc.Iterable[Path]) -> None:
"""Remove some files."""
for filename in filenames:
logging.debug("Removing %s" % filename)
try:
- os.remove(filename)
+ filename.unlink()
except OSError as detail:
logging.error("Couldn't remove %s: %s" % (filename, detail))
panic()
-def make_metapackage(name: str, depends: str, conflicts: str, arch: str = "all") -> str:
+def make_metapackage(name: str, depends: str, conflicts: str, arch: str = "all") -> Path:
"""Return the path to a .deb created just for satisfying dependencies
Caller is responsible for removing the temporary directory containing the
@@ -661,11 +661,11 @@ def make_metapackage(name: str, depends: str, conflicts: str, arch: str = "all")
"""
# Inspired by pbuilder's pbuilder-satisfydepends-aptitude
- tmpdir = tempfile.mkdtemp(dir=settings.tmpdir)
+ tmpdir = Path(tempfile.mkdtemp(dir=settings.tmpdir))
panic_handler_id = do_on_panic(lambda: shutil.rmtree(tmpdir))
- create_file(os.path.join(tmpdir, ".piuparts.tmpdir"), "metapackage creation")
+ create_file(tmpdir / ".piuparts.tmpdir", "metapackage creation")
old_umask = os.umask(0)
- os.makedirs(os.path.join(tmpdir, name, "DEBIAN"), mode=0o755)
+ (tmpdir / name / "DEBIAN").mkdir(mode=0o755, parents=True)
os.umask(old_umask)
control = deb822.Deb822()
control["Package"] = name
@@ -684,39 +684,23 @@ def make_metapackage(name: str, depends: str, conflicts: str, arch: str = "all")
"piuparts and can safely be removed"
)
- create_file(os.path.join(tmpdir, name, "DEBIAN", "control"), control.dump())
+ create_file(tmpdir / name / "DEBIAN" / "control", control.dump())
logging.debug("metapackage:\n" + indent_string(control.dump()))
- run(["dpkg-deb", "-b", "-Zgzip", "--nocheck", os.path.join(tmpdir, name)])
+ run(["dpkg-deb", "-b", "-Zgzip", "--nocheck", os.fspath(tmpdir / name)])
dont_do_on_panic(panic_handler_id)
- return os.path.join(tmpdir, name + ".deb")
-
-
-def split_path(pathname: str) -> list[str]:
- parts = []
- while pathname:
- head, tail = os.path.split(pathname)
- # print("split '%s' => '%s' + '%s'" % (pathname, head, tail))
- if tail:
- parts.append(tail)
- elif not head:
- break
- elif head == pathname:
- parts.append(head)
- break
- pathname = head
- return parts
+ return tmpdir / (name + ".deb")
@typing.overload
-def canonicalize_path(root: str, pathname: str, report_links: typing.Literal[False] = False) -> str: ...
+def canonicalize_path(root: Path, pathname: PurePath, report_links: typing.Literal[False] = False) -> PurePath: ...
@typing.overload
-def canonicalize_path(root: str, pathname: str, report_links: typing.Literal[True]) -> list[tuple[str, str]]: ...
+def canonicalize_path(root: Path, pathname: PurePath, report_links: typing.Literal[True]) -> list[tuple[str, str]]: ...
-def canonicalize_path(root: str, pathname: str, report_links: bool = False) -> str | list[tuple[str, str]]:
+def canonicalize_path(root: Path, pathname: PurePath, report_links: bool = False) -> PurePath | list[tuple[str, str]]:
"""Canonicalize a path name, simulating chroot at 'root'.
When resolving the symlink, pretend (similar to chroot) that
@@ -732,7 +716,7 @@ def canonicalize_path(root: str, pathname: str, report_links: bool = False) -> s
# print("\nCANONICALIZE %s %s" % (root, pathname))
links = []
seen: list[str] = []
- parts = split_path(pathname)
+ parts = list(reversed(pathname.parts))
# print("PARTS ", list(reversed(parts)))
path = "/"
while parts:
@@ -751,36 +735,42 @@ def canonicalize_path(root: str, pathname: str, report_links: bool = False) -> s
# meaning of 'path' because it contains no symlinks - they have been
# resolved already.
newpath = os.path.normpath(os.path.join(path, part))
- rootedpath = os.path.join(root, newpath[1:])
+ rootedpath = root / newpath[1:]
if newpath == "/":
path = "/"
- elif os.path.islink(rootedpath):
- target = os.readlink(rootedpath)
+ elif rootedpath.is_symlink():
+ target = rootedpath.readlink()
# print("LINK to '%s'" % target)
- links.append((newpath, target))
- if os.path.isabs(target):
+ links.append((newpath, os.fspath(target)))
+ if target.is_absolute():
path = "/"
- parts.extend(split_path(target))
+ parts.extend(reversed(target.parts))
else:
path = newpath
# print("FINAL '%s'" % path)
if report_links:
return links
- return path
+ return PurePath(path)
-def is_broken_symlink(root: str, dirpath: str, filename: str) -> bool:
+def is_broken_symlink(root: Path, dirpath: Path, filename: str) -> bool:
"""Is symlink dirpath+filename broken?"""
- if dirpath[: len(root)] == root:
- dirpath = dirpath[len(root) :]
- pathname = canonicalize_path(root, os.path.join(dirpath, filename))
- pathname = os.path.join(root, pathname[1:])
+ dirpath_inside = os.fspath(dirpath).removeprefix(os.fspath(root))
+ pathname = canonicalize_path(root, PurePath(dirpath_inside, filename))
# The symlink chain, if any, has now been resolved. Does the target
# exist?
# print("EXISTS ", pathname, os.path.exists(pathname))
- return not os.path.exists(pathname)
+ return not (root / os.fspath(pathname)[1:]).exists()
+
+
+def apt_path(filename: PurePath) -> str:
+ """Turn a path into a filename accepted by apt as a filename. The output
+ must start with a slash or a dot.
+ """
+ result = os.fspath(filename)
+ return result if result.startswith("/") else "./" + result
FileInfo = namedtuple("FileInfo", ["st", "target", "user", "group"])
@@ -798,21 +788,21 @@ class Chroot:
"""A chroot for testing things in."""
def __init__(self) -> None:
- self.name: str | None = None
+ self.name: Path | None = None
self.bootstrapped = False
- self.mounts: list[str] = []
+ self.mounts: list[Path] = []
self.initial_selections: dict[str, tuple[str, str | None]] | None = None
self.avail_md5_history: list[str] = []
self.systemd_tmpfiles: dict[str, tuple[str, str]] = {}
def create_temp_dir(self) -> None:
"""Create a temporary directory for the chroot."""
- self.name = tempfile.mkdtemp(dir=settings.tmpdir)
- create_file(os.path.join(self.name, ".piuparts.tmpdir"), "chroot")
- os.chmod(self.name, 0o755)
+ self.name = Path(tempfile.mkdtemp(dir=settings.tmpdir))
+ create_file(self.name / ".piuparts.tmpdir", "chroot")
+ self.name.chmod(0o755)
logging.debug("Created temporary directory %s" % self.name)
- def create(self, temp_tgz: str | None = None) -> None:
+ def create(self, temp_tgz: Path | None = None) -> None:
"""Create a chroot according to user's wishes."""
self.panic_handler_id = do_on_panic(self.remove)
if not settings.schroot and not settings.docker_image:
@@ -840,17 +830,17 @@ class Chroot:
# Copy scripts dirs into the chroot, merging all dirs together,
# later files overwriting earlier ones.
if settings.scriptsdirs:
- self.mkdir_p("tmp/scripts/")
- dest = self.relative("tmp/scripts/")
+ self.mkdir_p(PurePath("tmp/scripts/"))
+ dest = self.relative(PurePath("tmp/scripts/"))
for sdir in settings.scriptsdirs:
logging.debug("Copying scriptsdir %s to %s" % (sdir, dest))
- for sfile in os.listdir(sdir):
+ for sfile in sdir.iterdir():
if (
- (sfile.startswith("post_") or sfile.startswith("pre_") or sfile.startswith("is_testable_"))
- and ".dpkg-" not in sfile
- and os.path.isfile(os.path.join(sdir, sfile))
+ sfile.name.startswith(("post_", "pre_", "is_testable_"))
+ and ".dpkg-" not in sfile.name
+ and sfile.is_file()
):
- shutil.copy(os.path.join(sdir, sfile), dest)
+ shutil.copy(sfile, dest)
# Run custom scripts after chroot has been unpacked/debootstrapped
# Useful for adjusting apt configuration e.g. for internal mirror usage
@@ -876,12 +866,12 @@ class Chroot:
if not self.name:
return
- if not settings.keep_env and os.path.exists(self.name):
+ if not settings.keep_env and self.name.exists():
self.terminate_running_processes()
self.unmount_all()
if settings.lvm_volume:
logging.debug("Unmounting and removing LVM snapshot %s" % self.lvm_snapshot_name)
- run(["umount", self.name])
+ run(["umount", os.fspath(self.name)])
run(["lvremove", "-f", self.lvm_snapshot])
if settings.schroot:
logging.debug("Terminate schroot session '%s'" % self.name)
@@ -890,9 +880,9 @@ class Chroot:
logging.debug("Destroy docker container '%s'" % self.docker_container)
run(["docker", "rm", "-f", self.docker_container])
if not settings.schroot and not settings.docker_image:
- run(["rm", "-rf", "--one-file-system", self.name])
- if os.path.exists(self.name):
- create_file(os.path.join(self.name, ".piuparts.tmpdir"), "removal failed")
+ run(["rm", "-rf", "--one-file-system", os.fspath(self.name)])
+ if self.name.exists():
+ create_file(self.name / ".piuparts.tmpdir", "removal failed")
logging.debug("Removed directory tree at %s" % self.name)
elif settings.keep_env:
if settings.schroot:
@@ -906,7 +896,7 @@ class Chroot:
def was_bootstrapped(self) -> bool:
return self.bootstrapped
- def create_temp_tgz_file(self) -> str:
+ def create_temp_tgz_file(self) -> Path:
"""Return the path to a file to be used as a temporary tgz file"""
# Yes, create_temp_file() would work just as well, but putting it in
# the interface for Chroot allows the VirtServ hack to work.
@@ -914,21 +904,22 @@ class Chroot:
os.close(fd)
return temp_tgz
- def remove_temp_tgz_file(self, temp_tgz: str) -> None:
+ def remove_temp_tgz_file(self, temp_tgz: Path) -> None:
"""Remove the file that was used as a temporary tgz file"""
# Yes, remove_files() would work just as well, but putting it in
# the interface for Chroot allows the VirtServ hack to work.
remove_files([temp_tgz])
- def pack_into_tgz(self, result: str) -> None:
+ def pack_into_tgz(self, result: Path) -> None:
"""Tar and compress all files in the chroot."""
assert self.name is not None
self.run(["apt-get", "clean"])
logging.debug("Saving %s to %s." % (self.name, result))
- fd, tmpfile = tempfile.mkstemp(dir=os.path.dirname(result))
+ fd, tmpfile = tempfile.mkstemp(dir=result.parent)
os.close(fd)
- panic_handler_id = do_on_panic(lambda: os.remove(tmpfile))
+ tmpfilepath = Path(tmpfile)
+ panic_handler_id = do_on_panic(tmpfilepath.unlink)
run(
[
@@ -940,29 +931,29 @@ class Chroot:
"--exclude",
"tmp/scripts",
"-C",
- self.name,
+ os.fspath(self.name),
"./",
]
)
- os.chmod(tmpfile, 0o644)
- os.rename(tmpfile, result)
+ tmpfilepath.chmod(0o644)
+ tmpfilepath.rename(result)
dont_do_on_panic(panic_handler_id)
- def unpack_from_tgz(self, tarball: str) -> None:
+ def unpack_from_tgz(self, tarball: Path) -> None:
"""Unpack a tarball to a chroot."""
assert self.name is not None
logging.debug("Unpacking %s into %s" % (tarball, self.name))
prefix = []
- if settings.eatmydata and os.path.isfile("/usr/bin/eatmydata"):
+ if settings.eatmydata and Path("/usr/bin/eatmydata").is_file():
prefix.append("eatmydata")
- run(prefix + ["tar", "-C", self.name, "--auto-compress", "-xf", tarball])
+ run(prefix + ["tar", "-C", os.fspath(self.name), "--auto-compress", "-xf", os.fspath(tarball)])
def setup_from_schroot(self, schroot: str) -> None:
self.schroot_session = schroot.split(":", 1)[-1] + "-" + str(uuid.uuid1()) + "-piuparts"
run(["schroot", "--begin-session", "--chroot", schroot, "--session-name", self.schroot_session])
ret_code, output = run(["schroot", "--chroot", "session:" + self.schroot_session, "--location"])
- self.name = output.strip()
+ self.name = Path(output.strip())
logging.info("New schroot session in '%s'" % self.name)
@staticmethod
@@ -975,7 +966,7 @@ class Chroot:
def setup_from_docker(self, docker_image: str) -> None:
self.check_if_docker_storage_driver_is_supported()
with tempfile.TemporaryDirectory() as tmpdir:
- cidfile = pathlib.Path(tmpdir) / "cidfile"
+ cidfile = Path(tmpdir) / "cidfile"
ret_code, output = run(["docker", "run", "-d", "-it", "--cidfile", str(cidfile), docker_image, "bash"])
if ret_code != 0:
logging.error("Couldn't start the container from '%s'" % docker_image)
@@ -984,7 +975,7 @@ class Chroot:
self.docker_container = cidfile.read_text().strip()
ret_code, output = run(["docker", "inspect", "-f", "{{ .GraphDriver.Data.MergedDir }}", self.docker_container])
- self.name = output.strip()
+ self.name = Path(output.strip())
logging.info("New container created %r at %r", self.docker_container, self.name)
def setup_from_lvm(self, lvm_volume: str) -> None:
@@ -998,38 +989,36 @@ class Chroot:
logging.debug("Creating LVM snapshot %s from %s" % (self.lvm_snapshot, lvm_volume))
run(["lvcreate", "-n", self.lvm_snapshot, "-s", lvm_volume, "-L", settings.lvm_snapshot_size])
logging.info("Mounting LVM snapshot to %s" % self.name)
- run(["mount", self.lvm_snapshot, self.name])
+ run(["mount", self.lvm_snapshot, os.fspath(self.name)])
- def setup_from_dir(self, dirname: str) -> None:
+ def setup_from_dir(self, dirname: Path) -> None:
"""Create chroot from an existing one."""
assert self.name is not None
# if on same device, make hard link
cmd = ["cp"]
- if settings.hard_link and os.stat(dirname).st_dev == os.stat(self.name).st_dev:
+ if settings.hard_link and dirname.stat().st_dev == self.name.stat().st_dev:
cmd += ["-al"]
logging.debug("Hard linking %s to %s" % (dirname, self.name))
else:
cmd += ["-ax"]
logging.debug("Copying %s into %s" % (dirname, self.name))
- for name in os.listdir(dirname):
- src = os.path.join(dirname, name)
- dst = os.path.join(self.name, name)
- run(cmd + [src, dst])
+ for entry in dirname.iterdir():
+ run(cmd + [os.fspath(entry), os.fspath(self.name / entry.name)])
def interactive_shell(self) -> None:
assert self.name is not None
logging.info("Entering interactive shell in %s" % self.name)
env = os.environ.copy()
- env["debian_chroot"] = "piuparts:%s" % self.name
+ env["debian_chroot"] = "piuparts:%s" % os.fspath(self.name)
try:
- subprocess.call(["chroot", self.name, "bash", "-l"], env=env)
+ subprocess.call(["chroot", os.fspath(self.name), "bash", "-l"], env=env)
except Exception:
pass
def run(self, command: list[str], ignore_errors: bool = False) -> tuple[int, str]:
assert self.name is not None
prefix = []
- if settings.eatmydata and os.path.isfile(os.path.join(self.name, "usr/bin/eatmydata")):
+ if settings.eatmydata and (self.name / "usr/bin/eatmydata").is_file():
prefix.append("eatmydata")
if settings.schroot:
return run(
@@ -1062,15 +1051,13 @@ class Chroot:
)
else:
return run(
- ["chroot", self.name] + prefix + command,
+ ["chroot", os.fspath(self.name)] + prefix + command,
ignore_errors=ignore_errors,
timeout=settings.max_command_runtime,
)
- def mkdir_p(self, path: str) -> None:
- fullpath = self.relative(path)
- if not os.path.isdir(fullpath):
- os.makedirs(fullpath)
+ def mkdir_p(self, path: PurePath) -> None:
+ self.relative(path).mkdir(exist_ok=True, parents=True)
def create_apt_sources(self, distro: str) -> None:
"""Create an /etc/apt/sources.list with a given distro."""
@@ -1081,7 +1068,7 @@ class Chroot:
lines.append("deb %s %s %s" % (mirror, distro, " ".join(components)))
for repo in settings.extra_repos:
lines.append(repo)
- create_file(self.relative("etc/apt/sources.list"), "\n".join(lines) + "\n")
+ create_file(self.relative(PurePath("etc/apt/sources.list")), "\n".join(lines) + "\n")
logging.debug("sources.list:\n" + indent_string("\n".join(lines)))
def aptupdate_run(self) -> int | None:
@@ -1155,7 +1142,7 @@ class Chroot:
if settings.dpkg_force_confdef:
lines.append('Dpkg::Options {"--force-confdef";};\n')
- create_file(self.relative("etc/apt/apt.conf.d/piuparts"), "".join(lines))
+ create_file(self.relative(PurePath("etc/apt/apt.conf.d/piuparts")), "".join(lines))
def create_dpkg_conf(self) -> None:
"""Create /etc/dpkg/dpkg.cfg.d/piuparts inside the chroot."""
@@ -1169,12 +1156,12 @@ class Chroot:
"This will hide problems, see #466118."
)
if lines:
- self.mkdir_p("etc/dpkg/dpkg.cfg.d")
- create_file(self.relative("etc/dpkg/dpkg.cfg.d/piuparts"), "".join(lines))
+ self.mkdir_p(PurePath("etc/dpkg/dpkg.cfg.d"))
+ create_file(self.relative(PurePath("etc/dpkg/dpkg.cfg.d/piuparts")), "".join(lines))
def create_policy_rc_d(self) -> None:
"""Create a policy-rc.d that prevents daemons from running."""
- full_name = self.relative("usr/sbin/policy-rc.d")
+ full_name = self.relative(PurePath("usr/sbin/policy-rc.d"))
policy = "#!/bin/sh\n"
if settings.allow_database:
policy += 'test "$1" = "mariadb" && exit 0\n'
@@ -1186,7 +1173,7 @@ class Chroot:
policy += 'test "$1" = "firebird4.0" && exit 0\n'
policy += "exit 101\n"
create_file(full_name, policy)
- os.chmod(full_name, 0o755)
+ full_name.chmod(0o755)
logging.debug("Created policy-rc.d and chmodded it.")
def create_resolv_conf(self) -> None:
@@ -1194,14 +1181,14 @@ class Chroot:
if settings.docker_image:
# Docker takes care of this
return
- full_name = self.relative("etc/resolv.conf")
+ full_name = self.relative(PurePath("etc/resolv.conf"))
resolvconf = ""
with open("/etc/resolv.conf", "r") as f:
for line in f:
if line.strip() and not line.startswith(("#", ";")):
resolvconf += line.strip() + "\n"
create_file(full_name, resolvconf)
- os.chmod(full_name, 0o644)
+ full_name.chmod(0o644)
logging.debug("Created resolv.conf.")
def setup_minimal_chroot(self) -> None:
@@ -1210,7 +1197,7 @@ class Chroot:
assert settings.distro_config is not None
logging.debug("Setting up minimal chroot for %s at %s." % (settings.debian_distros[0], self.name))
prefix = []
- if settings.eatmydata and os.path.isfile("/usr/bin/eatmydata"):
+ if settings.eatmydata and Path("/usr/bin/eatmydata").is_file():
prefix.append("eatmydata")
options = []
if settings.do_not_verify_signatures:
@@ -1230,7 +1217,11 @@ class Chroot:
prefix
+ settings.bootstrapcmd
+ options
- + [settings.debian_distros[0], self.name, settings.distro_config.get_mirror(settings.debian_distros[0])]
+ + [
+ settings.debian_distros[0],
+ os.fspath(self.name),
+ settings.distro_config.get_mirror(settings.debian_distros[0]),
+ ]
)
self.bootstrapped = True
@@ -1244,7 +1235,7 @@ class Chroot:
if settings.eatmydata:
debfoster_command.append("eatmydata")
self.run(debfoster_command)
- remove_files([self.relative("var/lib/debfoster/keepers")])
+ remove_files([self.relative(PurePath("var/lib/debfoster/keepers"))])
self.run(["dpkg", "--purge", "debfoster"])
def configure_chroot(self) -> None:
@@ -1271,18 +1262,18 @@ class Chroot:
"tty": os.makedev(5, 0),
}
for devname, devnum in chardevices.items():
+ inner_path = self.name / "dev" / devname
devname = "/dev/" + devname
- inner_path = self.name + devname
isdevice = False
try:
- isdevice = stat.S_ISCHR(os.stat(inner_path).st_mode)
+ isdevice = stat.S_ISCHR(inner_path.stat().st_mode)
except FileNotFoundError:
# Try creating missing devices. If that fails with -EPERM, we
# likely are in an unprivileged namespace and resort to bind
# mounting them individually.
try:
os.mknod(inner_path, stat.S_IFCHR | 0o666, devnum)
- os.chmod(inner_path, 0o666) # Override umask
+ inner_path.chmod(0o666) # Override umask
isdevice = True
except OSError as err:
if err.errno != errno.EPERM:
@@ -1291,7 +1282,7 @@ class Chroot:
os.mknod(inner_path, stat.S_IFREG)
if isdevice:
try:
- open(inner_path, "rb").close()
+ inner_path.touch()
except PermissionError:
# Could not open the device, the filesystem is probably nodev.
isdevice = False
@@ -1302,7 +1293,7 @@ class Chroot:
else:
raise
if not isdevice:
- self.mount(devname, devname, opts=["bind"])
+ self.bind_mount(Path(devname), PurePath(devname))
symlinks = {
"fd": "/proc/self/fd",
@@ -1311,14 +1302,14 @@ class Chroot:
"stderr": "/proc/self/fd/2",
}
for linkname, linktarget in symlinks.items():
- linkname = self.name + "/dev/" + linkname
+ linkpath = self.name / "dev" / linkname
try:
- os.lstat(linkname)
+ linkpath.lstat()
except FileNotFoundError:
- os.symlink(linktarget, linkname)
+ linkpath.symlink_to(linktarget)
for bindmount in settings.bindmounts:
- self.mount(bindmount, bindmount, opts=["rbind"])
+ self.bind_mount(Path(bindmount), PurePath(bindmount), recursive=True)
def remember_available_md5(self) -> None:
"""Keep a history of 'apt-cache dumpavail | md5sum' after initial
@@ -1362,7 +1353,7 @@ class Chroot:
continue
if line.startswith("#"):
if line.startswith("# /"):
- if os.path.exists(self.relative(line[3:])):
+ if self.relative(PurePath(line[3:])).exists():
current_file = line[2:]
continue
flag, filename, *_ = line.split()
@@ -1418,16 +1409,16 @@ class Chroot:
logging.info("the following packages are not in the archive: " + ", ".join(new_packages))
return known_packages
- def copy_files(self, source_names: list[str], target_name: str) -> None:
+ def copy_files(self, source_names: list[Path], target_name: PurePath) -> None:
"""Copy files in 'source_name' to file/dir 'target_name', relative
to the root of the chroot."""
- target_name = self.relative(target_name)
- logging.debug("Copying %s to %s" % (", ".join(source_names), target_name))
+ target_path = self.relative(target_name)
+ logging.debug("Copying %s to %s" % (", ".join(map(str, source_names)), target_path))
for source_name in source_names:
try:
- shutil.copy(source_name, target_name)
+ shutil.copy(source_name, target_path)
except IOError as detail:
- logging.error("Error copying %s to %s: %s" % (source_name, target_name, detail))
+ logging.error("Error copying %s to %s: %s" % (source_name, target_path, detail))
panic()
def list_installed_files(self, pre_info: dict[str, FileInfo], post_info: dict[str, FileInfo]) -> None:
@@ -1464,7 +1455,7 @@ class Chroot:
return installed
def install_packages(
- self, package_files: list[str], packages: list[str], with_scripts: bool = True, reinstall: bool = False
+ self, package_files: list[Path], packages: list[str], with_scripts: bool = True, reinstall: bool = False
) -> None:
if package_files:
self.install_package_files(package_files, packages, with_scripts=with_scripts)
@@ -1472,7 +1463,7 @@ class Chroot:
self.install_packages_by_name(packages, with_scripts=with_scripts, reinstall=reinstall)
def install_package_files(
- self, package_files: list[str], packages: list[str] | None = None, with_scripts: bool = False
+ self, package_files: list[Path], packages: list[str] | None = None, with_scripts: bool = False
) -> None:
if packages and settings.testdebs_repo:
self.install_packages_by_name(packages, with_scripts=with_scripts)
@@ -1490,8 +1481,8 @@ class Chroot:
# This must look like a local path so that apt-get can
# distinguish it from a 'package/suite' request.
- self.copy_files(package_files, "tmp")
- tmp_files = [os.path.join("./tmp", os.path.basename(a)) for a in package_files]
+ self.copy_files(package_files, PurePath("tmp"))
+ tmp_files = [PurePath("tmp", a.name) for a in package_files]
if with_scripts:
self.run_scripts("pre_install")
@@ -1511,9 +1502,9 @@ class Chroot:
pre_info = self.get_tree_meta_data()
if apt_can_install_debs:
- self.run(apt_get_install + tmp_files)
+ self.run([*apt_get_install, *map(apt_path, tmp_files)])
else:
- ret, out = self.run(["dpkg", "-i"] + tmp_files, ignore_errors=True)
+ ret, out = self.run(["dpkg", "-i", *map(os.fspath, tmp_files)], ignore_errors=True)
if ret != 0:
if "dependency problems - leaving unconfigured" in out:
pass
@@ -1538,7 +1529,7 @@ class Chroot:
if with_scripts:
self.run_scripts("post_install")
- remove_files([self.relative(name) for name in tmp_files])
+ remove_files(map(self.relative, tmp_files))
def install_packages_by_name(self, packages: list[str], with_scripts: bool = True, reinstall: bool = False) -> None:
if packages:
@@ -1616,11 +1607,13 @@ class Chroot:
def check_debsums(self) -> None:
assert self.name is not None
- status, output = run(["debsums", "--root", self.name, "-ac", "--ignore-obsolete"], ignore_errors=True)
+ status, output = run(
+ ["debsums", "--root", os.fspath(self.name), "-ac", "--ignore-obsolete"], ignore_errors=True
+ )
if status != 0:
logging.error(
"FAIL: debsums reports modifications inside the chroot:\n%s"
- % indent_string(output.replace(self.name, ""))
+ % indent_string(output.replace(os.fspath(self.name), ""))
)
if not settings.warn_on_debsums_errors:
panic()
@@ -1629,7 +1622,7 @@ class Chroot:
"""Run adequate and categorize output according to our needs."""
assert self.name is not None
packages = unqualify([p for p in packages if not p.endswith("=None")])
- if packages and settings.adequate and os.path.isfile("/usr/bin/adequate"):
+ if packages and settings.adequate and Path("/usr/bin/adequate").is_file():
status, output = run(["dpkg-query", "-f", "${Version}\n", "-W", "adequate"], ignore_errors=True)
logging.info("Running adequate version %s now." % output.strip())
adequate_tags = [
@@ -1654,7 +1647,7 @@ class Chroot:
"broken-symlink",
]
ignored_tags: list[str] = []
- status, output = run(["adequate", "--root", self.name] + packages, ignore_errors=True)
+ status, output = run(["adequate", "--root", os.fspath(self.name)] + packages, ignore_errors=True)
for tag in ignored_tags:
# ignore some tags
_regex = "^[^:]+: " + tag + " .*\n"
@@ -1674,7 +1667,7 @@ class Chroot:
error_code = "FAIL"
logging.error(
"%s: Inadequate results from running adequate!\n%s"
- % (error_code, indent_string(output.replace(self.name, "")))
+ % (error_code, indent_string(output.replace(os.fspath(self.name), "")))
)
if inadequate_results:
logging.error(
@@ -1699,8 +1692,8 @@ class Chroot:
overwrites = False
usrmerge = set()
for f in sorted(file_owners.keys()):
- dn, fn = os.path.split(f)
- dc = canonicalize_path(self.name, dn)
+ dn = os.fspath(f.parent)
+ dc = os.fspath(canonicalize_path(self.name, f.parent))
if dn != dc:
# Allow the /usr merge to have taken place. For example, if
# f (the file recorded in the dpkg database) is /bin/cat,
@@ -1716,7 +1709,7 @@ class Chroot:
continue
- fc = os.path.join(dc, fn)
+ fc = PurePath(dc, f.name)
of = ", ".join(file_owners[f])
if fc in file_owners:
overwrites = True
@@ -1724,7 +1717,7 @@ class Chroot:
else:
ofc = "?"
bad.append("%s (%s) != %s (%s)" % (f, of, fc, ofc))
- for link, target in canonicalize_path(self.name, dn, report_links=True):
+ for link, target in canonicalize_path(self.name, f.parent, report_links=True):
bad.append(" %s -> %s" % (link, target))
if bad:
if overwrites:
@@ -1840,27 +1833,27 @@ class Chroot:
"""Return the filesystem meta data for all objects in the chroot."""
self.run(["apt-get", "clean"])
logging.debug("Recording chroot state")
- root = self.relative(".")
+ root = self.relative(PurePath("."))
uidmap = {}
- with open(self.relative("etc/passwd"), "r") as passwd:
+ with self.relative(PurePath("etc/passwd")).open("r") as passwd:
for line in passwd:
usr, x, uid = line.split(":")[0:3]
uidmap[int(uid)] = usr
gidmap = {}
- with open(self.relative("etc/group"), "r") as groupfile:
+ with self.relative(PurePath("etc/group")).open("r") as groupfile:
for line in groupfile:
grp, x, gid = line.split(":")[0:3]
gidmap[int(gid)] = grp
vdict = {}
- proc = os.path.join(root, "proc")
- devpts = os.path.join(root, "dev/pts")
- for dirpath, dirnames, filenames in os.walk(root):
- assert dirpath[: len(root)] == root
- if dirpath[: len(proc) + 1] in [proc, proc + "/"]:
+ proc = os.fspath(root / "proc")
+ devpts = os.fspath(root / "dev/pts")
+ for dirpath, dirnames, filenames in root.walk():
+ assert os.fspath(dirpath).startswith(os.fspath(root))
+ if os.fspath(dirpath)[: len(proc) + 1] in [proc, proc + "/"]:
continue
- if dirpath[: len(devpts) + 1] in [devpts, devpts + "/"]:
+ if os.fspath(dirpath)[: len(devpts) + 1] in [devpts, devpts + "/"]:
continue
- for name in [dirpath] + [os.path.join(dirpath, f) for f in filenames]:
+ for name in [os.fspath(dirpath)] + [os.fspath(dirpath / f) for f in filenames]:
st = os.lstat(name)
if stat.S_ISLNK(st.st_mode):
target = os.readlink(name)
@@ -1876,7 +1869,7 @@ class Chroot:
group = gidmap[st.st_gid]
else:
group = "#%d" % st.st_gid
- vdict[name[len(root) :]] = FileInfo(st, target, user, group)
+ vdict[name[len(os.fspath(root)) :]] = FileInfo(st, target, user, group)
return vdict
def get_state_meta_data(self) -> StateMetaData:
@@ -1888,21 +1881,19 @@ class Chroot:
"diversions": self.get_diversions(),
}
- def relative(self, pathname: str) -> str:
+ def relative(self, pathname: PurePath) -> Path:
assert self.name is not None
- if pathname.startswith("/"):
- return os.path.join(self.name, pathname[1:])
- return os.path.join(self.name, pathname)
+ return self.name / os.fspath(pathname).removeprefix("/")
- def get_files_owned_by_packages(self) -> dict[str, list[str]]:
+ def get_files_owned_by_packages(self) -> dict[PurePath, list[str]]:
"""Return dict[filename] = [packagenamelist]."""
- vdir = self.relative("var/lib/dpkg/info")
- vdict: dict[str, list[str]] = {}
- for basename in os.listdir(vdir):
- if basename.endswith(".list"):
- pkg = basename[: -len(".list")]
- for line in readlines_file(os.path.join(vdir, basename)):
- pathname = line.strip()
+ vdir = self.relative(PurePath("var/lib/dpkg/info"))
+ vdict: dict[PurePath, list[str]] = {}
+ for entry in vdir.iterdir():
+ if entry.name.endswith(".list"):
+ pkg = entry.name[: -len(".list")]
+ for line in readlines_file(entry):
+ pathname = PurePath(line.strip())
if pathname in vdict:
vdict[pathname].append(pkg)
else:
@@ -1916,7 +1907,7 @@ class Chroot:
count = len(output.strip().split("\n")) - 2 # header + bash launched on container creation
else:
assert self.name is not None
- status, output = run(["lsof", "-w", "+D", self.name], ignore_errors=True)
+ status, output = run(["lsof", "-w", "+D", os.fspath(self.name)], ignore_errors=True)
count = len(output.split("\n")) - 1
if count > 0:
if fail is None:
@@ -1937,7 +1928,10 @@ class Chroot:
seen = []
while True:
p = subprocess.Popen(
- ["lsof", "-t", "+D", self.name], stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True
+ ["lsof", "-t", "+D", os.fspath(self.name)],
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ universal_newlines=True,
)
stdout, _ = p.communicate()
if not stdout:
@@ -1962,33 +1956,53 @@ class Chroot:
# If /selinux is present, assume that this is the only supported
# location by libselinux. Otherwise use the new location.
# /selinux was shipped by the libselinux package until wheezy.
- def selinuxfs_path(self) -> str:
- if os.path.isdir(self.relative("/selinux")):
- return "/selinux"
+ def selinuxfs_path(self) -> PurePath:
+ if self.relative(PurePath("/selinux")).is_dir():
+ return PurePath("/selinux")
else:
- return "/sys/fs/selinux"
+ return PurePath("/sys/fs/selinux")
+
+ def bind_mount(
+ self, source: Path, path: PurePath, *, recursive: bool = False, readonly: bool = False, no_mkdir: bool = False
+ ) -> None:
+ """Bind mount something into the chroot and remember it for unmount_all()."""
+ assert self.name is not None
+ path = canonicalize_path(self.name, path)
+ fullpath = self.relative(path)
+ if not no_mkdir:
+ if source.is_dir():
+ self.mkdir_p(path)
+ elif not fullpath.exists():
+ self.mkdir_p(path.parent)
+ os.mknod(fullpath, stat.S_IFREG)
+ opts = ["rbind" if recursive else "bind"]
+ if readonly:
+ opts.append("ro")
+ run(["mount", "-o", ",".join(opts), os.fspath(source), os.fspath(fullpath)])
+ self.mounts.append(fullpath)
def mount(
- self, source: str, path: str, fstype: str | None = None, opts: list[str] | None = None, no_mkdir: bool = False
+ self,
+ source: str,
+ path: PurePath,
+ fstype: str | None = None,
+ opts: list[str] | None = None,
+ no_mkdir: bool = False,
) -> None:
"""Mount something into the chroot and remember it for unmount_all()."""
assert self.name is not None
if opts is None:
opts = []
- path = canonicalize_path(self.name, path)
+ path = canonicalize_path(self.name, PurePath(path))
fullpath = self.relative(path)
if not no_mkdir:
- if not ("bind" in opts or "rbind" in opts) or os.path.isdir(source):
- self.mkdir_p(path)
- elif not os.path.exists(fullpath):
- self.mkdir_p(os.path.dirname(path))
- os.mknod(fullpath, stat.S_IFREG)
+ self.mkdir_p(path)
command = ["mount"]
if fstype is not None:
command.extend(["-t", fstype])
if opts:
command.extend(["-o", ",".join(opts)])
- command.extend([source, fullpath])
+ command.extend([source, os.fspath(fullpath)])
run(command)
self.mounts.append(fullpath)
@@ -1998,46 +2012,46 @@ class Chroot:
# Workaround to unmount /proc/sys/fs/binfmt_misc which is mounted by
# update-binfmts but never unmounted
# This workaround can be removed once #847788 is fixed
- binfmt_misc = self.relative("/proc/sys/fs/binfmt_misc")
- if os.path.ismount(binfmt_misc):
+ binfmt_misc = self.relative(PurePath("/proc/sys/fs/binfmt_misc"))
+ if binfmt_misc.is_mount():
self.mounts.append(binfmt_misc)
for mountpoint in reversed(self.mounts):
- run(["umount", mountpoint], ignore_errors=True)
+ run(["umount", os.fspath(mountpoint)], ignore_errors=True)
def mount_proc(self) -> None:
"""Mount /proc etc. inside chroot."""
- self.mount("proc", "/proc", fstype="proc")
- etcmtab = self.relative("etc/mtab")
- if not os.path.lexists(etcmtab):
- os.symlink("../proc/mounts", etcmtab)
+ self.mount("proc", PurePath("/proc"), fstype="proc")
+ etcmtab = self.relative(PurePath("etc/mtab"))
+ if not etcmtab.exists(follow_symlinks=False):
+ etcmtab.symlink_to("../proc/mounts")
self.mount(
"devpts",
- "/dev/pts",
+ PurePath("/dev/pts"),
fstype="devpts",
opts=["newinstance", "noexec", "nosuid", "gid=5", "mode=0620", "ptmxmode=0666"],
)
- dev_ptmx_rel_path = self.relative("dev/ptmx")
- if not os.path.islink(dev_ptmx_rel_path):
- if not os.path.exists(dev_ptmx_rel_path):
+ dev_ptmx_rel_path = self.relative(PurePath("dev/ptmx"))
+ if not dev_ptmx_rel_path.is_symlink():
+ if not dev_ptmx_rel_path.exists():
# /dev/pts/ptmx has been created by the previous self.mount("devpts", ...),
# we can symlink /dev/ptmx to it.
- os.symlink("pts/ptmx", dev_ptmx_rel_path)
+ dev_ptmx_rel_path.symlink_to("pts/ptmx")
else:
# /dev/ptmx is an unknown entity. Override it with a safe bind mount of /dev/pts/ptmx.
# Unlinking the existing /dev/ptmx could fail, this is safer.
- self.mount(self.relative("dev/pts/ptmx"), "/dev/ptmx", opts=["bind"], no_mkdir=True)
+ self.bind_mount(self.relative(PurePath("dev/pts/ptmx")), PurePath("/dev/ptmx"), no_mkdir=True)
p = subprocess.Popen(["tty"], stdout=subprocess.PIPE, universal_newlines=True)
stdout, _ = p.communicate()
- current_tty = stdout.strip()
- if p.returncode == 0 and os.path.exists(current_tty):
- dev_console = self.relative("/dev/console")
- if not os.path.exists(dev_console):
+ current_tty = Path(stdout.strip())
+ if p.returncode == 0 and current_tty.exists():
+ dev_console = self.relative(PurePath("/dev/console"))
+ if not dev_console.exists():
os.mknod(dev_console, 0o0600, os.makedev(5, 1))
- self.mount(current_tty, "/dev/console", opts=["bind"], no_mkdir=True)
- self.mount("tmpfs", "/dev/shm", fstype="tmpfs", opts=["size=65536k"])
+ self.bind_mount(current_tty, PurePath("/dev/console"), no_mkdir=True)
+ self.mount("tmpfs", PurePath("/dev/shm"), fstype="tmpfs", opts=["size=65536k"])
if selinux_enabled():
- self.mount("/sys/fs/selinux", self.selinuxfs_path(), opts=["bind", "ro"])
+ self.bind_mount(Path("/sys/fs/selinux"), self.selinuxfs_path(), readonly=True)
def is_ignored(self, pathname: str, info: str = "PATH", quiet: bool = False) -> bool:
"""Is a file (or dir or whatever) to be ignored?"""
@@ -2068,8 +2082,8 @@ class Chroot:
def check_files_moved_usr(
self,
packages: list[str] = [],
- files_before: dict[str, list[str]] = {},
- files_after: dict[str, list[str]] = {},
+ files_before: dict[PurePath, list[str]] = {},
+ files_after: dict[PurePath, list[str]] = {},
warn_only: bool = False,
) -> None:
"""Check that no files were moved from /{bin|sbin|lib*} and /usr/{bin|sbin|lib*}"""
@@ -2085,21 +2099,21 @@ class Chroot:
broken = []
for old_path in files_before:
# '/' is a separate element in the parts list
- old_path_parts = pathlib.Path(old_path).parts
+ old_path_parts = old_path.parts
if len(old_path_parts) < 3:
continue
if old_path_parts[1] == "usr" and (
old_path_parts[2] in ["bin", "sbin"] or old_path_parts[2].startswith("lib")
):
- new_path = os.path.join("/", *old_path_parts[2:])
+ new_path = PurePath(*old_path_parts[2:])
elif old_path_parts[1] in ["bin", "sbin"] or old_path_parts[1].startswith("lib"):
- new_path = "/usr" + old_path
+ new_path = PurePath("/usr", *old_path_parts[1:])
else:
continue
# Skip over directories, multiple packages can ship files in the same directories
- if new_path in files_after and os.path.isfile(self.relative(new_path)):
+ if new_path in files_after and self.relative(new_path).is_file():
broken.append("%s %s => %s %s" % (old_path, files_before[old_path], new_path, files_after[new_path]))
if broken:
@@ -2117,22 +2131,21 @@ class Chroot:
else:
logging.debug("No file moved between /{bin|sbin|lib*} and /usr/{bin|sbin|lib*}.")
- def check_for_broken_symlinks(self, warn_only: bool = False, file_owners: dict[str, list[str]] = {}) -> None:
+ def check_for_broken_symlinks(self, warn_only: bool = False, file_owners: dict[PurePath, list[str]] = {}) -> None:
"""Check that all symlinks in chroot are non-broken."""
assert self.name is not None
if not settings.check_broken_symlinks:
return
broken = []
- for dirpath, dirnames, filenames in os.walk(self.name):
+ for dirpath, dirnames, filenames in self.name.walk():
# Remove /proc within chroot to avoid lots of spurious errors.
if dirpath == self.name and "proc" in dirnames:
dirnames.remove("proc")
for filename in filenames:
- full_name = name = os.path.join(dirpath, filename)
- if name.startswith(self.name):
- name = name[len(self.name) :]
+ full_name = dirpath / filename
+ name = PurePath("/", full_name.relative_to(self.name))
ret = is_broken_symlink(self.name, dirpath, filename)
- if ret and not self.is_ignored(name, info="broken symlink"):
+ if ret and not self.is_ignored(os.fspath(name), info="broken symlink"):
try:
target = os.readlink(full_name)
except os.error:
@@ -2155,19 +2168,19 @@ class Chroot:
it returns the list of files."""
# FIXME! Does not work for M-A: same packages
- vdir = self.relative("var/lib/dpkg/info")
+ vdir = self.relative(PurePath("var/lib/dpkg/info"))
vlist = []
for p in packages:
basename = p + ".list"
- if not os.path.exists(os.path.join(vdir, basename)):
+ if not (vdir / basename).exists():
continue
- for line in readlines_file(os.path.join(vdir, basename)):
+ for line in readlines_file(vdir / basename):
pathname = line.strip()
if pathname.startswith("/etc/cron."):
- if os.path.isfile(self.relative(pathname.strip("/"))):
- st = os.lstat(self.relative(pathname.strip("/")))
+ if self.relative(PurePath(pathname.strip("/"))).is_file():
+ st = self.relative(PurePath(pathname.strip("/"))).lstat()
mode = st[stat.ST_MODE]
# XXX /etc/cron.d/ files are NOT executables
if mode & stat.S_IEXEC:
@@ -2181,7 +2194,7 @@ class Chroot:
cron file as cron would do (except for SHELL)"""
failed = False
for vfile in list:
- if not os.path.exists(self.relative(vfile.strip("/"))):
+ if not self.relative(PurePath(vfile.strip("/"))).exists():
continue
retval, output = self.run([vfile])
@@ -2192,25 +2205,25 @@ class Chroot:
if failed:
panic()
- def check_if_logrotatefiles(self, packages: list[str]) -> list[str]:
+ def check_if_logrotatefiles(self, packages: list[str]) -> list[PurePath]:
"""Check if the packages have logrotate files under /etc/logrotate.d and in case positive,
it returns the list of files."""
# FIXME! Does not work for M-A: same packages
- vdir = self.relative("var/lib/dpkg/info")
+ vdir = self.relative(PurePath("var/lib/dpkg/info"))
vlist = []
for p in packages:
basename = p + ".list"
- if not os.path.exists(os.path.join(vdir, basename)):
+ if not (vdir / basename).exists():
continue
- for line in readlines_file(os.path.join(vdir, basename)):
- pathname = line.strip()
- if os.path.dirname(pathname) == "/etc/logrotate.d":
- if os.path.isfile(self.relative(pathname.strip("/"))):
+ for line in readlines_file(vdir / basename):
+ pathname = PurePath(line.strip())
+ if pathname.parent == PurePath("/etc/logrotate.d"):
+ if self.relative(pathname).is_file():
vlist.append(pathname)
- logging.info("Package " + p + " contains logrotate file: " + pathname)
+ logging.info("Package %s contains logrotate file: %s", p, pathname)
return vlist
@@ -2225,15 +2238,15 @@ class Chroot:
diff = diff_selections(self, old_selections)
return diff.keys()
- def check_output_logrotatefiles(self, list: list[str]) -> None:
+ def check_output_logrotatefiles(self, list: list[PurePath]) -> None:
"""Check if a given list of logrotatefiles has any output. Executes
logrotate file as logrotate would do from cron (except for SHELL)"""
failed = False
for vfile in list:
- if not os.path.exists(self.relative(vfile.strip("/"))):
+ if not self.relative(vfile).exists():
continue
- retval, output = self.run(["/usr/sbin/logrotate", vfile])
+ retval, output = self.run(["/usr/sbin/logrotate", os.fspath(vfile)])
if output or retval != 0:
failed = True
logging.error("FAIL: Logrotate file %s exits with error or has output with package removed" % vfile)
@@ -2248,15 +2261,15 @@ class Chroot:
if not settings.scriptsdirs:
return errorcodes
logging.info("Running scripts " + step)
- basepath = self.relative("tmp/scripts/")
- if not os.path.exists(basepath):
+ basepath = self.relative(PurePath("tmp/scripts/"))
+ if not basepath.exists():
logging.error("Scripts directory %s does not exist" % basepath)
panic()
- list_scripts = sorted(os.listdir(basepath))
+ list_scripts = sorted(basepath.iterdir())
for vfile in list_scripts:
- if vfile.startswith(step):
- script = os.path.join("tmp/scripts", vfile)
- errorcode, output = self.run([script], ignore_errors=ignore_errors)
+ if vfile.name.startswith(step):
+ script = PurePath("tmp/scripts", vfile.name)
+ errorcode, output = self.run([os.fspath(script)], ignore_errors=ignore_errors)
errorcodes = errorcodes | errorcode
return errorcodes
@@ -2364,7 +2377,7 @@ def diff_meta_data(
return new, removed, modified
-def file_list(meta_infos: list[tuple[str, FileInfo]], file_owners: dict[str, list[str]]) -> str:
+def file_list(meta_infos: list[tuple[str, FileInfo]], file_owners: dict[PurePath, list[str]]) -> str:
"""Return list of indented filenames."""
meta_infos = sorted(meta_infos[:])
vlist = []
@@ -2373,9 +2386,7 @@ def file_list(meta_infos: list[tuple[str, FileInfo]], file_owners: dict[str, lis
if obj.target is not None:
info = " -> %s" % obj.target
vlist.append(" %s%s\t" % (name, info))
- key = name
- if key.endswith("/"):
- key = key[:-1]
+ key = PurePath(name.removesuffix("/"))
if key in file_owners:
vlist.append(" owned by: %s\n" % ", ".join(file_owners[key]))
else:
@@ -2451,17 +2462,10 @@ def get_package_names_from_package_files(package_files: list[str]) -> list[str]:
# from the 'Files' stanza.
-def process_changes(changes: str) -> list[str] | None:
+def process_changes(changes: Path) -> list[Path] | None:
# Determine the path to the changes file, then check if it's readable.
- dir_path = ""
- changes_path = ""
- if not os.path.dirname(changes):
- changes_path = os.path.basename(changes)
- else:
- dir_path = os.path.dirname(changes) + "/"
- changes_path = os.path.abspath(changes)
- if not os.access(changes_path, os.R_OK):
- logging.warning(changes_path + " is not readable. Skipping.")
+ if not os.access(changes, os.R_OK):
+ logging.warning("%s is not readable. Skipping.", changes)
return None
# Determine the packages in the changes file through the 'Files' stanza.
@@ -2473,9 +2477,7 @@ def process_changes(changes: str) -> list[str] | None:
""",
re.MULTILINE | re.DOTALL | re.VERBOSE,
)
- with open(changes_path, "r") as f:
- file_text = f.read()
- matches = pattern.split(file_text)
+ matches = pattern.split(changes.read_text())
# Append all the packages found in the changes file to a package list.
package_list = []
@@ -2483,8 +2485,7 @@ def process_changes(changes: str) -> list[str] | None:
package_p = re.compile(r".*?([^ ]+\.deb)$")
for line in newline_p.split(matches[1]):
if package_p.match(line):
- package = dir_path + package_p.split(line)[1]
- package_list.append(package)
+ package_list.append(changes.parent / package_p.split(line)[1])
# Return the list.
return package_list
@@ -2493,7 +2494,7 @@ def process_changes(changes: str) -> list[str] | None:
def check_results(
chroot: Chroot,
chroot_state: StateMetaData,
- file_owners: dict[str, list[str]],
+ file_owners: dict[PurePath, list[str]],
deps_info: dict[str, FileInfo] | None = None,
) -> bool:
"""Check that current chroot state matches 'chroot_state'.
@@ -2581,7 +2582,7 @@ def check_results(
def install_purge_test(
chroot: Chroot,
chroot_state: StateMetaData,
- package_files: list[str],
+ package_files: list[Path],
packages: list[str],
extra_packages: list[str],
) -> bool:
@@ -2614,7 +2615,7 @@ def install_purge_test(
# We were given package files, so let's get the Depends and
# Conflicts directly from the .debs
for deb in package_files:
- returncode, output = run(["dpkg", "-f", deb])
+ returncode, output = run(["dpkg", "-f", os.fspath(deb)])
control = deb822.Deb822(output)
control_infos.append(control)
else:
@@ -2663,7 +2664,7 @@ def install_purge_test(
)
def cleanup_metapackage() -> None:
- shutil.rmtree(os.path.dirname(metapackage))
+ shutil.rmtree(metapackage.parent)
panic_handler_id = do_on_panic(cleanup_metapackage)
@@ -2677,7 +2678,7 @@ def install_purge_test(
# don't panic(), too many problems on old distros
# Now remove it
- metapackagename = os.path.basename(metapackage)[:-4]
+ metapackagename = metapackage.name[:-4]
chroot.purge_packages([metapackagename])
cleanup_metapackage()
dont_do_on_panic(panic_handler_id)
@@ -2739,7 +2740,7 @@ def install_purge_test(
def install_upgrade_test(
- chroot: Chroot, chroot_state: StateMetaData, package_files: list[str], packages: list[str], old_packages: list[str]
+ chroot: Chroot, chroot_state: StateMetaData, package_files: list[Path], packages: list[str], old_packages: list[str]
) -> bool:
"""Install old_packages via apt-get, then upgrade from package files.
Return True if successful, False if not."""
@@ -2788,21 +2789,21 @@ def install_upgrade_test(
return check_results(chroot, chroot_state, file_owners_after)
-def save_meta_data(filename: str, chroot_state: StateMetaData) -> None:
+def save_meta_data(filename: Path, chroot_state: StateMetaData) -> None:
"""Save directory tree meta data into a file for fast access later."""
logging.debug("Saving chroot meta data to %s" % filename)
- with open(filename, "wb") as f:
+ with filename.open("wb") as f:
pickle.dump(chroot_state, f)
-def load_meta_data(filename: str) -> StateMetaData:
+def load_meta_data(filename: Path) -> StateMetaData:
"""Load meta data saved by 'save_meta_data'."""
logging.debug("Loading chroot meta data from %s" % filename)
- with open(filename, "rb") as f:
+ with filename.open("rb") as f:
return pickle.load(f)
-def install_and_upgrade_between_distros(package_files: list[str], packages_qualified: list[str]) -> bool:
+def install_and_upgrade_between_distros(package_files: list[Path], packages_qualified: list[str]) -> bool:
"""Install package and upgrade it between distributions, then remove.
Return True if successful, False if not."""
@@ -2837,7 +2838,7 @@ def install_and_upgrade_between_distros(package_files: list[str], packages_quali
chroot_state = None
if settings.end_meta:
- if os.path.exists(settings.end_meta):
+ if settings.end_meta.exists():
chroot_state = load_meta_data(settings.end_meta)
else:
logging.info("Cannot load chroot state from %s - generating it on-the-fly." % settings.end_meta)
@@ -2977,7 +2978,7 @@ def find_default_debian_mirrors() -> list[tuple[str, list[str]]]:
"""Find the default Debian mirrors."""
mirrors = []
try:
- for line in readlines_file("/etc/apt/sources.list"):
+ for line in readlines_file(Path("/etc/apt/sources.list")):
line = re.sub(r"\[arch=.*\]", "", line)
parts = line.split()
if len(parts) > 2 and parts[0] == "deb":
@@ -3573,7 +3574,8 @@ def parse_command_line() -> list[str]:
settings.defaults = opts.defaults
defaults = DefaultsFactory().new_defaults()
- settings.tmpdir = opts.tmpdir
+ if opts.tmpdir:
+ settings.tmpdir = Path(opts.tmpdir)
settings.keep_env = opts.keep_env
settings.shell_on_error = opts.shell_on_error
if opts.max_command_output_size:
@@ -3592,7 +3594,7 @@ def parse_command_line() -> list[str]:
settings.bootstrapcmd = shlex.split(opts.bootstrapcmd)
settings.keep_sources_list = opts.keep_sources_list
if opts.keyring:
- settings.keyring = opts.keyring
+ settings.keyring = Path(opts.keyring)
else:
settings.keyring = defaults.get_keyring()
settings.do_not_verify_signatures = opts.do_not_verify_signatures
@@ -3606,21 +3608,26 @@ def parse_command_line() -> list[str]:
settings.eatmydata = not opts.no_eatmydata
settings.dpkg_force_unsafe_io = not opts.dpkg_noforce_unsafe_io
settings.dpkg_force_confdef = opts.dpkg_force_confdef
- settings.scriptsdirs = opts.scriptsdir
+ settings.scriptsdirs = [Path(d) for d in opts.scriptsdir]
settings.bindmounts += opts.bindmount
settings.allow_database = opts.allow_database
settings.update_retries = opts.update_retries
# chroot setup
settings.arch = opts.arch
- settings.basetgz = opts.basetgz
- settings.savetgz = opts.save
+ if opts.basetgz:
+ settings.basetgz = Path(opts.basetgz)
+ if opts.save:
+ settings.savetgz = Path(opts.save)
settings.lvm_volume = opts.lvm_volume
settings.lvm_snapshot_size = opts.lvm_snapshot_size
- settings.existing_chroot = opts.existing_chroot
+ if opts.existing_chroot:
+ settings.existing_chroot = Path(opts.existing_chroot)
settings.hard_link = opts.hard_link
settings.schroot = opts.schroot
- settings.end_meta = opts.end_meta
- settings.save_end_meta = opts.save_end_meta
+ if opts.end_meta:
+ settings.end_meta = Path(opts.end_meta)
+ if opts.save_end_meta:
+ settings.save_end_meta = Path(opts.save_end_meta)
settings.skip_minimize = opts.skip_minimize
settings.minimize = opts.minimize
if settings.minimize:
@@ -3672,18 +3679,15 @@ def parse_command_line() -> list[str]:
exitcode = None
if not settings.tmpdir:
- if "TMPDIR" in os.environ:
- settings.tmpdir = os.environ["TMPDIR"]
- else:
- settings.tmpdir = "/tmp"
+ settings.tmpdir = Path(os.environ.get("TMPDIR", "/tmp"))
assert settings.tmpdir is not None
- if not os.path.isdir(settings.tmpdir):
+ if not settings.tmpdir.is_dir():
logging.error("Temporary directory is not a directory: %s" % settings.tmpdir)
panic()
for sdir in settings.scriptsdirs:
- if not os.path.isdir(sdir):
+ if not sdir.is_dir():
logging.error("Scripts directory is not a directory: %s" % sdir)
panic()
@@ -3728,7 +3732,7 @@ def process_packages(package_list: list[str]) -> None:
# Find the names of packages.
if settings.args_are_package_files:
packages = get_package_names_from_package_files(package_list)
- package_files = package_list
+ package_files = list(map(Path, package_list))
else:
packages = package_list
package_files = []
@@ -3841,17 +3845,16 @@ def main() -> None:
del os.environ["DISPLAY"]
changes_packages_list = []
- regular_packages_list = []
+ regular_packages_list: list[str] = []
changes_p = re.compile(r".*\.changes$")
for arg in args:
if changes_p.match(arg):
- package_list = process_changes(arg)
- assert package_list is not None
+ package_paths = process_changes(Path(arg))
+ assert package_paths is not None
if settings.single_changes_list:
- for package in package_list:
- regular_packages_list.append(package)
+ regular_packages_list.extend(os.fspath(p) for p in package_paths)
else:
- changes_packages_list.append(package_list)
+ changes_packages_list.append([os.fspath(p) for p in package_paths])
else:
regular_packages_list.append(arg)
=====================================
tests/test_piuparts.py
=====================================
@@ -1,4 +1,5 @@
import os
+from piuparts import Path
import shutil
import unittest
from unittest.mock import patch
@@ -18,7 +19,7 @@ class DefaultsFactoryTests(unittest.TestCase):
defaults = self.df.new_defaults()
guess_flavor_mock.assert_called_once()
- self.assertEqual(defaults.get_keyring(), "/usr/share/keyrings/debian-archive-keyring.gpg")
+ self.assertEqual(defaults.get_keyring(), Path("/usr/share/keyrings/debian-archive-keyring.gpg"))
self.assertEqual(
defaults.get_components(),
["main", "contrib", "non-free", "non-free-firmware"],
@@ -39,7 +40,7 @@ class DefaultsFactoryTests(unittest.TestCase):
defaults = self.df.new_defaults()
guess_flavor_mock.assert_called_once()
- self.assertEqual(defaults.get_keyring(), "/usr/share/keyrings/ubuntu-archive-keyring.gpg")
+ self.assertEqual(defaults.get_keyring(), Path("/usr/share/keyrings/ubuntu-archive-keyring.gpg"))
self.assertEqual(defaults.get_components(), ["main", "universe", "restricted", "multiverse"])
self.assertEqual(
defaults.get_mirror(),
@@ -64,11 +65,11 @@ class DefaultsFactoryTests(unittest.TestCase):
class IsBrokenSymlinkTests(unittest.TestCase):
- testdir = "is-broken-symlink-testdir"
+ testdir = Path("is-broken-symlink-testdir")
def symlink(self, target, name):
- pathname = os.path.join(self.testdir, name)
- os.symlink(target, pathname)
+ pathname = self.testdir / name
+ pathname.symlink_to(target)
self.symlinks.append(pathname)
def setUp(self):
@@ -82,9 +83,9 @@ class IsBrokenSymlinkTests(unittest.TestCase):
self.symlink("absolute-broken", "absolute-broken-to-symlink")
self.symlink("/", "absolute-works")
self.symlink("/absolute-works", "absolute-works-to-symlink")
- os.mkdir(os.path.join(self.testdir, "dir"))
+ (self.testdir / "dir").mkdir()
self.symlink("dir", "dir-link")
- os.mkdir(os.path.join(self.testdir, "dir/subdir"))
+ (self.testdir / "dir/subdir").mkdir()
self.symlink("subdir", "dir/subdir-link")
self.symlink("notexist/", "trailing-slash-broken")
self.symlink("dir/", "trailing-slash-works")
@@ -151,7 +152,7 @@ class IsBrokenSymlinkTests(unittest.TestCase):
def testMultiLevelNestedSymlinks(self):
# target/first-link -> ../target/second-link -> ../target
- os.mkdir(os.path.join(self.testdir, "target"))
+ (self.testdir / "target").mkdir()
self.symlink("../target", "target/second-link")
self.symlink("../target/second-link", "target/first-link")
self.assertFalse(is_broken_symlink(self.testdir, self.testdir, "target/first-link"))
@@ -160,8 +161,8 @@ class IsBrokenSymlinkTests(unittest.TestCase):
# first-link -> /second-link/final-target
# second-link -> /target-dir
- os.mkdir(os.path.join(self.testdir, "final-dir"))
- os.mkdir(os.path.join(self.testdir, "final-dir/final-target"))
+ (self.testdir / "final-dir").mkdir()
+ (self.testdir / "final-dir/final-target").mkdir()
self.symlink("/second-link/final-target", "first-link")
self.symlink("/final-dir", "second-link")
self.assertFalse(is_broken_symlink(self.testdir, self.testdir, "first-link"))
View it on GitLab: https://salsa.debian.org/debian/piuparts/-/commit/71ae753c42161e27f85e1d737294460338ab1a57
--
View it on GitLab: https://salsa.debian.org/debian/piuparts/-/commit/71ae753c42161e27f85e1d737294460338ab1a57
You're receiving this email because of your account on salsa.debian.org. Manage all notifications: https://salsa.debian.org/-/profile/notifications | Help: https://salsa.debian.org/help
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://alioth-lists.debian.net/pipermail/piuparts-devel/attachments/20260705/d1b11737/attachment-0001.htm>
More information about the Piuparts-devel
mailing list