Coverage for src/debputy/plugins/debputy/package_processors.py: 54%
173 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-01-16 17:20 +0000
1import contextlib
2import functools
3import gzip
4import os
5import re
6import subprocess
7from contextlib import ExitStack
8from typing import IO, Any
9from collections.abc import Iterator, Callable
11from debputy.plugin.api import VirtualPath
12from debputy.util import (
13 _error,
14 xargs,
15 escape_shell,
16 _info,
17 assume_not_none,
18 _debug_log,
19)
22@contextlib.contextmanager
23def _open_maybe_gzip(path: VirtualPath) -> Iterator[IO[bytes] | gzip.GzipFile]:
24 if path.name.endswith(".gz"):
25 with gzip.GzipFile(path.fs_path, "rb") as fd:
26 yield fd
27 else:
28 with path.open(byte_io=True) as fd:
29 yield fd
32_SO_LINK_RE = re.compile(rb"[.]so\s+(.*)\s*")
33_LA_DEP_LIB_RE = re.compile(rb"'.+'")
36def _detect_so_link(path: VirtualPath) -> str | None:
37 so_link_re = _SO_LINK_RE
38 with _open_maybe_gzip(path) as fd:
39 for line in fd:
40 m = so_link_re.search(line)
41 if m:
42 return m.group(1).decode("utf-8")
43 return None
46def _replace_with_symlink(path: VirtualPath, so_link_target: str) -> None:
47 adjusted_target = so_link_target
48 parent_dir = path.parent_dir
49 assert parent_dir is not None # For the type checking
50 if parent_dir.name == os.path.dirname(adjusted_target):
51 # Avoid man8/../man8/foo links
52 adjusted_target = os.path.basename(adjusted_target)
53 elif "/" in so_link_target:
54 # symlinks and so links have a different base directory when the link has a "/".
55 # Adjust with an extra "../" to align the result
56 adjusted_target = "../" + adjusted_target
58 path.unlink()
59 parent_dir.add_symlink(path.name, adjusted_target)
62@functools.lru_cache(1)
63def _has_man_recode() -> bool:
64 # Ideally, we would just use shutil.which or something like that.
65 # Unfortunately, in debhelper, we experienced problems with which
66 # returning "yes" for a man tool that actually could not be run
67 # on salsa CI.
68 #
69 # Therefore, we adopt the logic of dh_installman to run the tool
70 # with --help to confirm it is not broken, because no one could
71 # figure out what happened in the salsa CI and my life is still
72 # too short to figure it out.
73 try:
74 subprocess.check_call(
75 ["man-recode", "--help"],
76 stdin=subprocess.DEVNULL,
77 stdout=subprocess.DEVNULL,
78 stderr=subprocess.DEVNULL,
79 restore_signals=True,
80 )
81 except subprocess.CalledProcessError:
82 return False
83 return True
86def process_manpages(fs_root: VirtualPath, _unused1: Any, _unused2: Any) -> None:
87 man_dir = fs_root.lookup("./usr/share/man")
88 if not man_dir:
89 return
91 re_encode = []
92 for path in (p for p in man_dir.all_paths() if p.is_file and p.has_fs_path):
93 size = path.size
94 if size == 0:
95 continue
96 so_link_target = None
97 if size <= 1024:
98 # debhelper has a 1024 byte guard on the basis that ".so file tend to be small".
99 # That guard worked well for debhelper, so lets keep it for now on that basis alone.
100 so_link_target = _detect_so_link(path)
101 if so_link_target:
102 _replace_with_symlink(path, so_link_target)
103 else:
104 re_encode.append(path)
106 if not re_encode or not _has_man_recode():
107 return
109 with ExitStack() as manager:
110 manpages = [
111 manager.enter_context(p.replace_fs_path_content()) for p in re_encode
112 ]
113 static_cmd = ["man-recode", "--to-code", "UTF-8", "--suffix", ".encoded"]
114 for cmd in xargs(static_cmd, manpages):
115 _info(f"Ensuring manpages have utf-8 encoding via: {escape_shell(*cmd)}")
116 try:
117 subprocess.check_call(
118 cmd,
119 stdin=subprocess.DEVNULL,
120 restore_signals=True,
121 )
122 except subprocess.CalledProcessError:
123 _error(
124 "The man-recode process failed. Please review the output of `man-recode` to understand"
125 " what went wrong."
126 )
127 for manpage in manpages:
128 dest_name = manpage
129 if dest_name.endswith(".gz"):
130 encoded_name = dest_name[:-3] + ".encoded"
131 with open(dest_name, "wb") as out:
132 _debug_log(
133 f"Recompressing {dest_name} via gzip -9nc {escape_shell(encoded_name)}"
134 )
135 try:
136 subprocess.check_call(
137 [
138 "gzip",
139 "-9nc",
140 encoded_name,
141 ],
142 stdin=subprocess.DEVNULL,
143 stdout=out,
144 )
145 except subprocess.CalledProcessError:
146 _error(
147 f"The command {escape_shell('gzip', '-nc', f'{encoded_name}')} > {dest_name} failed!"
148 )
149 else:
150 os.rename(f"{dest_name}.encoded", manpage)
153def _filter_compress_paths() -> Callable[[VirtualPath], Iterator[VirtualPath]]:
154 ignore_dir_basenames = {
155 "_sources",
156 }
157 ignore_basenames = {
158 ".htaccess",
159 "index.sgml",
160 "objects.inv",
161 "search_index.json",
162 "copyright",
163 }
164 ignore_extensions = {
165 ".htm",
166 ".html",
167 ".xhtml",
168 ".gif",
169 ".png",
170 ".jpg",
171 ".jpeg",
172 ".gz",
173 ".taz",
174 ".tgz",
175 ".z",
176 ".bz2",
177 ".epub",
178 ".jar",
179 ".zip",
180 ".odg",
181 ".odp",
182 ".odt",
183 ".css",
184 ".xz",
185 ".lz",
186 ".lzma",
187 ".haddock",
188 ".hs",
189 ".woff",
190 ".woff2",
191 ".svg",
192 ".svgz",
193 ".js",
194 ".devhelp2",
195 ".map", # Technically, dh_compress has this one case-sensitive
196 }
197 ignore_special_cases = ("-gz", "-z", "_z")
199 def _filtered_walk(path: VirtualPath) -> Iterator[VirtualPath]:
200 for path, children in path.walk():
201 if path.name in ignore_dir_basenames: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 children.clear()
203 continue
204 if path.is_dir and path.name == "examples": 204 ↛ 206line 204 didn't jump to line 206 because the condition on line 204 was never true
205 # Ignore anything beneath /usr/share/doc/*/examples
206 parent = path.parent_dir
207 grand_parent = parent.parent_dir if parent else None
208 if grand_parent and grand_parent.absolute == "/usr/share/doc":
209 children.clear()
210 continue
211 name = path.name
212 if (
213 path.is_symlink
214 or not path.is_file
215 or name in ignore_basenames
216 or not path.has_fs_path
217 ):
218 continue
220 name_lc = name.lower()
221 _, ext = os.path.splitext(name_lc)
223 if ext in ignore_extensions or name_lc.endswith(ignore_special_cases): 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true
224 continue
225 yield path
227 return _filtered_walk
230def _find_compressable_paths(fs_root: VirtualPath) -> Iterator[VirtualPath]:
231 path_filter = _filter_compress_paths()
233 for p, compress_size_threshold in (
234 ("./usr/share/info", 0),
235 ("./usr/share/man", 0),
236 ("./usr/share/doc", 4096),
237 ):
238 path = fs_root.lookup(p)
239 if path is None:
240 continue
241 paths = path_filter(path)
242 if compress_size_threshold: 242 ↛ 245line 242 didn't jump to line 245 because the condition on line 242 was never true
243 # The special-case for changelog and NEWS is from dh_compress. Generally these files
244 # have always been compressed regardless of their size.
245 paths = (
246 p
247 for p in paths
248 if p.size > compress_size_threshold
249 or p.name.startswith(("changelog", "NEWS"))
250 )
251 yield from paths
252 x11_path = fs_root.lookup("./usr/share/fonts/X11")
253 if x11_path: 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 yield from (
255 p for p in x11_path.all_paths() if p.is_file and p.name.endswith(".pcf")
256 )
259def apply_compression(fs_root: VirtualPath, _unused1: Any, _unused2: Any) -> None:
260 # TODO: Support hardlinks
261 compressed_files: dict[str, str] = {}
262 for path in _find_compressable_paths(fs_root):
263 parent_dir = assume_not_none(path.parent_dir)
264 with (
265 parent_dir.add_file(f"{path.name}.gz", mtime=path.mtime) as new_file,
266 open(new_file.fs_path, "wb") as fd,
267 ):
268 try:
269 subprocess.check_call(["gzip", "-9nc", path.fs_path], stdout=fd)
270 except subprocess.CalledProcessError:
271 full_command = f"gzip -9nc {escape_shell(path.fs_path)} > {escape_shell(new_file.fs_path)}"
272 _error(
273 f"The compression of {path.path} failed. Please review the error message from gzip to"
274 f" understand what went wrong. Full command was: {full_command}"
275 )
276 compressed_files[path.path] = new_file.path
277 del parent_dir[path.name]
279 all_remaining_symlinks = {p.path: p for p in fs_root.all_paths() if p.is_symlink}
280 changed = True
281 while changed:
282 changed = False
283 remaining: list[VirtualPath] = list(all_remaining_symlinks.values())
284 for symlink in remaining:
285 target = symlink.readlink()
286 dir_target, basename_target = os.path.split(target)
287 new_basename_target = f"{basename_target}.gz"
288 symlink_parent_dir = assume_not_none(symlink.parent_dir)
289 dir_path = (
290 symlink_parent_dir.lookup(dir_target)
291 if dir_target
292 else symlink_parent_dir
293 )
294 if ( 294 ↛ 299line 294 didn't jump to line 299 because the condition on line 294 was never true
295 not dir_path
296 or basename_target in dir_path
297 or new_basename_target not in dir_path
298 ):
299 continue
300 del all_remaining_symlinks[symlink.path]
301 changed = True
303 new_link_name = (
304 f"{symlink.name}.gz"
305 if not symlink.name.endswith(".gz")
306 else symlink.name
307 )
308 symlink_parent_dir.add_symlink(
309 new_link_name, os.path.join(dir_target, new_basename_target)
310 )
311 symlink.unlink()
314def _la_files(fs_root: VirtualPath) -> Iterator[VirtualPath]:
315 lib_dir = fs_root.lookup("/usr/lib")
316 if not lib_dir:
317 return
318 # Original code only iterators directly in /usr/lib. To be a faithful conversion, we do the same
319 # here.
320 # Eagerly resolve the list as the replacement can trigger a runtime error otherwise
321 paths = list(lib_dir.iterdir)
322 yield from (p for p in paths if p.is_file and p.name.endswith(".la"))
325# Conceptually, the same feature that dh_gnome provides.
326# The clean_la_files function based on the dh_gnome version written by Luca Falavigna in 2010,
327# who in turn references a Makefile version of the feature.
328# https://salsa.debian.org/gnome-team/gnome-pkg-tools/-/commit/2868e1e41ea45443b0fb340bf4c71c4de87d4a5b
329def clean_la_files(
330 fs_root: VirtualPath,
331 _unused1: Any,
332 _unused2: Any,
333) -> None:
334 for path in _la_files(fs_root):
335 buffer = []
336 with path.open(byte_io=True) as fd:
337 replace_file = False
338 for line in fd:
339 if line.startswith(b"dependency_libs"):
340 replacement = _LA_DEP_LIB_RE.sub(b"''", line)
341 if replacement != line:
342 replace_file = True
343 line = replacement
344 buffer.append(line)
346 if not replace_file:
347 continue
348 _info(f"Clearing the dependency_libs line in {path.path}")
349 with path.replace_fs_path_content() as fs_path, open(fs_path, "wb") as wfd:
350 wfd.writelines(buffer)